Update bot.py

bot.py

@@ -1,336 +1,281 @@
 import os
-import gc
 import asyncio
-import aiosqlite
-import time
+import logging
 import shutil
-import random
-import threading
-from pyrogram import Client, filters, idle
-from pyrogram.errors import FloodWait
+import time
+import gradio as gr
+from pyrogram import Client, filters
+from pyrogram.errors import FloodWait, RPCError
 from pyrogram.types import InputMediaPhoto, InputMediaVideo, InputMediaDocument, InputMediaAudio
 from pyrogram.enums import ChatAction
-import gradio as gr
 
-# ============ الإعدادات ============
-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)
-
-MAX_FILE_SIZE = 2000 * 1024 * 1024  # 2GB
-BASE_UPLOAD_DELAY = 4
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
 
+# ============ الإعدادات ============
 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))
+SOURCE_CHATS = os.environ.get("SOURCE_CHATS", "")
 
-download_queue = asyncio.Queue()
-
-# ============ قاعدة البيانات ============
-async def init_db():
-    async with aiosqlite.connect(DB_PATH) as db:
-        await db.execute("PRAGMA journal_mode=WAL;")
-        await db.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)
-        )''')
-        await db.commit()
-
-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
-
-# ============ محرك التنزيل ============
-async def download_worker_task(worker_id):
-    while True:
-        message = await download_queue.get()
-        try:
-            while shutil.disk_usage(DATA_DIR).free / (1024 * 1024) <= 500:
-                await asyncio.sleep(15)
-
-            unique_id, m_type, file_size = get_media_details(message)
-            if not unique_id or file_size > MAX_FILE_SIZE:
-                continue
+SOURCE_CHAT_IDS = []
+if SOURCE_CHATS:
+    try:
+        SOURCE_CHAT_IDS = [int(x.strip()) for x in SOURCE_CHATS.split(",") if x.strip()]
+    except:
+        logger.error("❌ تنسيق SOURCE_CHATS غير صحيح")
 
-            async with aiosqlite.connect(DB_PATH) as db:
-                async with db.execute("SELECT id FROM queue WHERE file_unique_id=?", (unique_id,)) as cursor:
-                    if await cursor.fetchone():
-                        continue
-            
-            file_prefix = f"{DOWNLOAD_DIR}/{message.chat.id}_{message.id}_"
-            
-            for attempt in range(3):
-                try:
-                    file_path = await asyncio.wait_for(message.download(file_name=file_prefix), timeout=300)
-                    if file_path and os.path.exists(file_path) and os.path.getsize(file_path) > 0:
-                        group_id = str(message.media_group_id) if message.media_group_id else None
-                        src_title = message.chat.title or "المصدر"
-                        raw_cap = message.caption or ""
-                        final_caption = f"{raw_cap}\n\n📌 *المصدر:* {src_title}".strip()
+DATA_DIR = "./data"
+DOWNLOAD_DIR = os.path.join(DATA_DIR, "downloads")
+os.makedirs(DOWNLOAD_DIR, exist_ok=True)
 
-                        async with aiosqlite.connect(DB_PATH) as db:
-                            await db.execute(
-                                "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())
-                            )
-                            await db.commit()
-                        print(f"⚡ [Worker-{worker_id}] تم تنزيل: {message.id}")
-                        break
-                    else:
-                        if file_path and os.path.exists(file_path):
-                            os.remove(file_path)
-                except Exception:
-                    await asyncio.sleep(2)
-        except Exception as e:
-            print(f"❌ خطأ في التنزيل: {e}")
-        finally:
-            download_queue.task_done()
+logger.info(f"🎯 الهدف: {TARGET_CHAT} | المصادر: {SOURCE_CHAT_IDS or 'الكل'}")
 
-# ============ صائد الرسائل ============
+# ============ العميل ============
 app = Client(
-    "memory_session",
+    "archiver",
     api_id=API_ID,
     api_hash=API_HASH,
     session_string=SESSION_STRING,
     in_memory=True,
     device_model="Samsung Galaxy S23 Ultra",
-    system_version="Android 14.0",
+    system_version="Android 14",
     app_version="10.14.1",
     lang_code="ar"
 )
 
-@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_queue.put(message)
-
-@app.on_message(filters.me & filters.command("status"))
-async def status_command(client, message):
-    async with aiosqlite.connect(DB_PATH) as db:
-        async with db.execute("SELECT COUNT(*) FROM queue WHERE status='DOWNLOADED'") as c1:
-            pending = (await c1.fetchone())[0]
-        async with db.execute("SELECT COUNT(*) FROM queue WHERE status='UPLOADED'") as c2:
-            uploaded = (await c2.fetchone())[0]
+# ============ دوال مساعدة ============
+def get_media_info(message):
+    """استخرج معلومات الملف من الرسالة"""
+    if message.photo:
+        return "photo", message.photo.file_size or 0
+    elif message.video:
+        return "video", message.video.file_size or 0
+    elif message.document:
+        return "doc", message.document.file_size or 0
+    elif message.audio:
+        return "audio", message.audio.file_size or 0
+    elif message.voice:
+        return "voice", message.voice.file_size or 0
+    elif message.video_note:
+        return "video_note", message.video_note.file_size or 0
+    elif message.animation:
+        return "animation", message.animation.file_size or 0
+    return None, 0
+
+# ============ التعامل مع الألبومات ============
+media_groups = {}
+mg_lock = asyncio.Lock()
+
+async def flush_media_group(chat_id, mg_id):
+    """انتظر 5 ثواني لجمع كل أجزاء الألبوم ثم أرسلها"""
+    await asyncio.sleep(5)
     
-    total, used, free = shutil.disk_usage(DATA_DIR)
-    free_mb = round(free / (1024 * 1024), 2)
+    async with mg_lock:
+        key = (chat_id, mg_id)
+        if key not in media_groups:
+            return
+        messages = media_groups.pop(key, [])
     
-    report = (
-        f"👑 **حالة الأرشفة:**\n\n"
-        f"📥 جاهز للرفع: `{pending}`\n"
-        f"📤 تم رفعه: `{uploaded}`\n"
-        f"🔄 قيد التنزيل: `{download_queue.qsize()}`\n"
-        f"💾 مساحة متاحة: `{free_mb} MB`"
-    )
-    await message.reply_text(report)
-
-# ============ محرك الرفع ============
-async def upload_worker():
-    global BASE_UPLOAD_DELAY
-    while True:
+    if not messages:
+        return
+    
+    messages.sort(key=lambda m: m.id)
+    logger.info(f"📦 ألبوم {mg_id} يحتوي على {len(messages)} ملفات")
+    
+    downloaded = []
+    for i, msg in enumerate(messages):
+        m_type, fsize = get_media_info(msg)
+        if not m_type:
+            continue
+        
+        cap = msg.caption if i == 0 else ""
+        src_title = msg.chat.title or "المصدر"
+        final_cap = f"{cap}\n\n📌 المصدر: {src_title}".strip() if cap else f"📌 المصدر: {src_title}"
+        
         try:
-            async with aiosqlite.connect(DB_PATH) as db:
-                async with db.execute(
-                    "SELECT id, chat_id, msg_id, media_group_id, media_type, caption, file_path, file_size, timestamp FROM queue WHERE status='DOWNLOADED' ORDER BY id ASC LIMIT 1"
-                ) as cursor:
-                    row = await cursor.fetchone()
-
-            if not row:
-                await asyncio.sleep(2)
-                continue
-
-            q_id, chat_id, msg_id, group_id, m_type, caption, file_path, file_size, ts = row
-
-            if not os.path.exists(file_path):
-                async with aiosqlite.connect(DB_PATH) as db:
-                    await db.execute("UPDATE queue SET status='FAILED' WHERE id=?", (q_id,))
-                    await db.commit()
-                continue
-
+            file_path = await msg.download(file_name=f"{DOWNLOAD_DIR}/{chat_id}_{msg.id}_")
+            if file_path and os.path.exists(file_path) and os.path.getsize(file_path) > 0:
+                downloaded.append((m_type, file_path, final_cap))
+                logger.info(f"⬇️ تم التنزيل: {msg.id}")
+            else:
+                logger.warning(f"⚠️ فشل تنزيل: {msg.id}")
+        except Exception as e:
+            logger.error(f"❌ خطأ تنزيل {msg.id}: {e}")
+    
+    if not downloaded:
+        return
+    
+    # إرسال الألبوم
+    try:
+        media_list = []
+        for idx, (m_type, fpath, cap) in enumerate(downloaded):
+            item_cap = cap if idx == 0 else ""
+            
+            if m_type == "photo":
+                media_list.append(InputMediaPhoto(media=fpath, caption=item_cap))
+            elif m_type == "video":
+                media_list.append(InputMediaVideo(media=fpath, caption=item_cap, supports_streaming=True))
+            elif m_type == "doc":
+                media_list.append(InputMediaDocument(media=fpath, caption=item_cap))
+            elif m_type == "audio":
+                media_list.append(InputMediaAudio(media=fpath, caption=item_cap))
+        
+        if media_list:
+            await app.send_media_group(TARGET_CHAT, media=media_list)
+            logger.info(f"✅ تم إرسال الألبوم {mg_id}")
+    except FloodWait as e:
+        logger.warning(f"⏳ FloodWait: {e.value} ثانية")
+        await asyncio.sleep(e.value + 5)
+        await app.send_media_group(TARGET_CHAT, media=media_list)
+    except Exception as e:
+        logger.error(f"❌ فشل إرسال الألبوم: {e}")
+        # احتياطي: أرسل كل ملف لوحده
+        for m_type, fpath, cap in downloaded:
             try:
                 if m_type == "photo":
-                    await app.send_chat_action(TARGET_CHAT, ChatAction.UPLOAD_PHOTO)
-                elif m_type in ["video", "animation"]:
-                    await app.send_chat_action(TARGET_CHAT, ChatAction.UPLOAD_VIDEO)
+                    await app.send_photo(TARGET_CHAT, photo=fpath, caption=cap)
+                elif m_type == "video":
+                    await app.send_video(TARGET_CHAT, video=fpath, caption=cap, supports_streaming=True)
                 elif m_type == "doc":
-                    await app.send_chat_action(TARGET_CHAT, ChatAction.UPLOAD_DOCUMENT)
-                elif m_type in ["audio", "voice"]:
-                    await app.send_chat_action(TARGET_CHAT, ChatAction.UPLOAD_AUDIO)
-                await asyncio.sleep(random.uniform(1.0, 2.0))
+                    await app.send_document(TARGET_CHAT, document=fpath, caption=cap)
+                elif m_type == "audio":
+                    await app.send_audio(TARGET_CHAT, audio=fpath, caption=cap)
+                await asyncio.sleep(2)
+            except Exception as e2:
+                logger.error(f"❌ فشل الإرسال الفردي: {e2}")
+    finally:
+        # حذف الملفات المؤقتة
+        for _, fpath, _ in downloaded:
+            try:
+                if os.path.exists(fpath):
+                    os.remove(fpath)
             except:
                 pass
 
-            if group_id:
-                time_passed = time.time() - ts
-                if time_passed < 10:
-                    await asyncio.sleep(10 - time_passed)
-
-                async with aiosqlite.connect(DB_PATH) as db:
-                    async with db.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)
-                    ) as cursor:
-                        group_rows = await cursor.fetchall()
-
-                if group_rows:
-                    for i in range(0, len(group_rows), 10):
-                        chunk = group_rows[i:i+10]
-                        media_list, ids_to_update, files_to_delete = [], [], []
-
-                        for idx, g_item in enumerate(chunk):
-                            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:
-                            await app.send_media_group(TARGET_CHAT, media=media_list)
-                            async with aiosqlite.connect(DB_PATH) as db:
-                                for item_id in ids_to_update:
-                                    await db.execute("UPDATE queue SET status='UPLOADED' WHERE id=?", (item_id,))
-                                await db.commit()
-
-                            for f_path in files_to_delete:
-                                if os.path.exists(f_path):
-                                    os.remove(f_path)
-                            
-                            gc.collect()
-                            await asyncio.sleep(BASE_UPLOAD_DELAY + random.uniform(2.0, 4.0))
-                    continue
-
-            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)
-            elif m_type == "audio":
-                await app.send_audio(TARGET_CHAT, audio=file_path, caption=caption)
-            elif m_type == "voice":
-                await app.send_voice(TARGET_CHAT, voice=file_path, caption=caption)
-            elif m_type == "video_note":
-                await app.send_video_note(TARGET_CHAT, video_note=file_path)
-            elif m_type == "animation":
-                await app.send_animation(TARGET_CHAT, animation=file_path, caption=caption)
-
-            async with aiosqlite.connect(DB_PATH) as db:
-                await db.execute("UPDATE queue SET status='UPLOADED' WHERE id=?", (q_id,))
-                await db.commit()
-
-            if os.path.exists(file_path):
-                os.remove(file_path)
-            
-            size_factor = min(file_size / (10 * 1024 * 1024), 10)
-            safe_sleep = BASE_UPLOAD_DELAY + size_factor + random.uniform(1.0, 3.0)
-            
-            gc.collect()
-            await asyncio.sleep(safe_sleep)
-
-        except FloodWait as e:
-            print(f"⚠️ FloodWait: {e.value} ثانية")
-            BASE_UPLOAD_DELAY += 5
-            await asyncio.sleep(e.value + random.uniform(10, 20))
-        except Exception as e:
-            print(f"⚠️ خطأ: {e}")
-            async with aiosqlite.connect(DB_PATH) as db:
-                await db.execute("UPDATE queue SET status='FAILED' WHERE id=?", (q_id,))
-                await db.commit()
-            if os.path.exists(file_path):
-                os.remove(file_path)
-            await asyncio.sleep(5)
-
-# ============ التشغيل الرئيسي ============
-async def main():
-    await init_db()
-    await app.start()
+# ============ صائد الرسائل المنفردة ============
+@app.on_message(filters.media & (filters.channel | filters.group))
+async def on_media(client, message):
+    chat_id = message.chat.id
+    
+    if SOURCE_CHAT_IDS and chat_id not in SOURCE_CHAT_IDS:
+        return
+    if chat_id == TARGET_CHAT:
+        return
     
-    print("🔄 جاري الاتصال...")
+    logger.info(f"📥 جديد من {chat_id} | رسالة {message.id}")
+    
+    # إذا ألبوم — اجمعه
+    if message.media_group_id:
+        key = (chat_id, message.media_group_id)
+        async with mg_lock:
+            if key not in media_groups:
+                media_groups[key] = []
+                asyncio.create_task(flush_media_group(chat_id, message.media_group_id))
+            media_groups[key].append(message)
+        return
+    
+    # رسالة منفردة — حمل وأرسل
+    m_type, fsize = get_media_info(message)
+    if not m_type:
+        return
+    
+    src_title = message.chat.title or "المصدر"
+    raw_cap = message.caption or ""
+    final_cap = f"{raw_cap}\n\n📌 المصدر: {src_title}".strip() if raw_cap else f"📌 المصدر: {src_title}"
+    
+    file_path = None
     try:
-        me = await app.get_me()
-        print(f"✅ متصل: {me.first_name}")
-        async for _ in app.get_dialogs(limit=150):
-            pass
+        file_path = await message.download(file_name=f"{DOWNLOAD_DIR}/{chat_id}_{message.id}_")
+        if not file_path or not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
+            logger.warning(f"⚠️ فشل تنزيل الرسالة {message.id}")
+            return
+        
+        # إرسال حسب النوع
+        if m_type == "photo":
+            await app.send_photo(TARGET_CHAT, photo=file_path, caption=final_cap)
+        elif m_type == "video":
+            await app.send_video(TARGET_CHAT, video=file_path, caption=final_cap, supports_streaming=True)
+        elif m_type == "doc":
+            await app.send_document(TARGET_CHAT, document=file_path, caption=final_cap)
+        elif m_type == "audio":
+            await app.send_audio(TARGET_CHAT, audio=file_path, caption=final_cap)
+        elif m_type == "voice":
+            await app.send_voice(TARGET_CHAT, voice=file_path, caption=final_cap)
+        elif m_type == "video_note":
+            await app.send_video_note(TARGET_CHAT, video_note=file_path)
+        elif m_type == "animation":
+            await app.send_animation(TARGET_CHAT, animation=file_path, caption=final_cap)
+        
+        logger.info(f"✅ تم الإرس��ل: {message.id}")
+        await asyncio.sleep(3)  # تأخير بسيط لتجنب الحظر
+        
+    except FloodWait as e:
+        logger.warning(f"⏳ FloodWait: {e.value} ثانية")
+        await asyncio.sleep(e.value + 5)
+        # إعادة المحاولة
+        if m_type == "photo":
+            await app.send_photo(TARGET_CHAT, photo=file_path, caption=final_cap)
+        elif m_type == "video":
+            await app.send_video(TARGET_CHAT, video=file_path, caption=final_cap)
+        elif m_type == "doc":
+            await app.send_document(TARGET_CHAT, document=file_path, caption=final_cap)
+    except RPCError as e:
+        logger.error(f"❌ خطأ Telegram: {e}")
     except Exception as e:
-        print(f"⚠️ تحذير: {e}")
+        logger.error(f"❌ خطأ: {e}")
+    finally:
+        if file_path and os.path.exists(file_path):
+            try:
+                os.remove(file_path)
+            except:
+                pass
 
-    print("🚀 البوت يعمل!")
-    
-    for i in range(2):
-        asyncio.create_task(download_worker_task(i + 1))
-        
-    asyncio.create_task(upload_worker())
+# ============ التشغيل ============
+async def startup():
+    await app.start()
+    me = await app.get_me()
+    logger.info("=" * 50)
+    logger.info(f"🚀 البوت يعمل!")
+    logger.info(f"👤 الحساب: {me.first_name} (@{me.username})")
+    logger.info(f"🎯 القناة الهدف: {TARGET_CHAT}")
+    logger.info(f"📡 القنوات المصدر: {SOURCE_CHAT_IDS or 'كل القنوات'}")
+    logger.info("=" * 50)
     
-    await idle()
-    await app.stop()
-
-def run_bot():
-    """تشغيل البوت في thread منفصلة"""
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
     while True:
-        try:
-            loop.run_until_complete(main())
-        except Exception as e:
-            print(f"💥 إعادة تشغيل: {e}")
-            time.sleep(10)
-
-# ============ Gradio Interface ============
-bot_status = {"running": False, "logs": "⏳ جاري التشغيل..."}
-
-def start_bot():
-    if not bot_status["running"]:
-        bot_status["running"] = True
-        bot_status["logs"] = "🟢 البوت يعمل في الخلفية"
-        thread = threading.Thread(target=run_bot, daemon=True)
-        thread.start()
-        return "✅ تم تشغيل البوت!"
-    return "⚠️ البوت يعمل بالفعل!"
-
-# تشغيل البوت تلقائياً
-start_bot()
+        await asyncio.sleep(3600)
 
+# ============ واجهة Gradio ============
 with gr.Blocks(title="Telegram Archiver") as demo:
-    gr.Markdown("# 📡 Telegram Archiver Bot")
-    gr.Markdown("البوت يعمل في الخلفية ويقوم بأرشفة الملفات تلقائياً")
+    gr.Markdown("# 📡 Telegram Archiver")
+    gr.Markdown("🟢 **الحالة:** البوت يعمل في الخلفية")
     
-    status_box = gr.Textbox(label="الحالة", value="🟢 البوت يعمل", interactive=False)
-    gr.Markdown("### المتغيرات المطلوبة في Settings:")
-    gr.Markdown("- `API_ID`\n- `API_HASH`\n- `SESSION_STRING`\n- `TARGET_CHAT`")
-
-demo.launch(server_name="0.0.0.0", server_port=7860)
+    gr.Textbox(
+        label="الحالة",
+        value="✅ نشط - افتح تبويب Logs لرؤية النشاط",
+        interactive=False
+    )
+    
+    gr.Markdown("""
+    ### 🔧 المتغيرات المطلوبة:
+    
+    | المتغير | الوصف |
+    |---------|-------|
+    | `API_ID` | من my.telegram.org |
+    | `API_HASH` | من my.telegram.org |
+    | `SESSION_STRING` | سلسلة الجلسة |
+    | `TARGET_CHAT` | معرف قناتك الخاصة |
+    | `SOURCE_CHATS` | (اختياري) قنوات المصدر |
+
+    ### ⚠️ ملاحظة مهمة:
+    الحساب لازم يكون **مشترك** في القناة المصدرية حتى يقدر يحمل منها.
+    """)
+
+if __name__ == "__main__":
+    loop = asyncio.get_event_loop()
+    loop.create_task(startup())
+    demo.launch(server_name="0.0.0.0", server_port=7860)