Update main.py

main.py

@@ -1,205 +1,285 @@
 import asyncio
-import os
-import re
 import time
-import glob
+import re
+import os
+import json
 import random
-from datetime import datetime
-from threading import Thread
-
-from flask import Flask
+import glob
 from telethon import TelegramClient, events, errors
 from telethon.sessions import StringSession
-from motor.motor_asyncio import AsyncIOMotorClient
+from flask import Flask
+from threading import Thread
 
 # ==========================================
-# --- CONFIGURATION (HF SECRETS) ---
-# ==========================================
-API_ID = int(os.environ.get('API_ID'))
-API_HASH = os.environ.get('API_HASH')
+# --- SECURE CONFIGURATION ---
+# Pulling all credentials securely from Hugging Face Secrets
+API_ID = int(os.environ.get('API_ID'))       
+API_HASH = os.environ.get('API_HASH')        
 SESSION_STRING = os.environ.get('SESSION_STRING')
-MONGO_URI = os.environ.get('MONGO_URI')
 
 owner_ids_string = os.environ.get('OWNER_IDS', '')
 OWNER_IDS = [int(user_id.strip()) for user_id in owner_ids_string.split(',') if user_id.strip()]
 DESTINATION_GROUP_ID = int(os.environ.get('DESTINATION_GROUP_ID'))
 SOURCE_CHAT_ID = int(os.environ.get('SOURCE_CHAT_ID'))
 
+MAX_FILE_SIZE_MB = 1500  # Skip files larger than 1.5GB
 # ==========================================
-# --- DATABASE SETUP (MongoDB) ---
-# ==========================================
-db_client = AsyncIOMotorClient(MONGO_URI)
-db = db_client['telegram_forwarder']
-settings_col = db['settings']
-processed_col = db['processed_messages']
-
-async def get_settings():
-    doc = await settings_col.find_one({"id": "bot_config"})
-    if not doc:
-        # Default settings
-        default = {
-            "id": "bot_config",
-            "last_id": 0,
-            "max_size_mb": 1500,
-            "delay_base": 10,
-            "sync_active": False
-        }
-        await settings_col.insert_one(default)
-        return default
-    return doc
-
-async def update_setting(key, value):
-    await settings_col.update_one({"id": "bot_config"}, {"$set": {key: value}})
 
-# ==========================================
-# --- FLASK & UTILS ---
-# ==========================================
+client = TelegramClient(StringSession(SESSION_STRING), API_ID, API_HASH)
+sync_active = False
+PROGRESS_FILE = "progress.json"
+
+# --- FLASK KEEP-ALIVE ---
 app = Flask('')
 @app.route('/')
-def home(): return "Elite Forwarder is Live!"
+def home():
+    return "Hugging Face Sync Bot is active! (V3 - Elite Mode)"
 
-def run_web(): app.run(host='0.0.0.0', port=7860)
+def run_web():
+    app.run(host='0.0.0.0', port=7860) 
 
-def get_progress_bar(current, total):
-    percentage = (current / total) * 100 if total > 0 else 0
-    completed = int(percentage / 10)
-    return f"|{'█' * completed}{'░' * (10 - completed)}| {percentage:.1f}%"
+# --- AUTO-CLEANUP ---
+def cleanup_leftovers():
+    print("🧹 Sweeping for leftover files...")
+    leftovers = glob.glob("*.zip") + glob.glob("*.jpg") + glob.glob("*.mp4") + glob.glob("*.mkv") + glob.glob("*.pdf") + glob.glob("*.png")
+    for f in leftovers:
+        try:
+            os.remove(f)
+            print(f"🗑️ Deleted leftover: {f}")
+        except Exception:
+            pass
 
-def cleanup():
-    for f in glob.glob("*.zip") + glob.glob("*.mp4") + glob.glob("*.mkv") + glob.glob("*.jpg"):
-        try: os.remove(f)
-        except: pass
+cleanup_leftovers()
 
-# ==========================================
-# --- CORE LOGIC ---
-# ==========================================
-client = TelegramClient(StringSession(SESSION_STRING), API_ID, API_HASH)
+# --- STATE MANAGEMENT & AUTO-RESUME ---
+def save_state(current_id):
+    with open(PROGRESS_FILE, 'w') as f:
+        json.dump({"current_id": current_id}, f)
 
-async def process_message(msg, config):
-    """Downloads and uploads message. Returns size_mb if successful."""
-    msg_id = msg.id
-    
-    # Duplicate Detection
-    if await processed_col.find_one({"source_id": msg_id}):
-        return 0, "duplicate"
+def load_state():
+    if os.path.exists(PROGRESS_FILE):
+        with open(PROGRESS_FILE, 'r') as f:
+            return json.load(f)
+    return None
+
+def clear_state():
+    if os.path.exists(PROGRESS_FILE):
+        os.remove(PROGRESS_FILE)
 
+async def get_last_synced_id():
+    """Scans the destination group to find the last uploaded ID (HF Reboot Savior)"""
     try:
-        caption = f"{msg.text or ''}\n\n🆔 **Source ID:** `{msg_id}`"
-        
-        # Handle Media
-        if msg.media:
-            size_mb = 0
+        async for msg in client.iter_messages(DESTINATION_GROUP_ID, limit=50):
+            if msg.text and "🆔 **Source ID:**" in msg.text:
+                match = re.search(r'Source ID:\*\* `?(\d+)`?', msg.text)
+                if match:
+                    return int(match.group(1))
+    except Exception as e:
+        print(f"❌ Auto-Resume Error: {e}")
+    return None
+
+# --- CORE PROCESSING LOGIC (BATCH MODE) ---
+async def process_single_message(msg):
+    """Returns (success_status, did_process_media, size_in_mb)"""
+    msg_id = msg.id
+    size_mb = 0
+
+    for attempt in range(3): 
+        try:
+            original_caption = msg.text or ""
+            tagged_caption = f"{original_caption}\n\n🆔 **Source ID:** `{msg_id}`".strip()
+                
+            # TEXT-ONLY MESSAGES
+            if not msg.media:
+                if original_caption: 
+                    await client.send_message(DESTINATION_GROUP_ID, tagged_caption)
+                return True, False, 0
+
+            # MEDIA MESSAGES
             if hasattr(msg.media, 'document'):
                 size_mb = msg.media.document.size / (1024 * 1024)
-            
-            if size_mb > config['max_size_mb']:
-                return 0, "too_large"
+                if size_mb > MAX_FILE_SIZE_MB:
+                    print(f"⚠️ Skipping ID {msg_id}: File too large ({size_mb:.2f}MB)")
+                    return True, False, 0 
 
-            # Forwarding is disabled in source, so we download
             path = await client.download_media(msg)
             if path:
-                await client.send_file(DESTINATION_GROUP_ID, path, caption=caption)
-                os.remove(path)
-                await processed_col.insert_one({"source_id": msg_id, "timestamp": datetime.now()})
-                return size_mb, "success"
-        
-        # Handle Text Only
-        elif msg.text:
-            await client.send_message(DESTINATION_GROUP_ID, caption)
-            await processed_col.insert_one({"source_id": msg_id, "timestamp": datetime.now()})
-            return 0, "success"
-
-    except errors.FloodWaitError as e:
-        return e.seconds, "flood"
-    except Exception as e:
-        print(f"Error: {e}")
-        return 0, "error"
-    return 0, "skipped"
+                await client.send_file(
+                    DESTINATION_GROUP_ID,
+                    path,
+                    caption=tagged_caption 
+                )
+                print(f"✅ UPLOADED: ID {msg_id}")
+                if os.path.exists(path):
+                    os.remove(path) 
+            return True, True, size_mb
+
+        except errors.FloodWaitError as e:
+            print(f"⚠️ Telegram Limit! Waiting {e.seconds}s...")
+            await asyncio.sleep(e.seconds) 
+        except Exception as e:
+            print(f"❌ Error on {msg_id} (Attempt {attempt+1}/3): {e}")
+            await asyncio.sleep(5) 
+            
+    print(f"❌ Skipping {msg_id} after 3 failed attempts.")
+    return True, False, 0
 
 # ==========================================
-# --- LISTENERS & LOOPS ---
+# --- BOT COMMANDS ---
 # ==========================================
 
-@client.on(events.NewMessage(chats=SOURCE_CHAT_ID))
-async def live_listener(event):
-    """Feature 1: Live Listener Mode"""
-    config = await get_settings()
-    if not config['sync_active']: return
-    
-    # Small delay to ensure DB order
-    await asyncio.sleep(2)
-    await process_message(event.message, config)
-    await update_setting("last_id", event.message.id)
+@client.on(events.NewMessage(pattern='/status'))
+async def status_handler(event):
+    global sync_active
+    if event.sender_id not in OWNER_IDS: return
 
-@client.on(events.NewMessage(pattern=r'/'))
-async def commands(event):
+    if not sync_active:
+        await event.respond("✅ **Idle.** Send `/sync <start_id>` to begin.")
+        return
+
+    state = load_state()
+    current_id = state['current_id'] if state else "Calculating..."
+
+    try:
+        latest_msgs = await client.get_messages(SOURCE_CHAT_ID, limit=1)
+        if latest_msgs:
+            latest_id = latest_msgs[0].id
+            remaining = latest_id - (current_id if isinstance(current_id, int) else 0)
+            
+            if remaining <= 0:
+                await event.respond(f"✅ **Fully Caught Up!**\nMonitoring for live drops at ID: `{current_id}`")
+            else:
+                await event.respond(
+                    f"📊 **Live Sync Status:**\n"
+                    f"➤ **Current ID:** `{current_id}`\n"
+                    f"➤ **Target ID:** `{latest_id}`\n"
+                    f"⏳ **Remaining:** `{remaining}` files."
+                )
+    except Exception as e:
+        await event.respond(f"📊 **Sync Progress:** ID `{current_id}` *(Error: {e})*")
+
+
+@client.on(events.NewMessage(pattern='/stop'))
+async def stop_handler(event):
+    global sync_active
     if event.sender_id not in OWNER_IDS: return
-    cmd = event.text.split()
     
-    if cmd[0] == '/status':
-        config = await get_settings()
-        latest = await client.get_messages(SOURCE_CHAT_ID, limit=1)
-        curr = config['last_id']
-        total = latest[0].id
-        bar = get_progress_bar(curr, total)
-        status = "🟢 Syncing" if config['sync_active'] else "🔴 Stopped"
-        await event.respond(f"**Bot Status:** {status}\n**Limit:** {config['max_size_mb']}MB\n**Progress:** `{curr}/{total}`\n{bar}")
-
-    elif cmd[0] == '/set_limit':
-        new_limit = int(cmd[1])
-        await update_setting("max_size_mb", new_limit)
-        await event.respond(f"✅ Max file size set to {new_limit}MB")
-
-    elif cmd[0] == '/set_delay':
-        new_delay = int(cmd[1])
-        await update_setting("delay_base", new_delay)
-        await event.respond(f"✅ Base delay set to {new_delay}s")
-
-    elif cmd[0] == '/resume':
-        await update_setting("sync_active", True)
-        await event.respond("🚀 Resuming Backlog Sync...")
-        asyncio.create_task(run_backlog_sync(event))
-
-    elif cmd[0] == '/stop':
-        await update_setting("sync_active", False)
-        await event.respond("🛑 Sync Paused.")
-
-async def run_backlog_sync(event):
-    """The Background Loop for Old Messages"""
-    while True:
-        config = await get_settings()
-        if not config['sync_active']: break
-        
-        # Get next batch of messages
-        messages = await client.get_messages(SOURCE_CHAT_ID, min_id=config['last_id'], limit=20, reverse=True)
-        
-        if not messages:
-            await event.respond("🏁 Backlog finished! Live listener active.")
-            break
-        
-        for m in messages:
-            # Re-check config inside loop
-            config = await get_settings()
-            if not config['sync_active']: break
-            
-            size, status = await process_message(m, config)
-            
-            if status == "flood":
-                await asyncio.sleep(size + 5)
-            elif status == "success":
-                await update_setting("last_id", m.id)
-                # Proportional delay
-                delay = config['delay_base'] + (size / 50)
-                await asyncio.sleep(delay)
+    sync_active = False 
+    state = load_state()
+    current_id = state['current_id'] if state else "Unknown"
+    await event.respond(f"🛑 **Sync Stopped!**\nSafely paused at ID: `{current_id}`.\nSend `/resume` to continue.")
+
+
+@client.on(events.NewMessage(pattern=r'/(sync|resume)'))
+async def command_handler(event):
+    global sync_active
+    if event.sender_id not in OWNER_IDS: return
+
+    if sync_active:
+        await event.respond("⚠️ Sync is already running! Use `/status`.")
+        return
+
+    sync_active = True
+    text = event.text
+
+    if text.startswith('/resume'):
+        state = load_state()
+        if state:
+            start_id = state['current_id']
+            await event.respond(f"🔄 **Resuming Sync** from local save at ID `{start_id}`...")
+        else:
+            await event.respond("🔍 **Scanning group for last uploaded ID...** (HF Reboot Savior Active)")
+            found_id = await get_last_synced_id()
+            if found_id:
+                start_id = found_id + 1
+                save_state(start_id)
+                await event.respond(f"✅ **Found it!** Resuming Sync from ID `{start_id}`...")
             else:
-                await update_setting("last_id", m.id)
-                await asyncio.sleep(1)
+                sync_active = False 
+                await event.respond("❌ Could not find any previous uploads in the group. Please use `/sync <start_id>`.")
+                return
+    else:
+        links = re.findall(r'(\d+)', text)
+        if not links:
+            sync_active = False 
+            await event.respond("❌ Provide a start ID. Example: `/sync 150`")
+            return
+        start_id = int(links[-1]) 
+        clear_state()
+        save_state(start_id)
+        await event.respond(f"🚀 **Starting Sync** from ID `{start_id}`.")
+
+    client.loop.create_task(sync_loop(event))
+
+# ==========================================
+# --- CONTINUOUS SYNC LOOP ---
+# ==========================================
+async def sync_loop(event):
+    global sync_active
+    counter = 0
+    batch_limit = random.randint(85, 115) 
+    has_notified_catchup = False
+
+    while sync_active:
+        state = load_state()
+        if not state: break
+        current_id = state['current_id']
+
+        try:
+            # 🔥 BATCH FETCHING: Grabs up to 50 messages at once chronologically
+            messages = await client.get_messages(SOURCE_CHAT_ID, min_id=current_id - 1, limit=50, reverse=True)
+
+            if not messages:
+                if not has_notified_catchup:
+                    await event.respond("✅ **Backlog Complete!**\nBot is now live-monitoring for new uploads.")
+                    has_notified_catchup = True
+                    
+                print("⏳ Fully caught up. Waiting for new files...")
+                await asyncio.sleep(random.randint(10, 20)) 
+                continue
+
+            has_notified_catchup = False
+
+            for msg in messages:
+                if not sync_active: break 
+                
+                success, did_process, size_mb = await process_single_message(msg)
+                
+                if success:
+                    # Save the ID for the *next* message so we don't repeat this one
+                    next_id = msg.id + 1 
+                    save_state(next_id)
+                    
+                    if did_process:
+                        counter += 1
+                        
+                        # ⚖️ PROPORTIONAL SIZE DELAYS
+                        base_delay = random.randint(8, 16)
+                        size_delay = (size_mb / 50) if size_mb > 0 else 0
+                        total_delay = base_delay + size_delay
+                        
+                        print(f"⏱️ Delay: {total_delay:.1f}s (Base: {base_delay}s + Size: {size_delay:.1f}s)")
+                        await asyncio.sleep(total_delay) 
+                        
+                        if counter >= batch_limit:
+                            break_time = random.randint(1500, 2100) 
+                            print(f"💤 Taking random safety break for {break_time//60} mins...")
+                            await asyncio.sleep(break_time)
+                            counter = 0
+                            batch_limit = random.randint(85, 115) 
+                    else:
+                        await asyncio.sleep(random.uniform(0.3, 1.2)) # Fast-forward text/skipped media
+                else:
+                    break # If FloodWait hit, break the batch loop and retry on the next while iteration
+
+        except Exception as e:
+            print(f"❌ Loop Error: {e}")
+            await asyncio.sleep(5)
 
 # ==========================================
-# --- STARTUP ---
+# --- EXECUTION ---
 # ==========================================
-print("Bot initializing...")
-cleanup()
+print("Starting HF Sync Bot (Elite Batch & Auto-Resume Mode)...")
 Thread(target=run_web).start()
 client.start()
 client.run_until_disconnected()