Update bot.py

bot.py

@@ -32,7 +32,6 @@ API_HASH = os.environ.get("API_HASH", "")
 SESSION_STRING = os.environ.get("SESSION_STRING", "")
 TARGET_CHAT_RAW = os.environ.get("TARGET_CHAT", "me")
 
-# 🔑 استخدام /data للـ persistent storage (1TB)
 DATA_DIR = "/data"
 DOWNLOAD_DIR = os.path.join(DATA_DIR, "downloads")
 DB_PATH = os.path.join(DATA_DIR, "queue.db")
@@ -54,10 +53,9 @@ app = Client(
 
 TARGET_CHAT = None
 
-# ============ Throttle Controller (منع FloodWait) ============
+# ============ Throttle Controller ============
 class ThrottleController:
-    """تنظيم الإرسال: 1 رسالة/ثانية لكل دردشة"""
-    def __init__(self, rate=1.0):
+    def __init__(self, rate=1.2):
         self.rate = rate
         self.last_send = defaultdict(float)
         self.lock = asyncio.Lock()
@@ -68,19 +66,17 @@ class ThrottleController:
             elapsed = now - self.last_send[chat_id]
             if elapsed < self.rate:
                 wait_time = self.rate - elapsed
-                logger.debug(f"⏳ Throttle: انتظار {wait_time:.2f}s للدردشة {chat_id}")
                 await asyncio.sleep(wait_time)
             self.last_send[chat_id] = time.time()
 
-throttle = ThrottleController(rate=1.2)  # 1 رسالة كل 1.2 ثانية
+throttle = ThrottleController(rate=1.2)
 
 # ============ Semaphores ============
-download_sem = asyncio.Semaphore(3)   # 3 تحميلات متوازية كحد أقصى
-send_sem = asyncio.Semaphore(2)       # 2 إرساليات متوازية
+download_sem = asyncio.Semaphore(2)
+send_sem = asyncio.Semaphore(2)
 
 # ============ قاعدة البيانات ============
 async def init_db():
-    """إنشاء جداول SQLite في /data (persistent)"""
     async with aiosqlite.connect(DB_PATH) as db:
         await db.execute("""
             CREATE TABLE IF NOT EXISTS jobs (
@@ -103,16 +99,6 @@ async def init_db():
                 UNIQUE(chat_id, message_id)
             )
         """)
-        await db.execute("""
-            CREATE TABLE IF NOT EXISTS albums (
-                media_group_id TEXT PRIMARY KEY,
-                chat_id INTEGER,
-                total_expected INTEGER DEFAULT 0,
-                collected INTEGER DEFAULT 0,
-                status TEXT DEFAULT 'collecting',
-                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-            )
-        """)
         await db.execute("""
             CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)
         """)
@@ -120,15 +106,13 @@ async def init_db():
             CREATE INDEX IF NOT EXISTS idx_jobs_media_group ON jobs(media_group_id)
         """)
         await db.commit()
-    logger.info("✅ قاعدة البيانات جاهزة")
+    logger.info("Database ready")
 
 async def add_job(message):
-    """إضافة مهمة جديدة للـ queue"""
     m_type, ext = get_media_info(message)
     if not m_type:
         return
     
-    # الحصول على file_id و file_size
     file_id = None
     file_size = 0
     file_name = None
@@ -160,7 +144,7 @@ async def add_job(message):
     
     src_name = get_source_name(message)
     raw_cap = message.caption or ""
-    final_cap = f"{raw_cap}\n\n📌 المصدر: {src_name}".strip() if raw_cap else f"📌 المصدر: {src_name}"
+    final_cap = f"{raw_cap}\n\nSource: {src_name}".strip() if raw_cap else f"Source: {src_name}"
     
     async with aiosqlite.connect(DB_PATH) as db:
         try:
@@ -174,10 +158,9 @@ async def add_job(message):
             ))
             await db.commit()
         except Exception as e:
-            logger.error(f"❌ خطأ في إضافة المهمة: {e}")
+            logger.error(f"Add job error: {e}")
 
 async def update_job_status(chat_id, message_id, status, file_path=None, error=None):
-    """تحديث حالة المهمة"""
     async with aiosqlite.connect(DB_PATH) as db:
         if file_path:
             await db.execute("""
@@ -196,8 +179,7 @@ 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=5):
-    """جلب المهام المعلقة"""
+async def get_pending_jobs(limit=3):
     async with aiosqlite.connect(DB_PATH) as db:
         db.row_factory = aiosqlite.Row
         cursor = await db.execute("""
@@ -210,7 +192,6 @@ async def get_pending_jobs(limit=5):
         return [dict(row) for row in rows]
 
 async def get_downloaded_album_jobs(media_group_id):
-    """جلب كل ملفات ألبوم محملة وجاهزة للإرسال"""
     async with aiosqlite.connect(DB_PATH) as db:
         db.row_factory = aiosqlite.Row
         cursor = await db.execute("""
@@ -222,7 +203,6 @@ async def get_downloaded_album_jobs(media_group_id):
         return [dict(row) for row in rows]
 
 async def get_album_pending_count(media_group_id):
-    """عدد الملفات المتبقية في الألبوم"""
     async with aiosqlite.connect(DB_PATH) as db:
         cursor = await db.execute("""
             SELECT COUNT(*) FROM jobs 
@@ -232,7 +212,6 @@ async def get_album_pending_count(media_group_id):
         return row[0]
 
 async def delete_job(chat_id, 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=?
@@ -241,7 +220,6 @@ async def delete_job(chat_id, message_id):
         if row and row[0] and os.path.exists(row[0]):
             try:
                 os.remove(row[0])
-                logger.info(f"🗑️ ملف محذوف: {row[0]}")
             except:
                 pass
         await db.execute("DELETE FROM jobs WHERE chat_id=? AND message_id=?", (chat_id, message_id))
@@ -270,32 +248,25 @@ def get_media_info(message):
 def get_source_name(message):
     chat = message.chat
     if chat.type in ["channel", "supergroup", "group"]:
-        return chat.title or "قناة/مجموعة"
+        return chat.title or "Channel/Group"
     elif chat.type == "private":
-        return chat.first_name or chat.username or "خاص"
-    return "مصدر"
-
-def get_file_id_from_job(job):
-    """استخراج file_id من الرسالة الأصلية"""
-    # نستخدم app.download_media مباشرة بالـ message_id
-    return job['file_id']
+        return chat.first_name or chat.username or "Private"
+    return "Source"
 
 # ============ التحميل مع التحقق ============
 async def verify_download(file_path, expected_size):
-    """التحقق من اكتمال التحميل"""
     if not file_path or not os.path.exists(file_path):
         return False
     actual_size = os.path.getsize(file_path)
     if expected_size > 0 and actual_size < expected_size:
-        logger.warning(f"⚠️ ملف ناقص: {actual_size}/{expected_size} بايت")
+        logger.warning(f"Incomplete file: {actual_size}/{expected_size} bytes")
         return False
     if actual_size == 0:
-        logger.warning(f"⚠️ ملف بحجم صفر: {file_path}")
+        logger.warning(f"Zero-byte file: {file_path}")
         return False
     return True
 
 async def download_job(job):
-    """تحميل ملف مهمة مع التحقق من الاكتمال"""
     chat_id = job['chat_id']
     message_id = job['message_id']
     expected_size = job['file_size'] or 0
@@ -309,10 +280,8 @@ async def download_job(job):
     file_path = os.path.join(DOWNLOAD_DIR, f"{chat_id}_{message_id}{ext}")
     
     try:
-        # تحديث الحالة إلى downloading
         await update_job_status(chat_id, message_id, 'downloading')
         
-        # 🔑 التحميل باستخدام file_id
         downloaded = await app.download_media(
             job['file_id'],
             file_name=file_path
@@ -321,10 +290,8 @@ 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"🔄 إعادة تحميل {message_id}")
+            logger.info(f"Retrying download {message_id}")
             if os.path.exists(downloaded):
                 os.remove(downloaded)
             downloaded = await app.download_media(
@@ -332,18 +299,18 @@ async def download_job(job):
                 file_name=file_path
             )
             if not await verify_download(downloaded, expected_size):
-                raise Exception(f"File incomplete after retry: {os.path.getsize(downloaded) if downloaded else 0}/{expected_size}")
+                raise Exception(f"Incomplete after retry")
         
         await update_job_status(chat_id, message_id, 'downloaded', file_path=downloaded)
-        logger.info(f"✅ تحميل مكتمل: {message_id} | {os.path.getsize(downloaded)} بايت")
+        logger.info(f"Download complete: {message_id} | {os.path.getsize(downloaded)} bytes")
         return True
         
     except Exception as e:
         await update_job_status(chat_id, message_id, 'pending', error=str(e))
-        logger.error(f"❌ فشل تحميل {message_id}: {e}")
+        logger.error(f"Download failed {message_id}: {e}")
         return False
 
-# ============ الإرسال مع Tenacity ============
+# ============ الإرسال ============
 @retry(
     stop=stop_after_attempt(5),
     wait=wait_exponential_jitter(initial=1, max=60),
@@ -354,13 +321,12 @@ async def safe_send(send_func):
     return await send_func()
 
 async def send_job(job):
-    """إرسال مهمة محملة"""
     chat_id = job['chat_id']
     message_id = job['message_id']
     file_path = job['file_path']
     
     if not file_path or not os.path.exists(file_path):
-        logger.error(f"❌ ملف غير موجود: {file_path}")
+        logger.error(f"File not found: {file_path}")
         await update_job_status(chat_id, message_id, 'pending', error="File not found")
         return False
     
@@ -385,42 +351,33 @@ async def send_job(job):
         elif m_type == 'animation':
             await safe_send(lambda: app.send_animation(TARGET_CHAT, animation=file_path, caption=caption))
         
-        logger.info(f"✅ إرسال ناجح: {message_id}")
+        logger.info(f"Sent: {message_id}")
         await delete_job(chat_id, message_id)
         return True
         
     except Exception as e:
-        logger.error(f"❌ فشل إرسال {message_id}: {e}")
+        logger.error(f"Send failed {message_id}: {e}")
         await update_job_status(chat_id, message_id, 'downloaded', error=str(e))
         return False
 
-# ============ إرسال الألبوم ============
 async def send_album_group(media_group_id):
-    """إرسال ألبوم كامل بعد تحميل كل ملفاته"""
     jobs = await get_downloaded_album_jobs(media_group_id)
     if not jobs:
         return
     
-    # التحقق من عدم وجود ملفات معلقة
     pending = await get_album_pending_count(media_group_id)
     if pending > 0:
-        logger.info(f"⏳ ألبوم {media_group_id} ينتظر {pending} ملفات")
+        logger.info(f"Album {media_group_id} waiting for {pending} files")
         return
     
-    # ترتيب الملفات
     jobs.sort(key=lambda j: j['message_id'])
     
-    # فصل المدعوم في الألبوم عن المنفرد
     album_inputs = []
     album_jobs = []
     standalone_jobs = []
     
     for i, job in enumerate(jobs):
-        if i == 0:
-            cap = job['caption']
-        else:
-            cap = None
-        
+        cap = job['caption'] if i == 0 else None
         m_type = job['media_type']
         file_path = job['file_path']
         
@@ -440,7 +397,6 @@ async def send_album_group(media_group_id):
             standalone_jobs.append(job)
     
     async with send_sem:
-        # إرسال الألبوم المدعوم (تقسيم إلى 10)
         chunk_size = 10
         for idx in range(0, len(album_inputs), chunk_size):
             chunk = album_inputs[idx:idx+chunk_size]
@@ -458,28 +414,33 @@ async def send_album_group(media_group_id):
                 
                 for j in chunk_jobs:
                     await delete_job(j['chat_id'], j['message_id'])
-                logger.info(f"✅ ألبوم {media_group_id} | جزء {idx//chunk_size + 1} تم الإرسال")
+                logger.info(f"Album {media_group_id} part {idx//chunk_size + 1} sent")
             except Exception as e:
-                logger.error(f"❌ فشل إرسال جزء من الألبوم: {e}")
+                logger.error(f"Album send failed: {e}")
         
-        # إرسال المنفردات
         for job in standalone_jobs:
             await send_job(job)
 
 # ============ Workers ============
 async def download_worker():
-    """معالج التحميل المستمر"""
-    logger.info("🔄 Download Worker started")
+    logger.info("Download Worker started")
+    await asyncio.sleep(3)
+    
     while True:
         try:
-            jobs = await get_pending_jobs(limit=5)
+            if not app.is_connected:
+                logger.warning("Pyrogram not connected, waiting...")
+                await asyncio.sleep(5)
+                continue
+            
+            jobs = await get_pending_jobs(limit=3)
             if not jobs:
                 await asyncio.sleep(2)
                 continue
             
             tasks = []
             for job in jobs:
-                if len(tasks) >= 3:
+                if len(tasks) >= 2:
                     break
                 tasks.append(download_job(job))
             
@@ -487,62 +448,68 @@ async def download_worker():
                 await asyncio.gather(*tasks, return_exceptions=True)
             
         except Exception as e:
-            logger.error(f"❌ خطأ في Download Worker: {e}")
+            logger.error(f"Download Worker error: {e}")
             await asyncio.sleep(5)
 
 async def send_worker():
-    """معالج الإرسال المستمر"""
-    logger.info("📤 Send Worker started")
+    logger.info("Send Worker started")
+    await asyncio.sleep(5)
+    
     while True:
         try:
-            # جلب المهام المحملة (غير ألبومات)
+            if not app.is_connected:
+                logger.warning("Pyrogram not connected, waiting...")
+                await asyncio.sleep(5)
+                continue
+            
             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 5
+                    ORDER BY created_at LIMIT 3
                 """)
-                rows = await cursor.fetchall()
-                single_jobs = [dict(row) for row in rows]
-            
-            # جلب الألبوم��ت الجاهزة
-            cursor = await db.execute("""
-                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
-            """)
-            album_rows = await cursor.fetchall()
+                single_jobs = [dict(row) for row in await cursor.fetchall()]
+                
+                cursor = await db.execute("""
+                    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
+                """)
+                album_rows = await cursor.fetchall()
             
-            # إرسال المنفردات
             for job in single_jobs:
+                if not app.is_connected:
+                    break
                 async with send_sem:
                     await send_job(job)
             
-            # معالجة الألبومات
             for (mg_id,) in album_rows:
+                if not app.is_connected:
+                    break
                 await send_album_group(mg_id)
             
             await asyncio.sleep(1)
             
         except Exception as e:
-            logger.error(f"❌ خطأ في Send Worker: {e}")
+            logger.error(f"Send Worker error: {e}")
             await asyncio.sleep(5)
 
-# ============ استخدام get_media_group للألبومات ============
+# ============ Album Fetch ============
 async def fetch_full_album(message):
-    """جلب كل رسائل الألبوم مرة واحدة باستخدام get_media_group"""
     try:
         messages = await app.get_media_group(message.chat.id, message.id)
-        logger.info(f"📦 get_media_group: {len(messages)} رسائل")
+        logger.info(f"get_media_group: {len(messages)} messages")
         for msg in messages:
             await add_job(msg)
         return True
     except Exception as e:
-        logger.warning(f"⚠️ فشل get_media_group: {e}")
+        logger.warning(f"get_media_group failed: {e}")
         return False
 
-# ============ المعالج الرئيسي ============
+# ============ Handlers ============
 @app.on_message(filters.incoming & ~filters.me & filters.media)
 async def on_media(client, message):
     chat_id = message.chat.id
@@ -551,13 +518,11 @@ async def on_media(client, message):
     
     src_name = get_source_name(message)
     m_type, _ = get_media_info(message)
-    logger.info(f"📥 جديد: {chat_id}/{message.id} | {m_type} | ألبوم: {message.media_group_id is not None}")
+    logger.info(f"New: {chat_id}/{message.id} | {m_type} | album: {message.media_group_id is not None}")
     
     if message.media_group_id:
-        # 🔑 محاولة جلب الألبوم كاملاً
         success = await fetch_full_album(message)
         if not success:
-            # Fallback: إضافة الرسالة الحالية فقط
             await add_job(message)
     else:
         await add_job(message)
@@ -566,7 +531,7 @@ async def on_media(client, message):
 async def monitor_all(client, message):
     if str(message.chat.id) == str(TARGET_CHAT):
         return
-    logger.info(f"👁️ رسالة نصية من {message.chat.id}")
+    logger.info(f"Text from {message.chat.id}")
 
 async def resolve_target_chat():
     global TARGET_CHAT
@@ -578,17 +543,17 @@ async def resolve_target_chat():
         chat_id_int = int(raw)
         chat = await app.get_chat(chat_id_int)
         TARGET_CHAT = chat_id_int
-        logger.info(f"🎯 الهدف: {chat.title}")
+        logger.info(f"Target: {chat.title}")
     except Exception as e:
-        logger.error(f"❌ فشل التحقق: {e}")
+        logger.error(f"Target check failed: {e}")
         TARGET_CHAT = "me"
 
 # ============ Gradio + FastAPI ============
 def run_gradio():
     with gr.Blocks(title="Telegram Archiver Pro") as demo:
-        gr.Markdown("# 📡 Telegram Archiver Pro")
-        gr.Markdown("🟢 **الحالة:** نظام Queue مضبوط 100%")
-        gr.Textbox(label="الحالة", value="✅ نشط | SQLite Queue | 3 Workers", interactive=False)
+        gr.Markdown("# Telegram Archiver Pro")
+        gr.Markdown("Status: Running")
+        gr.Textbox(label="Status", value="Active | SQLite Queue | 3 Workers", interactive=False)
     
     fastapi_app = FastAPI()
     
@@ -605,23 +570,34 @@ def run_gradio():
     import uvicorn
     uvicorn.run(fastapi_app, host="0.0.0.0", port=7860)
 
-# ============ التشغيل الرئيسي ============
+# ============ Main ============
 async def main():
     await app.start()
+    
+    retries = 0
+    while not app.is_connected and retries < 10:
+        await asyncio.sleep(1)
+        retries += 1
+        logger.info(f"Waiting for connection... ({retries}/10)")
+    
+    if not app.is_connected:
+        logger.error("Failed to connect to Telegram")
+        return
+    
     me = await app.get_me()
     await resolve_target_chat()
     await init_db()
     
-    # تشغيل Workers
+    await asyncio.sleep(2)
     asyncio.create_task(download_worker())
+    await asyncio.sleep(2)
     asyncio.create_task(send_worker())
     
     logger.info("=" * 60)
-    logger.info(f"🚀 Archiver Pro يعمل!")
-    logger.info(f"👤 الحساب: {me.first_name}")
-    logger.info(f"🎯 الهدف: {TARGET_CHAT}")
-    logger.info(f"💾 قاعدة البيانات: {DB_PATH}")
-    logger.info(f"📁 مجلد التحميل: {DOWNLOAD_DIR}")
+    logger.info(f"Archiver Pro running!")
+    logger.info(f"Account: {me.first_name}")
+    logger.info(f"Target: {TARGET_CHAT}")
+    logger.info(f"Database: {DB_PATH}")
     logger.info("=" * 60)
     
     stop_event = asyncio.Event()