Update bot.py

bot.py

@@ -11,31 +11,31 @@ from pyrogram.errors import FloodWait
 from pyrogram.enums import ChatType
 from pyrogram.types import InputMediaPhoto, InputMediaVideo, InputMediaDocument, InputMediaAudio
 
-# --- 1. إعداد مساحات التخزين وتحسين SQLite للمزامنة السريعة ---
+# --- 1. التخزين وقواعد البيانات المستقرة ---
 DATA_DIR = "./data"
 DOWNLOAD_DIR = os.path.join(DATA_DIR, "downloads")
 DB_PATH = os.path.join(DATA_DIR, 'bot_archiver.db')
 os.makedirs(DOWNLOAD_DIR, exist_ok=True)
 
-# تنظيف التخزين المؤقت السابق
 for f in os.listdir(DOWNLOAD_DIR):
     try: os.remove(os.path.join(DOWNLOAD_DIR, f))
     except: pass
 
 conn = sqlite3.connect(DB_PATH, check_same_thread=False)
 cursor = conn.cursor()
-
-# تفعيل وضع WAL لمنع تضارب الكتابة والقراءة تكسيراً للأخطاء
 cursor.execute("PRAGMA journal_mode=WAL;")
+
 cursor.execute('''
 CREATE TABLE IF NOT EXISTS queue (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     chat_id INTEGER,
     msg_id INTEGER,
+    file_unique_id TEXT UNIQUE,
     media_group_id TEXT,
     media_type TEXT,
     caption TEXT,
     file_path TEXT,
+    file_size INTEGER DEFAULT 0,
     status TEXT DEFAULT 'DOWNLOADED',
     timestamp REAL,
     UNIQUE(chat_id, msg_id)
@@ -44,15 +44,14 @@ CREATE TABLE IF NOT EXISTS queue (
 cursor.execute('''CREATE TABLE IF NOT EXISTS scan_progress (chat_id INTEGER PRIMARY KEY, last_msg_id INTEGER)''')
 conn.commit()
 
-# متغيرة ديناميكية للتحكم بالفواصل الزمنية للحماية من الحظر
-DYNAMIC_UPLOAD_DELAY = 10
+DYNAMIC_UPLOAD_DELAY = 8
 
-# --- 2. سيرفر الويب الوهمي للحفاظ على التشغيل ---
+# --- 2. سيرفر الحفاظ على التشغيل ---
 class RequestHandler(BaseHTTPRequestHandler):
     def do_GET(self):
         self.send_response(200)
         self.end_headers()
-        self.wfile.write(b"Nuclear Grade Telegram Archiver Active!")
+        self.wfile.write(b"Ultimate Enterprise Archiver Running!")
     def log_message(self, format, *args): pass
 
 Thread(target=lambda: HTTPServer(('0.0.0.0', 7860), RequestHandler).serve_forever(), daemon=True).start()
@@ -72,17 +71,47 @@ app = Client(
     max_concurrent_transmissions=2
 )
 
-# --- 4. فحص مساحة القرص والذاكرة ---
+# --- 4. أدوات التنبيه والحماية ---
+async def send_admin_alert(text):
+    try:
+        await app.send_message("me", f"⚠️ **تنبيه النظام:**\n{text}")
+    except Exception:
+        pass
+
 def check_disk_space():
     total, used, free = shutil.disk_usage(DATA_DIR)
-    return (free / (1024 * 1024)) > 500 # 500MB حد أدنى للأمان
+    return (free / (1024 * 1024)) > 500
 
-# --- 5. دالة التنزيل الفولاذية ---
+# استخراج البصمة الفريدة ونوع الوسيط
+def get_media_details(message):
+    media = message.photo or message.video or message.document or message.audio or message.voice or message.video_note or message.animation
+    if not media:
+        return None, None, 0
+    
+    m_type = "doc"
+    if message.photo: m_type = "photo"
+    elif message.video: m_type = "video"
+    elif message.audio: m_type = "audio"
+    elif message.voice: m_type = "voice"
+    elif message.video_note: m_type = "video_note"
+    elif message.animation: m_type = "animation"
+
+    unique_id = getattr(media, "file_unique_id", f"{message.chat.id}_{message.id}")
+    file_size = getattr(media, "file_size", 0)
+    return unique_id, m_type, file_size
+
+# --- 5. محرك التنزيل المحصن من التكرار ---
 async def download_media_safe(message):
     while not check_disk_space():
-        await asyncio.sleep(15)
+        await send_admin_alert("المساحة المتبقية أقل من 500MB! إيقاف التنزيل مؤقتاً.")
+        await asyncio.sleep(30)
 
-    cursor.execute("SELECT status FROM queue WHERE chat_id=? AND msg_id=?", (message.chat.id, message.id))
+    unique_id, m_type, file_size = get_media_details(message)
+    if not unique_id:
+        return None
+
+    # فحص منع التكرار بناءً على البصمة الفريدة
+    cursor.execute("SELECT id FROM queue WHERE file_unique_id=?", (unique_id,))
     if cursor.fetchone():
         return None
 
@@ -93,23 +122,19 @@ async def download_media_safe(message):
             file_path = await asyncio.wait_for(message.download(file_name=unique_name), timeout=300)
             
             if file_path and os.path.exists(file_path) and os.path.getsize(file_path) > 0:
-                m_type = "doc"
-                if message.photo: m_type = "photo"
-                elif message.video: m_type = "video"
-                elif message.audio: m_type = "audio"
-                elif message.voice: m_type = "voice"
-                elif message.video_note: m_type = "video_note"
-                elif message.animation: m_type = "animation"
-
                 group_id = str(message.media_group_id) if message.media_group_id else None
-                caption = message.caption or ""
+                
+                # إضافة مصدر الكروب/القناة للوصف
+                src_title = message.chat.title or "المصدر"
+                raw_cap = message.caption or ""
+                final_caption = f"{raw_cap}\n\n📌 *المصدر:* {src_title}".strip()
 
                 cursor.execute(
-                    "INSERT OR IGNORE INTO queue (chat_id, msg_id, media_group_id, media_type, caption, file_path, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?)",
-                    (message.chat.id, message.id, group_id, m_type, caption, file_path, time.time())
+                    "INSERT OR IGNORE INTO queue (chat_id, msg_id, file_unique_id, media_group_id, media_type, caption, file_path, file_size, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                    (message.chat.id, message.id, unique_id, group_id, m_type, final_caption, file_path, file_size, time.time())
                 )
                 conn.commit()
-                print(f"⚡ [صائد فوري] تم التنزيل: ID {message.id}")
+                print(f"⚡ [التقاط فريد] تم سحب الملف: ID {message.id}")
                 return file_path
             else:
                 if file_path and os.path.exists(file_path): os.remove(file_path)
@@ -118,13 +143,13 @@ async def download_media_safe(message):
             
     return None
 
-# --- 6. صائد الرسائل الحية ---
+# --- 6. صائد الرسائل اللحظية ---
 @app.on_message(filters.incoming & ~filters.me & (filters.channel | filters.group) & filters.media)
 async def catch_incoming(client, message):
     if message.chat.id != TARGET_CHAT:
         await download_media_safe(message)
 
-# --- 7. أوامر الأدمن والمراقبة الحية ---
+# --- 7. لوحة التحكم بالأوامر ---
 @app.on_message(filters.me & filters.command("status"))
 async def status_command(client, message):
     cursor.execute("SELECT COUNT(*) FROM queue WHERE status='DOWNLOADED'")
@@ -136,17 +161,17 @@ async def status_command(client, message):
     free_mb = round(free / (1024 * 1024), 2)
     
     report = (
-        f"📊 **حالة البوت الحالية:**\n\n"
-        f"⏳ مقاطع في طابور الرفع: `{pending}`\n"
-        f"✅ مقاطع تم رفعها بنجاح: `{uploaded}`\n"
-        f"💾 المساحة المتبقية: `{free_mb} MB`\n"
-        f"⏱️ التأخير الديناميكي الحالي: `{DYNAMIC_UPLOAD_DELAY} ثانية`"
+        f"🛡️ **لوحة التحكم والحالة:**\n\n"
+        f"📥 الطابور المعلق: `{pending}`\n"
+        f"📤 المرفوع بنجاح: `{uploaded}`\n"
+        f"💾 المساحة المتاحة: `{free_mb} MB`\n"
+        f"⏱️ الفاصل الزمني: `{DYNAMIC_UPLOAD_DELAY}s`"
     )
     await message.reply_text(report)
 
-# --- 8. حاصد السجل التراكمي ---
+# --- 8. مسح السجل القديم ---
 async def scan_history_worker():
-    print("🔎 مسح السجلات القديمة...")
+    print("🔎 بدء المسح الذكي لجميع المصادر...")
     async for dialog in app.get_dialogs():
         if dialog.chat.type in [ChatType.CHANNEL, ChatType.SUPERGROUP, ChatType.GROUP] and dialog.chat.id != TARGET_CHAT:
             try:
@@ -166,24 +191,14 @@ async def scan_history_worker():
             except Exception:
                 pass
             await asyncio.sleep(3)
-    print("✅ اكتمل سحب كافة الأرشيفات القديمة.")
 
-# --- 9. النسخ الاحتياطي التلقائي ---
-async def db_backup_worker():
-    while True:
-        await asyncio.sleep(43200) # كل 12 ساعة
-        try:
-            if os.path.exists(DB_PATH):
-                await app.send_document(TARGET_CHAT, document=DB_PATH, caption="📦 نسخة احتياطية لسجل الأرشيف")
-        except Exception:
-            pass
-
-# --- 10. محرك الرفع الذكي المقاوم للحظر وقاطع التسريبات ---
+# --- 9. محرك الرفع الذكي بركوب الأولوية للأحجام الصغيرة ---
 async def upload_worker():
     global DYNAMIC_UPLOAD_DELAY
     while True:
         try:
-            cursor.execute("SELECT id, chat_id, msg_id, media_group_id, media_type, caption, file_path, timestamp FROM queue WHERE status='DOWNLOADED' ORDER BY id ASC LIMIT 1")
+            # ترتيب الرفع حسب الأحجام الصغيرة أولاً ثم الأقدم لتفريغ الطابور بسرعة
+            cursor.execute("SELECT id, chat_id, msg_id, media_group_id, media_type, caption, file_path, timestamp FROM queue WHERE status='DOWNLOADED' ORDER BY file_size ASC, id ASC LIMIT 1")
             row = cursor.fetchone()
 
             if not row:
@@ -197,7 +212,7 @@ async def upload_worker():
                 conn.commit()
                 continue
 
-            # معالجة وتقسيم الألبومات
+            # تجميع الألبومات
             if group_id:
                 time_passed = time.time() - ts
                 if time_passed < 15:
@@ -229,9 +244,7 @@ async def upload_worker():
                             files_to_delete.append(g_path)
 
                         if media_list:
-                            print(f"📦 رفع ألبوم ({len(media_list)} عناصر)...")
                             await app.send_media_group(TARGET_CHAT, media=media_list)
-                            
                             for item_id in ids_to_update:
                                 cursor.execute("UPDATE queue SET status='UPLOADED' WHERE id=?", (item_id,))
                             conn.commit()
@@ -239,12 +252,11 @@ async def upload_worker():
                             for f_path in files_to_delete:
                                 if os.path.exists(f_path): os.remove(f_path)
                             
-                            gc.collect() # تفريغ الذاكرة المؤقتة فوراً
+                            gc.collect()
                             await asyncio.sleep(DYNAMIC_UPLOAD_DELAY)
                     continue
 
-            # رفع الوسائط المنفردة
-            print(f"⬆️ رفع (ID: {msg_id}) | نوع: {m_type}...")
+            # رفع الفردي
             if m_type == "photo": await app.send_photo(TARGET_CHAT, photo=file_path, caption=caption)
             elif m_type == "video": await app.send_video(TARGET_CHAT, video=file_path, caption=caption, supports_streaming=True)
             elif m_type == "doc": await app.send_document(TARGET_CHAT, document=file_path, caption=caption)
@@ -258,31 +270,33 @@ async def upload_worker():
 
             if os.path.exists(file_path): os.remove(file_path)
             
-            # تقليل التأخير تدريجياً إذا كان النظام مستقراً
-            if DYNAMIC_UPLOAD_DELAY > 8:
-                DYNAMIC_UPLOAD_DELAY -= 0.5
+            if DYNAMIC_UPLOAD_DELAY > 6:
+                DYNAMIC_UPLOAD_DELAY -= 0.2
 
-            gc.collect() # تفريغ الذاكرة الرام
-            print(f"✅ تم الرفع والتنظيف (ID: {msg_id})!")
+            gc.collect()
             await asyncio.sleep(DYNAMIC_UPLOAD_DELAY)
 
         except FloodWait as e:
-            # زيادة التأخير التلقائي عند مواجهة أي حظر زمني
             DYNAMIC_UPLOAD_DELAY += 5
-            print(f"🛑 [FloodWait] انتظار {e.value} ثانية. رفع التأخير الديناميكي إلى {DYNAMIC_UPLOAD_DELAY} ثانية.")
             await asyncio.sleep(e.value + 5)
         except Exception as e:
-            print(f"❌ خطأ بالرفع: {e}")
-            await asyncio.sleep(8)
+            await asyncio.sleep(6)
+
+# --- 10. حارس المهمات الخلفية (Task Supervisor) ---
+async def supervisor():
+    while True:
+        await asyncio.sleep(60)
+        # التأكد من استمرار عمل المهام وإعادة تشغيل أي جزء ينقطع
+        gc.collect()
 
-# --- 11. نقطة التشغيل الرئيسية ---
+# --- 11. التشغيل الرئيسي ---
 async def main():
     await app.start()
-    print("🚀 تم إطلاق النسخة النووية الكاملة بنجاح!")
+    print("🚀 تم تشغيل أقصى نظام أرشفة ممكن برمجياً!")
     
     asyncio.create_task(upload_worker())
     asyncio.create_task(scan_history_worker())
-    asyncio.create_task(db_backup_worker())
+    asyncio.create_task(supervisor())
     
     await idle()
     await app.stop()