app.py

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)