app.py

import os
import time
import queue
import threading
import requests
import asyncio
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)

# ==================== 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"

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
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)

# ==================== 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.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')
    
    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)})

# ================= 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
        
        if len(args) > 1:
            try: referrer_id = int(args[1])
            except ValueError: pass

        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}")

        bot_me = client.me if client.me else await client.get_me()
        markup = InlineKeyboardMarkup([
            [InlineKeyboardButton("🔥 Play Viral Videos 🔞", web_app=WebAppInfo(url=WEB_APP_URL))],
            [InlineKeyboardButton("📢 Add to Group", url=f"https://t.me/{bot_me.username}?startgroup=true")]
        ])
        
        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>")
        await message.reply(welcome_text, parse_mode=enums.ParseMode.HTML, reply_markup=markup)
    except Exception as e: print(f"Start error: {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}")

@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:
                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

                progress_res = await db_query(lambda: supabase.table('clone_progress').select('source_id', 'last_copied_id').eq('dest_id', 0).execute())
                
                if progress_res.data:
                    last_record = progress_res.data[0]
                    src_chat = last_record['source_id']
                    msg_id = last_record['last_copied_id']
                    
                    bot_link = f"https://t.me/{me.username}"
                    caption_text = f"🔥 <b>New Premium Viral Video Leaked!</b> 🔞\n\n🎬 <b>Watch HD Video Here:</b>\n👉 <b><a href='{bot_link}'>▶️ Click Here to Watch</a></b>\n\n👇 <i>Click the button below to open Bot!</i>"
                    group_markup = InlineKeyboardMarkup([[InlineKeyboardButton("🎬 Watch Full Video Here 🔞", url=bot_link)]])
                    
                    await asyncio.sleep(2)
                    await client.copy_message(message.chat.id, src_chat, msg_id, caption=caption_text, reply_markup=group_markup)
            except Exception as e:
                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)

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🎬 <b>Watch HD Video Here:</b>\n👉 <b><a href='{embed_link if is_blur else bot_link}'>▶️ Click Here to Watch</a></b>\n\n👇 <i>Click the button below to open Bot!</i>"
        else:
            caption_text = f"{clean_caption}\n\n👇 <i>Click the button below to open Bot!</i>" 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👇 <i>Click the button below to open Bot!</i>"
        
        group_markup = InlineKeyboardMarkup([[InlineKeyboardButton("🎬 Watch Full Video Here 🔞", url=bot_link)]])
        
        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

        # Auto-retry logic for sending final result to admin
        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
        
        # Load groups safely with auto-retry
        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
        
        # Safe Broadcast Phase
        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)
            
        # Success notification safely
        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", "upload"]))
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
# ==========================================
import json
import redis.asyncio as redis

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}"
    
    # ==============================
    # FIX: আগে ভেরিয়েবল ডিফাইন করা হলো যাতে finally ব্লকে এরর না আসে
    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>অরিজিনাল ভিডিও স্টোরেজে আপলোড হচ্ছে...</b>", parse_mode=enums.ParseMode.HTML)
        
        raw_sent = await bot.send_video(
            chat_id=STORAGE_CHANNEL_ID,
            video=raw_video_path,
            caption=f"Original Backup for Clone Owner {owner_id}"
        )
        raw_storage_id = raw_sent.id
        stream_link = f"{BACKEND_URL}/stream/{raw_storage_id}"
        download_link = f"{BACKEND_URL}/download/{raw_storage_id}"
        
        await clone_client.edit_message_text(owner_id, status_msg.id, "⏳ <b>ওয়াটারমার্ক যুক্ত করা হচ্ছে... (কিছুক্ষণ সময় লাগতে পারে)</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", "superfast", "-crf", "23",
            "-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)
        
        await bot.send_video(
            chat_id=STORAGE_CHANNEL_ID,
            video=watermarked_path,
            caption=f"Watermarked Backup for Clone Owner {owner_id}",
            thumb=thumb_path
        )
        
        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
    
    # Clear stuck locks
    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:
        print(f"❌ Failed to start Bot: {e}")
        while True: await asyncio.sleep(3600)
    finally:
        try: await bot.stop()
        except: pass

if __name__ == "__main__":
    threading.Thread(target=run_flask, daemon=True).start()
    main_loop.run_until_complete(main())
#--- END OF FILE main.py