import gradio as gr from pydub import AudioSegment import requests import os import uuid import re import time # برای sleep در Polling # مسیر ذخیره فایل‌های موقت TEMP_DIR = "temp_audio" if not os.path.exists(TEMP_DIR): os.makedirs(TEMP_DIR) # تنظیمات Polling POLLING_INTERVAL_SECONDS = 2 MAX_POLLING_ATTEMPTS = 30 # مثلاً 30 * 2 = 60 ثانیه صبر می‌کنیم def download_file(url, output_path): """فایل را از یک URL دانلود می‌کند.""" try: response = requests.get(url, stream=True) response.raise_for_status() with open(output_path, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) return True except requests.exceptions.RequestException as e: print(f"Error downloading {url}: {e}") return False except Exception as e: print(f"An unexpected error occurred during download of {url}: {e}") return False def get_audio_from_input(input_source): """ منبع ورودی را پردازش کرده و یک شی AudioSegment برمی‌گرداند. می‌تواند یک فایل محلی یا یک URL باشد. """ unique_filename = os.path.join(TEMP_DIR, str(uuid.uuid4())) if input_source.startswith("http://") or input_source.startswith("https://"): file_extension = os.path.splitext(input_source.split('?')[0])[1] if not file_extension: file_extension = ".mp3" temp_filepath = unique_filename + "_downloaded" + file_extension if not download_file(input_source, temp_filepath): return None, f"خطا در دانلود فایل از لینک: {input_source}" audio_path = temp_filepath else: audio_path = input_source try: audio = AudioSegment.from_file(audio_path) return audio, None except Exception as e: return None, f"خطا در بارگذاری فایل صوتی ({audio_path}): {e}. مطمئن شوید فایل MP3 یا WAV معتبر است." finally: if 'temp_filepath' in locals() and os.path.exists(temp_filepath): try: os.remove(temp_filepath) except OSError as e: print(f"Error removing temporary file {temp_filepath}: {e}") def merge_audio_files(input_sources): """ چندین فایل صوتی را (از لینک یا فایل) ادغام می‌کند و یک فایل MP3 خروجی می‌دهد. input_sources: یک لیست از URLها یا مسیرهای فایل‌ها. """ if not input_sources: return None, "لیست ورودی‌های صوتی خالی است." combined_audio = AudioSegment.empty() errors = [] for source in input_sources: audio_segment, error = get_audio_from_input(source) if audio_segment: combined_audio += audio_segment else: errors.append(error) print(f"Skipping {source} due to error: {error}") # برای debugging کنسول if not combined_audio.duration_seconds > 0: return None, "هیچ فایل صوتی معتبری برای ادغام پیدا نشد. " + "\n".join(errors) if errors else "" output_filename = os.path.join(TEMP_DIR, f"merged_audio_{uuid.uuid4()}.mp3") try: combined_audio.export(output_filename, format="mp3") return output_filename, "عملیات موفقیت آمیز بود!" except Exception as e: return None, f"خطا در ذخیره فایل خروجی: {e}" def tts_and_merge(text_input): """ متن ورودی را تجزیه می‌کند، با Talkbot API صدا تولید می‌کند و سپس آن‌ها را ادغام می‌کند. """ if not text_input.strip(): return None, "لطفاً متنی برای پردازش وارد کنید." lines = text_input.strip().split('\n') audio_urls_to_merge = [] errors = [] # برای Gradio که بتواند در حین Polling وضعیت را نشان دهد yield None, "در حال شروع پردازش TTS..." for line_idx, line in enumerate(lines): match = re.match(r'^\s*\((\d+)\)(.*)$', line) if match: speaker_number = match.group(1) text_for_tts = match.group(2).strip() if not text_for_tts: errors.append(f"خطا: متن خالی برای گوینده {speaker_number} در خط '{line}'") continue # فرض: این Endpoint برای درخواست TTS و دریافت event_id است # (ممکن است نیاز به تغییر به POST و body JSON داشته باشد) tts_request_url = f"https://talkbot.ir/api/TTS-S{speaker_number}/request?text={requests.utils.quote(text_for_tts)}" # URL Encode text yield None, f"در حال تولید صدا برای خط {line_idx+1} (گوینده {speaker_number})..." try: # مرحله 1: ارسال درخواست TTS و دریافت event_id response = requests.get(tts_request_url) # یا requests.post(...) response.raise_for_status() # فرض: پاسخ یک JSON با event_id است response_data = response.json() event_id = response_data.get("event_id") if not event_id: errors.append(f"API برای گوینده {speaker_number} در خط {line_idx+1} یک event_id معتبر برنگرداند: {response.text}") continue print(f"درخواست TTS برای {speaker_number} (خط {line_idx+1}) ارسال شد، Event ID: {event_id}") # مرحله 2: Polling برای وضعیت audio_link = None polling_attempts = 0 while polling_attempts < MAX_POLLING_ATTEMPTS: polling_attempts += 1 status_url = f"https://talkbot.ir/api/TTS-S{speaker_number}/status/{event_id}" # یا یک Endpoint کلی تر مثل /api/tts/status/{event_id} yield None, f"خط {line_idx+1} (گوینده {speaker_number}): در حال بررسی وضعیت (تلاش {polling_attempts}/{MAX_POLLING_ATTEMPTS})..." status_response = requests.get(status_url) status_response.raise_for_status() status_data = status_response.json() status = status_data.get("status") if status == "completed": audio_link = status_data.get("audio_url") if audio_link and audio_link.startswith("http"): audio_urls_to_merge.append(audio_link) print(f"صدا برای {speaker_number} (خط {line_idx+1}) آماده شد: {audio_link}") break # از حلقه Polling خارج می‌شویم else: errors.append(f"لینک صوتی نامعتبر در پاسخ وضعیت برای {speaker_number} (خط {line_idx+1}): {status_data}") break elif status == "failed": error_msg = status_data.get("error", "خطای ناشناخته از API.") errors.append(f"تولید صدا برای گوینده {speaker_number} (خط {line_idx+1}) با خطا مواجه شد: {error_msg}") break elif status == "processing": time.sleep(POLLING_INTERVAL_SECONDS) else: errors.append(f"وضعیت ناشناخته از API برای {speaker_number} (خط {line_idx+1}): {status_data}") break if not audio_link: # اگر Polling به پایان رسید و لینکی دریافت نشد errors.append(f"تولید صدا برای گوینده {speaker_number} (خط {line_idx+1}) در زمان تعیین شده به اتمام نرسید یا لینکی دریافت نشد.") except requests.exceptions.RequestException as e: errors.append(f"خطا در ارتباط با Talkbot API برای گوینده {speaker_number} (خط {line_idx+1}): {e}") except Exception as e: errors.append(f"خطای غیرمنتظره در پردازش Talkbot API برای گوینده {speaker_number} (خط {line_idx+1}): {e}") else: if line.strip(): errors.append(f"فرمت نامعتبر در خط: '{line}'. انتظار می‌رود (شماره)متن.") if not audio_urls_to_merge: final_message = "هیچ فایل صوتی برای ادغام تولید نشد." if errors: final_message += "\n\nخطاهای رخ داده:\n" + "\n".join(errors) yield None, final_message return yield None, "در حال ادغام فایل‌های صوتی..." merged_output_path, merge_message = merge_audio_files(audio_urls_to_merge) final_message = merge_message if errors: final_message += "\n\nخطاهای رخ داده:\n" + "\n".join(errors) yield merged_output_path, final_message # ایجاد رابط کاربری Gradio with gr.Blocks() as demo: gr.Markdown( """ # ابزار ادغام فایل‌های صوتی (MP3/WAV) و تولید صدا از متن در اینجا دو بخش اصلی وجود دارد: 1. **ادغام فایل‌های صوتی موجود:** می‌توانید لینک‌های فایل‌های صوتی (MP3/WAV) را وارد کنید. 2. **تولید صدا از متن با Talkbot API و ادغام:** متنی را با فرمت `(شماره)متن` وارد کنید. (مثال: `(1)سلام\n(2)بله`) این بخش با استفاده از Talkbot API صدا تولید کرده و آن‌ها را ادغام می‌کند. """ ) with gr.Tab("ادغام فایل‌های صوتی موجود"): gr.Markdown("## ادغام فایل‌های صوتی موجود (از لینک یا فایل محلی)") audio_links_input = gr.Textbox( label="لینک یا مسیر فایل‌های صوتی (هر کدام در یک خط جدید)", placeholder="مثال:\nhttps://example.com/audio1.mp3\n./local_audio.wav\nhttps://example.com/audio2.wav", lines=10 ) audio_merge_output_message = gr.Textbox(label="پیام", interactive=False) audio_merge_output_audio = gr.Audio(label="فایل صوتی ادغام شده", type="filepath") merge_button = gr.Button("ادغام فایل‌های صوتی") merge_button.click( fn=lambda x: merge_audio_files([s.strip() for s in x.split('\n') if s.strip()]), inputs=[audio_links_input], outputs=[audio_merge_output_audio, audio_merge_output_message] ) gr.Examples( examples=[ ["https://www.soundhelix.com/examples/mp3/SoundHelix-Song-1.mp3\nhttps://www.soundhelix.com/examples/mp3/SoundHelix-Song-2.mp3"], ], inputs=audio_links_input, label="نمونه‌ها" ) with gr.Tab("تولید صدا از متن (Talkbot API) و ادغام"): gr.Markdown("## تولید صدا با Talkbot API و ادغام فایل‌ها") tts_text_input = gr.Textbox( label="متن برای تولید صدا (فرمت: (شماره)متن - هر پرسوناژ در یک خط جدید)", placeholder="(1)سلام این تست صحبت اولین نفر است.\n(2)سلام، بله این هم یک تست است و من کاراکتر دوم هستم.\n(1)خب از کجا شروع کنیم\n(2)بهتره از اول شروع کنیم", lines=10 ) tts_output_message = gr.Textbox(label="پیام", interactive=False) tts_output_audio = gr.Audio(label="فایل صوتی تولید و ادغام شده", type="filepath") tts_merge_button = gr.Button("تولید و ادغام صدا") tts_merge_button.click( fn=tts_and_merge, inputs=[tts_text_input], outputs=[tts_output_audio, tts_output_message] ) gr.Examples( examples=[ ["(1)سلام این تست صحبت اولین نفر است.\n(2)سلام، بله این هم یک تست است و من کاراکتر دوم هستم."], ["(1)امروز هوا چطوره؟\n(2)فکر کنم آفتابیه."] ], inputs=tts_text_input, label="نمونه‌ها" ) if __name__ == "__main__": demo.launch(debug=True, show_api=False, inline=False, share=True) # برای اینکه Gradio بتواند خروجی‌های میانی (yield) را مدیریت کند، نیاز به queue دارید. # demo.launch(share=True, enable_queue=True)