app.py

import gradio as gr
from pydub import AudioSegment
import requests
import os
import uuid
import re
import mimetypes
import tempfile # برای مدیریت فایل‌های موقت

# --- تنظیمات FFmpeg/ffprobe (برای محیط‌های محلی) ---
# در محیط‌های Production مانند Hugging Face Spaces، معمولاً FFmpeg از قبل نصب شده است.
# این خطوط تنها در صورتی نیاز است که با مشکل "FFmpeg or AVConvNotFound" مواجه شوید
# و FFmpeg را در مسیرهای غیر استاندارد نصب کرده باشید.
# PARENT_DIR = os.path.dirname(os.path.abspath(__file__))
# FFMPEG_PATH = os.path.join(PARENT_DIR, "ffmpeg", "bin", "ffmpeg.exe") # مثال برای ویندوز
# FFPROBE_PATH = os.path.join(PARENT_DIR, "ffmpeg", "bin", "ffprobe.exe") # مثال برای ویندوز
#
# if os.path.exists(FFMPEG_PATH):
#     AudioSegment.converter = FFMPEG_PATH
#     print(f"FFmpeg path set to: {FFMPEG_PATH}")
# if os.path.exists(FFPROBE_PATH):
#     AudioSegment.ffprobe = FFPROBE_PATH
#     print(f"FFprobe path set to: {FFPROBE_PATH}")
# else:
#     print("FFmpeg/FFprobe paths not explicitly set. Pydub will look in system PATH.")

# یک تابع کمکی برای بررسی وجود FFmpeg
def check_ffmpeg_presence():
    try:
        from pydub.utils import get_prober_name, get_encoder_name
        # تلاش برای فراخوانی ffprobe تا بررسی کنیم آیا در دسترس است یا نه
        output = os.popen(f"{get_prober_name()} -version").read()
        if "ffprobe" in output or "avprobe" in output:
            print("FFprobe is available.")
            return True
        else:
            print("FFprobe not found in system PATH or configured path.")
            return False
    except Exception as e:
        print(f"Error checking FFmpeg/FFprobe: {e}")
        return False

# بررسی در ابتدای اجرا
if not check_ffmpeg_presence():
    print("WARNING: FFmpeg/ffprobe might not be correctly installed or configured. Audio processing may fail.")
    # می‌توانید اینجا یک خطای Gradio به کاربر نشان دهید
    # gr.Warning("FFmpeg/ffprobe not found. Please ensure it's installed and accessible.")


def download_audio(url: str, output_dir: str) -> str | None:
    """
    فایل صوتی را از URL دانلود می‌کند.
    :param url: URL فایل صوتی.
    :param output_dir: دایرکتوری برای ذخیره فایل دانلود شده.
    :return: مسیر کامل فایل دانلود شده یا None در صورت خطا.
    """
    try:
        response = requests.get(url, stream=True, timeout=30)
        response.raise_for_status()

        # استفاده از Content-Disposition برای نام فایل اگر موجود باشد
        content_disposition = response.headers.get('Content-Disposition')
        if content_disposition:
            fname_match = re.search(r'filename\*?=(?:UTF-8\'\')?\"?([^\"]+)\"?', content_disposition)
            if fname_match:
                original_filename = fname_match.group(1).encode('latin-1').decode('utf-8')
                # پاکسازی نام فایل برای جلوگیری از مشکلات مسیر
                original_filename = re.sub(r'[\\/:*?"<>|]', '_', original_filename)
                # اطمینان حاصل شود که پسوند دارد
                if not os.path.splitext(original_filename)[1]:
                    original_filename += mimetypes.guess_extension(response.headers.get('Content-Type', ''), strict=False) or '.tmp'
                temp_filename = f"{uuid.uuid4().hex}_{original_filename}"
            else:
                temp_filename = f"temp_download_{uuid.uuid4().hex}.tmp"
        else:
            # تخمین پسوند از Content-Type یا URL
            content_type = response.headers.get('Content-Type', '')
            ext = mimetypes.guess_extension(content_type, strict=False)
            if not ext:
                path_parts = os.path.splitext(url.split('?')[0])
                if len(path_parts) > 1 and path_parts[1]:
                    ext = path_parts[1].lower()
                else:
                    ext = ".tmp" # پسوند موقت برای pydub تا خودش تشخیص دهد

            # ساخت نام فایل موقت (با پسوند مربوطه)
            temp_filename = f"temp_download_{uuid.uuid4().hex}{ext}"

        output_path = os.path.join(output_dir, temp_filename)


        with open(output_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:
                    f.write(chunk)
        print(f"Successfully downloaded {url} to {output_path}")
        return output_path
    except requests.exceptions.Timeout:
        print(f"Error downloading {url}: Request timed out after 30 seconds.")
        return None
    except requests.exceptions.RequestException as e:
        print(f"Network or HTTP error downloading {url}: {e}")
        return None
    except Exception as e:
        print(f"An unexpected error occurred during download of {url}: {e}")
        return None

def merge_audio_files_sequential(audio_file_paths: list[str], output_file_path: str) -> bool:
    """
    چندین فایل صوتی را به ترتیب ادغام می‌کند. همه فایل‌ها به MP3 خروجی گرفته می‌شوند.
    :param audio_file_paths: لیستی از مسیرهای فایل‌های صوتی.
    :param output_file_path: مسیر کامل برای ذخیره فایل ادغام شده (خروجی همیشه MP3 خواهد بود).
    :return: True اگر ادغام موفق بود، False در غیر این صورت.
    """
    if not audio_file_paths:
        print("No audio files provided for merging.")
        return False

    combined_audio = None
    successful_merges = 0

    try:
        # بارگذاری اولین فایل
        try:
            combined_audio = AudioSegment.from_file(audio_file_paths[0])
            print(f"Successfully loaded initial audio from: {audio_file_paths[0]}")
            successful_merges = 1
        except Exception as e:
            print(f"Error loading first audio file {audio_file_paths[0]}: {e}")
            return False

        # ادغام بقیه فایل‌ها
        for i, file_path in enumerate(audio_file_paths[1:]):
            try:
                audio = AudioSegment.from_file(file_path)
                combined_audio += audio
                print(f"Successfully merged {file_path}")
                successful_merges += 1
            except Exception as e:
                print(f"Error merging audio from {file_path}: {e}. Skipping this file.")
                # در اینجا می‌توانیم انتخاب کنیم که آیا کل عملیات را متوقف کنیم یا ادامه دهیم.
                # برای این کاربرد، ادامه دادن معقول‌تر است (ادغام بقیه فایل‌های موفق).
                pass # ادامه به فایل بعدی حتی اگر این یکی موفق نبود

        if successful_merges < len(audio_file_paths):
            print(f"Warning: Only {successful_merges} out of {len(audio_file_paths)} files were successfully loaded/merged.")
            if successful_merges == 0:
                print("No audio segments were successfully processed for merging.")
                return False

        # خروجی گرفتن نهایی به فرمت MP3
        combined_audio.export(output_file_path, format="mp3")
        print(f"Successfully exported final merged file to {output_file_path}")
        return True
    except Exception as e:
        print(f"An unexpected error occurred during audio merging process: {e}")
        return False

def process_audio_links_gradio(urls_input: str) -> tuple[str, str | None]:
    """
    تابع اصلی Gradio برای پردازش لینک‌های صوتی.
    """
    urls = [url.strip() for url in re.split(r'[\n,]+', urls_input) if url.strip()]

    if not urls:
        return "لطفاً حداقل یک لینک فایل صوتی معتبر وارد کنید.", None

    if len(urls) < 2:
        return "برای ادغام، لطفاً حداقل دو لینک فایل صوتی وارد کنید.", None

    status_message = ""
    output_audio_path = None
    downloaded_files_paths = []

    # استفاده از NamedTemporaryFile برای دایرکتوری موقت
    # این دایرکتوری به طور خودکار پس از خروج از context manager حذف می‌شود.
    with tempfile.TemporaryDirectory() as request_temp_dir:
        print(f"Using temporary directory: {request_temp_dir}")
        try:
            for i, url in enumerate(urls):
                status_message += f"در حال دانلود فایل {i+1} از: {url}\n"
                gr.Info(f"Downloading file {i+1} of {len(urls)}...")
                downloaded_path = download_audio(url, request_temp_dir)
                if downloaded_path:
                    downloaded_files_paths.append(downloaded_path)
                else:
                    return f"خطا در دانلود فایل {i+1} از '{url}'. لطفاً لینک را بررسی کنید. ممکن است فایل صوتی نباشد یا دسترسی به آن ممکن نباشد.", None

            if not downloaded_files_paths:
                return "هیچ فایلی با موفقیت دانلود نشد. لطفاً لینک‌ها را بررسی کنید.", None
            
            # ترتیب دهی فایل ها بر اساس نام (برای تضمین ترتیب ادغام)
            downloaded_files_paths.sort()

            status_message += "فایل‌ها با موفقیت دانلود شدند. در حال ادغام...\n"
            gr.Info("Files downloaded. Merging audio segments...")

            output_file_name_final = f"merged_output_{uuid.uuid4().hex}.mp3"
            output_file_path_final = os.path.join(request_temp_dir, output_file_name_final)

            if merge_audio_files_sequential(downloaded_files_paths, output_file_path_final):
                status_message += f"فایل‌ها با موفقیت ادغام شدند. فایل خروجی: {output_file_path_final}\n"
                output_audio_path = output_file_path_final
            else:
                status_message += "خطا در ادغام فایل‌ها. (جزئیات خطا در کنسول سرور)\n"

        except Exception as e:
            status_message += f"خطای پیش‌بینی نشده در طول پردازش: {e}\n"
            print(f"Unhandled exception in process_audio_links_gradio: {e}")
        finally:
            # tempfile.TemporaryDirectory به صورت خودکار دایرکتوری را پاک می‌کند.
            # فقط باید مطمئن شویم که Gradio به فایل `output_audio_path` اشاره می‌کند
            # و gradigo خودش آن فایل را برای دانلود در اختیار قرار می‌دهد.
            print(f"Temporary directory {request_temp_dir} will be removed.")

    return status_message, output_audio_path

# تعریف رابط Gradio
# (بخش interface بدون تغییرات عمده)
iface = gr.Interface(
    fn=process_audio_links_gradio,
    inputs=[
        gr.Textbox(
            label="لینک‌های فایل صوتی (MP3, WAV و غیره) - هر لینک در یک خط یا با کاما جدا کنید",
            placeholder="مثال:\nhttps://www.soundhelix.com/examples/mp3/SoundHelix-Song-1.mp3\nhttps://file-examples.com/storage/fe/2017/11/file_example_WAV_1MG.wav\n...",
            lines=5
        ),
    ],
    outputs=[
        gr.Textbox(label="وضعیت", interactive=False),
        gr.Audio(label="فایل صوتی ادغام شده", type="filepath")
    ],
    title="🎙️ ادغام کننده چندین فایل صوتی از طریق لینک (MP3, WAV و غیره) 🎵",
    description="لینک فایل‌های صوتی خود را (هر لینک در یک خط جدید یا با کاما جدا شده) در کادر زیر وارد کنید تا به صورت یک فایل **MP3** ادغام شوند.",
    allow_flagging="never",
    theme=gr.themes.Soft(),
    css="""
    body { font-family: 'Vazirmatn', sans-serif; direction: rtl; text-align: right; }
    h1, h2, h3, h4, h5, h6 { text-align: center; }
    .gr-textbox label, .gr-audio label { text-align: right; width: 100%; display: block; }
    .gradio-container {
        max-width: 800px;
        margin: auto;
        padding: 20px;
        box-shadow: 0 4px 8px rgba(0, 0, 0, 0.1);
        border-radius: 8px;
    }
    .gr-button {
      background-color: #4CAF50;
      color: white;
    }
    """
)

if __name__ == "__main__":
    iface.launch(share=True)