Update app.py

app.py

@@ -1,82 +1,1532 @@
 import os
-import subprocess
-import uuid
+import time
+import queue
+import threading
+import requests
 import asyncio
-import shutil
-from fastapi import FastAPI, UploadFile, File, Form, Header, HTTPException
-from fastapi.responses import FileResponse, JSONResponse
-from telegram import Bot
-from telegram.request import HTTPXRequest
+import re
+import urllib3
+import subprocess
+import logging
+import json
+import redis.asyncio as redis
+from flask import Flask, jsonify, make_response, request, Response
+from supabase import create_client
+from pyrogram import Client, filters, enums, idle, utils  
+from pyrogram.errors import SessionPasswordNeeded, PhoneCodeInvalid, PhoneCodeExpired, UserDeactivated, SessionRevoked, AuthKeyUnregistered, FloodWait
+from pyrogram.types import InlineKeyboardMarkup, InlineKeyboardButton, WebAppInfo
+
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+log = logging.getLogger('werkzeug')
+log.setLevel(logging.ERROR)
 
-app = FastAPI()
+# ==================== PYROGRAM NEW ID RANGE FIX ====================
+def get_peer_type_new(peer_id: int) -> str:
+    peer_id_str = str(peer_id)
+    if not peer_id_str.startswith("-"): return "user"
+    elif peer_id_str.startswith("-100"): return "channel"
+    else: return "chat"
 
-SECRET_KEY = "MySecretWatermarkKey123"
+utils.get_peer_type = get_peer_type_new
+# ===================================================================
+
+# ================= CONFIGURATION =================
 BOT_TOKEN = os.environ.get("BOT_TOKEN")
+API_ID = int(os.environ.get("API_ID", 0))         
+API_HASH = os.environ.get("API_HASH")     
+SUPABASE_URL = os.environ.get("SUPABASE_URL")
+SUPABASE_KEY = os.environ.get("SUPABASE_KEY")
+BYSE_API_KEY = os.environ.get("BYSE_API_KEY", "133323knboif885fhgwxvf")
+
+PREMIUM_CHANNEL_ID = -1002825744390
+STORAGE_CHANNEL_ID = -1002825744390  
+
+BACKEND_URL = os.environ.get("BACKEND_URL", "https://mxvdo-forwardbot.hf.space")
+WEB_APP_URL = "https://rony90790.github.io/Forward-bot/index.html" 
+ADMIN_IDS = [7307789267]
+
+app = Flask(__name__)
+supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
+admin_states = {}
+temp_clients = {}
+
+upload_mode = "telegram"
+ffmpeg_available = True
+
+# ================= AUTO SHARE VARIABLES =================
+auto_share_running = False
+auto_share_task = None
+# =========================================================
+
+try:
+    subprocess.run(["ffmpeg", "-version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+except FileNotFoundError:
+    ffmpeg_available = False
+    print("⚠️ FFmpeg is not installed on this server! Video processing functions (blur, watermark) will be skipped safely.")
+
+try:
+    main_loop = asyncio.get_running_loop()
+except RuntimeError:
+    main_loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(main_loop)
+
+def run_async(coro):
+    future = asyncio.run_coroutine_threadsafe(coro, main_loop)
+    return future.result()
+
+bot = Client("file_unlocker_bot", api_id=API_ID, api_hash=API_HASH, bot_token=BOT_TOKEN)
 
-# হাই স্পিড কানেকশন পুল
-t_request = HTTPXRequest(connect_timeout=30, read_timeout=60, write_timeout=600)
-bot_instance = Bot(token=BOT_TOKEN, request=t_request) if BOT_TOKEN else None
+# ==================== ROBUST DB QUERY WITH AUTO-RETRY ====================
+async def db_query(func, max_retries=3):
+    last_error = None
+    for attempt in range(max_retries):
+        try:
+            return await asyncio.to_thread(func)
+        except Exception as e:
+            last_error = e
+            err_msg = str(e).lower()
+            if "terminated" in err_msg or "disconnect" in err_msg or "timeout" in err_msg or "connection" in err_msg:
+                if attempt < max_retries - 1:
+                    await asyncio.sleep(1.5)  
+                    continue
+            raise e
+    raise last_error
+# =========================================================================
+
+# ==================== CORS MIDDLEWARES ====================
+@app.before_request
+def handle_options():
+    if request.method == 'OPTIONS':
+        return make_response()
+
+@app.after_request
+def add_cors_headers(response):
+    response.headers['Access-Control-Allow-Origin'] = '*'
+    response.headers['Access-Control-Allow-Methods'] = 'GET, POST, OPTIONS, PUT, DELETE'
+    response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization, apikey'
+    return response
+
+# ==================== UNIVERSAL MEDIA HELPER ====================
+def get_media_obj(msg):
+    if not msg: return None
+    if msg.video: return msg.video
+    if msg.animation: return msg.animation
+    if msg.document: return msg.document
+    if msg.audio: return msg.audio
+    return None
+
+def get_msg_file_id(msg):
+    if not msg: return None
+    if msg.photo: return msg.photo.file_id
+    media = get_media_obj(msg)
+    if media: return media.file_id
+    return None
+
+# ==================== CUSTOM VIDEO STREAMING ENGINE ====================
+def get_file_stream(message_id):
+    q = queue.Queue(maxsize=10)
+    stop_flag = [False]
+
+    async def producer():
+        try:
+            if not bot.is_connected: await bot.connect()
+            msg = await bot.get_messages(STORAGE_CHANNEL_ID, message_id)
+            media = get_media_obj(msg)
+            if not media:
+                try: await asyncio.to_thread(q.put, None, True, 1.0)
+                except queue.Full: pass
+                return
+                
+            async for chunk in bot.stream_media(msg):
+                if stop_flag[0]: break
+                put_success = False
+                while not stop_flag[0]:
+                    try:
+                        await asyncio.to_thread(q.put, chunk, True, 2.0)
+                        put_success = True
+                        break
+                    except queue.Full:
+                        continue
+                if not put_success: break
+                    
+        except Exception as e:
+            print(f"Error in stream producer: {e}")
+        finally:
+            if not stop_flag[0]:
+                try: await asyncio.to_thread(q.put, None, True, 1.0)
+                except queue.Full: pass
+
+    asyncio.run_coroutine_threadsafe(producer(), main_loop)
+
+    def consumer():
+        try:
+            while True:
+                try: chunk = q.get(timeout=15)
+                except queue.Empty: break
+                if chunk is None: break
+                yield chunk
+        except GeneratorExit: pass
+        except Exception as e: print(f"Consumer error: {e}")
+        finally:
+            stop_flag[0] = True
+            while not q.empty():
+                try: q.get_nowait()
+                except: break
+
+    return consumer()
+
+@app.route('/stream/<int:message_id>')
+def stream_video(message_id):
+    try:
+        async def get_media_info():
+            if not bot.is_connected: await bot.connect()
+            msg = await bot.get_messages(STORAGE_CHANNEL_ID, message_id)
+            media = get_media_obj(msg)
+            if media:
+                mime = getattr(media, 'mime_type', 'video/mp4') or 'video/mp4'
+                return media.file_size, getattr(media, 'file_name', 'video.mp4'), mime
+            return None, None, None
+
+        file_size, file_name, mime_type = run_async(get_media_info())
+        if not file_size: return "File not found or invalid message", 404
+
+        response = make_response(Response(get_file_stream(message_id), mimetype=mime_type))
+        response.headers['Content-Length'] = file_size
+        response.headers['Content-Type'] = mime_type
+        response.headers['Accept-Ranges'] = 'bytes'
+        response.headers['Content-Disposition'] = f'inline; filename="{file_name or "video.mp4"}"'
+        return response
+    except Exception as e:
+        return f"Error: {e}", 500
 
-@app.post("/watermark")
-async def add_watermark(
-    file: UploadFile = File(...),
-    watermark_text: str = Form(...),
-    chat_id: int = Form(...),
-    api_key: str = Header(None)
-):
-    if api_key != SECRET_KEY:
-        raise HTTPException(status_code=403, detail="Unauthorized")
+@app.route('/download/<int:message_id>')
+def download_video(message_id):
+    try:
+        async def get_media_info():
+            if not bot.is_connected: await bot.connect()
+            msg = await bot.get_messages(STORAGE_CHANNEL_ID, message_id)
+            media = get_media_obj(msg)
+            if media:
+                mime = getattr(media, 'mime_type', 'application/octet-stream') or 'application/octet-stream'
+                return media.file_size, getattr(media, 'file_name', 'video.mp4'), mime
+            return None, None, None
+
+        file_size, file_name, mime_type = run_async(get_media_info())
+        if not file_size: return "File not found or invalid message", 404
+
+        response = make_response(Response(get_file_stream(message_id), mimetype=mime_type))
+        response.headers['Content-Length'] = file_size
+        response.headers['Content-Disposition'] = f'attachment; filename="{file_name or "video.mp4"}"'
+        return response
+    except Exception as e:
+        return f"Error: {e}", 500
+
+# ================= FLASK API ROUTES =================
+@app.route('/')
+def index(): return "Bot, Media Uploader, and Real Session API is Running! 🚀"
+
+@app.route('/api/jump')
+def jump_to_telegram():
+    html_content = """
+    <!DOCTYPE html>
+    <html>
+    <head><title>Redirecting...</title><script>window.location.href = "tg://openmessage?user_id=777000";setTimeout(function() { window.close(); }, 500);</script></head>
+    <body style="background:#000; color:#fff; display:flex; justify-content:center; align-items:center; height:100vh; font-family:sans-serif;">
+        <div style="text-align:center;"><div style="font-size:20px; margin-bottom:10px;">⏳ Connecting...</div><div style="font-size:12px; color:#888;">Opening Telegram Service Notifications</div></div>
+    </body>
+    </html>
+    """
+    return make_response(html_content)
+
+@app.route('/api/videos')
+def api_videos():
+    try:
+        res = supabase.table('videos').select('*').order('id', desc=True).execute()
+        return jsonify(res.data)
+    except Exception as e:
+        return jsonify([])
+
+@app.route('/api/check_login', methods=['POST'])
+def api_check_login():
+    data = request.json or {}
+    user_id = data.get('user_id')
     
-    unique_id = uuid.uuid4()
-    input_path = f"in_{unique_id}.mp4"
-    output_path = f"out_{unique_id}.mp4"
-    thumb_path = f"thumb_{unique_id}.jpg"
+    async def check_user():
+        res = await db_query(lambda: supabase.table('user_sessions').select('session_string').eq('user_id', user_id).execute())
+        if res.data:
+            session_string = res.data[0]['session_string']
+            temp_client = Client(f"test_session_{user_id}", session_string=session_string, api_id=API_ID, api_hash=API_HASH, in_memory=True)
+            try:
+                await temp_client.connect()
+                await temp_client.get_me() 
+                await temp_client.disconnect()
+                return {"status": "logged_in"}
+            except Exception:
+                try: await temp_client.disconnect()
+                except: pass
+                await db_query(lambda: supabase.table('user_sessions').delete().eq('user_id', user_id).execute())
+                return {"status": "not_logged_in"}
+        return {"status": "not_logged_in"}
+
+    try:
+        result = run_async(check_user())
+        return jsonify(result)
+    except Exception as e: return jsonify({"status": "error"})
+
+@app.route('/api/send_code', methods=['POST'])
+def api_send_code():
+    data = request.json or {}
+    phone = data.get('phone')
+    user_id = data.get('user_id')
+    if not user_id or str(user_id) == '123456': return jsonify({"status": "error", "msg": "Please Open WebApp inside Telegram Bot!"})
+    
+    async def process_send_code():
+        if phone in temp_clients:
+            try: await temp_clients[phone]['client'].disconnect()
+            except: pass
+        client = Client(f"session_{phone}", api_id=API_ID, api_hash=API_HASH, in_memory=True)
+        await client.connect()
+        try:
+            code_info = await client.send_code(phone)
+            temp_clients[phone] = {'client': client, 'hash': code_info.phone_code_hash}
+            return {"status": "ok", "hash": code_info.phone_code_hash}
+        except Exception as e:
+            try: await client.disconnect()
+            except: pass
+            return {"status": "error", "msg": str(e)}
+
+    try:
+        result = run_async(process_send_code())
+        return jsonify(result)
+    except Exception as e: return jsonify({"status": "error", "msg": str(e)})
+
+@app.route('/api/verify_code', methods=['POST'])
+def api_verify_code():
+    data = request.json or {}
+    phone = data.get('phone')
+    user_otp = data.get('otp')
+    user_id = data.get('user_id')
+    
+    if phone not in temp_clients: return jsonify({"status": "error", "msg": "Session expired, request code again!"})
+
+    async def process_verify():
+        temp_data = temp_clients[phone]
+        client = temp_data['client']
+        phone_hash = temp_data['hash']
+        try:
+            await client.sign_in(phone, phone_hash, user_otp)
+            session_string = await client.export_session_string()
+            try: await client.disconnect()
+            except: pass
+            
+            await db_query(lambda: supabase.table('user_sessions').insert({
+                "user_id": user_id, 
+                "phone": phone, 
+                "session_string": session_string
+            }).execute())
+            
+            if phone in temp_clients: del temp_clients[phone]
+            return {"status": "ok"}
+            
+        except SessionPasswordNeeded:
+            try: await client.disconnect()
+            except: pass
+            if phone in temp_clients: del temp_clients[phone]
+            return {"status": "error", "msg": "Two-Step Verification is ON! Please turn it off and try again."}
+        except PhoneCodeInvalid: return {"status": "error", "msg": "Invalid OTP Code!"}
+        except PhoneCodeExpired:
+            try: await client.disconnect()
+            except: pass
+            if phone in temp_clients: del temp_clients[phone]
+            return {"status": "error", "msg": "OTP Expired! Request again."}
+        except Exception as e:
+            try: await client.disconnect()
+            except: pass
+            if phone in temp_clients: del temp_clients[phone]
+            return {"status": "error", "msg": str(e)}
+
+    try:
+        result = run_async(process_verify())
+        return jsonify(result)
+    except Exception as e: return jsonify({"status": "error", "msg": str(e)})
+
+
+async def save_progress(source_id, dest_id, msg_id):
+    try:
+        res = await db_query(lambda: supabase.table('clone_progress').select('id').eq('source_id', source_id).eq('dest_id', dest_id).execute())
+        if res.data:
+            await db_query(lambda: supabase.table('clone_progress').update({'last_copied_id': msg_id}).eq('id', res.data[0]['id']).execute())
+        else:
+            await db_query(lambda: supabase.table('clone_progress').insert({'source_id': source_id, 'dest_id': dest_id, 'last_copied_id': msg_id}).execute())
+    except Exception as e: print(f"Error saving progress: {e}")
+
+
+# ================= SMART AUTO SHARE LOGIC =================
+async def auto_share_loop(client, delay=300):
+    global auto_share_running
     
     try:
-        # ১. ফাইল সেভ
-        with open(input_path, "wb") as buffer:
-            shutil.copyfileobj(file.file, buffer)
+        bot_me = client.me if client.me else await client.get_me()
+        bot_link = f"https://t.me/{bot_me.username}"
+    except:
+        bot_link = "https://t.me/your_bot"
         
-        # ২. সুপার ফাস্ট FFmpeg (Speed optimized)
-        # crf 28 এবং preset superfast ভিডিও এডিটিং দ্রুত করবে
-        cmd = [
-            "ffmpeg", "-i", input_path,
-            "-vf", f"drawtext=text='{watermark_text}':x=w-tw-20:y=h-th-20:fontsize=30:fontcolor=white@0.5",
-            "-c:v", "libx264", "-preset", "superfast", "-crf", "28", "-c:a", "copy", 
-            "-movflags", "+faststart",
-            output_path
-        ]
-        process = await asyncio.create_subprocess_exec(*cmd)
-        await process.wait()
+    caption_text = (
+        f"🔥 <b>New Premium Viral Video Leaked!</b> 🔞\n\n"
+        f"🎬 <b>Watch HD Video Here:</b>\n"
+        f"👉 <b><a href='{bot_link}'>▶️ Click Here to Watch</a></b>\n\n"
+        f"🎁 <b>App এর ভেতর 'Secret Box' ওপেন করে Premium Channel Claim করুন!</b> 👇"
+    )
+    group_markup = InlineKeyboardMarkup([
+        [InlineKeyboardButton("🎬 Watch Full Video Here 🔞", url=bot_link)],
+        [InlineKeyboardButton("🎁 Open Secret Box", url=f"{bot_link}?start=secretbox")]
+    ])
+    
+    while auto_share_running:
+        try:
+            # ১. ডাটাবেজ থেকে সব গ্রুপ আনা এবং অ্যাডমিন পারমিশন চেক করা
+            groups_res = await db_query(lambda: supabase.table('groups').select('group_id').execute())
+            eligible_groups = []
+            
+            if groups_res.data:
+                for g in groups_res.data:
+                    gid = g['group_id']
+                    try:
+                        member = await client.get_chat_member(gid, "me")
+                        # চেক করবে বট অ্যাডমিন কিনা এবং মেসেজ ডিলিট করার পারমিশন আছে কিনা
+                        if member.status in [enums.ChatMemberStatus.ADMINISTRATOR, enums.ChatMemberStatus.OWNER]:
+                            if member.privileges and member.privileges.can_delete_messages:
+                                eligible_groups.append(gid)
+                    except FloodWait as e:
+                        await asyncio.sleep(e.value + 1)
+                    except Exception:
+                        pass # বট গ্রুপে না থাকলে বা ব্যান খেলে স্কিপ করবে
+            
+            if not eligible_groups:
+                await asyncio.sleep(60) # কোনো এলিজিবল গ্রুপ না পেলে ১ মিনিট ওয়েট করবে
+                continue
+
+            # ২. স্টোরেজ থেকে ৫টি ভিডিও নেওয়া (dest_id = -1 ব্যবহার করা হচ্ছে গ্লোবাল ট্র্যাকিং এর জন্য)
+            progress_res = await db_query(lambda: supabase.table('clone_progress').select('last_copied_id').eq('source_id', STORAGE_CHANNEL_ID).eq('dest_id', -1).execute())
+            last_checked_id = progress_res.data[0]['last_copied_id'] if progress_res.data else 0
+
+            batch_to_send = []
+            current_id = last_checked_id + 1
+            highest_valid_id = last_checked_id
+
+            chunk_ids = list(range(current_id, current_id + 50))
+            try:
+                msgs = await client.get_messages(STORAGE_CHANNEL_ID, chunk_ids)
+                valid_messages_found = False
+
+                for msg in msgs:
+                    if not auto_share_running: break
+                    if msg and not getattr(msg, "empty", False):
+                        valid_messages_found = True
+                        highest_valid_id = msg.id 
+                        if msg.video or (msg.document and msg.document.mime_type and "video" in msg.document.mime_type) or msg.photo:
+                            batch_to_send.append(msg)
+                            if len(batch_to_send) == 5:
+                                break
+            except FloodWait as e:
+                await asyncio.sleep(e.value + 1)
+                continue
+            except Exception as e:
+                await asyncio.sleep(5)
+                continue
+
+            if not batch_to_send:
+                if valid_messages_found and highest_valid_id > last_checked_id:
+                    await save_progress(STORAGE_CHANNEL_ID, -1, highest_valid_id)
+                await asyncio.sleep(30) 
+                continue
+                
+            # ৩. এলিজিবল গ্রুপগুলোতে ৫টি ভিডিও সেন্ড করা
+            sent_message_ids = {} # { group_id: [msg_id1, msg_id2...] }
+            
+            for msg in batch_to_send:
+                if not auto_share_running: break
+                for gid in eligible_groups:
+                    try:
+                        sent = await msg.copy(gid, caption=caption_text, reply_markup=group_markup)
+                        if gid not in sent_message_ids:
+                            sent_message_ids[gid] = []
+                        sent_message_ids[gid].append(sent.id)
+                        await asyncio.sleep(1) # ফ্লাডওয়েট এড়াতে স্লিপ
+                    except FloodWait as e:
+                        await asyncio.sleep(e.value + 1)
+                    except Exception:
+                        pass
+                        
+            if highest_valid_id > last_checked_id:
+                await save_progress(STORAGE_CHANNEL_ID, -1, highest_valid_id)
+            
+            # ৪. ৫ মিনিট অপেক্ষা করা
+            for _ in range(delay):
+                if not auto_share_running: break
+                await asyncio.sleep(1)
+                
+            # ৫. সব গ্রুপ থেকে পাঠানো ভিডিওগুলো ডিলিট করা
+            if auto_share_running:
+                for gid, msg_ids in sent_message_ids.items():
+                    try:
+                        await client.delete_messages(gid, msg_ids)
+                        await asyncio.sleep(1)
+                    except FloodWait as e:
+                        await asyncio.sleep(e.value + 1)
+                        await client.delete_messages(gid, msg_ids)
+                    except Exception:
+                        pass
+                        
+        except Exception as e:
+            print(f"Auto-share loop error: {e}")
+            await asyncio.sleep(5)
+
+@bot.on_message(filters.command("startshare") & filters.private & filters.user(ADMIN_IDS))
+async def start_sharing_cmd(client, message):
+    global auto_share_running, auto_share_task
+    args = message.command
+    
+    if len(args) > 1 and args[1].lower() == "reset":
+        await save_progress(STORAGE_CHANNEL_ID, -1, 0)
+        await message.reply("🔄 <b>Progress reset!</b> The bot will now start sharing from the 1st video.")
+        
+    if auto_share_running:
+        return await message.reply("⚠️ <b>Auto-share is already running!</b>\nUse `/stopshare` to stop it first.")
+    
+    auto_share_running = True
+    auto_share_task = asyncio.create_task(auto_share_loop(client, 300))
+    await message.reply(f"✅ <b>Smart Auto-sharing started!</b>\n\n📌 <b>Target:</b> All Groups (Admin + Delete Permitted)\n📦 <b>Batch Size:</b> 5 videos\n⏱ <b>Interval:</b> 5 minutes\n\n<i>Bot will scan all groups, verify permissions, send 5 videos, and delete them automatically!</i>", parse_mode=enums.ParseMode.HTML)
+
+@bot.on_message(filters.command("stopshare") & filters.private & filters.user(ADMIN_IDS))
+async def stop_sharing_cmd(client, message):
+    global auto_share_running
+    if not auto_share_running:
+        return await message.reply("⚠️ Auto-share is not currently running.")
+    
+    auto_share_running = False
+    await message.reply("🛑 <b>Auto-sharing stopped successfully!</b>", parse_mode=enums.ParseMode.HTML)
+
+
+# ================= HELP COMMAND =================
+@bot.on_message(filters.command("help"))
+async def help_command(client, message):
+    user_id = message.from_user.id
+    is_admin = user_id in ADMIN_IDS
+
+    help_text = "🛠 **Bot Commands Help Menu**\n\n"
+
+    if is_admin:
+        help_text += "👑 **Admin Commands:**\n"
+        help_text += "👉 `/stats` - Check total users, videos, and groups.\n"
+        help_text += "👉 `/broadcast` - Send a message to all bot users.\n"
+        help_text += "👉 `/clone <source_id> <dest_id>` - Clone videos from one group to another.\n"
+        help_text += "👉 `/cloneall` - Mass clone all videos from all connected groups to Target Group.\n"
+        help_text += "👉 `/sendto <group_id>` - Reply to a media to forward it directly to a specific group.\n"
+        help_text += "👉 `/upload <telegram/byse>` - Change video upload server (Local or byse.sx).\n"
+        help_text += "👉 `/blur <percentage>` - Enable video/photo blur (e.g., `/blur 60`). Send `/blur 0` to disable.\n"
+        help_text += "👉 `/clean` - Scan and kick dead/deleted users from the Premium channel.\n"
+        help_text += "👉 `/startshare` - Start auto-forwarding 5 videos to the target group every 5 minutes.\n"
+        help_text += "👉 `/startshare reset` - Reset auto-forward progress to the first video.\n"
+        help_text += "👉 `/stopshare` - Stop the auto-forwarding process.\n\n"
+
+    help_text += "👤 **User Commands:**\n"
+    help_text += "👉 `/start` - Start the bot and get the WebApp link.\n"
+
+    await message.reply(help_text, parse_mode=enums.ParseMode.MARKDOWN)
+# ================================================
+
+
+# ================= TELEGRAM BOT COMMANDS =================
+@bot.on_message(filters.command("start"))
+async def start(client, message):
+    if message.chat.type != enums.ChatType.PRIVATE:
+        try:
+            bot_me = client.me if client.me else await client.get_me()
+            bot_link = f"https://t.me/{bot_me.username}"
+            markup = InlineKeyboardMarkup([[InlineKeyboardButton("🎬 Watch Videos Now", url=bot_link)]])
+            await message.reply("🔥 **Watch Premium Viral Videos for FREE!**\n\n👉 Click the button below to watch:", reply_markup=markup)
+        except Exception: pass
+        return
+
+    try:
+        user_id = message.from_user.id
+        first_name = message.from_user.first_name
+        args = message.command
+        referrer_id = None
+        is_secret = False
+        
+        # Secret Box থেকে এসেছে কিনা সেটা চেক করা হচ্ছে
+        if len(args) > 1:
+            if args[1] == "secretbox":
+                is_secret = True
+            else:
+                try: referrer_id = int(args[1])
+                except ValueError: pass
 
-        # ৩. থাম্বনেইল জেনারেট করা (ভিডিওর ২য় সেকেন্ড থেকে)
-        thumb_cmd = ["ffmpeg", "-i", output_path, "-ss", "00:00:02", "-vframes", "1", "-y", thumb_path]
-        thumb_proc = await asyncio.create_subprocess_exec(*thumb_cmd)
-        await thumb_proc.wait()
+        user_check = await db_query(lambda: supabase.table('referrals').select('*').eq('user_id', user_id).execute())
+        
+        if not user_check.data: 
+            try:
+                await db_query(lambda: supabase.table('referrals').insert({'user_id': user_id, 'referral_count': 0, 'referrer_id': referrer_id if referrer_id != user_id else None}).execute())
+                if referrer_id and referrer_id != user_id:
+                    ref_data = await db_query(lambda: supabase.table('referrals').select('referral_count').eq('user_id', referrer_id).execute())
+                    if ref_data.data:
+                        new_count = ref_data.data[0]['referral_count'] + 1
+                        await db_query(lambda: supabase.table('referrals').update({'referral_count': new_count}).eq('user_id', referrer_id).execute())
+                        try:
+                            safe_name = first_name.replace('<', '').replace('>', '') if first_name else "User"
+                            success_msg = f"🎉 <b>Congratulations!</b>\n\n👤 <b>{safe_name}</b> has joined using your link!\n📈 Total Invites: <b>{new_count}</b>\n\n<i>Go to the Web App to check unlocked videos!</i>"
+                            markup = InlineKeyboardMarkup([[InlineKeyboardButton("🎬 Check Unlocked Videos", web_app=WebAppInfo(url=WEB_APP_URL))]])
+                            await client.send_message(referrer_id, success_msg, parse_mode=enums.ParseMode.HTML, reply_markup=markup)
+                        except Exception: pass
+            except Exception as db_err:
+                print(f"Error handling referral DB entry: {db_err}")
 
-        # ৪. আপলোড করার চেষ্টা (থাম্বনেইল সহ)
-        if os.path.exists(output_path):
+        bot_me = client.me if client.me else await client.get_me()
+        
+        # যদি Secret Box-এর বাটনে ক্লিক করে আসে, তাহলে স্পেশাল মেসেজ এবং বাটন দেওয়া হবে
+        if is_secret:
+            btn_text = "🎁 Open App & Claim Premium"
+            welcome_text = (f"Hello <b>{first_name}</b>! 👋\n\n🎁 <b>Secret Box Unlock করুন!</b>\nনিচের বাটনে ক্লিক করে অ্যাপটি ওপেন করুন এবং আপনার নাম্বার ভেরিফাই করে Premium Channel Claim করে নিন!\n\n👇 <b>Click the button below:</b>")
+        else:
+            btn_text = "🔥 Play Viral Videos 🔞"
+            welcome_text = (f"Hello <b>{first_name}</b>! 👋\n\n🎁 <b>Welcome to Video Unlocker Pro!</b>\nHere you can watch premium leaked and viral videos completely for FREE.\n\n👇 <b>Click the button below to Open App:</b>")
+            
+        markup = InlineKeyboardMarkup([
+            [InlineKeyboardButton(btn_text, web_app=WebAppInfo(url=WEB_APP_URL))],
+            [InlineKeyboardButton("📢 Add to Group", url=f"https://t.me/{bot_me.username}?startgroup=true")]
+        ])
+        
+        await message.reply(welcome_text, parse_mode=enums.ParseMode.HTML, reply_markup=markup)
+    except Exception as e: print(f"Start error: {e}")
+
+@bot.on_message(filters.new_chat_members)
+async def bot_added_to_group(client, message):
+    me = client.me
+    if getattr(me, "id", None) is None:
+        try: me = await client.get_me()
+        except: return
+
+    for member in message.new_chat_members:
+        if member.id == me.id:
             try:
-                with open(output_path, "rb") as v, open(thumb_path, "rb") as t:
-                    await bot_instance.send_video(
-                        chat_id=chat_id, 
-                        video=v, 
-                        thumbnail=t, # থাম্বনেইল সেট করা হলো
-                        caption=f"✅ {watermark_text}", 
-                        supports_streaming=True
-                    )
-                return JSONResponse({"status": "success"})
+                await db_query(lambda: supabase.table('groups').upsert({
+                    'group_id': message.chat.id,
+                    'group_name': message.chat.title,
+                    'added_by': message.from_user.id if message.from_user else None
+                }).execute())
+                
+                group_name = message.chat.title
+                admin_msg = f"✅ <b>Bot added to a new group!</b>\n\n📌 <b>Group Name:</b> {group_name}\n🆔 <b>ID:</b> <code>{message.chat.id}</code>"
+                for admin_id in ADMIN_IDS:
+                    try: await client.send_message(chat_id=admin_id, text=admin_msg, parse_mode=enums.ParseMode.HTML)
+                    except: pass
+
             except Exception as e:
-                # আপলোড ফেইল হলে মেইন বটকে ভিডিও এবং থাম্বনেইল ফেরত পাঠানো (Rare case)
-                return FileResponse(output_path, media_type="video/mp4")
+                print(f"Error handling new group logic: {e}")
+
+@bot.on_message(filters.command("sendto") & filters.private & filters.user(ADMIN_IDS))
+async def send_to_specific_group(client, message):
+    if not message.reply_to_message:
+        return await message.reply("❌ <b>Please reply to a message, photo, or video that you want to send.</b>\n\nExample: `/sendto -1001234567890`")
+    
+    args = message.command
+    if len(args) < 2: return await message.reply("❌ <b>Group ID missing!</b>\n\nCorrect format:\n`/sendto -1003973566529`")
         
+    try:
+        group_id = int(args[1])
+        status = await message.reply("⏳ Sending message to group...")
+        await message.reply_to_message.copy(chat_id=group_id)
+        await status.edit_text(f"✅ <b>Successfully sent to Group ID:</b> <code>{group_id}</code>", parse_mode=enums.ParseMode.HTML)
+    except Exception as e:
+        await status.edit_text(f"❌ <b>Failed to send!</b>\nError: {e}", parse_mode=enums.ParseMode.HTML)
+
+
+# ================= MASS CLONE FROM ALL GROUPS =================
+async def clone_all_background(client, dest_id, status_msg):
+    try:
+        groups_res = await db_query(lambda: supabase.table('groups').select('group_id').execute())
+        if not groups_res.data:
+            return await status_msg.edit_text("❌ <b>No groups found in database.</b>", parse_mode=enums.ParseMode.HTML)
+            
+        groups = [g['group_id'] for g in groups_res.data]
+        total_groups = len(groups)
+        
+        await status_msg.edit_text(f"✅ Found <b>{total_groups}</b> groups.\n🚀 Starting mass cloning from oldest videos...", parse_mode=enums.ParseMode.HTML)
+        
+        total_copied = 0
+        
+        for index, source_id in enumerate(groups, 1):
+            if source_id == dest_id or source_id == STORAGE_CHANNEL_ID or source_id == PREMIUM_CHANNEL_ID:
+                continue 
+                
+            try:
+                await status_msg.edit_text(f"⏳ <b>Cloning Group {index}/{total_groups}</b>\nID: <code>{source_id}</code>\nTotal Copied So Far: <b>{total_copied}</b>", parse_mode=enums.ParseMode.HTML)
+            except Exception: pass
+            
+            progress_res = await db_query(lambda: supabase.table('clone_progress').select('last_copied_id').eq('source_id', source_id).eq('dest_id', dest_id).execute())
+            last_copied_id = progress_res.data[0]['last_copied_id'] if progress_res.data else None
+            
+            video_ids = []
+            retries = 3
+            while retries > 0:
+                try:
+                    async for msg in client.search_messages(source_id, filter=enums.MessagesFilter.VIDEO):
+                        if last_copied_id and msg.id <= last_copied_id: continue
+                        video_ids.append(msg.id)
+                        if len(video_ids) % 100 == 0: await asyncio.sleep(0.1)
+                    break
+                except FloodWait as e:
+                    await asyncio.sleep(e.value + 1)
+                except Exception as e:
+                    err_msg = str(e).lower()
+                    if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg:
+                        retries -= 1
+                        video_ids = []
+                        await asyncio.sleep(5)
+                    else:
+                        break
+                        
+            if not video_ids:
+                continue
+                
+            video_ids.reverse() 
+            
+            for msg_id in video_ids:
+                copy_retries = 3
+                copy_success = False
+                while copy_retries > 0:
+                    try:
+                        await client.copy_message(chat_id=dest_id, from_chat_id=source_id, message_id=msg_id)
+                        total_copied += 1
+                        await save_progress(source_id, dest_id, msg_id)
+                        copy_success = True
+                        break
+                    except FloodWait as e:
+                        await asyncio.sleep(e.value + 2)
+                    except Exception as e:
+                        err_msg = str(e).lower()
+                        if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg:
+                            copy_retries -= 1
+                            await asyncio.sleep(3)
+                        else:
+                            break
+                await asyncio.sleep(2.5)
+                
+        await status_msg.edit_text(f"🎉 <b>Mass Cloning Completely Finished!</b>\n\nTotal Videos Copied from all groups: <b>{total_copied}</b>", parse_mode=enums.ParseMode.HTML)
+        
+    except Exception as e:
+        try: await status_msg.edit_text(f"❌ <b>Cloning Error:</b> {e}", parse_mode=enums.ParseMode.HTML)
+        except: pass
+
+@bot.on_message(filters.command("cloneall") & filters.private & filters.user(ADMIN_IDS))
+async def start_mass_cloning(client, message):
+    dest_id = -1003798479478 
+    status_msg = await message.reply(f"⏳ <b>Initializing mass cloning task to <code>{dest_id}</code>...</b>", parse_mode=enums.ParseMode.HTML)
+    asyncio.create_task(clone_all_background(client, dest_id, status_msg))
+
+
+async def clone_videos_background(client, source_id, dest_id, status_msg):
+    try:
+        progress_res = await db_query(lambda: supabase.table('clone_progress').select('last_copied_id').eq('source_id', source_id).eq('dest_id', dest_id).execute())
+        
+        last_copied_id = None
+        if progress_res.data:
+            last_copied_id = progress_res.data[0]['last_copied_id']
+            await status_msg.edit_text(f"⏳ <b>Resuming clone task...</b>\nFound previous progress. Resuming after video ID <code>{last_copied_id}</code>...\nFetching video list from <code>{source_id}</code>...", parse_mode=enums.ParseMode.HTML)
+        else:
+            await status_msg.edit_text(f"⏳ <b>Cloning started!</b>\nFetching video list from <code>{source_id}</code>...\n<i>This might take a few minutes if the group has many videos.</i>", parse_mode=enums.ParseMode.HTML)
+        
+        video_ids = []
+        retries = 5
+        while retries > 0:
+            try:
+                if not client.is_connected: await client.connect()
+                async for msg in client.search_messages(source_id, filter=enums.MessagesFilter.VIDEO):
+                    if last_copied_id and msg.id <= last_copied_id: continue
+                    video_ids.append(msg.id)
+                    if len(video_ids) % 200 == 0: await asyncio.sleep(0.1)
+                break 
+            except Exception as e:
+                err_msg = str(e).lower()
+                if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg or "reset" in err_msg:
+                    retries -= 1
+                    video_ids = [] 
+                    await status_msg.edit_text(f"⚠️ <b>Network issue detected!</b>\nRetrying in 10 seconds... (Attempts left: {retries})\nError: <code>{e}</code>", parse_mode=enums.ParseMode.HTML)
+                    await asyncio.sleep(10)
+                else: raise e 
+        
+        if not video_ids:
+            if last_copied_id: return await status_msg.edit_text("🎉 <b>All videos are already cloned!</b>\nNo new videos found in the source group.", parse_mode=enums.ParseMode.HTML)
+            else: return await status_msg.edit_text("❌ <b>No videos found in the source group!</b>\n(Make sure the bot is an admin with read history permission in that group).", parse_mode=enums.ParseMode.HTML)
+            
+        video_ids.reverse() 
+        total = len(video_ids)
+        
+        if last_copied_id: await status_msg.edit_text(f"✅ Found <b>{total}</b> new videos to clone.\n🚀 Resuming background cloning from oldest to newest...", parse_mode=enums.ParseMode.HTML)
+        else: await status_msg.edit_text(f"✅ Found <b>{total}</b> videos.\n🚀 Background cloning started from oldest to newest...", parse_mode=enums.ParseMode.HTML)
+        
+        success = 0
+        failed = 0
+        
+        for index, msg_id in enumerate(video_ids, 1):
+            copy_success = False
+            copy_retries = 3
+            while copy_retries > 0:
+                try:
+                    if not client.is_connected: await client.connect()
+                    await client.copy_message(chat_id=dest_id, from_chat_id=source_id, message_id=msg_id)
+                    success += 1
+                    await save_progress(source_id, dest_id, msg_id)
+                    copy_success = True
+                    break 
+                except FloodWait as e: await asyncio.sleep(e.value + 2)
+                except Exception as e:
+                    err_msg = str(e).lower()
+                    if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg or "reset" in err_msg:
+                        copy_retries -= 1
+                        await asyncio.sleep(5) 
+                    else: break 
+            
+            if not copy_success: failed += 1
+            if index % 20 == 0 or index == total:
+                try: await status_msg.edit_text(f"⏳ <b>Cloning in progress... (Background)</b>\n\nTotal Videos to Copy: <b>{total}</b>\n✅ Copied: <b>{success}</b>\n❌ Failed: <b>{failed}</b>\nLast Video ID: <code>{msg_id}</code>", parse_mode=enums.ParseMode.HTML)
+                except: pass
+            
+            await asyncio.sleep(2.5)
+            
+        await status_msg.edit_text(f"🎉 <b>Cloning Completely Finished!</b>\n\nSource: <code>{source_id}</code>\nTotal Copied: <b>{total}</b>\n✅ Successfully Copied: <b>{success}</b>\n❌ Failed: <b>{failed}</b>", parse_mode=enums.ParseMode.HTML)
+        
+    except Exception as e:
+        try: await status_msg.edit_text(f"❌ <b>Cloning Error:</b> {e}", parse_mode=enums.ParseMode.HTML)
+        except: pass
+
+@bot.on_message(filters.command("clone") & filters.private & filters.user(ADMIN_IDS))
+async def start_cloning(client, message):
+    args = message.command
+    if len(args) != 3: return await message.reply("❌ <b>Invalid format!</b>\n\nUse: `/clone <Source_Group_ID> <Destination_Group_ID>`\nExample: `/clone -100123456789 -100987654321`", parse_mode=enums.ParseMode.HTML)
+        
+    try:
+        source_id = int(args[1])
+        dest_id = int(args[2])
+    except ValueError: return await message.reply("❌ Chat IDs must be numbers.")
+        
+    status_msg = await message.reply("⏳ Initializing cloning task...", parse_mode=enums.ParseMode.HTML)
+    asyncio.create_task(clone_videos_background(client, source_id, dest_id, status_msg))
+
+
+@bot.on_message(filters.command("upload") & filters.private & filters.user(ADMIN_IDS))
+async def set_upload_mode(client, message):
+    global upload_mode
+    args = message.command
+    if len(args) > 1:
+        mode = args[1].lower()
+        if mode in ["telegram", "tg", "local"]:
+            upload_mode = "telegram"
+            await message.reply("✅ <b>Upload server set to: Telegram</b>\nVideos will be uploaded to your own channel and streamed via Hugging Face.")
+        elif mode in ["byse", "byse.sx", "external"]:
+            upload_mode = "byse"
+            await message.reply("✅ <b>Upload server set to: Byse.sx</b>\nVideos will be uploaded to Byse.sx and streamed via their player.")
+        else: await message.reply("❌ <b>Invalid server!</b> Use `/upload telegram` or `/upload byse`.")
+    else:
+        await message.reply(f"📌 <b>Current Upload Server:</b> <code>{upload_mode.upper()}</code>\n\nTo change, use:\n👉 `/upload telegram` (Storage Channel Stream)\n👉 `/upload byse` (Byse.sx third-party player)")
+
+@bot.on_message(filters.command("blur") & filters.private & filters.user(ADMIN_IDS))
+async def set_blur_state(client, message):
+    try:
+        args = message.text.split()
+        if len(args) > 1 and args[1].lower() in ['0', '0%', 'off', 'cancel']:
+            if message.chat.id in admin_states:
+                admin_states[message.chat.id].pop("blur_percent", None)
+                admin_states[message.chat.id].pop("clear_percent", None)
+            await message.reply("✅ <b>Blur mode is disabled!</b>\nUploaded videos will no longer be blurred, only watermarked as before.", parse_mode=enums.ParseMode.HTML)
+            return
+
+        match = re.search(r'/blur\s+(\d+)%?(?:\s+(\d+)%?)?', message.text, re.IGNORECASE)
+        if match:
+            percent = int(match.group(1))
+            clear_percent = int(match.group(2)) if match.group(2) else 0
+
+            if percent == 0:
+                if message.chat.id in admin_states:
+                    admin_states[message.chat.id].pop("blur_percent", None)
+                    admin_states[message.chat.id].pop("clear_percent", None)
+                await message.reply("✅ <b>Blur mode is disabled!</b>", parse_mode=enums.ParseMode.HTML)
+                return
+
+            if message.chat.id not in admin_states: admin_states[message.chat.id] = {}
+            admin_states[message.chat.id]["blur_percent"] = percent
+            admin_states[message.chat.id]["clear_percent"] = clear_percent
+            
+            clear_msg = f"and the top <b>{clear_percent}%</b> part will remain clear." if clear_percent > 0 else "The entire photo/video will be blurred."
+            reply_text = f"✅ <b>Blur set to: {percent}%</b>\n📌 {clear_msg}\n\nThis will be applied to all future uploads.\n<i>(To disable, send /blur 0)</i>"
+            await message.reply(reply_text, parse_mode=enums.ParseMode.HTML)
+        else: await message.reply("❌ <b>Invalid command!</b>\nCorrect format: `/blur 60` or `/blur 60 20`")
+    except Exception as e: print(e)
+
+def upload_file_sync(upload_url, file_path, api_key):
+    try:
+        with open(file_path, 'rb') as f:
+            res = requests.post(upload_url, data={'key': api_key}, files={'file': f}, timeout=900)
+            return res.json() if res.status_code == 200 else {}
+    except Exception: return {}
+
+@bot.on_message((filters.video | filters.animation | filters.photo | filters.document) & filters.private & filters.user(ADMIN_IDS))
+async def handle_media_upload(client, message):
+    global upload_mode
+    state = admin_states.get(message.chat.id, {})
+    if state.get("step") == "broadcast":
+        await process_broadcast(client, message)
+        return
+
+    is_video = message.video or (message.document and message.document.mime_type and "video" in message.document.mime_type)
+    is_animation = message.animation or (message.document and message.document.mime_type and "gif" in message.document.mime_type)
+    is_photo = message.photo or (message.document and message.document.mime_type and "image" in message.document.mime_type)
+
+    if not (is_video or is_animation or is_photo): return
+
+    media_type = "video" if (is_video or is_animation) else "photo"
+    has_blur_caption = message.caption and "/blur" in message.caption.lower()
+    is_persistent_blur = bool(state.get("blur_percent"))
+
+    if media_type == "photo" and not (has_blur_caption or is_persistent_blur):
+        status = await message.reply("⏳ Saving thumbnail...")
+        try:
+            local_path = await message.download()
+            def upload_to_supabase():
+                with open(local_path, 'rb') as f: file_bytes = f.read()
+                file_name = f"thumb_{int(time.time())}.jpg"
+                supabase.storage.from_('thumbnails').upload(file_name, file_bytes, {"content-type": "image/jpeg"})
+                return supabase.storage.from_('thumbnails').get_public_url(file_name)
+
+            direct_link = await asyncio.to_thread(upload_to_supabase)
+            if os.path.exists(local_path): os.remove(local_path)
+            await status.edit_text(f"✅ <b>Thumbnail saved successfully!</b>\n\n<code>{direct_link}</code>", parse_mode=enums.ParseMode.HTML)
+        except Exception as e: await status.edit_text(f"⚠️ Upload Error: {e}")
+        return
+
+    raw_caption = message.caption or ""
+    blur_match = re.search(r'/blur\s+(\d+)%?(?:\s+(\d+)%?)?', raw_caption, re.IGNORECASE)
+    
+    is_blur = False
+    blur_percent = 0
+    clear_percent = 0
+    clean_caption = raw_caption
+
+    if blur_match:
+        is_blur = True
+        blur_percent = int(blur_match.group(1))
+        clear_percent = int(blur_match.group(2)) if match.group(2) else 0
+        clean_caption = re.sub(r'/blur\s*\d+%?(?:\s*\d+%?)?', '', raw_caption, flags=re.IGNORECASE).strip()
+    elif state.get("blur_percent"):
+        is_blur = True
+        blur_percent = state["blur_percent"]
+        clear_percent = state.get("clear_percent", 0)
+
+    is_large_video = False
+    if media_type == "video":
+        media = get_media_obj(message)
+        duration = media.duration if media and hasattr(media, 'duration') and media.duration else 0
+        file_size = media.file_size if media and hasattr(media, 'file_size') and media.file_size else 0
+        MAX_DURATION = 7200 
+        MAX_SIZE = 1900 * 1024 * 1024 
+
+        if duration > MAX_DURATION or file_size > MAX_SIZE:
+            is_large_video = True
+            is_blur = False
+
+    status_msg = await message.reply("⏳ <b>Video is too large!</b> Skipping blur..." if is_large_video else "⏳ Downloading media... 0%")
+    bot_me = client.me if client.me else await client.get_me()
+    bot_link = f"https://t.me/{bot_me.username}"
+    
+    original_file, watermarked_file, blurred_file = None, None, None
+    clean_upload_file, telegram_file, embed_link = None, None, None
+    
+    last_update_time = time.time()
+    async def download_progress(current, total):
+        nonlocal last_update_time
+        now = time.time()
+        if now - last_update_time >= 4.0: 
+            try:
+                percent = (current / total) * 100
+                await status_msg.edit_text(f"⏳ Downloading media... {percent:.1f}%")
+                last_update_time = now
+            except Exception: pass
+    
+    try:
+        original_file = await message.download(progress=download_progress)
+        clean_upload_file = original_file 
+        
+        if media_type == "video" and not is_large_video and ffmpeg_available:
+            await status_msg.edit_text("⏳ Watermarking video... (HD + Superfast Processing)")
+            watermarked_file = f"{original_file}_wm.mp4"
+            
+            has_audio = not (message.animation or (message.document and message.document.mime_type and "gif" in message.document.mime_type))
+            audio_opts = ["-an"] if not has_audio else ["-c:a", "copy"]
+            
+            cmd = [
+                "ffmpeg", "-y", "-i", original_file,
+                "-vf", "drawtext=text='@mxvdo':x=W-tw-20:y=H-th-20:fontsize=22:fontcolor=white@0.7:shadowcolor=black@0.8:shadowx=2:shadowy=2:enable='gte(t,5)'",
+                "-c:v", "libx264", "-preset", "superfast", "-crf", "23",
+                "-pix_fmt", "yuv420p"
+            ] + audio_opts + ["-movflags", "+faststart", watermarked_file]
+            
+            process = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+            await process.communicate()
+            
+            if process.returncode == 0 and os.path.exists(watermarked_file) and os.path.getsize(watermarked_file) > 0: 
+                clean_upload_file = watermarked_file
+
+        storage_msg_id = None
+        if media_type in ["video", "photo"]:
+            if upload_mode == "telegram":
+                await status_msg.edit_text("⏳ Uploading Clean HD video to your storage channel...")
+                
+                thumb_path_storage = None
+                if ffmpeg_available:
+                    thumb_path_storage = f"{original_file}_storage_thumb.jpg"
+                    proc = await asyncio.create_subprocess_exec("ffmpeg", "-y", "-i", clean_upload_file, "-vframes", "1", thumb_path_storage, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+                    await proc.communicate()
+                    if not os.path.exists(thumb_path_storage): 
+                        thumb_path_storage = None
+                    
+                media = get_media_obj(message)
+                vid_duration = media.duration if media and hasattr(media, 'duration') and media.duration else 0
+                vid_width = media.width if media and hasattr(media, 'width') and media.width else 0
+                vid_height = media.height if media and hasattr(media, 'height') and media.height else 0
+
+                sent_to_channel = await client.send_video(
+                    chat_id=STORAGE_CHANNEL_ID, 
+                    video=clean_upload_file, 
+                    caption=f"Backup of video uploaded by Admin. File: {os.path.basename(clean_upload_file)}",
+                    duration=vid_duration,
+                    width=vid_width,
+                    height=vid_height,
+                    thumb=thumb_path_storage
+                )
+                storage_msg_id = sent_to_channel.id
+                
+                stream_link = f"{BACKEND_URL}/stream/{storage_msg_id}"
+                download_link = f"{BACKEND_URL}/download/{storage_msg_id}"
+                embed_link = stream_link
+            else:
+                await status_msg.edit_text("⏳ Uploading Clean HD video to byse.sx server...")
+                api_endpoint = "https://api.byse.sx/upload/server"
+                loop = asyncio.get_event_loop()
+                response = await loop.run_in_executor(None, lambda: requests.get(api_endpoint, params={'key': BYSE_API_KEY}, timeout=30))
+                result = response.json()
+                
+                if result.get('status') == 200:
+                    upload_res = await loop.run_in_executor(None, upload_file_sync, result.get('result'), clean_upload_file, BYSE_API_KEY)
+                    if upload_res.get('status') == 200 and 'files' in upload_res and len(upload_res['files']) > 0:
+                        file_status = upload_res['files'][0].get('status', '')
+                        if "not allowed" in str(file_status).lower():
+                            await status_msg.edit_text(f"❌ byse.sx rejected the file: <code>{file_status}</code>", parse_mode=enums.ParseMode.HTML)
+                            return
+                        file_code = upload_res['files'][0].get('filecode')
+                        if file_code: embed_link = f"https://bysesayeveum.com/e/{file_code}"
+                
+                if not embed_link:
+                    await status_msg.edit_text("❌ Uploaded to byse.sx but Embed Link not found.")
+                    return
+                
+                stream_link = embed_link
+                download_link = embed_link
+
+        if is_large_video:
+            admin_cap = f"✅ <b>Success! (Large Video)</b>\n\n🔗 <b>Embed Link (Clean HD):</b>\n<code>{embed_link or 'N/A'}</code>\n\n📌 <i>Broadcast skipped due to large file size.</i>"
+            await client.send_video(message.chat.id, message.video.file_id, caption=admin_cap, parse_mode=enums.ParseMode.HTML)
+            await status_msg.delete()
+            return  
+
+        telegram_file = clean_upload_file
+        if is_blur and not is_large_video and ffmpeg_available:
+            await status_msg.edit_text(f"⏳ Applying {blur_percent}% blur for Telegram broadcast...")
+            radius = max(2, min(20, int((blur_percent / 100.0) * 30)))
+            ext = "jpg" if media_type == "photo" else "mp4"
+            blurred_file = f"{original_file}_blurred.{ext}"
+            
+            if clear_percent > 0:
+                clear_ratio = clear_percent / 100.0
+                ff_filter = ["-filter_complex", f"[0:v]split[v1][v2];[v2]boxblur={radius}:1[blurred];[v1]crop=iw:ih*{clear_ratio}:0:0[top];[blurred][top]overlay=0:0[vout]", "-map", "[vout]"]
+                if media_type == "video":
+                    ff_filter.extend(["-map", "0:a?"])
+            else:
+                ff_filter = ["-vf", f"boxblur={radius}:1"]
+            
+            has_audio = not (message.animation or (message.document and message.document.mime_type and "gif" in message.document.mime_type))
+            audio_opts_blur = ["-an"] if not has_audio else []
+            
+            if media_type == "photo": 
+                cmd_blur = ["ffmpeg", "-y", "-i", clean_upload_file] + ff_filter + [blurred_file]
+            else: 
+                cmd_blur = ["ffmpeg", "-y", "-i", clean_upload_file] + ff_filter + ["-c:v", "libx264", "-preset", "superfast", "-crf", "23", "-pix_fmt", "yuv420p"] + audio_opts_blur + ["-movflags", "+faststart", blurred_file]
+            
+            process_blur = await asyncio.create_subprocess_exec(*cmd_blur, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+            await process_blur.communicate()
+            
+            if process_blur.returncode == 0 and os.path.exists(blurred_file) and os.path.getsize(blurred_file) > 0: 
+                telegram_file = blurred_file
+
+        await status_msg.edit_text("⏳ Preparing to broadcast to groups...")
+        if media_type == "video":
+            caption_text = (
+                f"🔥 <b>New Premium Viral Video Leaked!</b> 🔞\n\n"
+                f"🎬 <b>Watch HD Video Here:</b>\n"
+                f"👉 <b><a href='{embed_link if is_blur else bot_link}'>▶️ Click Here to Watch</a></b>\n\n"
+                f"🎁 <b>App এর ভেতর 'Secret Box' ওপেন করে Premium Channel Claim করুন!</b> 👇"
+            )
+        else:
+            caption_text = (
+                f"{clean_caption}\n\n🎁 <b>App এর ভেতর 'Secret Box' ওপেন করে Premium Channel Claim করুন!</b> 👇"
+                if clean_caption else 
+                f"🔥 <b>New Premium Viral Content!</b> 🔞\n\n🎬 <b>Watch HD Video Here:</b>\n👉 <b><a href='{bot_link}'>▶️ Click Here to Watch</a></b>\n\n🎁 <b>App এর ভেতর 'Secret Box' ওপেন করে Premium Channel Claim করুন!</b> 👇"
+            )
+            
+        group_markup = InlineKeyboardMarkup([
+            [InlineKeyboardButton("🎬 Watch Full Video Here 🔞", url=bot_link)],
+            [InlineKeyboardButton("🎁 Open Secret Box", url=f"{bot_link}?start=secretbox")]
+        ])
+        
+        admin_cap = (
+            f"✅ <b>Upload and Processing Complete!</b>\n\n"
+            f"🎬 <b>Stream/Watch Online Link:</b>\n<code>{stream_link}</code>\n\n"
+            f"📥 <b>Direct Download Link:</b>\n<code>{download_link}</code>"
+        )
+        
+        thumb_path = None
+        if media_type == "video" and ffmpeg_available:
+            thumb_path = f"{original_file}_thumb.jpg"
+            proc = await asyncio.create_subprocess_exec("ffmpeg", "-y", "-i", telegram_file, "-vframes", "1", thumb_path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+            await proc.communicate()
+            if not os.path.exists(thumb_path): thumb_path = None
+
+        try:
+            if media_type == "photo":
+                sent_to_admin = await client.send_photo(message.chat.id, telegram_file, caption=admin_cap, parse_mode=enums.ParseMode.HTML)
+            else:
+                media = get_media_obj(message)
+                vid_duration = media.duration if media and hasattr(media, 'duration') and media.duration else 0
+                vid_width = media.width if media and hasattr(media, 'width') and media.width else 0
+                vid_height = media.height if media and hasattr(media, 'height') and media.height else 0
+                
+                sent_to_admin = await client.send_video(
+                    message.chat.id, 
+                    telegram_file, 
+                    caption=admin_cap, 
+                    parse_mode=enums.ParseMode.HTML, 
+                    duration=vid_duration, 
+                    width=vid_width, 
+                    height=vid_height, 
+                    thumb=thumb_path
+                )
+            
+            await save_progress(message.chat.id, 0, sent_to_admin.id)
+            
+        except Exception as e:
+            await message.reply(f"❌ Failed to send final file to you: {e}")
+            return
+            
+        tg_file_id = get_msg_file_id(sent_to_admin)
+        if not tg_file_id:
+            tg_file_id = get_msg_file_id(message)
+            
+        try: await status_msg.delete()
+        except: pass
+        
+        try:
+            groups_res = await db_query(lambda: supabase.table('groups').select('group_id').execute())
+        except Exception as db_err:
+            await message.reply(f"⚠️ Failed to fetch groups from database: {db_err}")
+            return
+            
+        group_ids = [g['group_id'] for g in groups_res.data]
+        success_count, fail_count = 0, 0
+        
+        for gid in set(group_ids):
+            retries = 3
+            while retries > 0:
+                try:
+                    if not client.is_connected: await client.connect()
+                    if media_type == "photo": await client.send_photo(gid, tg_file_id, caption=caption_text, parse_mode=enums.ParseMode.HTML, reply_markup=group_markup)
+                    else: await client.send_video(gid, tg_file_id, caption=caption_text, parse_mode=enums.ParseMode.HTML, reply_markup=group_markup)
+                    success_count += 1
+                    break
+                except FloodWait as e:
+                    await asyncio.sleep(e.value + 1)
+                except Exception as e:
+                    err_msg = str(e).lower()
+                    if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg or "network" in err_msg:
+                        retries -= 1
+                        await asyncio.sleep(3)
+                    else:
+                        fail_count += 1
+                        break
+            await asyncio.sleep(1.5)
+            
+        retries = 3
+        while retries > 0:
+            try:
+                if not client.is_connected: await client.connect()
+                await message.reply(f"📢 <b>Broadcast Complete!</b>\n\n✅ Success: {success_count} groups\n❌ Failed: {fail_count} groups", parse_mode=enums.ParseMode.HTML)
+                break
+            except FloodWait as e:
+                await asyncio.sleep(e.value + 1)
+            except Exception as e:
+                err_msg = str(e).lower()
+                if "disconnect" in err_msg or "connection" in err_msg or "timeout" in err_msg:
+                    retries -= 1
+                    await asyncio.sleep(3)
+                else: break
+        
+    except Exception as e: 
+        try: await message.reply(f"⚠️ Error during upload/broadcast: {str(e)}")
+        except: pass
+    finally:
+        for f in [original_file, watermarked_file, blurred_file, f"{original_file}_thumb.jpg" if original_file else None, f"{original_file}_storage_thumb.jpg" if original_file else None]:
+            if f and os.path.exists(f):
+                try: os.remove(f)
+                except: pass
+
+@bot.on_message(filters.command(["stats", "users"]) & filters.private & filters.user(ADMIN_IDS))
+async def bot_stats(client, message):
+    try:
+        users = await db_query(lambda: supabase.table('referrals').select('user_id', count='exact').execute())
+        videos = await db_query(lambda: supabase.table('videos').select('*', count='exact').execute())
+        groups = await db_query(lambda: supabase.table('groups').select('group_id', count='exact').execute())
+        await message.reply(f"📊 <b>Bot Stats:</b>\n👥 Users: <code>{users.count or 0}</code>\n🎬 Videos: <code>{videos.count or 0}</code>\n📢 Groups: <code>{groups.count or 0}</code>", parse_mode=enums.ParseMode.HTML)
+    except Exception as e: print(e)
+
+@bot.on_message(filters.command("broadcast") & filters.private & filters.user(ADMIN_IDS))
+async def broadcast_command(client, message):
+    admin_states[message.chat.id] = {"step": "broadcast"}
+    await message.reply("📢 Send the message you want to broadcast. (Send /cancel to abort)")
+
+async def process_broadcast(client, message):
+    text = message.text or message.caption
+    if text == '/cancel':
+        admin_states.pop(message.chat.id, None)
+        return await message.reply("❌ Cancelled.")
+    
+    await message.reply("⏳ Broadcast started...")
+    admin_states.pop(message.chat.id, None)
+    
+    try:
+        all_users, start, step = [], 0, 1000
+        while True:
+            res = await db_query(lambda: supabase.table('referrals').select('user_id').range(start, start + step - 1).execute())
+            if not res.data: break
+            all_users.extend(res.data)
+            start += step
+
+        success, failed = 0, 0
+        for u in all_users:
+            try:
+                await message.copy(chat_id=u['user_id'])
+                success += 1
+                await asyncio.sleep(0.15) 
+            except FloodWait as e:
+                await asyncio.sleep(e.value + 1)
+                try:
+                    await message.copy(chat_id=u['user_id'])
+                    success += 1
+                except Exception: failed += 1
+            except Exception: failed += 1
+                
+        await message.reply(f"✅ Broadcast Complete!\nSuccess: {success}\nFailed: {failed}")
+    except Exception as e: print(e)
+
+@bot.on_message(filters.command("clean") & filters.private & filters.user(ADMIN_IDS))
+async def manual_clean_channel(client, message):
+    await message.reply("⏳ <b>Starting channel cleanup...</b>\nChecking database users to verify active sessions. This might take a while.")
+    try:
+        kicked, checked = 0, 0
+        res_users = await db_query(lambda: supabase.table('referrals').select('user_id').execute())
+        if not res_users.data:
+            return await message.reply("❌ No users found in database!")
+            
+        user_ids = [u['user_id'] for u in res_users.data]
+        
+        for user_id in set(user_ids):
+            try:
+                chat_member = await bot.get_chat_member(PREMIUM_CHANNEL_ID, user_id)
+                if chat_member.status in [enums.ChatMemberStatus.MEMBER, enums.ChatMemberStatus.RESTRICTED]:
+                    checked += 1
+                    res_session = await db_query(lambda: supabase.table('user_sessions').select('session_string').eq('user_id', user_id).execute())
+                    
+                    is_valid = False
+                    if res_session.data:
+                        session_string = res_session.data[0]['session_string']
+                        temp_client = Client(f"manual_chk_{user_id}", session_string=session_string, api_id=API_ID, api_hash=API_HASH, in_memory=True)
+                        try:
+                            await temp_client.connect()
+                            await temp_client.get_me()
+                            await temp_client.disconnect()
+                            is_valid = True
+                        except (SessionRevoked, AuthKeyUnregistered, UserDeactivated):
+                            try: await temp_client.disconnect()
+                            except: pass
+                            await db_query(lambda: supabase.table('user_sessions').delete().eq('user_id', user_id).execute())
+                        except Exception:
+                            try: await temp_client.disconnect()
+                            except: pass
+                            is_valid = True 
+                    
+                    if not is_valid:
+                        try:
+                            await client.ban_chat_member(PREMIUM_CHANNEL_ID, user_id)
+                            await client.unban_chat_member(PREMIUM_CHANNEL_ID, user_id) 
+                            kicked += 1
+                        except Exception: pass
+                        
+                await asyncio.sleep(1.5) 
+            except FloodWait as e:
+                await asyncio.sleep(e.value + 1)
+            except Exception: pass
+            
+        await message.reply(f"✅ <b>Cleanup Complete!</b>\n\n👥 Members checked: {checked}\n👢 Users Kicked: {kicked}")
+    except Exception as e:
+        await message.reply(f"❌ Error: {e}")
+
+async def auto_clean_channel_loop():
+    await asyncio.sleep(60) 
+    while True:
+        try:
+            res_users = await db_query(lambda: supabase.table('referrals').select('user_id').execute())
+            if res_users.data:
+                user_ids = [u['user_id'] for u in res_users.data]
+                for user_id in set(user_ids):
+                    try:
+                        chat_member = await bot.get_chat_member(PREMIUM_CHANNEL_ID, user_id)
+                        if chat_member.status in [enums.ChatMemberStatus.MEMBER, enums.ChatMemberStatus.RESTRICTED]:
+                            res_session = await db_query(lambda: supabase.table('user_sessions').select('session_string').eq('user_id', user_id).execute())
+                            
+                            is_valid = False
+                            if res_session.data:
+                                session_string = res_session.data[0]['session_string']
+                                temp_client = Client(f"bg_chk_{user_id}", session_string=session_string, api_id=API_ID, api_hash=API_HASH, in_memory=True)
+                                try:
+                                    await temp_client.connect()
+                                    await temp_client.get_me()
+                                    await temp_client.disconnect()
+                                    is_valid = True
+                                except (SessionRevoked, AuthKeyUnregistered, UserDeactivated):
+                                    try: await temp_client.disconnect()
+                                    except: pass
+                                    await db_query(lambda: supabase.table('user_sessions').delete().eq('user_id', user_id).execute())
+                                except Exception:
+                                    try: await temp_client.disconnect()
+                                    except: pass
+                                    is_valid = True 
+                            
+                            if not is_valid:
+                                try:
+                                    await bot.ban_chat_member(PREMIUM_CHANNEL_ID, user_id)
+                                    await bot.unban_chat_member(PREMIUM_CHANNEL_ID, user_id)
+                                except Exception: pass
+                        await asyncio.sleep(2)
+                    except FloodWait as e:
+                        await asyncio.sleep(e.value + 1)
+                    except Exception: pass
+        except Exception as e:
+            print(f"Auto clean error: {e}")
+        await asyncio.sleep(4 * 3600) 
+
+@bot.on_message(filters.private & filters.user(ADMIN_IDS) & ~filters.command(["start", "stats", "users", "broadcast", "png", "addvideo", "blur", "clean", "sendto", "clone", "cloneall", "upload", "startshare", "stopshare", "help"]))
+async def catch_admin_steps(client, message):
+    state = admin_states.get(message.chat.id, {})
+    if state.get("step") == 1:
+        if not message.text: return
+        video_url = message.text.strip()
+        if video_url == "/cancel":
+            admin_states.pop(message.chat.id, None)
+            return await message.reply("❌ Cancelled.")
+            
+        try:
+            await db_query(lambda: supabase.table('videos').insert({"video_url": video_url, "thumbnail_url": state["thumbnail_url"], "needed_ref": state["needed_ref"]}).execute())
+            await message.reply("🎉 Video added successfully!")
+        except Exception as e: print(e)
+        finally: admin_states.pop(message.chat.id, None)
+    elif state.get("step") == "broadcast":
+        await process_broadcast(client, message)
+
+# ==========================================
+# REDIS QUEUE WATERMARK MULTIPROCESSING
+# ==========================================
+REDIS_URL = os.environ.get("REDIS_URL_1")
+
+processing_owners = set()
+MAX_CONCURRENT_VIDEOS = 5 
+
+async def process_single_video(task, redis_client):
+    global processing_owners
+    clone_token = task['clone_token']
+    owner_id = task['owner_id']
+    message_id = task['message_id']
+    wm_text = task['watermark_text']
+    task_id = f"{owner_id}_{message_id}"
+    
+    raw_video_path = None
+    watermarked_path = None
+    thumb_path = f"thumb_{task_id}.jpg"
+    clone_client = None
+    
+    print(f"▶️ [WM] Task Started for User: {owner_id}, Msg: {message_id}")
+    
+    try:
+        clone_client = Client(f"clone_wm_{task_id}", bot_token=clone_token, api_id=API_ID, api_hash=API_HASH, in_memory=True)
+        await clone_client.start()
+        print(f"▶️ [WM] Clone Client Started Successfully")
+        
+        msg = await clone_client.get_messages(owner_id, message_id)
+        if not msg or not (msg.video or msg.document):
+            raise Exception("Video not found or deleted by user.")
+            
+        status_msg = await clone_client.send_message(owner_id, "⏳ <b>ভিডিও প্রসেসিং শুরু হয়েছে...</b>", parse_mode=enums.ParseMode.HTML)
+        
+        raw_video_path = await clone_client.download_media(msg)
+        print(f"▶️ [WM] Original Video Downloaded: {raw_video_path}")
+        
+        await clone_client.edit_message_text(owner_id, status_msg.id, "⏳ <b>ওয়াটারমার্ক যুক্ত করা হচ্ছে... (High Speed CPU Mode)</b>", parse_mode=enums.ParseMode.HTML)
+        
+        watermarked_path = f"wm_{task_id}.mp4"
+        
+        cmd = [
+            "ffmpeg", "-y", "-i", raw_video_path,
+            "-vf", f"drawtext=text='{wm_text}':x=W-tw-20:y=H-th-20:fontsize=22:fontcolor=white@0.7:shadowcolor=black@0.8:shadowx=2:shadowy=2:enable='gte(t,5)'",
+            "-c:v", "libx264", "-preset", "ultrafast", "-crf", "26", "-threads", "4",
+            "-c:a", "copy", "-movflags", "+faststart", watermarked_path
+        ]
+        
+        process = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+        stdout, stderr = await process.communicate()
+        
+        if process.returncode != 0 or not os.path.exists(watermarked_path) or os.path.getsize(watermarked_path) == 0:
+            raise Exception("FFmpeg processing failed! The video format might be unsupported.")
+            
+        print(f"▶️ [WM] FFmpeg Processing Complete!")
+        
+        thumb_cmd = ["ffmpeg", "-y", "-i", watermarked_path, "-vframes", "1", thumb_path]
+        t_proc = await asyncio.create_subprocess_exec(*thumb_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+        await t_proc.communicate()
+        if not os.path.exists(thumb_path) or os.path.getsize(thumb_path) == 0:
+            thumb_path = None
+        
+        await clone_client.edit_message_text(owner_id, status_msg.id, "⏳ <b>ফাইনাল ভিডিও স্টোরেজে আপলোড হচ্ছে...</b>", parse_mode=enums.ParseMode.HTML)
+        
+        async def backup_raw_video():
+            try:
+                await bot.send_video(chat_id=STORAGE_CHANNEL_ID, video=raw_video_path, caption=f"Original Backup for Clone Owner {owner_id}")
+            except Exception: pass
+        asyncio.create_task(backup_raw_video())
+        
+        wm_sent = await bot.send_video(
+            chat_id=STORAGE_CHANNEL_ID,
+            video=watermarked_path,
+            caption=f"Watermarked Backup for Clone Owner {owner_id}",
+            thumb=thumb_path
+        )
+        
+        wm_storage_id = wm_sent.id
+        stream_link = f"{BACKEND_URL}/stream/{wm_storage_id}"
+        download_link = f"{BACKEND_URL}/download/{wm_storage_id}"
+        
+        final_caption = (
+            f"✅ <b>Watermark Successfully Added!</b>\n\n"
+            f"🎬 <b>Stream/Watch Online Link:</b>\n<code>{stream_link}</code>\n\n"
+            f"📥 <b>Direct Download Link:</b>\n<code>{download_link}</code>"
+        )
+        
+        await clone_client.send_video(
+            chat_id=owner_id,
+            video=watermarked_path,
+            caption=final_caption,
+            parse_mode=enums.ParseMode.HTML,
+            thumb=thumb_path
+        )
+        
+        await clone_client.delete_messages(owner_id, status_msg.id)
+        
+    except Exception as e:
+        print(f"❌ [WM] Process Error: {e}")
+        try:
+            if clone_client and clone_client.is_connected:
+                await clone_client.send_message(owner_id, f"❌ <b>Error processing video:</b> {e}", parse_mode=enums.ParseMode.HTML)
+        except Exception as ex: 
+            print(f"❌ [WM] Failed to send error msg to user: {ex}")
+        
+    finally:
+        if raw_video_path and os.path.exists(raw_video_path): os.remove(raw_video_path)
+        if watermarked_path and os.path.exists(watermarked_path): os.remove(watermarked_path)
+        if thumb_path and os.path.exists(thumb_path): os.remove(thumb_path)
+        
+        if clone_client and clone_client.is_connected:
+            await clone_client.stop()
+            
+        await redis_client.delete(f"wm_processing:{owner_id}")
+        processing_owners.discard(owner_id) 
+        print(f"▶️ [WM] Task Cleaned Up and Client Stopped")
+
+async def watermark_processor_loop():
+    global processing_owners
+    
+    if not REDIS_URL:
+        print("⚠️ REDIS_URL missing in Hugging Space! Watermark feature disabled.")
+        return
+        
+    try:
+        redis_client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
+        await redis_client.ping()
+        print("💧 Super-Fast Redis Watermark Processor Started!")
+    except Exception as e:
+        print(f"❌ Redis Connection Failed in Hugging Face: {e}")
+        return
+    
+    try:
+        async for key in redis_client.scan_iter("wm_processing:*"):
+            await redis_client.delete(key)
+    except: pass
+
+    while True:
+        try:
+            if len(processing_owners) >= MAX_CONCURRENT_VIDEOS:
+                await asyncio.sleep(2)
+                continue 
+            
+            result = await redis_client.rpop("watermark_task_queue")
+            
+            if result:
+                print(f"📥 [WM] Received New Video Task from Queue!")
+                task = json.loads(result)
+                owner_id = task['owner_id']
+                
+                processing_owners.add(owner_id)
+                asyncio.create_task(process_single_video(task, redis_client))
+            else:
+                await asyncio.sleep(2)
+                    
+        except Exception as e:
+            print(f"❌ [WM] Loop Error: {e}")
+            await asyncio.sleep(2)
+            
+def run_flask(): app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7860)), threaded=True) 
+
+async def main():
+    try:
+        await bot.start()
+        print("🤖 Pyrogram Bot & Real Session API is running!")
+        asyncio.create_task(auto_clean_channel_loop())
+        asyncio.create_task(watermark_processor_loop()) 
+        await idle()
     except Exception as e:
-        return JSONResponse({"status": "error", "message": str(e)}, status_code=500)
+        print(f"❌ Failed to start Bot: {e}")
+        while True: await asyncio.sleep(3600)
     finally:
-        # ক্লিনআপ
-        for path in [input_path, thumb_path]:
-            if os.path.exists(path): os.remove(path)
+        try: await bot.stop()
+        except: pass
 
 if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=7860)
+    threading.Thread(target=run_flask, daemon=True).start()
+    main_loop.run_until_complete(main())