Update bot.py

bot.py

@@ -12,7 +12,18 @@ from hydrogram.types import (
     InputMediaAudio
 )
 from fastapi import FastAPI
-from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
+
+# ============ CRITICAL FIX: Monkey-patch FloodPremiumWait ============
+# Older Hydrogram versions don't have FloodPremiumWait. Telegram sends it → AttributeError → infinite crash loop.
+import hydrogram.errors.rpc_error as _rpc_error
+if "FLOOD_PREMIUM_WAIT" not in [k for d in _rpc_error.exceptions.values() for k in d.keys()]:
+    class FloodPremiumWait(FloodWait):
+        CODE = 420
+        NAME = "FLOOD_PREMIUM_WAIT"
+    setattr(hydrogram.errors, "FloodPremiumWait", FloodPremiumWait)
+    _rpc_error.exceptions.setdefault(420, {})["FLOOD_PREMIUM_WAIT"] = "FloodPremiumWait"
+    logging.getLogger(__name__).warning("Patched missing FloodPremiumWait into Hydrogram")
+# ===================================================================
 
 logging.basicConfig(
     level=logging.INFO,
@@ -38,19 +49,42 @@ app = Client(
     api_hash=API_HASH,
     session_string=SESSION_STRING,
     in_memory=True,
-    workers=2,  # Hydrogram يتحمل workers أعلى
+    workers=2,
     device_model="Samsung Galaxy S23 Ultra",
     system_version="Android 14",
     app_version="10.14.1",
     lang_code="ar",
-    takeout=False,  # 🔑 مهم: لا يستخدم Takeout (يسبب مشاكل)
+    takeout=False,
 )
 
 TARGET_CHAT = None
 
+# ============ Global Flood Control ============
+class GlobalFloodControl:
+    """Pause ALL workers when ANY worker hits a FloodWait."""
+    def __init__(self):
+        self.until = 0
+        self._lock = asyncio.Lock()
+    
+    async def hit(self, seconds):
+        async with self._lock:
+            target = time.time() + seconds
+            if target > self.until:
+                self.until = target
+                logger.warning(f"Global flood control activated: {seconds}s")
+    
+    async def wait(self):
+        async with self._lock:
+            remaining = self.until - time.time()
+        if remaining > 0:
+            logger.info(f"Global flood pause: {remaining:.0f}s remaining")
+            await asyncio.sleep(remaining)
+
+flood_control = GlobalFloodControl()
+
 # ============ Throttle Controller ============
 class ThrottleController:
-    def __init__(self, rate=2.0):
+    def __init__(self, rate=3.0):  # slightly slower to avoid bans
         self.rate = rate
         self.last_send = {}
         self.lock = asyncio.Lock()
@@ -64,11 +98,11 @@ class ThrottleController:
                 await asyncio.sleep(self.rate - elapsed)
             self.last_send[chat_id] = time.time()
 
-throttle = ThrottleController(rate=2.0)
+throttle = ThrottleController(rate=3.0)
 
 # ============ Locks ============
 download_lock = asyncio.Lock()
-send_sem = asyncio.Semaphore(2)
+send_sem = asyncio.Semaphore(1)  # reduce to 1 to be gentler on API
 
 # ============ قاعدة البيانات ============
 async def init_db():
@@ -133,6 +167,14 @@ async def add_job(message):
         file_id = message.animation.file_id
         file_size = message.animation.file_size or 0
     
+    # CRITICAL FIX: Explicit typing to prevent "unsupported type" SQLite errors
+    chat_id = int(message.chat.id)
+    message_id = int(message.id)
+    file_size = int(file_size) if file_size else 0
+    file_id = str(file_id) if file_id else None
+    file_name = str(file_name) if file_name else None
+    media_group_id = str(message.media_group_id) if message.media_group_id else None
+    
     src_name = get_source_name(message)
     raw_cap = message.caption or ""
     final_cap = f"{raw_cap}\n\nSource: {src_name}".strip() if raw_cap else f"Source: {src_name}"
@@ -144,7 +186,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, str(message.media_group_id) if message.media_group_id else None,
+                chat_id, message_id, media_group_id,
                 m_type, file_id, file_name, file_size, final_cap, src_name
             ))
             await db.commit()
@@ -152,17 +194,21 @@ async def add_job(message):
             logger.error(f"Add job error: {e}")
 
 async def update_job_status(chat_id, message_id, status, file_path=None, error=None):
+    # CRITICAL FIX: Explicit int typing
+    chat_id = int(chat_id)
+    message_id = int(message_id)
+    
     async with aiosqlite.connect(DB_PATH) as db:
         if file_path:
             await db.execute("""
                 UPDATE jobs SET status=?, file_path=?, updated_at=CURRENT_TIMESTAMP 
                 WHERE chat_id=? AND message_id=?
-            """, (status, file_path, chat_id, message_id))
+            """, (status, str(file_path), chat_id, message_id))
         elif error:
             await db.execute("""
                 UPDATE jobs SET status=?, error=?, attempts=attempts+1, updated_at=CURRENT_TIMESTAMP 
                 WHERE chat_id=? AND message_id=?
-            """, (status, error, chat_id, message_id))
+            """, (status, str(error)[:500], chat_id, message_id))  # cap error length
         else:
             await db.execute("""
                 UPDATE jobs SET status=?, updated_at=CURRENT_TIMESTAMP 
@@ -178,7 +224,7 @@ async def get_pending_jobs(limit=1):
             WHERE status='pending' AND attempts < 10
             ORDER BY created_at 
             LIMIT ?
-        """, (limit,))
+        """, (int(limit),))
         rows = await cursor.fetchall()
         return [dict(row) for row in rows]
 
@@ -189,7 +235,7 @@ async def get_downloaded_album_jobs(media_group_id):
             SELECT * FROM jobs 
             WHERE media_group_id=? AND status='downloaded'
             ORDER BY message_id
-        """, (media_group_id,))
+        """, (str(media_group_id),))
         rows = await cursor.fetchall()
         return [dict(row) for row in rows]
 
@@ -198,11 +244,13 @@ async def get_album_pending_count(media_group_id):
         cursor = await db.execute("""
             SELECT COUNT(*) FROM jobs 
             WHERE media_group_id=? AND status IN ('pending', 'downloading')
-        """, (media_group_id,))
+        """, (str(media_group_id),))
         row = await cursor.fetchone()
         return row[0]
 
 async def delete_job(chat_id, message_id):
+    chat_id = int(chat_id)
+    message_id = int(message_id)
     async with aiosqlite.connect(DB_PATH) as db:
         cursor = await db.execute("""
             SELECT file_path FROM jobs WHERE chat_id=? AND message_id=?
@@ -261,7 +309,7 @@ async def download_with_message_object(chat_id, message_id, file_path):
     """إعادة جلب الرسالة وتحميلها (الأكثر موثوقية)"""
     try:
         logger.info(f"Re-fetching message {chat_id}/{message_id}")
-        message = await app.get_messages(chat_id, message_id)
+        message = await app.get_messages(int(chat_id), int(message_id))
         if not message or not message.media:
             raise Exception("Message not found or no media")
         
@@ -276,10 +324,11 @@ async def download_with_message_object(chat_id, message_id, file_path):
 
 async def download_job(job):
     """3 محاولات: file_id → message object → final retry"""
-    chat_id = job['chat_id']
-    message_id = job['message_id']
-    expected_size = job['file_size'] or 0
-    file_id = job['file_id']
+    # CRITICAL FIX: Explicit int typing to prevent SQLite binding errors
+    chat_id = int(job['chat_id'])
+    message_id = int(job['message_id'])
+    expected_size = int(job['file_size'] or 0)
+    file_id = str(job['file_id']) if job['file_id'] else None
     
     ext = {
         'photo': '.jpg', 'video': '.mp4', 'doc': '',
@@ -295,6 +344,9 @@ async def download_job(job):
         except:
             pass
     
+    # Respect global flood waits before starting
+    await flood_control.wait()
+    
     try:
         await update_job_status(chat_id, message_id, 'downloading')
         downloaded = None
@@ -304,6 +356,9 @@ async def download_job(job):
             logger.info(f"Attempt 1 (file_id): {message_id}")
             try:
                 downloaded = await app.download_media(file_id, file_name=file_path)
+            except FloodWait as e:
+                await flood_control.hit(e.value)
+                raise
             except Exception as e:
                 logger.warning(f"file_id failed: {e}")
         
@@ -315,10 +370,13 @@ async def download_job(job):
         # المحاولة 3: إعادة أخيرة
         if not await verify_download(downloaded, expected_size):
             logger.info(f"Attempt 3 (retry): {message_id}")
-            await asyncio.sleep(3)
+            await asyncio.sleep(5)
             if file_id:
                 try:
                     downloaded = await app.download_media(file_id, file_name=file_path)
+                except FloodWait as e:
+                    await flood_control.hit(e.value)
+                    raise
                 except:
                     pass
             if not await verify_download(downloaded, expected_size):
@@ -334,26 +392,39 @@ async def download_job(job):
     except FloodWait as e:
         logger.warning(f"FloodWait: {e.value}s")
         await update_job_status(chat_id, message_id, 'pending', error=f"FloodWait:{e.value}")
-        await asyncio.sleep(min(e.value, 60))
+        await flood_control.hit(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}")
+        err_str = str(e)
+        await update_job_status(chat_id, message_id, 'pending', error=err_str)
+        logger.error(f"Download failed {message_id}: {err_str}")
         return False
 
 # ============ الإرسال ============
-@retry(
-    stop=stop_after_attempt(5),
-    wait=wait_exponential_jitter(initial=1, max=60),
-    retry=retry_if_exception_type((FloodWait, ConnectionError, TimeoutError)),
-    reraise=True
-)
 async def safe_send(send_func):
-    return await send_func()
+    """Manual retry that RESPECTS FloodWait from Telegram."""
+    for attempt in range(5):
+        try:
+            return await send_func()
+        except FloodWait as e:
+            wait_time = getattr(e, 'value', getattr(e, 'x', 30))
+            logger.warning(f"FloodWait {wait_time}s on attempt {attempt+1}/5")
+            await flood_control.hit(wait_time)
+            await asyncio.sleep(min(wait_time, 300))
+        except (ConnectionError, TimeoutError) as e:
+            wait_time = min(2 ** attempt, 60)
+            logger.warning(f"Connection error, retry in {wait_time}s: {e}")
+            await asyncio.sleep(wait_time)
+        except Exception as e:
+            if attempt == 4:
+                raise
+            logger.warning(f"Send error (attempt {attempt+1}): {e}")
+            await asyncio.sleep(min(2 ** attempt, 30))
+    return None
 
 async def send_job(job):
-    chat_id = job['chat_id']
-    message_id = job['message_id']
+    chat_id = int(job['chat_id'])
+    message_id = int(job['message_id'])
     file_path = job['file_path']
     
     if not file_path or not os.path.exists(file_path):
@@ -361,6 +432,9 @@ async def send_job(job):
         await update_job_status(chat_id, message_id, 'pending', error="File not found")
         return False
     
+    # Respect global flood waits
+    await flood_control.wait()
+    
     try:
         await throttle.wait(TARGET_CHAT)
         
@@ -401,7 +475,7 @@ async def send_album_group(media_group_id):
         logger.info(f"Album {media_group_id} waiting for {pending} files")
         return
     
-    jobs.sort(key=lambda j: j['message_id'])
+    jobs.sort(key=lambda j: int(j['message_id']))
     
     album_inputs = []
     album_jobs = []
@@ -427,6 +501,8 @@ async def send_album_group(media_group_id):
         else:
             standalone_jobs.append(job)
     
+    await flood_control.wait()
+    
     async with send_sem:
         chunk_size = 10
         for idx in range(0, len(album_inputs), chunk_size):
@@ -461,12 +537,14 @@ async def download_worker():
         try:
             if not app.is_connected:
                 logger.warning("Not connected, waiting...")
-                await asyncio.sleep(5)
+                await asyncio.sleep(10)
                 continue
             
+            await flood_control.wait()
+            
             jobs = await get_pending_jobs(limit=1)
             if not jobs:
-                await asyncio.sleep(1)
+                await asyncio.sleep(2)
                 continue
             
             job = jobs[0]
@@ -480,7 +558,7 @@ async def download_worker():
             
         except Exception as e:
             logger.error(f"Download Worker error: {e}")
-            await asyncio.sleep(5)
+            await asyncio.sleep(10)
 
 async def send_worker():
     logger.info("Send Worker started")
@@ -490,16 +568,18 @@ async def send_worker():
         try:
             if not app.is_connected:
                 logger.warning("Not connected, waiting...")
-                await asyncio.sleep(5)
+                await asyncio.sleep(10)
                 continue
             
+            await flood_control.wait()
+            
             async with aiosqlite.connect(DB_PATH) as db:
                 db.row_factory = aiosqlite.Row
                 
                 cursor = await db.execute("""
                     SELECT * FROM jobs 
                     WHERE status='downloaded' AND (media_group_id IS NULL OR media_group_id='')
-                    ORDER BY created_at LIMIT 3
+                    ORDER BY created_at LIMIT 2
                 """)
                 single_jobs = [dict(row) for row in await cursor.fetchall()]
                 
@@ -507,7 +587,7 @@ async def send_worker():
                     SELECT media_group_id FROM jobs 
                     WHERE status='downloaded' AND media_group_id IS NOT NULL AND media_group_id!=''
                     GROUP BY media_group_id
-                    LIMIT 2
+                    LIMIT 1
                 """)
                 album_rows = await cursor.fetchall()
             
@@ -516,19 +596,19 @@ async def send_worker():
                     break
                 async with send_sem:
                     await send_job(job)
-                    await asyncio.sleep(2)
+                    await asyncio.sleep(3)
             
             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(3)
             
-            await asyncio.sleep(1)
+            await asyncio.sleep(2)
             
         except Exception as e:
             logger.error(f"Send Worker error: {e}")
-            await asyncio.sleep(5)
+            await asyncio.sleep(10)
 
 # ============ Album Fetch ============
 async def fetch_full_album(message):
@@ -545,7 +625,7 @@ async def fetch_full_album(message):
 # ============ Handlers ============
 @app.on_message(filters.incoming & ~filters.me & filters.media)
 async def on_media(client, message):
-    chat_id = message.chat.id
+    chat_id = int(message.chat.id)
     if str(chat_id) == str(TARGET_CHAT):
         return
     
@@ -656,4 +736,4 @@ if __name__ == "__main__":
         try:
             loop.run_until_complete(app.stop())
         except:
-            pass
+            pass