Update bot.py

bot.py

@@ -6,149 +6,215 @@ from http.server import BaseHTTPRequestHandler, HTTPServer
 from pyrogram import Client, filters, idle
 from pyrogram.errors import FloodWait
 from pyrogram.enums import ChatType
+from pyrogram.types import InputMediaPhoto, InputMediaVideo, InputMediaDocument, InputMediaAudio
 
-# --- 1. إعداد مساحات التخزين (متوافق مع Hugging Face) ---
-DATA_DIR = "./data" 
+# --- 1. إعداد مساحات التخزين وقاعدة البيانات المستدامة ---
+DATA_DIR = "./data"
 DOWNLOAD_DIR = os.path.join(DATA_DIR, "downloads")
-DB_PATH = os.path.join(DATA_DIR, 'bot_memory.db')
+DB_PATH = os.path.join(DATA_DIR, 'bot_archiver.db')
 os.makedirs(DOWNLOAD_DIR, exist_ok=True)
 
 conn = sqlite3.connect(DB_PATH, check_same_thread=False)
 cursor = conn.cursor()
-cursor.execute('''CREATE TABLE IF NOT EXISTS processed (chat_id INTEGER, msg_id INTEGER, UNIQUE(chat_id, msg_id))''')
-conn.commit()
-
-def is_processed(chat_id, msg_id):
-    cursor.execute("SELECT 1 FROM processed WHERE chat_id=? AND msg_id=?", (chat_id, msg_id))
-    return cursor.fetchone() is not None
 
-def mark_processed(chat_id, msg_id):
-    cursor.execute("INSERT OR IGNORE INTO processed VALUES (?, ?)", (chat_id, msg_id))
-    conn.commit()
+# جدول إدارة الطابور وحفظ الألبومات
+cursor.execute('''
+CREATE TABLE IF NOT EXISTS queue (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    chat_id INTEGER,
+    msg_id INTEGER,
+    media_group_id TEXT,
+    media_type TEXT,
+    caption TEXT,
+    file_path TEXT,
+    status TEXT DEFAULT 'DOWNLOADED',
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    UNIQUE(chat_id, msg_id)
+)
+''')
+conn.commit()
 
-# --- 2. سيرفر الويب الوهمي (لمنع Hugging Face من إغلاق البوت) ---
+# --- 2. سيرفر الويب لمنع إغلاق منصة Hugging Face ---
 class RequestHandler(BaseHTTPRequestHandler):
     def do_GET(self):
         self.send_response(200)
         self.end_headers()
-        self.wfile.write(b"Ultimate Safe Archiver is Running Perfectly!")
+        self.wfile.write(b"Dual-Engine Telegram Archiver Active!")
     def log_message(self, format, *args): pass
 
 Thread(target=lambda: HTTPServer(('0.0.0.0', 7860), RequestHandler).serve_forever(), daemon=True).start()
 
-# --- 3. إعدادات الحساب (بالذاكرة المؤقتة لمنع التعارض) ---
+# --- 3. إعدادات الحساب ---
 API_ID = int(os.environ.get("API_ID", 0))
 API_HASH = os.environ.get("API_HASH", "")
 SESSION_STRING = os.environ.get("SESSION_STRING", "")
 TARGET_CHAT = int(os.environ.get("TARGET_CHAT", 0))
 
 app = Client(
-    "memory_session", 
-    api_id=API_ID, 
-    api_hash=API_HASH, 
+    "memory_session",
+    api_id=API_ID,
+    api_hash=API_HASH,
     session_string=SESSION_STRING,
-    in_memory=True, 
-    max_concurrent_transmissions=1 # إرسال ملف واحد في نفس الوقت للأمان
+    in_memory=True,
+    max_concurrent_transmissions=3 # تسريع التحميل المتوازي
 )
 
-process_queue = asyncio.Queue()
+# --- 4. دالة التنزيل الآمن مع إعادة المحاولة لكسر خطأ 0 بايت ---
+async def download_media_safe(message):
+    cursor.execute("SELECT status FROM queue WHERE chat_id=? AND msg_id=?", (message.chat.id, message.id))
+    if cursor.fetchone():
+        return None # تم تنزيله سابقاً
 
-# --- 4. الرادار اللحظي (يصطاد المقطع فور نشره حتى لو حُذف لاحقاً) ---
-@app.on_message(filters.incoming & ~filters.me & (filters.channel | filters.group) & filters.media)
-async def catch_new_messages(client, message):
-    if not is_processed(message.chat.id, message.id):
-        await process_queue.put(message)
-
-# --- 5. حاصد الأرشيف القديم ---
-async def scan_old_history():
-    print("🔎 [جيمي] بدء مسح الأرشيف للتأكد من عدم تفويت أي مقطع...")
-    try:
-        async for dialog in app.get_dialogs():
-            if dialog.chat.type in [ChatType.CHANNEL, ChatType.SUPERGROUP, ChatType.GROUP] and dialog.chat.id != TARGET_CHAT:
-                print(f"📂 يتم الآن فحص: {dialog.chat.title}...")
-                try:
-                    async for message in app.get_chat_history(dialog.chat.id):
-                        if message.media and not is_processed(message.chat.id, message.id):
-                            await process_queue.put(message)
-                except Exception:
-                    pass
-                
-                await asyncio.sleep(7) # فاصل زمني لتجنب حظر المسح
-        print("✅ اكتمل سحب كافة الرسائل إلى طابور المعالجة المحكم!")
-    except Exception as e:
-        print(f"❌ خطأ بالمسح (لن يؤثر على العمليات الحالية): {e}")
-
-# --- 6. المحرك المدرع للكسر والرفع (لحل مشكلة 0 بايت وضمان النقل) ---
-async def restricted_bypass_worker():
-    while True:
-        message = await process_queue.get()
-        
-        if is_processed(message.chat.id, message.id):
-            process_queue.task_done()
-            continue
-            
-        file_path = None
+    unique_name = f"{DOWNLOAD_DIR}/{message.chat.id}_{message.id}"
+    
+    for attempt in range(3): # 3 محاولات تنزيل عند حدوث خلل شبكي
         try:
-            # إعطاء اسم فريد للملف لتجنب التداخل
-            unique_file_name = f"{DOWNLOAD_DIR}/{message.chat.id}_{message.id}"
-            print(f"⬇️ جاري سحب الملف (ID: {message.id})...")
-            
-            # محاولة التنزيل
-            file_path = await message.download(file_name=unique_file_name)
+            file_path = await message.download(file_name=unique_name)
+            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"
+
+                group_id = str(message.media_group_id) if message.media_group_id else None
+                caption = message.caption or ""
+
+                cursor.execute(
+                    "INSERT OR IGNORE INTO queue (chat_id, msg_id, media_group_id, media_type, caption, file_path) VALUES (?, ?, ?, ?, ?, ?)",
+                    (message.chat.id, message.id, group_id, m_type, caption, file_path)
+                )
+                conn.commit()
+                print(f"⚡ [صائد لحظي] تم سحب وحفظ الملف: ID {message.id}")
+                return file_path
+            else:
+                if file_path and os.path.exists(file_path):
+                    os.remove(file_path)
+        except Exception as e:
+            print(f"⚠️ محاولة تنزيل فاشلة ({attempt+1}/3) للرسالة {message.id}: {e}")
+            await asyncio.sleep(2)
             
-            # فحص رقم 1: هل تم إنشاء الملف؟
-            if not file_path or not os.path.exists(file_path):
-                raise Exception("فشل تيليجرام في إنشاء الملف.")
+    return None
 
-            # فحص رقم 2 (المهم جداً لحل مشكلتك): هل الملف فارغ؟
-            if os.path.getsize(file_path) == 0:
-                os.remove(file_path) # حذف الملف التالف
-                raise Exception("حجم الملف 0 بايت (سيتم الإعادة لاحقاً).")
+# --- 5. صائد الرسائل اللحظي (التقاط فوري قبل الحذف) ---
+@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)
+
+# --- 6. حاصد السجل القديم ---
+async def scan_history_worker():
+    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:
+                async for message in app.get_chat_history(dialog.chat.id):
+                    if message.media:
+                        await download_media_safe(message)
+            except Exception as e:
+                print(f"تجاوز القناة {dialog.chat.title} بسبب: {e}")
+            await asyncio.sleep(5)
+    print("✅ اكتمل مسح كافة الأرشيفات القديمة.")
 
-            print(f"⬆️ جاري الرفع إلى قناتك (حجم الملف سليم)...")
-            caption = message.caption or ""
-            
-            if message.photo:
+# --- 7. محرك الرفع الذكي وتجميع الألبومات (Uploader & Album Engine) ---
+async def upload_worker():
+    while True:
+        try:
+            # البحث عن أقدم عنصر نزل ولم يرفع بعد
+            cursor.execute("SELECT id, chat_id, msg_id, media_group_id, media_type, caption, file_path FROM queue WHERE status='DOWNLOADED' ORDER BY id ASC LIMIT 1")
+            row = cursor.fetchone()
+
+            if not row:
+                await asyncio.sleep(5)
+                continue
+
+            q_id, chat_id, msg_id, group_id, m_type, caption, file_path = row
+
+            # التأكد من وجود الملف على القرص
+            if not os.path.exists(file_path):
+                cursor.execute("UPDATE queue SET status='FAILED' WHERE id=?", (q_id,))
+                conn.commit()
+                continue
+
+            # معالجة الألبومات (Media Groups)
+            if group_id:
+                await asyncio.sleep(4) # مهلة زمنية للتأكد من اكتمال تنزيل جميع عناصر الألبوم
+                cursor.execute(
+                    "SELECT id, media_type, caption, file_path FROM queue WHERE chat_id=? AND media_group_id=? AND status='DOWNLOADED' ORDER BY msg_id ASC",
+                    (chat_id, group_id)
+                )
+                group_rows = cursor.fetchall()
+
+                if group_rows:
+                    media_list = []
+                    ids_to_update = []
+                    files_to_delete = []
+
+                    for idx, g_item in enumerate(group_rows[:10]): # تيليجرام يدعم حتى 10 وسائط في الألبوم الواحد
+                        g_id, g_type, g_cap, g_path = g_item
+                        # وضع النص الوصفي على العنصر الأول فقط في الألبوم
+                        item_caption = g_cap if idx == 0 else ""
+
+                        if g_type == "photo":
+                            media_list.append(InputMediaPhoto(media=g_path, caption=item_caption))
+                        elif g_type == "video":
+                            media_list.append(InputMediaVideo(media=g_path, caption=item_caption, supports_streaming=True))
+                        elif g_type == "doc":
+                            media_list.append(InputMediaDocument(media=g_path, caption=item_caption))
+                        elif g_type == "audio":
+                            media_list.append(InputMediaAudio(media=g_path, caption=item_caption))
+
+                        ids_to_update.append(g_id)
+                        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()
+
+                        for f_path in files_to_delete:
+                            if os.path.exists(f_path): os.remove(f_path)
+                        
+                        print("✅ تم رفع الألبوم بنجاح!")
+                        await asyncio.sleep(15) # فاصل حماية بعد رفع الألبوم
+                        continue
+
+            # معالجة الرسائل المنفردة
+            print(f"⬆️ جاري رفع ملف منفرد (ID: {msg_id})...")
+            if m_type == "photo":
                 await app.send_photo(TARGET_CHAT, photo=file_path, caption=caption)
-            elif message.video:
+            elif m_type == "video":
                 await app.send_video(TARGET_CHAT, video=file_path, caption=caption, supports_streaming=True)
-            elif message.document:
+            elif m_type == "doc":
                 await app.send_document(TARGET_CHAT, document=file_path, caption=caption)
-            else:
+            elif m_type == "audio":
                 await app.send_audio(TARGET_CHAT, audio=file_path, caption=caption)
-                
-            # لن يتم التوثيق كـ "مكتمل" إلا إذا نجح التنزيل والرفع معاً 100%
-            mark_processed(message.chat.id, message.id)
-            print(f"✅ تم النقل بنجاح (ID: {message.id})!")
-            
-            await asyncio.sleep(12) # استراحة لتهدئة السيرفر
+
+            cursor.execute("UPDATE queue SET status='UPLOADED' WHERE id=?", (q_id,))
+            conn.commit()
+
+            if os.path.exists(file_path): os.remove(file_path)
+            print(f"✅ تم رفع الملف بنجاح (ID: {msg_id})!")
+            await asyncio.sleep(12)
 
         except FloodWait as e:
-            print(f"🛑 تيليجرام يطلب التوقف {e.value} ثانية! (الملف بأمان في الطابور)")
+            print(f"🛑 [FloodWait] طلب تيليجرام التوقف لـ {e.value} ثانية. سيتم الانتظار دون فقدان البيانات.")
             await asyncio.sleep(e.value + 5)
-            await process_queue.put(message) # إعادة المقطع للطابور لعدم فقدانه
-            
         except Exception as e:
-            print(f"⚠️ خطأ مؤقت: {e} | سيتم إعادة محاولة نقل (ID: {message.id}) بعد قليل.")
-            await asyncio.sleep(5)
-            await process_queue.put(message) # إعادة المقطع للطابور لحين نجاحه
-            
-        finally:
-            # التنظيف الذكي: حذف الملف من السيرفر بعد رفعه أو فشله لتوفير المساحة
-            if file_path and os.path.exists(file_path):
-                try:
-                    os.remove(file_path)
-                except:
-                    pass
-            process_queue.task_done()
+            print(f"❌ خطأ أثناء الرفع: {e}")
+            await asyncio.sleep(10)
 
-# --- 7. التشغيل ---
+# --- 8. نقطة الانطلاق ---
 async def main():
     await app.start()
-    print("🚀 البوت المدرع يعمل الآن! لن يفوتك أي مقطع بإذن الله.")
+    print("🚀 تم تشغيل نظام الأرشيف المدرع (سحب فوري + رفع ألبومات)...")
     
-    asyncio.create_task(restricted_bypass_worker())
-    asyncio.create_task(scan_old_history())
+    asyncio.create_task(upload_worker())
+    asyncio.create_task(scan_history_worker())
     
     await idle()
     await app.stop()