Update app.py

app.py

@@ -3,164 +3,226 @@ from pydub import AudioSegment
 import requests
 import os
 import uuid
-import re # برای پردازش لینک‌های ورودی
-import mimetypes # برای استفاده از Content-Type برای تخمین پسوند فایل
+import re
+import mimetypes
+import tempfile # برای مدیریت فایل‌های موقت
 
-# پوشه برای ذخیره فایل‌های موقت
-TEMP_DIR = "temp_audio_files" # تغییر نام پوشه برای انعطاف‌پذیری بیشتر
-os.makedirs(TEMP_DIR, exist_ok=True) # اطمینان از وجود پوشه
+# --- تنظیمات 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.")
 
-def download_audio(url: str, output_dir: str) -> tuple[str | None, str | None]:
+# یک تابع کمکی برای بررسی وجود 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, None) در صورت خطا.
+    :return: مسیر کامل فایل دانلود شده یا None در صورت خطا.
     """
     try:
         response = requests.get(url, stream=True, timeout=30)
         response.raise_for_status()
 
-        content_type = response.headers.get('Content-Type', '')
-        # تخمین پسوند فایل بر اساس Content-Type
-        # اگر نتوانستیم پسوند را از Content-Type حدس بزنیم، از 'mp3' به عنوان پیش‌فرض استفاده می‌کنیم
-        # این کار ریسک ایجاد می‌کند ولی pydub ممکن است بتواند آن را تشخیص دهد.
-        estimated_extension = mimetypes.guess_extension(content_type)
-        if not estimated_extension:
-            # سعی می‌کنیم پسوند را از URL بگیریم
-            path_parts = os.path.splitext(url.split('?')[0])
-            if len(path_parts) > 1 and path_parts[1]:
-                estimated_extension = path_parts[1].lower()
+        # استفاده از 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:
-                # اگر هیچ کدام جواب نداد، mp3 را به عنوان پیش‌فرض قرار می‌دهیم
-                # pydub در زمان بارگذاری فایل واقعی، فرمت را بررسی می‌کند.
-                estimated_extension = ".mp3"
+                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}"
 
-        # اطمینان از پاک بودن پسوند (بدون نقطه اضافی)
-        if estimated_extension.startswith('.'):
-            estimated_extension = estimated_extension[1:]
+        output_path = os.path.join(output_dir, temp_filename)
 
-        # ساخت نام فایل موقت
-        temp_file_name = f"temp_download_{uuid.uuid4().hex}.{estimated_extension}"
-        output_path = os.path.join(output_dir, temp_file_name)
 
         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} (estimated type: {estimated_extension})")
-        return output_path, estimated_extension
+        print(f"Successfully downloaded {url} to {output_path}")
+        return output_path
     except requests.exceptions.Timeout:
-        print(f"Error downloading {url}: Request timed out.")
-        return None, None
+        print(f"Error downloading {url}: Request timed out after 30 seconds.")
+        return None
     except requests.exceptions.RequestException as e:
-        print(f"Error downloading {url}: {e}")
-        return None, None
+        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, None
+        return None
 
 def merge_audio_files_sequential(audio_file_paths: list[str], output_file_path: str) -> bool:
     """
     چندین فایل صوتی را به ترتیب ادغام می‌کند. همه فایل‌ها به MP3 خروجی گرفته می‌شوند.
-    :param audio_file_paths: لیستی از مسیرهای فایل‌های صوتی (MP3, WAV, و غیره).
+    :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:
         # بارگذاری اولین فایل
-        # pydub به طور خودکار فرمت را از پسوند فایل تشخیص می‌دهد.
-        combined_audio = AudioSegment.from_file(audio_file_paths[0])
+        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:]):
-            audio = AudioSegment.from_file(file_path)
-            combined_audio += audio
-            print(f"Merged {file_path}")
+            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 merged all files to {output_file_path}")
+        print(f"Successfully exported final merged file to {output_file_path}")
         return True
     except Exception as e:
-        print(f"Error merging audio files: {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 است که رشته‌ای از لینک‌های صوتی را دریافت کرده،
-    آن‌ها را دانلود و ادغام می‌کند.
-    :param urls_input: رشته‌ای شامل لینک‌های صوتی، جدا شده با خط جدید یا کاما.
-    :return: یک تاپل شامل پیام وضعیت و مسیر فایل خروجی (یا 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
+        return "برای ادغام، لطفاً حداقل دو لینک فایل صوتی وارد کنید.", None
 
     status_message = ""
     output_audio_path = None
     downloaded_files_paths = []
 
-    # برای تولید یک unique_id برای هر درخواست
-    request_unique_id = str(uuid.uuid4())
-    request_temp_dir = os.path.join(TEMP_DIR, request_unique_id)
-    os.makedirs(request_temp_dir, exist_ok=True)
-
-    try:
-        for i, url in enumerate(urls):
-            status_message += f"در حال دانلود فایل {i+1} از: {url}\n"
-            gr.Info(f"Downloading file {i+1}...")
-            # از تابع download_audio جدید استفاده می‌کنیم
-            downloaded_path, _ = download_audio(url, request_temp_dir)
-            if downloaded_path:
-                downloaded_files_paths.append(downloaded_path)
-            else:
-                return f"خطا در دانلود فایل {i+1} از '{url}'. لطفاً لینک را بررسی کنید. ممکن است فایل صوتی نباشد.", None
+    # استفاده از 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
+            if not downloaded_files_paths:
+                return "هیچ فایلی با موفقیت دانلود نشد. لطفاً لینک‌ها را بررسی کنید.", None
+            
+            # ترتیب دهی فایل ها بر اساس نام (برای تضمین ترتیب ادغام)
+            downloaded_files_paths.sort()
 
-        status_message += "فایل‌ها با موفقیت دانلود شدند. در حال ادغام...\n"
-        gr.Info("Files downloaded. Merging...")
+            status_message += "فایل‌ها با موفقیت دانلود شدند. در حال ادغام...\n"
+            gr.Info("Files downloaded. Merging audio segments...")
 
-        output_file_name = f"merged_output_{uuid.uuid4().hex}.mp3" # خروجی همیشه MP3 خواهد بود
-        output_file_path = os.path.join(request_temp_dir, output_file_name)
+            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):
-            status_message += f"فایل‌ها با موفقیت ادغام شدند. فایل خروجی: {output_file_path}\n"
-            output_audio_path = output_file_path
-        else:
-            status_message += "خطا در ادغام فایل‌ها.\n"
+            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"
-    finally:
-        # حذف فایل‌های موقت پس از اتمام کار
-        for f in os.listdir(request_temp_dir):
-            os.remove(os.path.join(request_temp_dir, f))
-        os.rmdir(request_temp_dir)
-        print(f"Removed temporary directory: {request_temp_dir}")
+        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
-# برای بهبود تجربه کاربری در فضای هاگینگ‌فیس، بهتر است از theme=gr.themes.Soft() استفاده کنید.
-# و همچنین css را با دقت بیشتری برای RTL بنویسید.
+# (بخش interface بدون تغییرات عمده)
 iface = gr.Interface(
     fn=process_audio_links_gradio,
     inputs=[
         gr.Textbox(
             label="لینک‌های فایل صوتی (MP3, WAV و غیره) - هر لینک در یک خط یا با کاما جدا کنید",
-            placeholder="مثال:\nhttps://example.com/audio1.mp3\nhttps://example.com/audio2.wav\n...",
+            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
         ),
     ],
@@ -171,7 +233,7 @@ iface = gr.Interface(
     title="🎙️ ادغام کننده چندین فایل صوتی از طریق لینک (MP3, WAV و غیره) 🎵",
     description="لینک فایل‌های صوتی خود را (هر لینک در یک خط جدید یا با کاما جدا شده) در کادر زیر وارد کنید تا به صورت یک فایل **MP3** ادغام شوند.",
     allow_flagging="never",
-    theme=gr.themes.Soft(), # استفاده از تم جدید Gradio
+    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; }
@@ -184,12 +246,11 @@ iface = gr.Interface(
         border-radius: 8px;
     }
     .gr-button {
-      background-color: #4CAF50; /* Example button color */
+      background-color: #4CAF50;
       color: white;
     }
     """
 )
 
-# تابع راه اندازی:
 if __name__ == "__main__":
     iface.launch(share=True)