Update bot.py

bot.py

@@ -47,7 +47,7 @@ CHANNELS_TO_SCRAPE = os.environ.get("CHANNELS_TO_SCRAPE", "")
 
 DATA_DIR = "/data" if os.path.exists("/data") else "/tmp/data"
 DOWNLOAD_DIR = os.path.join(DATA_DIR, "downloads")
-DB_PATH = os.path.join(DATA_DIR, "queue_v2.db")
+DB_PATH = os.path.join(DATA_DIR, "archiver_queue.db")
 os.makedirs(DOWNLOAD_DIR, exist_ok=True)
 
 # ============ إنشاء العميل ============
@@ -102,31 +102,80 @@ class RateLimiter:
 
 rate_limiter = RateLimiter(delay=2.0)
 
-# ============ قاعدة البيانات ============
+# ============ المعالجة التلقائية لقاعدة البيانات (Self-Healing DB) ============
+def cleanup_corrupt_db():
+    logger.warning("Cleaning up corrupted database files...")
+    for ext in ["", "-wal", "-shm", "-journal"]:
+        file_to_del = f"{DB_PATH}{ext}"
+        if os.path.exists(file_to_del):
+            try:
+                os.remove(file_to_del)
+                logger.info(f"Deleted: {file_to_del}")
+            except Exception as e:
+                logger.error(f"Failed to delete {file_to_del}: {e}")
+
 async def init_db():
-    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,
-                caption TEXT,
-                source_name TEXT,
-                album_total INTEGER DEFAULT 1,
-                album_index INTEGER DEFAULT 1,
-                status TEXT DEFAULT 'pending',
-                attempts INTEGER DEFAULT 0,
-                error TEXT,
-                created_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 initialized successfully.")
+    try:
+        # فحص سلامة الملف إذا كان موجوداً
+        if os.path.exists(DB_PATH):
+            try:
+                async with aiosqlite.connect(DB_PATH) as db:
+                    cursor = await db.execute("PRAGMA integrity_check;")
+                    res = await cursor.fetchone()
+                    if not res or res[0] != "ok":
+                        raise Exception("Integrity check failed")
+            except Exception as e:
+                logger.error(f"Detected corrupt database ({e}), resetting...")
+                cleanup_corrupt_db()
+
+        async with aiosqlite.connect(DB_PATH) as db:
+            await db.execute("PRAGMA journal_mode=WAL;")
+            await db.execute("PRAGMA synchronous=NORMAL;")
+            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,
+                    caption TEXT,
+                    source_name TEXT,
+                    album_total INTEGER DEFAULT 1,
+                    album_index INTEGER DEFAULT 1,
+                    status TEXT DEFAULT 'pending',
+                    attempts INTEGER DEFAULT 0,
+                    error TEXT,
+                    created_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 initialized successfully.")
+    except Exception as e:
+        logger.error(f"Critical error initializing DB: {e}, recreating fresh DB...")
+        cleanup_corrupt_db()
+        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,
+                    caption TEXT,
+                    source_name TEXT,
+                    album_total INTEGER DEFAULT 1,
+                    album_index INTEGER DEFAULT 1,
+                    status TEXT DEFAULT 'pending',
+                    attempts INTEGER DEFAULT 0,
+                    error TEXT,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    UNIQUE(chat_id, message_id)
+                )
+            """)
+            await db.commit()
 
 async def add_message_to_queue(message, album_info=None):
     if not message.media:
@@ -162,8 +211,8 @@ async def add_message_to_queue(message, album_info=None):
         album_total = album_info.get("total", 1)
         album_index = album_info.get("index", 1)
 
-    async with aiosqlite.connect(DB_PATH) as db:
-        try:
+    try:
+        async with aiosqlite.connect(DB_PATH) as db:
             await db.execute("""
                 INSERT OR IGNORE INTO jobs 
                 (chat_id, message_id, media_group_id, media_type, caption, source_name, album_total, album_index, status)
@@ -172,9 +221,11 @@ async def add_message_to_queue(message, album_info=None):
             await db.commit()
             stats["total_queued"] += 1
             return True
-        except Exception as e:
-            logger.error(f"Error adding message to DB queue: {e}")
-            return False
+    except Exception as e:
+        logger.error(f"Error adding message to DB queue: {e}")
+        if "malformed" in str(e).lower():
+            await init_db()
+        return False
 
 # ============ إعداد الكابشن ============
 def format_caption(raw_cap, source_name, album_index, album_total):
@@ -202,7 +253,7 @@ async def process_single_job(job):
 
     await rate_limiter.wait()
     
-    # محاولة النسخ المباشر
+    # 1. محاولة النسخ المباشر من السيرفر
     try:
         await app.copy_message(
             chat_id=TARGET_CHAT,
@@ -224,7 +275,7 @@ async def process_single_job(job):
     except Exception as e:
         logger.warning(f"Direct copy failed for {msg_id} ({e}), falling back to download & upload...")
 
-    # التحميل والرفع في حال الحماية
+    # 2. التحميل والرفع في حال الحماية
     try:
         msg = await app.get_messages(chat_id, msg_id)
         if not msg or not msg.media:
@@ -276,39 +327,52 @@ async def queue_worker():
                 await asyncio.sleep(5)
                 continue
 
-            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 < 5
-                    ORDER BY id ASC LIMIT 1
-                """)
-                job_row = await cursor.fetchone()
-                
-                if not job_row:
+            job = None
+            try:
+                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 < 5
+                        ORDER BY id ASC LIMIT 1
+                    """)
+                    job_row = await cursor.fetchone()
+                    
+                    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()
+            except Exception as db_err:
+                if "malformed" in str(db_err).lower():
+                    logger.error("Database malformed in worker, resetting database...")
+                    await init_db()
                     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()
+                else:
+                    raise db_err
 
             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()
+            try:
+                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 Exception as db_err:
+                logger.error(f"Failed to update job status: {db_err}")
 
         except (FloodWait, FloodPremiumWait) as e:
             wait_val = getattr(e, 'value', 30)