Update bot.py

bot.py

@@ -5,21 +5,15 @@ import time
 import threading
 import aiosqlite
 import gradio as gr
-from collections import defaultdict
-from pyrogram import Client, filters
-from pyrogram.errors import FloodWait
-from pyrogram.types import (
+from hydrogram import Client, filters
+from hydrogram.errors import FloodWait
+from hydrogram.types import (
     InputMediaPhoto, InputMediaVideo, InputMediaDocument, 
     InputMediaAudio
 )
 from fastapi import FastAPI
 from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
 
-# ============ إصلاح MIN_CHANNEL_ID ============
-import pyrogram
-pyrogram.utils.MIN_CHANNEL_ID = -1007852516352
-pyrogram.utils.MIN_CHAT_ID = -999999999999
-
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(levelname)s - %(message)s'
@@ -37,14 +31,14 @@ DOWNLOAD_DIR = os.path.join(DATA_DIR, "downloads")
 DB_PATH = os.path.join(DATA_DIR, "queue.db")
 os.makedirs(DOWNLOAD_DIR, exist_ok=True)
 
-# ============ العميل ============
+# ============ العميل (Hydrogram) ============
 app = Client(
     "archiver",
     api_id=API_ID,
     api_hash=API_HASH,
     session_string=SESSION_STRING,
     in_memory=True,
-    workers=1,
+    workers=4,  # Hydrogram يتعامل أفضل مع workers أعلى
     device_model="Samsung Galaxy S23 Ultra",
     system_version="Android 14",
     app_version="10.14.1",
@@ -55,24 +49,26 @@ TARGET_CHAT = None
 
 # ============ Throttle Controller ============
 class ThrottleController:
-    def __init__(self, rate=1.2):
+    def __init__(self, rate=2.0):
         self.rate = rate
-        self.last_send = defaultdict(float)
+        self.last_send = {}
         self.lock = asyncio.Lock()
     
     async def wait(self, chat_id):
         async with self.lock:
             now = time.time()
-            elapsed = now - self.last_send[chat_id]
+            last = self.last_send.get(chat_id, 0)
+            elapsed = now - last
             if elapsed < self.rate:
-                wait_time = self.rate - elapsed
-                await asyncio.sleep(wait_time)
+                await asyncio.sleep(self.rate - elapsed)
             self.last_send[chat_id] = time.time()
 
-throttle = ThrottleController(rate=1.2)
+throttle = ThrottleController(rate=2.0)
 
 # ============ Semaphores ============
-download_sem = asyncio.Semaphore(2)
+# 🔑 Serial downloading: ملف واحد فقط يُحمل في كل وقت (لمنع Media DC storm)
+download_lock = asyncio.Lock()
+# إرسال: 2 متوازي كحد أقصى
 send_sem = asyncio.Semaphore(2)
 
 # ============ قاعدة البيانات ============
@@ -99,12 +95,8 @@ async def init_db():
                 UNIQUE(chat_id, message_id)
             )
         """)
-        await db.execute("""
-            CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)
-        """)
-        await db.execute("""
-            CREATE INDEX IF NOT EXISTS idx_jobs_media_group ON jobs(media_group_id)
-        """)
+        await db.execute("CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)")
+        await db.execute("CREATE INDEX IF NOT EXISTS idx_jobs_media_group ON jobs(media_group_id)")
         await db.commit()
     logger.info("Database ready")
 
@@ -153,7 +145,7 @@ async def add_job(message):
                 (chat_id, message_id, media_group_id, media_type, file_id, file_name, file_size, caption, source_name, status)
                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending')
             """, (
-                message.chat.id, message.id, message.media_group_id, 
+                message.chat.id, message.id, str(message.media_group_id) if message.media_group_id else None,
                 m_type, file_id, file_name, file_size, final_cap, src_name
             ))
             await db.commit()
@@ -179,7 +171,8 @@ async def update_job_status(chat_id, message_id, status, file_path=None, error=N
             """, (status, chat_id, message_id))
         await db.commit()
 
-async def get_pending_jobs(limit=3):
+async def get_pending_jobs(limit=1):
+    """🔑 جلب مهمة واحدة فقط (Serial downloading)"""
     async with aiosqlite.connect(DB_PATH) as db:
         db.row_factory = aiosqlite.Row
         cursor = await db.execute("""
@@ -253,7 +246,7 @@ def get_source_name(message):
         return chat.first_name or chat.username or "Private"
     return "Source"
 
-# ============ التحميل مع التحقق ============
+# ============ التحميل المتسلسل (Serial) ============
 async def verify_download(file_path, expected_size):
     if not file_path or not os.path.exists(file_path):
         return False
@@ -267,6 +260,7 @@ async def verify_download(file_path, expected_size):
     return True
 
 async def download_job(job):
+    """🔑 تحميل متسلسل: ملف واحد في كل وقت"""
     chat_id = job['chat_id']
     message_id = job['message_id']
     expected_size = job['file_size'] or 0
@@ -282,6 +276,7 @@ async def download_job(job):
     try:
         await update_job_status(chat_id, message_id, 'downloading')
         
+        # 🔑 التحميل باستخدام Hydrogram
         downloaded = await app.download_media(
             job['file_id'],
             file_name=file_path
@@ -290,6 +285,7 @@ async def download_job(job):
         if not downloaded:
             raise Exception("download_media returned None")
         
+        # التحقق من الاكتمال
         if not await verify_download(downloaded, expected_size):
             logger.info(f"Retrying download {message_id}")
             if os.path.exists(downloaded):
@@ -305,6 +301,11 @@ async def download_job(job):
         logger.info(f"Download complete: {message_id} | {os.path.getsize(downloaded)} bytes")
         return True
         
+    except FloodWait as e:
+        logger.warning(f"FloodWait during download: {e.value}s")
+        await update_job_status(chat_id, message_id, 'pending', error=f"FloodWait:{e.value}")
+        await asyncio.sleep(e.value)
+        return False
     except Exception as e:
         await update_job_status(chat_id, message_id, 'pending', error=str(e))
         logger.error(f"Download failed {message_id}: {e}")
@@ -423,29 +424,33 @@ async def send_album_group(media_group_id):
 
 # ============ Workers ============
 async def download_worker():
-    logger.info("Download Worker started")
-    await asyncio.sleep(3)
+    """🔑 Serial Download Worker: ملف واحد في كل وقت"""
+    logger.info("Serial Download Worker started")
+    await asyncio.sleep(5)
     
     while True:
         try:
             if not app.is_connected:
-                logger.warning("Pyrogram not connected, waiting...")
+                logger.warning("Hydrogram not connected, waiting...")
                 await asyncio.sleep(5)
                 continue
             
-            jobs = await get_pending_jobs(limit=3)
+            jobs = await get_pending_jobs(limit=1)
             if not jobs:
-                await asyncio.sleep(2)
+                await asyncio.sleep(1)
                 continue
             
-            tasks = []
-            for job in jobs:
-                if len(tasks) >= 2:
-                    break
-                tasks.append(download_job(job))
+            job = jobs[0]
             
-            if tasks:
-                await asyncio.gather(*tasks, return_exceptions=True)
+            # 🔑 Serial lock: ملف واحد فقط يُحمل
+            async with download_lock:
+                success = await download_job(job)
+                
+                # 🔑 تأخير 3 ثواني بين كل تحميل (لمنع FloodWait)
+                if success:
+                    await asyncio.sleep(3)
+                else:
+                    await asyncio.sleep(5)  # تأخير أطول إذا فشل
             
         except Exception as e:
             logger.error(f"Download Worker error: {e}")
@@ -453,12 +458,12 @@ async def download_worker():
 
 async def send_worker():
     logger.info("Send Worker started")
-    await asyncio.sleep(5)
+    await asyncio.sleep(7)
     
     while True:
         try:
             if not app.is_connected:
-                logger.warning("Pyrogram not connected, waiting...")
+                logger.warning("Hydrogram not connected, waiting...")
                 await asyncio.sleep(5)
                 continue
             
@@ -485,11 +490,13 @@ async def send_worker():
                     break
                 async with send_sem:
                     await send_job(job)
+                    await asyncio.sleep(2)  # تأخير بين الإرساليات
             
             for (mg_id,) in album_rows:
                 if not app.is_connected:
                     break
                 await send_album_group(mg_id)
+                await asyncio.sleep(2)
             
             await asyncio.sleep(1)
             
@@ -552,8 +559,8 @@ async def resolve_target_chat():
 def run_gradio():
     with gr.Blocks(title="Telegram Archiver Pro") as demo:
         gr.Markdown("# Telegram Archiver Pro")
-        gr.Markdown("Status: Running")
-        gr.Textbox(label="Status", value="Active | SQLite Queue | 3 Workers", interactive=False)
+        gr.Markdown("Status: Running | Hydrogram | Serial Download")
+        gr.Textbox(label="Status", value="Active | SQLite Queue | Serial DL", interactive=False)
     
     fastapi_app = FastAPI()