Update bot.py

bot.py

@@ -3,6 +3,7 @@ import asyncio
 import logging
 import time
 import threading
+import aiosqlite
 import gradio as gr
 from collections import defaultdict
 from pyrogram import Client, filters
@@ -31,8 +32,10 @@ API_HASH = os.environ.get("API_HASH", "")
 SESSION_STRING = os.environ.get("SESSION_STRING", "")
 TARGET_CHAT_RAW = os.environ.get("TARGET_CHAT", "me")
 
-DATA_DIR = "./data"
+# 🔑 استخدام /data للـ persistent storage (1TB)
+DATA_DIR = "/data"
 DOWNLOAD_DIR = os.path.join(DATA_DIR, "downloads")
+DB_PATH = os.path.join(DATA_DIR, "queue.db")
 os.makedirs(DOWNLOAD_DIR, exist_ok=True)
 
 # ============ العميل ============
@@ -51,20 +54,200 @@ app = Client(
 
 TARGET_CHAT = None
 
+# ============ Throttle Controller (منع FloodWait) ============
+class ThrottleController:
+    """تنظيم الإرسال: 1 رسالة/ثانية لكل دردشة"""
+    def __init__(self, rate=1.0):
+        self.rate = rate
+        self.last_send = defaultdict(float)
+        self.lock = asyncio.Lock()
+    
+    async def wait(self, chat_id):
+        async with self.lock:
+            now = time.time()
+            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 ثانية
+
 # ============ Semaphores ============
-# Semaphore منفصل للألبومات والرسائل المنفردة لعدم التأثير على بعض
-album_sem = asyncio.Semaphore(1)
-single_sem = asyncio.Semaphore(2)
+download_sem = asyncio.Semaphore(3)   # 3 تحميلات متوازية كحد أقصى
+send_sem = asyncio.Semaphore(2)       # 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 (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                chat_id INTEGER NOT NULL,
+                message_id INTEGER NOT NULL,
+                media_group_id TEXT,
+                media_type TEXT,
+                file_id TEXT,
+                file_name TEXT,
+                file_size INTEGER DEFAULT 0,
+                file_path TEXT,
+                caption TEXT,
+                source_name TEXT,
+                status TEXT DEFAULT 'pending',
+                attempts INTEGER DEFAULT 0,
+                error TEXT,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                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)
+        """)
+        await db.execute("""
+            CREATE INDEX IF NOT EXISTS idx_jobs_media_group ON jobs(media_group_id)
+        """)
+        await db.commit()
+    logger.info("✅ قاعدة البيانات جاهزة")
+
+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
+    
+    if message.photo:
+        file_id = message.photo.file_id
+        file_size = message.photo.file_size or 0
+    elif message.video:
+        file_id = message.video.file_id
+        file_size = message.video.file_size or 0
+        file_name = message.video.file_name
+    elif message.document:
+        file_id = message.document.file_id
+        file_size = message.document.file_size or 0
+        file_name = message.document.file_name
+    elif message.audio:
+        file_id = message.audio.file_id
+        file_size = message.audio.file_size or 0
+        file_name = message.audio.file_name
+    elif message.voice:
+        file_id = message.voice.file_id
+        file_size = message.voice.file_size or 0
+    elif message.video_note:
+        file_id = message.video_note.file_id
+        file_size = message.video_note.file_size or 0
+    elif message.animation:
+        file_id = message.animation.file_id
+        file_size = message.animation.file_size or 0
+    
+    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}"
+    
+    async with aiosqlite.connect(DB_PATH) as db:
+        try:
+            await db.execute("""
+                INSERT OR IGNORE INTO jobs 
+                (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, message.media_group_id, 
+                m_type, file_id, file_name, file_size, final_cap, src_name
+            ))
+            await db.commit()
+        except Exception as e:
+            logger.error(f"❌ خطأ في إضافة المهمة: {e}")
 
-# ============ نظام الألبومات ============
-albums_storage = defaultdict(lambda: defaultdict(list))
-albums_lock = asyncio.Lock()
-albums_timers = {}
+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("""
+                UPDATE jobs SET status=?, file_path=?, updated_at=CURRENT_TIMESTAMP 
+                WHERE chat_id=? AND message_id=?
+            """, (status, 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))
+        else:
+            await db.execute("""
+                UPDATE jobs SET status=?, updated_at=CURRENT_TIMESTAMP 
+                WHERE chat_id=? AND message_id=?
+            """, (status, chat_id, message_id))
+        await db.commit()
+
+async def get_pending_jobs(limit=5):
+    """جلب المهام المعلقة"""
+    async with aiosqlite.connect(DB_PATH) as db:
+        db.row_factory = aiosqlite.Row
+        cursor = await db.execute("""
+            SELECT * FROM jobs 
+            WHERE status='pending' AND attempts < 10
+            ORDER BY created_at 
+            LIMIT ?
+        """, (limit,))
+        rows = await cursor.fetchall()
+        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("""
+            SELECT * FROM jobs 
+            WHERE media_group_id=? AND status='downloaded'
+            ORDER BY message_id
+        """, (media_group_id,))
+        rows = await cursor.fetchall()
+        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 
+            WHERE media_group_id=? AND status IN ('pending', 'downloading')
+        """, (media_group_id,))
+        row = await cursor.fetchone()
+        return row[0]
 
-# ============ Deduplication (منع التكرار) ============
-sent_messages = set()  # تتبع (chat_id, message_id) المرسلة
-dedup_lock = asyncio.Lock()
+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=?
+        """, (chat_id, message_id))
+        row = await cursor.fetchone()
+        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))
+        await db.commit()
 
+# ============ معلومات الملف ============
 def get_media_info(message):
     if message.photo:
         return "photo", ".jpg"
@@ -92,38 +275,75 @@ def get_source_name(message):
         return chat.first_name or chat.username or "خاص"
     return "مصدر"
 
-def create_input_media(message, caption=None):
-    """إنشاء InputMedia للألبوم — يدعم جميع الأنواع المسموح بها في send_media_group"""
-    cap = caption or message.caption or ""
+def get_file_id_from_job(job):
+    """استخراج file_id من الرسالة الأصلية"""
+    # نستخدم app.download_media مباشرة بالـ message_id
+    return job['file_id']
+
+# ============ التحميل مع التحقق ============
+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} بايت")
+        return False
+    if actual_size == 0:
+        logger.warning(f"⚠️ ملف بحجم صفر: {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
     
-    if message.photo:
-        return InputMediaPhoto(
-            media=message.photo.file_id,
-            caption=cap,
-            caption_entities=message.caption_entities
-        )
-    elif message.video:
-        return InputMediaVideo(
-            media=message.video.file_id,
-            caption=cap,
-            caption_entities=message.caption_entities,
-            supports_streaming=message.video.supports_streaming
-        )
-    elif message.document:
-        return InputMediaDocument(
-            media=message.document.file_id,
-            caption=cap,
-            caption_entities=message.caption_entities
-        )
-    elif message.audio:
-        return InputMediaAudio(
-            media=message.audio.file_id,
-            caption=cap,
-            caption_entities=message.caption_entities
+    ext = {
+        'photo': '.jpg', 'video': '.mp4', 'doc': '',
+        'audio': '.mp3', 'voice': '.ogg', 
+        'video_note': '.mp4', 'animation': '.mp4'
+    }.get(job['media_type'], '')
+    
+    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
         )
-    return None
+        
+        if not downloaded:
+            raise Exception("download_media returned None")
+        
+        # 🔑 التحقق من اكتمال الملف
+        if not await verify_download(downloaded, expected_size):
+            # إعادة المحاولة مرة واحدة
+            logger.info(f"🔄 إعادة تحميل {message_id}")
+            if os.path.exists(downloaded):
+                os.remove(downloaded)
+            downloaded = await app.download_media(
+                job['file_id'],
+                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}")
+        
+        await update_job_status(chat_id, message_id, 'downloaded', file_path=downloaded)
+        logger.info(f"✅ تحميل مكتمل: {message_id} | {os.path.getsize(downloaded)} بايت")
+        return True
+        
+    except Exception as e:
+        await update_job_status(chat_id, message_id, 'pending', error=str(e))
+        logger.error(f"❌ فشل تحميل {message_id}: {e}")
+        return False
 
-# ============ دالة إرسال متقدمة مع Tenacity ============
+# ============ الإرسال مع Tenacity ============
 @retry(
     stop=stop_after_attempt(5),
     wait=wait_exponential_jitter(initial=1, max=60),
@@ -131,338 +351,252 @@ def create_input_media(message, caption=None):
     reraise=True
 )
 async def safe_send(send_func):
-    """إرسال مع exponential backoff + jitter + 5 محاولات"""
     return await send_func()
 
-async def send_with_retry(send_func, max_retries=5):
-    """نسخة يدوية إذا أردت التحكم — لكن safe_send أفضل"""
-    last_error = None
-    for attempt in range(max_retries):
-        try:
-            return await send_func()
-        except FloodWait as e:
-            wait = min(e.value, 60)  # كاب 60 ثانية
-            logger.warning(f"⏳ FloodWait: انتظار {wait} ثانية (محاولة {attempt+1}/{max_retries})")
-            await asyncio.sleep(wait)
-            last_error = e
-        except (ConnectionError, TimeoutError) as e:
-            wait = min(2 ** attempt, 60)
-            logger.warning(f"🌐 Connection error: انتظار {wait} ثانية (محاولة {attempt+1}/{max_retries})")
-            await asyncio.sleep(wait)
-            last_error = e
-        except Exception as e:
-            logger.error(f"❌ خطأ غير متوقع: {e}")
-            raise
-    raise last_error or Exception("Failed after retries")
-
-# ============ إرسال الألبوم ============
-async def send_album(chat_id, media_group_id):
-    """إرسال الألبوم مع الحفاظ على الترتيب وعدم فقدان أي ملف"""
-    await asyncio.sleep(2.5)  # انتظار أطول قليلاً لجمع كل الرسائل
+async def send_job(job):
+    """إرسال مهمة محملة"""
+    chat_id = job['chat_id']
+    message_id = job['message_id']
+    file_path = job['file_path']
     
-    async with albums_lock:
-        key = (chat_id, media_group_id)
-        if key not in albums_timers:
-            return
-        albums_timers.pop(key, None)
-        messages = albums_storage[chat_id].pop(media_group_id, [])
+    if not file_path or not os.path.exists(file_path):
+        logger.error(f"❌ ملف غير موجود: {file_path}")
+        await update_job_status(chat_id, message_id, 'pending', error="File not found")
+        return False
     
-    if not messages:
+    try:
+        await throttle.wait(TARGET_CHAT)
+        
+        m_type = job['media_type']
+        caption = job['caption'] or ""
+        
+        if m_type == 'photo':
+            await safe_send(lambda: app.send_photo(TARGET_CHAT, photo=file_path, caption=caption))
+        elif m_type == 'video':
+            await safe_send(lambda: app.send_video(TARGET_CHAT, video=file_path, caption=caption, supports_streaming=True))
+        elif m_type == 'doc':
+            await safe_send(lambda: app.send_document(TARGET_CHAT, document=file_path, caption=caption))
+        elif m_type == 'audio':
+            await safe_send(lambda: app.send_audio(TARGET_CHAT, audio=file_path, caption=caption))
+        elif m_type == 'voice':
+            await safe_send(lambda: app.send_voice(TARGET_CHAT, voice=file_path, caption=caption))
+        elif m_type == 'video_note':
+            await safe_send(lambda: app.send_video_note(TARGET_CHAT, video_note=file_path))
+        elif m_type == 'animation':
+            await safe_send(lambda: app.send_animation(TARGET_CHAT, animation=file_path, caption=caption))
+        
+        logger.info(f"✅ إرسال ناجح: {message_id}")
+        await delete_job(chat_id, message_id)
+        return True
+        
+    except Exception as e:
+        logger.error(f"❌ فشل إرسال {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
     
-    messages.sort(key=lambda m: m.id)
-    total = len(messages)
-    logger.info(f"📦 ألبوم {media_group_id} | {total} رسائل | ترتيب: {[m.id for m in messages]}")
+    # التحقق من عدم وجود ملفات معلقة
+    pending = await get_album_pending_count(media_group_id)
+    if pending > 0:
+        logger.info(f"⏳ ألبوم {media_group_id} ينتظر {pending} ملفات")
+        return
     
-    src_name = get_source_name(messages[0])
+    # ترتيب الملفات
+    jobs.sort(key=lambda j: j['message_id'])
     
-    # فصل الرسائل: المدعومة في الألبوم vs المنفردة
-    album_media = []
-    album_messages = []
-    standalone_msgs = []
+    # فصل المدعوم في الألبوم عن المنفرد
+    album_inputs = []
+    album_jobs = []
+    standalone_jobs = []
     
-    for i, msg in enumerate(messages):
+    for i, job in enumerate(jobs):
         if i == 0:
-            raw_cap = msg.caption or ""
-            final_cap = f"{raw_cap}\n\n📌 المصدر: {src_name}".strip() if raw_cap else f"📌 المصدر: {src_name}"
+            cap = job['caption']
         else:
-            final_cap = None
+            cap = None
         
-        input_media = create_input_media(msg, final_cap)
-        if input_media:
-            album_media.append(input_media)
-            album_messages.append(msg)
+        m_type = job['media_type']
+        file_path = job['file_path']
+        
+        if m_type == 'photo':
+            album_inputs.append(InputMediaPhoto(media=file_path, caption=cap))
+            album_jobs.append(job)
+        elif m_type == 'video':
+            album_inputs.append(InputMediaVideo(media=file_path, caption=cap, supports_streaming=True))
+            album_jobs.append(job)
+        elif m_type == 'doc':
+            album_inputs.append(InputMediaDocument(media=file_path, caption=cap))
+            album_jobs.append(job)
+        elif m_type == 'audio':
+            album_inputs.append(InputMediaAudio(media=file_path, caption=cap))
+            album_jobs.append(job)
         else:
-            standalone_msgs.append(msg)
-            logger.info(f"📎 رسالة {msg.id} من نوع {get_media_info(msg)[0]} ستُرسل منفردة")
+            standalone_jobs.append(job)
     
-    # 🔑 إرسال الألبوم الرئيسي (المدعوم) — تقسيم إذا >10
-    if album_media:
-        async with album_sem:
-            # تقسيم الألبوم إلى مجموعات من 10 (حد Telegram)
-            chunk_size = 10
-            for chunk_idx in range(0, len(album_media), chunk_size):
-                chunk_media = album_media[chunk_idx:chunk_idx + chunk_size]
-                chunk_msgs = album_messages[chunk_idx:chunk_idx + chunk_size]
-                
-                try:
-                    if len(chunk_media) == 1:
-                        await safe_send(lambda: chunk_msgs[0].copy(chat_id=TARGET_CHAT))
-                        logger.info(f"✅ ألبوم {media_group_id} | جزء {chunk_idx//chunk_size + 1} | 1 ملف تم النسخ")
-                    else:
-                        await safe_send(
-                            lambda: app.send_media_group(
-                                chat_id=TARGET_CHAT,
-                                media=chunk_media,
-                                disable_notification=True
-                            )
-                        )
-                        logger.info(f"✅ ألبوم {media_group_id} | جزء {chunk_idx//chunk_size + 1} | {len(chunk_media)} ملفات (file_id)")
-                except Exception as e:
-                    logger.warning(f"⚠️ فشل النسخ السريع للجزء {chunk_idx//chunk_size + 1}: {e}")
-                    await download_and_send_album(chunk_msgs, src_name)
-    
-    # 🔑 إرسال الرسائل المنفردة (voice, video_note, animation)
-    for msg in standalone_msgs:
-        async with single_sem:
+    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]
+            chunk_jobs = album_jobs[idx:idx+chunk_size]
+            
             try:
-                await safe_send(lambda: msg.copy(chat_id=TARGET_CHAT))
-                logger.info(f"✅ رسالة منفردة {msg.id} تم النسخ")
+                if len(chunk) == 1:
+                    await safe_send(lambda: app.send_photo(
+                        TARGET_CHAT, photo=chunk[0].media, caption=chunk[0].caption
+                    ))
+                else:
+                    await safe_send(lambda: app.send_media_group(
+                        TARGET_CHAT, media=chunk, disable_notification=True
+                    ))
+                
+                for j in chunk_jobs:
+                    await delete_job(j['chat_id'], j['message_id'])
+                logger.info(f"✅ ألبوم {media_group_id} | جزء {idx//chunk_size + 1} تم الإرسال")
             except Exception as e:
-                logger.warning(f"⚠️ فشل نسخ منفرد {msg.id}: {e}")
-                await download_and_send_single(msg)
+                logger.error(f"❌ فشل إرسال جزء من الألبوم: {e}")
+        
+        # إرسال المنفردات
+        for job in standalone_jobs:
+            await send_job(job)
 
-async def download_and_send_album(messages, src_name=None):
-    """تحميل وإرسال الألبوم — مع FloodWait handling"""
-    if not messages:
-        return
-    
-    messages.sort(key=lambda m: m.id)
-    if src_name is None:
-        src_name = get_source_name(messages[0])
-    
-    downloaded_files = []
-    media_inputs = []
-    msg_map = []
-    
-    try:
-        for i, msg in enumerate(messages):
-            m_type, ext = get_media_info(msg)
-            if not m_type:
+# ============ Workers ============
+async def download_worker():
+    """معالج التحميل المستمر"""
+    logger.info("🔄 Download Worker started")
+    while True:
+        try:
+            jobs = await get_pending_jobs(limit=5)
+            if not jobs:
+                await asyncio.sleep(2)
                 continue
             
-            file_path = await app.download_media(
-                msg,
-                file_name=f"{DOWNLOAD_DIR}/album_{msg.chat.id}_{msg.media_group_id}_{msg.id}{ext}"
-            )
+            tasks = []
+            for job in jobs:
+                if len(tasks) >= 3:
+                    break
+                tasks.append(download_job(job))
             
-            if file_path and os.path.exists(file_path) and os.path.getsize(file_path) > 0:
-                downloaded_files.append(file_path)
-                msg_map.append(msg)
-                
-                if i == 0:
-                    raw_cap = msg.caption or ""
-                    final_cap = f"{raw_cap}\n\n📌 المصدر: {src_name}".strip() if raw_cap else f"📌 المصدر: {src_name}"
-                else:
-                    final_cap = None
-                
-                if m_type == "photo":
-                    media_inputs.append(InputMediaPhoto(media=file_path, caption=final_cap))
-                elif m_type == "video":
-                    media_inputs.append(InputMediaVideo(media=file_path, caption=final_cap, supports_streaming=True))
-                elif m_type == "doc":
-                    media_inputs.append(InputMediaDocument(media=file_path, caption=final_cap))
-                elif m_type == "audio":
-                    media_inputs.append(InputMediaAudio(media=file_path, caption=final_cap))
-        
-        # تقسيم إلى مجموعات من 10
-        chunk_size = 10
-        for chunk_idx in range(0, len(media_inputs), chunk_size):
-            chunk_inputs = media_inputs[chunk_idx:chunk_idx + chunk_size]
+            if tasks:
+                await asyncio.gather(*tasks, return_exceptions=True)
             
-            if len(chunk_inputs) >= 2:
-                await safe_send(
-                    lambda ci=chunk_inputs: app.send_media_group(
-                        chat_id=TARGET_CHAT,
-                        media=ci,
-                        disable_notification=True
-                    )
-                )
-                logger.info(f"✅ ألبوم (تحميل) | جزء {chunk_idx//chunk_size + 1} | {len(chunk_inputs)} ملفات")
-            elif len(chunk_inputs) == 1:
-                msg = msg_map[chunk_idx]
-                m_type, _ = get_media_info(msg)
-                file_path = downloaded_files[chunk_idx]
-                raw_cap = msg.caption or ""
-                final_cap = f"{raw_cap}\n\n📌 المصدر: {src_name}".strip() if raw_cap else f"📌 المصدر: {src_name}"
-                
-                if m_type == "photo":
-                    await safe_send(lambda: app.send_photo(TARGET_CHAT, photo=file_path, caption=final_cap))
-                elif m_type == "video":
-                    await safe_send(lambda: app.send_video(TARGET_CHAT, video=file_path, caption=final_cap, supports_streaming=True))
-                elif m_type == "doc":
-                    await safe_send(lambda: app.send_document(TARGET_CHAT, document=file_path, caption=final_cap))
-                elif m_type == "audio":
-                    await safe_send(lambda: app.send_audio(TARGET_CHAT, audio=file_path, caption=final_cap))
-                elif m_type == "voice":
-                    await safe_send(lambda: app.send_voice(TARGET_CHAT, voice=file_path, caption=final_cap))
-                elif m_type == "video_note":
-                    await safe_send(lambda: app.send_video_note(TARGET_CHAT, video_note=file_path))
-                elif m_type == "animation":
-                    await safe_send(lambda: app.send_animation(TARGET_CHAT, animation=file_path, caption=final_cap))
-                
-    except Exception as e:
-        logger.error(f"❌ فشل إرسال الألبوم: {e}")
-    finally:
-        for f in downloaded_files:
-            try:
-                if os.path.exists(f):
-                    os.remove(f)
-            except:
-                pass
+        except Exception as e:
+            logger.error(f"❌ خطأ في Download Worker: {e}")
+            await asyncio.sleep(5)
 
-async def download_and_send_single(message):
-    """تحميل وإرسال رسالة واحدة — مع FloodWait handling"""
-    m_type, ext = get_media_info(message)
-    if not m_type:
-        return
-    
-    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}"
-    file_path = None
-    
+async def send_worker():
+    """معالج الإرسال المستمر"""
+    logger.info("📤 Send Worker started")
+    while True:
+        try:
+            # جلب المهام المحملة (غير ألبومات)
+            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
+                """)
+                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()
+            
+            # إرسال المنفردات
+            for job in single_jobs:
+                async with send_sem:
+                    await send_job(job)
+            
+            # معالجة الألبومات
+            for (mg_id,) in album_rows:
+                await send_album_group(mg_id)
+            
+            await asyncio.sleep(1)
+            
+        except Exception as e:
+            logger.error(f"❌ خطأ في Send Worker: {e}")
+            await asyncio.sleep(5)
+
+# ============ استخدام get_media_group للألبومات ============
+async def fetch_full_album(message):
+    """جلب كل رسائل الألبوم مرة واحدة باستخدام get_media_group"""
     try:
-        file_path = await app.download_media(
-            message,
-            file_name=f"{DOWNLOAD_DIR}/{message.chat.id}_{message.id}{ext}"
-        )
-        if not file_path or not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
-            logger.error(f"❌ فشل تحميل الرسالة {message.id}")
-            return
-        
-        if m_type == "photo":
-            await safe_send(lambda: app.send_photo(TARGET_CHAT, photo=file_path, caption=final_cap))
-        elif m_type == "video":
-            await safe_send(lambda: app.send_video(TARGET_CHAT, video=file_path, caption=final_cap, supports_streaming=True))
-        elif m_type == "doc":
-            await safe_send(lambda: app.send_document(TARGET_CHAT, document=file_path, caption=final_cap))
-        elif m_type == "audio":
-            await safe_send(lambda: app.send_audio(TARGET_CHAT, audio=file_path, caption=final_cap))
-        elif m_type == "voice":
-            await safe_send(lambda: app.send_voice(TARGET_CHAT, voice=file_path, caption=final_cap))
-        elif m_type == "video_note":
-            await safe_send(lambda: app.send_video_note(TARGET_CHAT, video_note=file_path))
-        elif m_type == "animation":
-            await safe_send(lambda: app.send_animation(TARGET_CHAT, animation=file_path, caption=final_cap))
-        
-        logger.info(f"✅ تم الإرسال (تحميل): {message.id}")
+        messages = await app.get_media_group(message.chat.id, message.id)
+        logger.info(f"📦 get_media_group: {len(messages)} رسائل")
+        for msg in messages:
+            await add_job(msg)
+        return True
     except Exception as e:
-        logger.error(f"❌ فشل الإرسال النهائي {message.id}: {e}")
-    finally:
-        if file_path and os.path.exists(file_path):
-            try:
-                os.remove(file_path)
-            except:
-                pass
+        logger.warning(f"⚠️ فشل get_media_group: {e}")
+        return False
 
+# ============ المعالج الرئيسي ============
 @app.on_message(filters.incoming & ~filters.me & filters.media)
 async def on_media(client, message):
     chat_id = message.chat.id
-    msg_id = message.id
-    
-    # Deduplication
-    dedup_key = (chat_id, msg_id)
-    async with dedup_lock:
-        if dedup_key in sent_messages:
-            logger.debug(f"⏭️ تخطي رسالة مكررة: {msg_id}")
-            return
-        sent_messages.add(dedup_key)
-    
     if str(chat_id) == str(TARGET_CHAT):
         return
     
     src_name = get_source_name(message)
     m_type, _ = get_media_info(message)
-    logger.info(f"📥 جديد من {chat_id} | {src_name} | رسالة {msg_id} | نوع: {m_type} | ألبوم: {message.media_group_id is not None}")
+    logger.info(f"📥 جديد: {chat_id}/{message.id} | {m_type} | ألبوم: {message.media_group_id is not None}")
     
-    # إذا ألبوم
     if message.media_group_id:
-        mg_id = message.media_group_id
-        async with albums_lock:
-            albums_storage[chat_id][mg_id].append(message)
-            key = (chat_id, mg_id)
-            if key in albums_timers:
-                albums_timers[key].cancel()
-            
-            async def delayed_send():
-                await asyncio.sleep(2.5)
-                await send_album(chat_id, mg_id)
-            
-            timer_task = asyncio.create_task(delayed_send())
-            albums_timers[key] = timer_task
-        return
-    
-    # رسالة منفردة
-    async with single_sem:
-        try:
-            await safe_send(lambda: message.copy(chat_id=TARGET_CHAT))
-            logger.info(f"✅ تم النسخ: {msg_id}")
-            return
-        except Exception as e:
-            logger.warning(f"⚠️ فشل النسخ {msg_id}: {e}")
-        
-        # Fallback: تحميل وإرسال
-        await download_and_send_single(message)
+        # 🔑 محاولة جلب الألبوم كاملاً
+        success = await fetch_full_album(message)
+        if not success:
+            # Fallback: إضافة الرسالة الحالية فقط
+            await add_job(message)
+    else:
+        await add_job(message)
 
 @app.on_message(filters.incoming & ~filters.me)
 async def monitor_all(client, message):
     if str(message.chat.id) == str(TARGET_CHAT):
         return
-    src_name = get_source_name(message)
-    logger.info(f"👁️ رسالة من {message.chat.id} | نوع: {message.chat.type} | اسم: {src_name}")
+    logger.info(f"👁️ رسالة نصية من {message.chat.id}")
 
 async def resolve_target_chat():
     global TARGET_CHAT
     raw = TARGET_CHAT_RAW
     if raw.lower() == "me":
         TARGET_CHAT = "me"
-        logger.info("🎯 الهدف: Saved Messages (me)")
         return
     try:
         chat_id_int = int(raw)
         chat = await app.get_chat(chat_id_int)
         TARGET_CHAT = chat_id_int
-        logger.info(f"🎯 تم التحقق من القناة: {chat.title} | id: {chat_id_int}")
+        logger.info(f"🎯 الهدف: {chat.title}")
     except Exception as e:
-        logger.error(f"❌ فشل التحقق من القناة {raw}: {e}")
+        logger.error(f"❌ فشل التحقق: {e}")
         TARGET_CHAT = "me"
 
-async def self_ping():
-    while True:
-        try:
-            await asyncio.sleep(600)
-            await app.invoke(pyrogram.raw.functions.Ping(ping_id=0))
-            logger.debug("🏓 Self-ping")
-        except Exception as e:
-            logger.debug(f"Self-ping error: {e}")
-
 # ============ Gradio + FastAPI ============
 def run_gradio():
-    with gr.Blocks(title="Telegram Archiver") as demo:
-        gr.Markdown("# 📡 Telegram Archiver")
-        gr.Markdown("🟢 **الحالة:** البوت يعمل في الخلفية")
-        gr.Textbox(label="الحالة", value="✅ نشط", interactive=False)
+    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)
     
     fastapi_app = FastAPI()
     
     @fastapi_app.get("/health")
-    @fastapi_app.head("/health")
     async def health_check():
         return {"status": "ok", "timestamp": time.time()}
     
     @fastapi_app.get("/ping")
-    @fastapi_app.head("/ping")
     async def ping():
         return {"status": "alive"}
     
@@ -471,17 +605,24 @@ def run_gradio():
     import uvicorn
     uvicorn.run(fastapi_app, host="0.0.0.0", port=7860)
 
+# ============ التشغيل الرئيسي ============
 async def main():
     await app.start()
     me = await app.get_me()
     await resolve_target_chat()
-    asyncio.create_task(self_ping())
+    await init_db()
+    
+    # تشغيل Workers
+    asyncio.create_task(download_worker())
+    asyncio.create_task(send_worker())
     
-    logger.info("=" * 50)
-    logger.info(f"🚀 البوت يعمل!")
-    logger.info(f"👤 الحساب: {me.first_name} (@{me.username})")
-    logger.info(f"🎯 القناة الهدف: {TARGET_CHAT}")
-    logger.info("=" * 50)
+    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("=" * 60)
     
     stop_event = asyncio.Event()
     await stop_event.wait()