Update bot.py

bot.py

@@ -6,105 +6,87 @@ import threading
 import aiosqlite
 import gradio as gr
 from hydrogram import Client, filters
-from hydrogram.errors import FloodWait
+from hydrogram.errors import FloodWait, FloodPremiumWait, ChatForwardsRestricted, MessageIdInvalid
 from hydrogram.types import (
     InputMediaPhoto, InputMediaVideo, InputMediaDocument, 
     InputMediaAudio
 )
 from fastapi import FastAPI
+import uvicorn
 
-# ============ CRITICAL FIX: Monkey-patch FloodPremiumWait ============
-import hydrogram
-import hydrogram.errors.rpc_error as _rpc_error
-
-if "FLOOD_PREMIUM_WAIT" not in [k for d in _rpc_error.exceptions.values() for k in d.keys()]:
-    class FloodPremiumWait(FloodWait):
-        CODE = 420
-        NAME = "FLOOD_PREMIUM_WAIT"
-    setattr(hydrogram.errors, "FloodPremiumWait", FloodPremiumWait)
-    _rpc_error.exceptions.setdefault(420, {})["FLOOD_PREMIUM_WAIT"] = "FloodPremiumWait"
-    logging.getLogger(__name__).warning("Patched missing FloodPremiumWait into Hydrogram")
-# ===================================================================
-
+# ============ إعداد السجلات (Logging) ============
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(levelname)s - %(message)s'
 )
-logger = logging.getLogger(__name__)
+logger = logging.getLogger("ArchiverPro")
 
-# ============ الإعدادات ============
+# ============ الإعدادات والمتغيرات ============
 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_RAW = os.environ.get("TARGET_CHAT", "me")
+# يمكنك وضع قنوات مفصولة بفواصل مثل: "@channel1, -100123456789, @channel2"
+CHANNELS_TO_SCRAPE = os.environ.get("CHANNELS_TO_SCRAPE", "")
 
-DATA_DIR = "/data"
+DATA_DIR = "/tmp/data" if not os.path.exists("/data") else "/data"
 DOWNLOAD_DIR = os.path.join(DATA_DIR, "downloads")
-DB_PATH = os.path.join(DATA_DIR, "queue.db")
+DB_PATH = os.path.join(DATA_DIR, "queue_v2.db")
 os.makedirs(DOWNLOAD_DIR, exist_ok=True)
 
-# ============ العميل (Hydrogram) ============
+# ============ إنشاء العميل (Hydrogram Client) ============
 app = Client(
-    "archiver",
+    "archiver_pro",
     api_id=API_ID,
     api_hash=API_HASH,
     session_string=SESSION_STRING,
     in_memory=True,
-    workers=2,
-    device_model="Samsung Galaxy S23 Ultra",
-    system_version="Android 14",
-    app_version="10.14.1",
-    lang_code="ar",
-    takeout=False,
+    workers=4,
+    device_model="Desktop Archiver",
+    system_version="Linux Archiver 64bit",
+    app_version="2.0.0",
+    lang_code="ar"
 )
 
-TARGET_CHAT = None
+TARGET_CHAT = "me"
+stats = {
+    "total_queued": 0,
+    "total_transferred": 0,
+    "total_failed": 0,
+    "status": "Starting..."
+}
+
+# ============ إدارة الـ FloodWait والتحكم بالسرعة ============
+class RateLimiter:
+    def __init__(self, delay=2.5):
+        self.delay = delay
+        self.last_call = 0
+        self.lock = asyncio.Lock()
+        self.flood_until = 0
 
-# ============ Global Flood Control ============
-class GlobalFloodControl:
-    def __init__(self):
-        self.until = 0
-        self._lock = asyncio.Lock()
-    
-    async def hit(self, seconds):
-        async with self._lock:
-            target = time.time() + seconds
-            if target > self.until:
-                self.until = target
-                logger.warning(f"Global flood control activated: {seconds}s")
-    
     async def wait(self):
-        async with self._lock:
-            remaining = self.until - time.time()
-        if remaining > 0:
-            logger.info(f"Global flood pause: {remaining:.0f}s remaining")
-            await asyncio.sleep(remaining)
-
-flood_control = GlobalFloodControl()
-
-# ============ Throttle Controller ============
-class ThrottleController:
-    def __init__(self, rate=3.0):
-        self.rate = rate
-        self.last_send = {}
-        self.lock = asyncio.Lock()
-    
-    async def wait(self, chat_id):
         async with self.lock:
+            # التحقق من وجود حظر FloodWait
             now = time.time()
-            last = self.last_send.get(chat_id, 0)
-            elapsed = now - last
-            if elapsed < self.rate:
-                await asyncio.sleep(self.rate - elapsed)
-            self.last_send[chat_id] = time.time()
+            if self.flood_until > now:
+                wait_sec = self.flood_until - now
+                logger.warning(f"⏳ FloodWait فعال، انتظار {wait_sec:.1f} ثانية...")
+                await asyncio.sleep(wait_sec)
+            
+            # التحقق من الفاصل الزمني العادي لتفادي الحظر
+            now = time.time()
+            elapsed = now - self.last_call
+            if elapsed < self.delay:
+                await asyncio.sleep(self.delay - elapsed)
+            self.last_call = time.time()
 
-throttle = ThrottleController(rate=3.0)
+    def set_flood(self, seconds):
+        self.flood_until = time.time() + seconds + 2
+        logger.warning(f"🚨 تم استلام FloodWait: {seconds}s")
 
-# ============ Locks ============
-download_lock = asyncio.Lock()
-send_sem = asyncio.Semaphore(1)
+rate_limiter = RateLimiter(delay=2.0)
 
-# ============ قاعدة البيانات ============
+# ============ قاعدة البيانات وحفظ الطابور ============
 async def init_db():
     async with aiosqlite.connect(DB_PATH) as db:
         await db.execute("""
@@ -114,606 +96,377 @@ async def init_db():
                 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',
+                album_total INTEGER DEFAULT 1,
+                album_index INTEGER DEFAULT 1,
+                status TEXT DEFAULT 'pending', -- pending, processing, completed, failed
                 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 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("Database ready")
+    logger.info("✅ تم تهيئة قاعدة البيانات بنجاح.")
 
-async def add_job(message):
-    m_type, ext = get_media_info(message)
-    if not m_type:
-        return
-    
-    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
+async def add_message_to_queue(message, album_info=None):
+    if not message.media:
+        return False
     
     chat_id = int(message.chat.id)
     message_id = int(message.id)
-    file_size = int(file_size) if file_size else 0
-    file_id = str(file_id) if file_id else None
-    file_name = str(file_name) if file_name else None
     media_group_id = str(message.media_group_id) if message.media_group_id else None
     
-    src_name = get_source_name(message)
-    raw_cap = message.caption or ""
-    final_cap = f"{raw_cap}\n\nSource: {src_name}".strip() if raw_cap else f"Source: {src_name}"
+    # تحديد نوع الميديا
+    media_type = "unknown"
+    if message.photo: media_type = "photo"
+    elif message.video: media_type = "video"
+    elif message.document: media_type = "document"
+    elif message.audio: media_type = "audio"
+    elif message.voice: media_type = "voice"
+    elif message.video_note: media_type = "video_note"
+    elif message.animation: media_type = "animation"
+    
+    # اسم القناة أو المجموعة
+    chat = message.chat
+    source_name = chat.title or chat.first_name or chat.username or f"Chat_{chat_id}"
+    
+    raw_caption = message.caption or ""
     
+    # تفاصيل الألبوم إن وجدت
+    album_total = 1
+    album_index = 1
+    if album_info:
+        album_total = album_info.get("total", 1)
+        album_index = album_info.get("index", 1)
+
     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')
-            """, (
-                chat_id, message_id, media_group_id,
-                m_type, file_id, file_name, file_size, final_cap, src_name
-            ))
+                (chat_id, message_id, media_group_id, media_type, caption, source_name, album_total, album_index, status)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending')
+            """, (chat_id, message_id, media_group_id, media_type, raw_caption, source_name, album_total, album_index))
             await db.commit()
+            stats["total_queued"] += 1
+            return True
         except Exception as e:
-            logger.error(f"Add job error: {e}")
-
-async def update_job_status(chat_id, message_id, status, file_path=None, error=None):
-    chat_id = int(chat_id)
-    message_id = int(message_id)
-    
-    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, str(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, str(error)[:500], 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=1):
-    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 ?
-        """, (int(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
-        """, (str(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')
-        """, (str(media_group_id),))
-        row = await cursor.fetchone()
-        return row[0]
-
-async def delete_job(chat_id, message_id):
-    chat_id = int(chat_id)
-    message_id = int(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])
-            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"
-    elif message.video:
-        return "video", ".mp4"
-    elif message.document:
-        ext = os.path.splitext(message.document.file_name or "")[1] or ""
-        return "doc", ext
-    elif message.audio:
-        ext = os.path.splitext(message.audio.file_name or "")[1] or ".mp3"
-        return "audio", ext
-    elif message.voice:
-        return "voice", ".ogg"
-    elif message.video_note:
-        return "video_note", ".mp4"
-    elif message.animation:
-        return "animation", ".mp4"
-    return None, ""
-
-def get_source_name(message):
-    chat = message.chat
-    if chat.type in ["channel", "supergroup", "group"]:
-        return chat.title or "Channel/Group"
-    elif chat.type == "private":
-        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"Incomplete file: {actual_size}/{expected_size} bytes")
-        return False
-    if actual_size == 0:
-        logger.warning(f"Zero-byte file: {file_path}")
-        return False
-    return True
+            logger.error(f"خطأ في إضافة الرسالة لقاعدة البيانات: {e}")
+            return False
 
-async def download_with_message_object(chat_id, message_id, file_path):
-    try:
-        logger.info(f"Re-fetching message {chat_id}/{message_id}")
-        message = await app.get_messages(int(chat_id), int(message_id))
-        if not message or not message.media:
-            raise Exception("Message not found or no media")
-        
-        downloaded = await app.download_media(
-            message,
-            file_name=file_path
-        )
-        return downloaded
-    except Exception as e:
-        logger.error(f"Re-fetch download failed: {e}")
-        return None
-
-async def download_job(job):
-    chat_id = int(job['chat_id'])
-    message_id = int(job['message_id'])
-    expected_size = int(job['file_size'] or 0)
-    file_id = str(job['file_id']) if job['file_id'] else None
+# ============ إعداد الكابشن الذكي ============
+def format_caption(raw_cap, source_name, album_index, album_total):
+    lines = []
+    if raw_cap and raw_cap.strip():
+        lines.append(raw_cap.strip())
+        lines.append("")
     
-    ext = {
-        'photo': '.jpg', 'video': '.mp4', 'doc': '',
-        'audio': '.mp3', 'voice': '.ogg', 
-        'video_note': '.mp4', 'animation': '.mp4'
-    }.get(job['media_type'], '')
+    lines.append(f"📢 **المصدر:** {source_name}")
+    if album_total > 1:
+        lines.append(f"🔢 **الجزء:** {album_index} من {album_total}")
     
-    file_path = os.path.join(DOWNLOAD_DIR, f"{chat_id}_{message_id}{ext}")
-    
-    if os.path.exists(file_path):
-        try:
-            os.remove(file_path)
-        except:
-            pass
+    return "\n".join(lines)
+
+# ============ النقل الذكي (Fast Copy مع Fallback للتحميل) ============
+async def process_single_job(job):
+    chat_id = job["chat_id"]
+    msg_id = job["message_id"]
+    raw_caption = job["caption"]
+    source_name = job["source_name"]
+    album_index = job["album_index"]
+    album_total = job["album_total"]
     
-    await flood_control.wait()
+    caption = format_caption(raw_caption, source_name, album_index, album_total)
+
+    await rate_limiter.wait()
     
+    # 1. المحاولة الأولى: النسخ المباشر فائق السرعة بدون تحميل
     try:
-        await update_job_status(chat_id, message_id, 'downloading')
-        downloaded = None
-        
-        if file_id:
-            logger.info(f"Attempt 1 (file_id): {message_id}")
-            try:
-                downloaded = await app.download_media(file_id, file_name=file_path)
-            except FloodWait as e:
-                await flood_control.hit(e.value)
-                raise
-            except Exception as e:
-                logger.warning(f"file_id failed: {e}")
-        
-        if not await verify_download(downloaded, expected_size):
-            logger.info(f"Attempt 2 (message): {message_id}")
-            downloaded = await download_with_message_object(chat_id, message_id, file_path)
-        
-        if not await verify_download(downloaded, expected_size):
-            logger.info(f"Attempt 3 (retry): {message_id}")
-            await asyncio.sleep(5)
-            if file_id:
-                try:
-                    downloaded = await app.download_media(file_id, file_name=file_path)
-                except FloodWait as e:
-                    await flood_control.hit(e.value)
-                    raise
-                except:
-                    pass
-            if not await verify_download(downloaded, expected_size):
-                downloaded = await download_with_message_object(chat_id, message_id, file_path)
-        
-        if not await verify_download(downloaded, expected_size):
-            raise Exception(f"All attempts failed")
-        
-        await update_job_status(chat_id, message_id, 'downloaded', file_path=downloaded)
-        logger.info(f"Download complete: {message_id} | {os.path.getsize(downloaded)} bytes")
+        await app.copy_message(
+            chat_id=TARGET_CHAT,
+            from_chat_id=chat_id,
+            message_id=msg_id,
+            caption=caption
+        )
+        logger.info(f"⚡ تم النقل السريع بنجاح للرسالة {msg_id} من {source_name}")
         return True
-        
-    except FloodWait as e:
-        logger.warning(f"FloodWait: {e.value}s")
-        await update_job_status(chat_id, message_id, 'pending', error=f"FloodWait:{e.value}")
-        await flood_control.hit(e.value)
-        return False
+    except (FloodWait, FloodPremiumWait) as e:
+        rate_limiter.set_flood(e.value)
+        raise e
+    except ChatForwardsRestricted:
+        logger.warning(f"⚠️ القناة محمية المحتوى! التحويل إلى وضع التحميل والرفع للرسالة {msg_id}...")
+    except MessageIdInvalid:
+        logger.error(f"❌ الرسالة {msg_id} محذوفة أو غير صالحة.")
+        return "DELETED"
     except Exception as e:
-        err_str = str(e)
-        await update_job_status(chat_id, message_id, 'pending', error=err_str)
-        logger.error(f"Download failed {message_id}: {err_str}")
-        return False
+        logger.warning(f"⚠️ فشل النسخ المباشر للرسالة {msg_id} ({e})، جاري محاولة التحميل والرفع...")
 
-# ============ الإرسال ============
-async def safe_send(send_func):
-    for attempt in range(5):
-        try:
-            return await send_func()
-        except FloodWait as e:
-            wait_time = getattr(e, 'value', getattr(e, 'x', 30))
-            logger.warning(f"FloodWait {wait_time}s on attempt {attempt+1}/5")
-            await flood_control.hit(wait_time)
-            await asyncio.sleep(min(wait_time, 300))
-        except (ConnectionError, TimeoutError) as e:
-            wait_time = min(2 ** attempt, 60)
-            logger.warning(f"Connection error, retry in {wait_time}s: {e}")
-            await asyncio.sleep(wait_time)
-        except Exception as e:
-            if attempt == 4:
-                raise
-            logger.warning(f"Send error (attempt {attempt+1}): {e}")
-            await asyncio.sleep(min(2 ** attempt, 30))
-    return None
-
-async def send_job(job):
-    chat_id = int(job['chat_id'])
-    message_id = int(job['message_id'])
-    file_path = job['file_path']
-    
-    if not file_path or not os.path.exists(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
-    
-    await flood_control.wait()
-    
+    # 2. المحاولة الثانية: التحميل والرفع (في حال كانت القناة تمنع النقل المباشر)
     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))
+        msg = await app.get_messages(chat_id, msg_id)
+        if not msg or not msg.media:
+            return "DELETED"
+            
+        file_path = os.path.join(DOWNLOAD_DIR, f"temp_{chat_id}_{msg_id}")
+        downloaded = await app.download_media(msg, file_name=file_path)
         
-        logger.info(f"Sent: {message_id}")
-        await delete_job(chat_id, message_id)
+        if not downloaded or not os.path.exists(downloaded):
+            raise Exception("فشل تنزيل الملف")
+
+        # إعادة الإرسال حسب النوع
+        mtype = job["media_type"]
+        if mtype == "photo":
+            await app.send_photo(TARGET_CHAT, photo=downloaded, caption=caption)
+        elif mtype == "video":
+            await app.send_video(TARGET_CHAT, video=downloaded, caption=caption, supports_streaming=True)
+        elif mtype == "document":
+            await app.send_document(TARGET_CHAT, document=downloaded, caption=caption)
+        elif mtype == "audio":
+            await app.send_audio(TARGET_CHAT, audio=downloaded, caption=caption)
+        elif mtype == "voice":
+            await app.send_voice(TARGET_CHAT, voice=downloaded, caption=caption)
+        elif mtype == "video_note":
+            await app.send_video_note(TARGET_CHAT, video_note=downloaded)
+        elif mtype == "animation":
+            await app.send_animation(TARGET_CHAT, animation=downloaded, caption=caption)
+        else:
+            await app.send_document(TARGET_CHAT, document=downloaded, caption=caption)
+
+        # تنظيف الملف المؤقت فوراً لتوفير المساحة
+        if os.path.exists(downloaded):
+            os.remove(downloaded)
+            
+        logger.info(f"📥 تم التحميل والرفع بنجاح للرسالة {msg_id}")
         return True
-        
+    except (FloodWait, FloodPremiumWait) as e:
+        rate_limiter.set_flood(e.value)
+        raise e
     except Exception as e:
-        logger.error(f"Send failed {message_id}: {e}")
-        await update_job_status(chat_id, message_id, 'downloaded', error=str(e))
+        logger.error(f"❌ فشل تام في معالجة الرسالة {msg_id}: {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"Album {media_group_id} waiting for {pending} files")
-        return
-    
-    jobs.sort(key=lambda j: int(j['message_id']))
-    
-    album_inputs = []
-    album_jobs = []
-    standalone_jobs = []
-    
-    for i, job in enumerate(jobs):
-        cap = job['caption'] if i == 0 else None
-        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_jobs.append(job)
-    
-    await flood_control.wait()
-    
-    async with send_sem:
-        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:
-                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"Album {media_group_id} part {idx//chunk_size + 1} sent")
-            except Exception as 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")
-    await asyncio.sleep(5)
-    
+# ============ عامل معالجة الطابور (Worker) ============
+async def queue_worker():
+    logger.info("🚀 بدء تشغيل عامل الطابور...")
     while True:
         try:
             if not app.is_connected:
-                logger.warning("Not connected, waiting...")
-                await asyncio.sleep(10)
+                await asyncio.sleep(5)
                 continue
-            
-            await flood_control.wait()
-            
-            jobs = await get_pending_jobs(limit=1)
-            if not jobs:
-                await asyncio.sleep(2)
-                continue
-            
-            job = jobs[0]
-            
-            async with download_lock:
-                success = await download_job(job)
-                if success:
-                    await asyncio.sleep(3)
-                else:
-                    await asyncio.sleep(5)
-            
-        except Exception as e:
-            logger.error(f"Download Worker error: {e}")
-            await asyncio.sleep(10)
 
-async def send_worker():
-    logger.info("Send Worker started")
-    await asyncio.sleep(7)
-    
-    while True:
-        try:
-            if not app.is_connected:
-                logger.warning("Not connected, waiting...")
-                await asyncio.sleep(10)
-                continue
-            
-            await flood_control.wait()
-            
             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 2
+                    WHERE status = 'pending' AND attempts < 5
+                    ORDER BY id ASC LIMIT 1
                 """)
-                single_jobs = [dict(row) for row in await cursor.fetchall()]
+                job_row = await cursor.fetchone()
                 
-                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 1
-                """)
-                album_rows = await cursor.fetchall()
-            
-            for job in single_jobs:
-                if not app.is_connected:
-                    break
-                async with send_sem:
-                    await send_job(job)
-                    await asyncio.sleep(3)
-            
-            for (mg_id,) in album_rows:
-                if not app.is_connected:
-                    break
-                await send_album_group(mg_id)
-                await asyncio.sleep(3)
-            
-            await asyncio.sleep(2)
+                if not job_row:
+                    await asyncio.sleep(2)
+                    continue
+                
+                job = dict(job_row)
+                job_id = job["id"]
+                
+                await db.execute("UPDATE jobs SET status='processing' WHERE id=?", (job_id,))
+                await db.commit()
+
+            # المعالجة
+            result = await process_single_job(job)
             
+            async with aiosqlite.connect(DB_PATH) as db:
+                if result is True:
+                    await db.execute("UPDATE jobs SET status='completed' WHERE id=?", (job_id,))
+                    stats["total_transferred"] += 1
+                elif result == "DELETED":
+                    await db.execute("UPDATE jobs SET status='failed', error='Message Deleted/Invalid' WHERE id=?", (job_id,))
+                    stats["total_failed"] += 1
+                else:
+                    await db.execute("""
+                        UPDATE jobs SET status='pending', attempts=attempts+1, error='Retry later' WHERE id=?
+                    """, (job_id,))
+                await db.commit()
+
+        except (FloodWait, FloodPremiumWait) as e:
+            logger.warning(f"انتظار FloodWait: {e.value}s")
+            await asyncio.sleep(e.value + 1)
         except Exception as e:
-            logger.error(f"Send Worker error: {e}")
-            await asyncio.sleep(10)
+            logger.error(f"خطأ غير متوقع في الـ Worker: {e}")
+            await asyncio.sleep(5)
 
-# ============ Album Fetch ============
-async def fetch_full_album(message):
+# ============ ماسح الرسائل القديمة (History Scraper) ============
+async def scrape_channel_history(chat_identifier, limit=0):
+    """
+    يقوم بجلب كافة الرسائل القديمة من قناة أو مجموعة وترتيبها في الطابور
+    """
     try:
-        messages = await app.get_media_group(message.chat.id, message.id)
-        logger.info(f"get_media_group: {len(messages)} messages")
-        for msg in messages:
-            await add_job(msg)
-        return True
+        chat = await app.get_chat(chat_identifier)
+        chat_id = chat.id
+        chat_title = chat.title or chat.first_name or chat_identifier
+        logger.info(f"🔄 بدء جلب الرسائل القديمة من: {chat_title} ({chat_id})")
+        
+        count = 0
+        media_groups = {}
+        
+        # قراءة الأرشيف من الأقدم للأحدث أو من الأحدث للأقدم
+        async for message in app.get_chat_history(chat_id, limit=limit if limit > 0 else 0):
+            if message.media:
+                # إذا كان ألبوم
+                if message.media_group_id:
+                    mg_id = str(message.media_group_id)
+                    media_groups.setdefault(mg_id, []).append(message)
+                else:
+                    await add_message_to_queue(message)
+                    count += 1
+                    
+            if count % 50 == 0 and count > 0:
+                logger.info(f"📦 تم جدولة {count} رسالة من {chat_title}...")
+                await asyncio.sleep(0.1)
+
+        # إضافة الألبومات مع حساب عدد المقاطع لكل ألبوم
+        for mg_id, msgs in media_groups.items():
+            # ترتيب رسائل الألبوم بالترتيب الزمني الصحيح
+            msgs.sort(key=lambda m: m.id)
+            total = len(msgs)
+            for idx, msg in enumerate(msgs, start=1):
+                await add_message_to_queue(msg, album_info={"total": total, "index": idx})
+                count += 1
+
+        logger.info(f"✅ تم الانتهاء من فحص {chat_title}! إجمالي الوسائط المجدولة: {count}")
+        return f"تم فحص {chat_title} بنجاح وجدولة {count} ملف."
     except Exception as e:
-        logger.warning(f"get_media_group failed: {e}")
-        return False
+        logger.error(f"❌ فشل فحص القناة {chat_identifier}: {e}")
+        return f"خطأ في فحص {chat_identifier}: {str(e)}"
 
-# ============ Handlers ============
+# ============ رصد الرسائل الجديدة لحظياً (Real-time Listener) ============
 @app.on_message(filters.incoming & ~filters.me & filters.media)
-async def on_media(client, message):
-    chat_id = int(message.chat.id)
-    if str(chat_id) == str(TARGET_CHAT):
-        return
-    
-    src_name = get_source_name(message)
-    m_type, _ = get_media_info(message)
-    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:
-            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
-    logger.info(f"Text from {message.chat.id}")
+async def realtime_media_handler(client, message):
+    try:
+        # تجنب رصد الرسائل التي ترسلها الوجهة
+        if str(message.chat.id) == str(TARGET_CHAT):
+            return
+
+        if message.media_group_id:
+            # إذا كان ألبوم، انتظر ثانية واحدة لتكتمل حزمة الألبوم ثم جدولها
+            await asyncio.sleep(1.5)
+            try:
+                group_messages = await client.get_media_group(message.chat.id, message.id)
+                group_messages.sort(key=lambda m: m.id)
+                total = len(group_messages)
+                for idx, msg in enumerate(group_messages, start=1):
+                    await add_message_to_queue(msg, album_info={"total": total, "index": idx})
+            except Exception:
+                # إذا فشل جلب المجموعة كاملة، أضف الرسالة مفردة
+                await add_message_to_queue(message)
+        else:
+            await add_message_to_queue(message)
+            
+        logger.info(f"✨ رصد رسالة جديدة من {message.chat.title or message.chat.id}")
+    except Exception as e:
+        logger.error(f"خطأ في الرصد اللحظي: {e}")
 
-async def resolve_target_chat():
+# ============ تهيئة الهدف (TARGET_CHAT) ============
+async def resolve_target():
     global TARGET_CHAT
-    raw = TARGET_CHAT_RAW
-    if raw.lower() == "me":
+    raw = TARGET_CHAT_RAW.strip()
+    if raw.lower() in ["me", "saved", ""]:
         TARGET_CHAT = "me"
+        logger.info("🎯 الوجهة المحددة: الرسائل المحفوظة (Saved Messages)")
         return
     try:
         chat_id_int = int(raw)
         chat = await app.get_chat(chat_id_int)
         TARGET_CHAT = chat_id_int
-        logger.info(f"Target: {chat.title}")
-    except Exception as e:
-        logger.error(f"Target check failed: {e}")
-        TARGET_CHAT = "me"
+        logger.info(f"🎯 الوجهة المحددة: {chat.title} ({TARGET_CHAT})")
+    except ValueError:
+        try:
+            chat = await app.get_chat(raw)
+            TARGET_CHAT = chat.id
+            logger.info(f"🎯 الوجهة المحددة: {chat.title} ({TARGET_CHAT})")
+        except Exception as e:
+            logger.error(f"فشل التحقق من الوجهة ({raw}): {e}. تم التحويل للرسائل المحفوظة.")
+            TARGET_CHAT = "me"
 
-# ============ Gradio + FastAPI ============
-def run_gradio():
+# ============ واجهة التحكم السريعة (Gradio UI) ============
+def create_ui():
     with gr.Blocks(title="Telegram Archiver Pro") as demo:
-        gr.Markdown("# Telegram Archiver Pro")
-        gr.Markdown("Status: Running | Hydrogram | Serial | Verified")
-        gr.Textbox(label="Status", value="Active | SQLite Queue | 3-Attempt DL", interactive=False)
-    
+        gr.Markdown("## 🚀 محول وأرشيف تيليجرام الاحترافي (بدون ضياع ملفات)")
+        
+        with gr.Row():
+            status_box = gr.Textbox(label="حالة النظام", value="جاري التشغيل...", interactive=False)
+            queued_box = gr.Number(label="الملفات في الطابور", value=0, interactive=False)
+            transferred_box = gr.Number(label="الملفات المنقولة بنجاح", value=0, interactive=False)
+        
+        gr.Markdown("---")
+        gr.Markdown("### 📥 سحب الأرشيف القديم من قناة أو كروب:")
+        with gr.Row():
+            channel_input = gr.Textbox(
+                label="معرف أو رابط القناة / الكروب", 
+                placeholder="@my_channel أو -100123456789 أو رابط القناة"
+            )
+            limit_input = gr.Number(label="أقصى عدد رسائل (0 = الكل)", value=0)
+            scrape_btn = gr.Button("بدء السحب الآن 📥", variant="primary")
+        
+        scrape_output = gr.Textbox(label="نتيجة الفحص", interactive=False)
+        
+        def do_scrape(ch, lim):
+            if not ch:
+                return "يرجى كتابة معرف القناة أولاً"
+            loop = asyncio.new_event_loop()
+            res = loop.run_until_complete(scrape_channel_history(ch.strip(), int(lim)))
+            loop.close()
+            return res
+
+        scrape_btn.click(fn=do_scrape, inputs=[channel_input, limit_input], outputs=[scrape_output])
+        
+    return demo
+
+# ============ تشغيل خادم الويب و Hydrogram ============
+def run_web():
     fastapi_app = FastAPI()
-    
-    @fastapi_app.get("/health")
-    async def health_check():
-        return {"status": "ok", "timestamp": time.time()}
+    demo = create_ui()
+    gr.mount_gradio_app(fastapi_app, demo, path="/")
     
     @fastapi_app.get("/ping")
     async def ping():
-        return {"status": "alive"}
-    
-    gr.mount_gradio_app(fastapi_app, demo, path="/")
-    
-    import uvicorn
+        return {"status": "alive", "stats": stats}
+
     uvicorn.run(fastapi_app, host="0.0.0.0", port=7860)
 
-# ============ Main ============
 async def main():
-    try:
-        if app.is_connected:
-            await app.stop()
-            await asyncio.sleep(2)
-    except:
-        pass
-    
+    logger.info("جاري تسجيل الدخول إلى تيليجرام...")
     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")
-        return
-    
     me = await app.get_me()
-    await resolve_target_chat()
+    logger.info(f"✅ تم الدخول بحساب: {me.first_name} (@{me.username or me.id})")
+    
+    await resolve_target()
     await init_db()
     
-    await asyncio.sleep(2)
-    asyncio.create_task(download_worker())
-    await asyncio.sleep(2)
-    asyncio.create_task(send_worker())
+    # تشغيل العامل في الخلفية
+    asyncio.create_task(queue_worker())
     
-    logger.info("=" * 60)
-    logger.info(f"Archiver Pro running!")
-    logger.info(f"Account: {me.first_name}")
-    logger.info(f"Target: {TARGET_CHAT}")
-    logger.info("=" * 60)
+    # فحص القنوات المحددة في المتغيرات البيئية إن وجدت
+    if CHANNELS_TO_SCRAPE:
+        channels = [c.strip() for c in CHANNELS_TO_SCRAPE.split(",") if c.strip()]
+        for ch in channels:
+            asyncio.create_task(scrape_channel_history(ch))
+
+    stats["status"] = "يعمل بكفاءة ✅"
     
+    # الإبقاء قيد التشغيل
     stop_event = asyncio.Event()
     await stop_event.wait()
 
 if __name__ == "__main__":
-    gradio_thread = threading.Thread(target=run_gradio, daemon=True)
-    gradio_thread.start()
+    # تشغيل واجهة الويب في Thread مستقل لمنع تعليق البوت وتلبية متطلبات Hugging Face
+    web_thread = threading.Thread(target=run_web, daemon=True)
+    web_thread.start()
     
     loop = asyncio.get_event_loop()
     try:
@@ -721,7 +474,5 @@ if __name__ == "__main__":
     except KeyboardInterrupt:
         pass
     finally:
-        try:
-            loop.run_until_complete(app.stop())
-        except:
-            pass
+        if app.is_connected:
+            loop.run_until_complete(app.stop())