from __future__ import annotations import asyncio import html import json import logging import os import re import time import warnings from datetime import datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo from telegram import InlineKeyboardButton, InlineKeyboardMarkup, ReplyKeyboardMarkup, Update from telegram.warnings import PTBUserWarning from telegram.ext import ( Application, CallbackQueryHandler, CommandHandler, ContextTypes, ConversationHandler, MessageHandler, filters, ) from config import Settings, ensure_data_dirs, load_settings from auto_uploader import upload_token_file from jwt_generator import generate_jwts_for_accounts from scheduler import BotScheduler, parse_schedule from storage import AccountFile, Storage, atomic_write_json, safe_slug, validate_accounts logging.basicConfig( format="%(message)s", level=logging.WARNING, ) logger = logging.getLogger("jwt-bot") logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("telegram").setLevel(logging.WARNING) logging.getLogger("telegram.ext").setLevel(logging.WARNING) logging.getLogger("apscheduler").setLevel(logging.WARNING) warnings.filterwarnings( "ignore", message=r"If 'per_message=False', 'CallbackQueryHandler' will not be tracked.*", category=PTBUserWarning, ) UPLOAD_FILE, UPLOAD_SERVER, UPLOAD_PURPOSE, UPLOAD_META = range(4) BD_TZ = ZoneInfo("Asia/Dhaka") ICON_BOT = "\U0001f48e" ICON_UPLOAD = "\U0001f4e4" ICON_FILE = "\U0001f5c2" ICON_JWT = "\u26a1" ICON_OK = "\u2705" ICON_FAIL = "\u274c" ICON_QUEUE = "\U0001f501" ICON_LOCK = "\U0001f510" ICON_DELETE = "\U0001f5d1" ICON_DOT_GREEN = "\U0001f7e2" ICON_DOT_BLUE = "\U0001f535" ICON_DOT_PURPLE = "\U0001f7e3" ICON_CLOCK = "\u23f1" BTN_UPLOAD = f"{ICON_UPLOAD} Upload" BTN_GENERATE = f"{ICON_JWT} Generate" BTN_FILES = f"{ICON_FILE} Files" BTN_STATUS = f"{ICON_BOT} Status" BTN_SCHEDULES = f"{ICON_CLOCK} Schedules" BTN_HELP = f"{ICON_DOT_BLUE} ? Help" def premium_icon(name: str, fallback: str) -> str: emoji_id = os.getenv(f"PREMIUM_EMOJI_{name}_ID", "").strip() if not emoji_id: return fallback return f'{fallback}' def premium_keyboard() -> ReplyKeyboardMarkup: return ReplyKeyboardMarkup( [ [BTN_UPLOAD, BTN_GENERATE], [BTN_FILES, BTN_STATUS], [BTN_SCHEDULES, BTN_HELP], ], resize_keyboard=True, is_persistent=True, ) def help_topic_keyboard() -> InlineKeyboardMarkup: return InlineKeyboardMarkup( [ [ InlineKeyboardButton(f"{ICON_UPLOAD} Upload", callback_data="help:upload"), InlineKeyboardButton(f"{ICON_JWT} Generate", callback_data="help:generate"), ], [ InlineKeyboardButton(f"{ICON_FILE} Files", callback_data="help:files"), InlineKeyboardButton(f"{ICON_CLOCK} Schedule", callback_data="help:schedule"), ], [ InlineKeyboardButton(f"{ICON_QUEUE} Queue", callback_data="help:queue"), InlineKeyboardButton(f"{ICON_DELETE} Advanced", callback_data="help:advanced"), ], ] ) def schedule_menu_keyboard(interval: str = "7h") -> InlineKeyboardMarkup: return InlineKeyboardMarkup( [ [ InlineKeyboardButton(f"{ICON_CLOCK} bd like100 every {interval}", callback_data=f"schedquick:bd:like100:{interval}"), ], [ InlineKeyboardButton(f"{ICON_CLOCK} bd like200 every {interval}", callback_data=f"schedquick:bd:like200:{interval}"), ], [ InlineKeyboardButton(f"{ICON_DOT_GREEN} View schedules", callback_data="schedquick:list"), ], ] ) def h(value: Any) -> str: return html.escape(str(value), quote=False) def is_admin(update: Update, settings: Settings) -> bool: user = update.effective_user return bool(user and user.id in settings.admin_ids) def is_allowed_chat(update: Update, settings: Settings) -> bool: chat = update.effective_chat if not chat: return False return not settings.allowed_group_ids or chat.id in settings.allowed_group_ids async def guard(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool: settings: Settings = context.application.bot_data["settings"] if not is_allowed_chat(update, settings): if update.effective_message: await update.effective_message.reply_text(f"{ICON_LOCK} This chat is not allowed.") return False if not is_admin(update, settings): if update.effective_message: await update.effective_message.reply_text(f"{ICON_LOCK} Admin only.") return False return True def storage_from(context: ContextTypes.DEFAULT_TYPE) -> Storage: return context.application.bot_data["storage"] def settings_from(context: ContextTypes.DEFAULT_TYPE) -> Settings: return context.application.bot_data["settings"] def scheduler_from_app(application: Application) -> BotScheduler | None: return application.bot_data.get("scheduler") def schedule_line(item: dict[str, Any], scheduler: BotScheduler | None = None) -> str: next_run = scheduler.next_run_text(int(item["id"])) if scheduler else "not loaded" return ( f"{ICON_CLOCK} #{item['id']} " f"{h(item['server'])} {h(item['purpose'])} " f"{h(item['mode'])} {h(item['value'])}\n" f"{ICON_JWT} next refresh: {h(next_run)}" ) def schedule_menu_for(context: ContextTypes.DEFAULT_TYPE) -> InlineKeyboardMarkup: settings = settings_from(context) return schedule_menu_keyboard(f"{settings.token_refresh_hours}h") def format_duration(seconds: float) -> str: total = max(0, int(seconds)) days, remainder = divmod(total, 86400) hours, remainder = divmod(remainder, 3600) minutes, secs = divmod(remainder, 60) parts: list[str] = [] if days: parts.append(f"{days}d") if hours or parts: parts.append(f"{hours}h") if minutes or parts: parts.append(f"{minutes}m") parts.append(f"{secs}s") return " ".join(parts) def command_success_text(settings: Settings, server: str, purpose: str) -> str: command = f"/gennext {server} {purpose}" if command == settings.command_like100: return settings.success_text_like100 if command == settings.command_like200: return settings.success_text_like200 if purpose == "like100": return settings.success_text_like100 if purpose == "like200": return settings.success_text_like200 return "jwt generate success" def generated_at_bd() -> str: return datetime.now(BD_TZ).isoformat(timespec="seconds") def plain_value(value: Any) -> str: if value is None: return "none" return str(value) def generation_contract_message(header: str, metadata: dict[str, Any], upload_result: Any = None) -> str: lines = [ header, f"endpoint: {plain_value(metadata['endpoint'])}", f"server: {plain_value(metadata['server'])}", f"purpose: {plain_value(metadata['purpose'])}", f"file_index: {plain_value(metadata['file_index'])}", f"file_total: {plain_value(metadata['file_total'])}", f"file_id: {plain_value(metadata['file_id'])}", f"file_name: {plain_value(metadata['file_name'])}", f"accounts: {plain_value(metadata['accounts'])}", f"success: {plain_value(metadata['success'])}", f"failed: {plain_value(metadata['failed'])}", f"queue_mode: {plain_value(metadata['queue_mode'])}", f"next_file_index: {plain_value(metadata['next_file_index'])}", f"generated_at: {plain_value(metadata['generated_at'])}", ] if upload_result: lines.append(f"upload_status: {'success' if upload_result.success else 'failed'}") lines.append(f"upload_path: {plain_value(upload_result.remote_path)}") if upload_result.success: lines.append(f"upload_tokens: {plain_value(upload_result.tokens)}") lines.append(f"upload_size: {plain_value(upload_result.size)}") lines.append(f"upload_saved_to: {plain_value(upload_result.saved_to)}") else: lines.append(f"upload_reason: {plain_value(upload_result.reason)}") return "\n".join(lines) def last_generation_message(metadata: dict[str, Any]) -> str: return "\n".join( f"{key}: {plain_value(metadata.get(key))}" for key in ( "endpoint", "server", "purpose", "file_index", "file_total", "file_id", "file_name", "accounts", "success", "failed", "queue_mode", "next_file_index", "generated_at", ) ) def file_summary(file: AccountFile) -> str: return ( f"{ICON_FILE} #{file.id} {h(file.display_name)}\n" f"{ICON_DOT_BLUE} server: {h(file.server)} | purpose: {h(file.purpose)}\n" f"{ICON_QUEUE} accounts: {file.account_count} | status: {h(file.status)} | order: {file.order_index}\n" f"{ICON_LOCK} file: {h(file.filename)}" ) def pretty_button(text: str, icon: str) -> str: return f"{icon} {text}" async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return await update.message.reply_text( f"{premium_icon('BOT', ICON_BOT)} JWT Generator Bot\n" f"{premium_icon('STAR', ICON_DOT_PURPLE)} Easy control panel\n\n" f"{ICON_UPLOAD} Upload - account JSON add korte\n" f"{ICON_JWT} Generate - JWT banate\n" f"{ICON_FILE} Files - upload list dekhte\n" f"{ICON_BOT} Status - bot er obostha dekhte\n\n" f"{ICON_DOT_BLUE} Details bujhte ? Help button চাপুন.", parse_mode="HTML", reply_markup=premium_keyboard(), ) async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return await update.message.reply_text( f"{ICON_DOT_BLUE} JWT Bot Help\n" "Kon topic bujhte chan, niche button চাপুন.\n\n" f"{ICON_UPLOAD} Upload - account file add\n" f"{ICON_JWT} Generate - JWT banano\n" f"{ICON_FILE} Files - uploaded file list\n" f"{ICON_CLOCK} Schedule - auto generate\n" f"{ICON_QUEUE} Queue - next file system\n" f"{ICON_DELETE} Advanced - file control commands", parse_mode="HTML", reply_markup=premium_keyboard(), ) await update.message.reply_text( "Help topic select করুন:", reply_markup=help_topic_keyboard(), ) def help_topic_text(topic: str) -> str: topics = { "upload": ( f"{ICON_UPLOAD} Upload Guide\n\n" "কাজ: guest account JSON file bot-e save kore.\n\n" "Steps:\n" "1. Upload button চাপুন\n" "2. JSON file পাঠান\n" "3. server select করুন, যেমন bd\n" "4. purpose select করুন, যেমন like100\n" "5. name/order দিন, যেমন file_2 2\n\n" "JSON format:\n" "[{\"uid\":\"4782815656\",\"password\":\"pass\"}]" ), "generate": ( f"{ICON_JWT} Generate Guide\n\n" "কাজ: uploaded account file theke JWT token banay.\n\n" "Easy way: Generate button চাপুন.\n\n" "Command:\n" "/gennext bd like100\n" "/gennext bd like200\n" "/genfile 1\n" "/genall bd like100\n\n" "Last result dekhte:\n" "/lastgen bd like100\n\n" "Output file format:\n" "[{\"token\":\"JWT_HERE\"}]" ), "files": ( f"{ICON_FILE} Files Guide\n\n" "কাজ: kon file upload ache seta dekhay.\n\n" "Button: Files\n\n" "Commands:\n" "/files - sob file\n" "/files bd - bd server file\n" "/files bd like100 - bd like100 file\n\n" "File id diye later generate/delete/move kora jay." ), "schedule": ( f"{ICON_CLOCK} Schedule Guide\n\n" "কাজ: auto JWT generate.\n\n" "Commands:\n" "/schedule bd like100 every 30m\n" "/schedule bd like100 daily 04:00\n" "/schedule all daily 03:30\n" "/schedules\n" "/delschedule 1" ), "queue": ( f"{ICON_QUEUE} Queue Guide\n\n" "Same server + purpose e multiple file thakle bot order moto file use kore.\n\n" "Example:\n" "file 1 -> first /gennext\n" "file 2 -> second /gennext\n" "file 3 -> third /gennext\n\n" "Reset korte:\n" "/resetqueue bd like100" ), "advanced": ( f"{ICON_DELETE} Advanced File Control\n\n" "Regular use-e dorkar nai, problem hole use korben.\n\n" "/enablefile 1 - file active\n" "/disablefile 1 - file off\n" "/deletefile 1 - file delete\n" "/movefile 1 2 - order change\n\n" f"{ICON_LOCK} Password Telegram message/output-e show hoy na." ), } return topics.get(topic, f"{ICON_FAIL} Help topic not found.") async def help_topic_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: query = update.callback_query await query.answer() if not is_admin(update, settings_from(context)): await query.edit_message_text(f"{ICON_LOCK} Admin only.") return topic = query.data.split(":", 1)[1] await query.edit_message_text( help_topic_text(topic), parse_mode="HTML", reply_markup=help_topic_keyboard(), ) async def premium_button_router(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.message or not update.message.text: return text = update.message.text.strip() if text == BTN_UPLOAD: await upload_start(update, context) elif text == BTN_GENERATE: await generate_menu(update, context) elif text == BTN_FILES: if not await guard(update, context): return files = storage_from(context).list_files() if not files: await update.message.reply_text(f"{ICON_FILE} No files found.", reply_markup=premium_keyboard()) return await update.message.reply_text( "\n\n".join(file_summary(file) for file in files[:20]), parse_mode="HTML", reply_markup=premium_keyboard(), ) elif text == BTN_STATUS: await status(update, context) elif text == BTN_SCHEDULES: await schedules_command(update, context) elif text == BTN_HELP: await help_command(update, context) async def status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: ping_started = time.perf_counter() if not await guard(update, context): return data = storage_from(context).status() started_monotonic = float(context.application.bot_data.get("started_monotonic", time.monotonic())) uptime = format_duration(time.monotonic() - started_monotonic) ping_ms = (time.perf_counter() - ping_started) * 1000 lines = [ f"{ICON_BOT} Bot status", f"{ICON_DOT_GREEN} ping: {ping_ms:.0f} ms", f"{ICON_CLOCK} uptime: {h(uptime)}", f"{ICON_FILE} files: {data['file_count']}", f"{ICON_LOCK} accounts: {data['account_count']}", ] latest = data["latest"] if latest: lines.extend( [ f"\n{ICON_JWT} Last generation", f"server: {h(latest['server'])}", f"purpose: {h(latest['purpose'])}", f"success: {latest['success_count']}", f"failed: {latest['failed_count']}", f"output: {h(latest['output_filename'])}", ] ) if data["pointers"]: lines.append(f"\n{ICON_QUEUE} Queue pointers") lines.extend( f"{h(row['server'])} {h(row['purpose'])}: {row['next_index']}" for row in data["pointers"] ) schedules = storage_from(context).list_schedules() if schedules: scheduler = scheduler_from_app(context.application) lines.append(f"\n{ICON_CLOCK} Next refresh") lines.extend(schedule_line(item, scheduler) for item in schedules[:5]) await update.message.reply_text("\n".join(lines), parse_mode="HTML", reply_markup=premium_keyboard()) async def lastgen(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return if len(context.args) < 2: await update.message.reply_text("usage: /lastgen ") return server, purpose = context.args[0].lower(), context.args[1].lower() metadata = storage_from(context).get_last_generation(server, purpose) if not metadata: await update.message.reply_text(f"no last generation found for {server} {purpose}") return await update.message.reply_text(last_generation_message(metadata)) async def upload_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: if not await guard(update, context): return ConversationHandler.END context.user_data.pop("upload_accounts", None) context.user_data.pop("upload_server", None) context.user_data.pop("upload_purpose", None) await update.message.reply_text( f"{ICON_UPLOAD} Upload account file\nSend the guest account JSON file now.", parse_mode="HTML", ) return UPLOAD_FILE async def upload_file(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: if not await guard(update, context): return ConversationHandler.END settings = settings_from(context) document = update.message.document if not document: await update.message.reply_text(f"{ICON_FAIL} Please send a JSON document.") return UPLOAD_FILE if document.file_size and document.file_size > settings.max_upload_size: await update.message.reply_text(f"{ICON_FAIL} File too large.") return UPLOAD_FILE if document.file_name and not document.file_name.lower().endswith(".json"): await update.message.reply_text(f"{ICON_FAIL} Only .json files are accepted.") return UPLOAD_FILE tg_file = await document.get_file() raw = bytearray() await tg_file.download_as_bytearray(buf=raw) try: payload = json.loads(raw.decode("utf-8")) accounts = validate_accounts(payload) except Exception as exc: await update.message.reply_text(f"{ICON_FAIL} Invalid JSON: {h(exc)}", parse_mode="HTML") return UPLOAD_FILE context.user_data["upload_accounts"] = accounts keyboard = [ [InlineKeyboardButton(pretty_button(server.upper(), ICON_DOT_GREEN), callback_data=f"upload_server:{server}")] for server in settings.servers ] await update.message.reply_text( f"{ICON_OK} Valid file\nAccounts: {len(accounts)}\nChoose server:", reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="HTML", ) return UPLOAD_SERVER async def upload_server(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: query = update.callback_query await query.answer() settings = settings_from(context) if not is_admin(update, settings): await query.edit_message_text(f"{ICON_LOCK} Admin only.") return ConversationHandler.END server = query.data.split(":", 1)[1] context.user_data["upload_server"] = server keyboard = [ [InlineKeyboardButton(pretty_button(purpose, ICON_DOT_BLUE), callback_data=f"upload_purpose:{purpose}")] for purpose in settings.purposes ] await query.edit_message_text( f"{ICON_DOT_GREEN} server: {h(server)}\nChoose purpose:", reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="HTML", ) return UPLOAD_PURPOSE async def upload_purpose(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: query = update.callback_query await query.answer() settings = settings_from(context) if not is_admin(update, settings): await query.edit_message_text(f"{ICON_LOCK} Admin only.") return ConversationHandler.END purpose = query.data.split(":", 1)[1] context.user_data["upload_purpose"] = purpose await query.edit_message_text( f"{ICON_FILE} File metadata\n" "Send display name and optional order.\n\n" "file_2\n" "file_2 2", parse_mode="HTML", ) return UPLOAD_META async def upload_meta(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: if not await guard(update, context): return ConversationHandler.END accounts = context.user_data.get("upload_accounts") server = context.user_data.get("upload_server") purpose = context.user_data.get("upload_purpose") if not accounts or not server or not purpose: await update.message.reply_text(f"{ICON_FAIL} Upload session expired. Run /upload again.") return ConversationHandler.END parts = update.message.text.strip().split() display_name = parts[0] if parts else f"{server}_{purpose}" order_index = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 0 file = storage_from(context).add_account_file( display_name=display_name, server=server, purpose=purpose, order_index=order_index, accounts=accounts, ) context.user_data.clear() message = ( f"{ICON_OK} Saved\n" f"accounts: {len(accounts)}\n" "split: off\n\n" + file_summary(file) ) await update.message.reply_text(message, parse_mode="HTML") return ConversationHandler.END async def upload_cancel(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: context.user_data.clear() await update.message.reply_text(f"{ICON_FAIL} Upload cancelled.") return ConversationHandler.END async def files_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return args = [arg.lower() for arg in context.args] server = args[0] if len(args) >= 1 else None purpose = args[1] if len(args) >= 2 else None files = storage_from(context).list_files(server, purpose) if not files: await update.message.reply_text(f"{ICON_FILE} No files found.", reply_markup=premium_keyboard()) return chunks = [file_summary(file) for file in files[:20]] if len(files) > 20: chunks.append(f"...and {len(files) - 20} more") await update.message.reply_text("\n\n".join(chunks), parse_mode="HTML", reply_markup=premium_keyboard()) async def generate_file( *, context: ContextTypes.DEFAULT_TYPE, chat_id: int, file: AccountFile, success_header: str | None = None, queue_meta: dict[str, Any] | None = None, ) -> dict[str, Any]: settings = settings_from(context) storage = storage_from(context) accounts = storage.load_accounts(file) output, failures = await generate_jwts_for_accounts( accounts, file.server, debug=settings.debug, ) base = safe_slug(Path(file.filename).stem) output_filename = f"jwt_{file.server}_{file.purpose}_{base}.json" failed_filename = f"failed_{file.server}_{file.purpose}_{base}.json" if failures else None output_path = settings.generated_dir / output_filename atomic_write_json(output_path, output) failed_path = None if failures: failed_path = settings.failed_dir / failed_filename atomic_write_json(failed_path, failures) upload_result = None if settings.auto_upload_enabled and output: upload_result = await upload_token_file( base_url=settings.auto_upload_base_url, file_path=output_path, server=file.server, purpose=file.purpose, timeout=settings.auto_upload_timeout, auth_token=settings.auto_upload_token, ) storage.mark_generation( file_id=file.id, server=file.server, purpose=file.purpose, output_filename=output_filename, failed_filename=failed_filename, success_count=len(output), failed_count=len(failures), ) header = success_header or "jwt generate success" if queue_meta is None: queue_meta = storage.queue_metadata_for_file(file, settings.queue_mode) metadata = { "endpoint": file.purpose, "server": file.server, "purpose": file.purpose, "file_index": queue_meta.get("file_index", "none"), "file_total": queue_meta.get("file_total", 0), "file_id": file.id, "file_name": file.filename, "accounts": len(accounts), "success": len(output), "failed": len(failures), "queue_mode": settings.queue_mode, "next_file_index": queue_meta.get("next_file_index", "none"), "generated_at": generated_at_bd(), } storage.record_last_generation(metadata) summary = generation_contract_message(header, metadata, upload_result) await context.bot.send_message(chat_id=chat_id, text=summary) with output_path.open("rb") as handle: await context.bot.send_document(chat_id=chat_id, document=handle) if failed_path: with failed_path.open("rb") as handle: await context.bot.send_document(chat_id=chat_id, document=handle) return { "file": file, "success": len(output), "failed": len(failures), "output": output_filename, "upload": upload_result, "metadata": metadata, } async def gennext(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return if len(context.args) < 2: await update.message.reply_text(f"{ICON_JWT} Usage: /gennext <server> <purpose>", parse_mode="HTML") return settings = settings_from(context) server, purpose = context.args[0].lower(), context.args[1].lower() queue_result = storage_from(context).next_file_with_meta(server, purpose, settings.queue_mode) if not queue_result: await update.message.reply_text(f"{ICON_FAIL} No enabled queued file found.") return file = queue_result["file"] await update.message.reply_text( f"{ICON_JWT} Generating JWT for #{file.id} {h(file.filename)}...", parse_mode="HTML", ) await generate_file( context=context, chat_id=update.effective_chat.id, file=file, success_header=command_success_text(settings, server, purpose), queue_meta=queue_result, ) async def genfile(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return if not context.args or not context.args[0].isdigit(): await update.message.reply_text(f"{ICON_JWT} Usage: /genfile <file_id>", parse_mode="HTML") return file = storage_from(context).get_file(int(context.args[0])) if not file: await update.message.reply_text(f"{ICON_FAIL} File not found.") return await update.message.reply_text( f"{ICON_JWT} Generating JWT for #{file.id} {h(file.filename)}...", parse_mode="HTML", ) await generate_file( context=context, chat_id=update.effective_chat.id, file=file, success_header=command_success_text(settings_from(context), file.server, file.purpose), ) async def genall(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return if len(context.args) < 2: await update.message.reply_text(f"{ICON_JWT} Usage: /genall <server> <purpose>", parse_mode="HTML") return server, purpose = context.args[0].lower(), context.args[1].lower() files = storage_from(context).enabled_files(server, purpose) if not files: await update.message.reply_text(f"{ICON_FAIL} No enabled files found.") return await update.message.reply_text(f"{ICON_JWT} Generating {len(files)} file(s)...", parse_mode="HTML") storage = storage_from(context) settings = settings_from(context) for file in files: await generate_file( context=context, chat_id=update.effective_chat.id, file=file, success_header=command_success_text(settings, file.server, file.purpose), queue_meta=storage.queue_metadata_for_file(file, settings.queue_mode), ) async def generate_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return settings = settings_from(context) keyboard: list[list[InlineKeyboardButton]] = [] for purpose in ("like100", "like200"): row = [ InlineKeyboardButton( pretty_button(f"{ICON_DOT_BLUE} {server.upper()} {purpose}", ICON_JWT), callback_data=f"gen_next:{server}:{purpose}", ) for server in settings.servers[:3] ] keyboard.append(row) keyboard.append([InlineKeyboardButton(pretty_button(f"{ICON_DOT_GREEN} List files", ICON_FILE), callback_data="gen_files")]) await update.message.reply_text( f"{ICON_BOT} Choose a generation target", reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="HTML", ) async def generate_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: query = update.callback_query await query.answer() if not is_admin(update, settings_from(context)): await query.edit_message_text(f"{ICON_LOCK} Admin only.") return if query.data == "gen_files": files = storage_from(context).list_files() if not files: await query.edit_message_text(f"{ICON_FILE} No files found.") return await query.edit_message_text("\n\n".join(file_summary(file) for file in files[:10]), parse_mode="HTML") return _, server, purpose = query.data.split(":", 2) settings = settings_from(context) queue_result = storage_from(context).next_file_with_meta(server, purpose, settings.queue_mode) if not queue_result: await query.edit_message_text(f"{ICON_FAIL} No enabled queued file found.") return file = queue_result["file"] await query.edit_message_text( f"{ICON_JWT} Generating JWT for #{file.id} {h(file.filename)}...", parse_mode="HTML", ) await generate_file( context=context, chat_id=update.effective_chat.id, file=file, success_header=command_success_text(settings, server, purpose), queue_meta=queue_result, ) async def resetqueue(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return if len(context.args) < 2: await update.message.reply_text(f"{ICON_QUEUE} Usage: /resetqueue <server> <purpose>", parse_mode="HTML") return server, purpose = context.args[0].lower(), context.args[1].lower() storage_from(context).reset_queue(server, purpose) await update.message.reply_text( f"{ICON_QUEUE} Queue reset: {h(server)} {h(purpose)}", parse_mode="HTML", ) async def set_file_status(update: Update, context: ContextTypes.DEFAULT_TYPE, status_value: str) -> None: if not await guard(update, context): return if not context.args or not context.args[0].isdigit(): await update.message.reply_text(f"{ICON_FILE} Usage: /{status_value}file <file_id>", parse_mode="HTML") return ok = storage_from(context).set_file_status(int(context.args[0]), status_value) await update.message.reply_text(f"{ICON_OK} Updated." if ok else f"{ICON_FAIL} File not found.") async def enablefile(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: await set_file_status(update, context, "enabled") async def disablefile(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: await set_file_status(update, context, "disabled") async def deletefile(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return if not context.args or not context.args[0].isdigit(): await update.message.reply_text(f"{ICON_DELETE} Usage: /deletefile <file_id>", parse_mode="HTML") return file_id = int(context.args[0]) keyboard = [[InlineKeyboardButton(pretty_button("Confirm delete", ICON_DELETE), callback_data=f"delete:{file_id}")]] await update.message.reply_text( f"{ICON_DELETE} Confirm delete file #{file_id}", reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="HTML", ) async def delete_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: query = update.callback_query await query.answer() if not is_admin(update, settings_from(context)): await query.edit_message_text(f"{ICON_LOCK} Admin only.") return file_id = int(query.data.split(":", 1)[1]) ok = storage_from(context).delete_file(file_id) await query.edit_message_text(f"{ICON_DELETE} Deleted." if ok else f"{ICON_FAIL} File not found.") async def movefile(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return if len(context.args) < 2 or not context.args[0].isdigit() or not context.args[1].isdigit(): await update.message.reply_text(f"{ICON_QUEUE} Usage: /movefile <file_id> <new_order>", parse_mode="HTML") return ok = storage_from(context).move_file(int(context.args[0]), int(context.args[1])) await update.message.reply_text(f"{ICON_QUEUE} Moved." if ok else f"{ICON_FAIL} File not found.") async def schedule_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return if len(context.args) < 3: await update.message.reply_text( f"{ICON_CLOCK} Usage:\n/schedule bd like100 every 30m\n/schedule bd like100 daily 04:00", parse_mode="HTML", ) return if context.args[0].lower() == "all" and len(context.args) == 3: server, purpose, mode, value = "all", "all", context.args[1].lower(), context.args[2].lower() elif len(context.args) >= 4: server, purpose, mode, value = ( context.args[0].lower(), context.args[1].lower(), context.args[2].lower(), context.args[3].lower(), ) else: await update.message.reply_text( f"{ICON_CLOCK} Usage:\n/schedule bd like100 every 30m\n/schedule all daily 03:30", parse_mode="HTML", ) return try: parse_schedule(mode, value) except ValueError as exc: await update.message.reply_text(f"{ICON_FAIL} {h(exc)}", parse_mode="HTML") return schedule_id = storage_from(context).add_schedule( server=server, purpose=purpose, mode=mode, value=value, chat_id=update.effective_chat.id, ) scheduler: BotScheduler = context.application.bot_data["scheduler"] scheduler.add_runtime_job( { "id": schedule_id, "server": server, "purpose": purpose, "mode": mode, "value": value, "chat_id": update.effective_chat.id, } ) await update.message.reply_text(f"{ICON_CLOCK} Schedule saved: #{schedule_id}", parse_mode="HTML") async def schedule_quick_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: query = update.callback_query await query.answer() if not is_admin(update, settings_from(context)): await query.edit_message_text(f"{ICON_LOCK} Admin only.") return storage = storage_from(context) scheduler = scheduler_from_app(context.application) if query.data == "schedquick:list": schedules = storage.list_schedules() if not schedules: await query.edit_message_text( f"{ICON_CLOCK} No schedules yet\nSet auto refresh from menu:", parse_mode="HTML", reply_markup=schedule_menu_for(context), ) return await query.edit_message_text( "\n\n".join(schedule_line(item, scheduler) for item in schedules), parse_mode="HTML", reply_markup=schedule_menu_for(context), ) return _, server, purpose, interval = query.data.split(":", 3) schedule_id = storage.add_schedule( server=server, purpose=purpose, mode="every", value=interval, chat_id=query.message.chat_id, ) schedule = { "id": schedule_id, "server": server, "purpose": purpose, "mode": "every", "value": interval, "chat_id": query.message.chat_id, } if scheduler: scheduler.add_runtime_job(schedule) await query.edit_message_text( f"{ICON_OK} Auto refresh schedule saved\n" + schedule_line(schedule, scheduler), parse_mode="HTML", reply_markup=schedule_menu_for(context), ) async def schedules_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return schedules = storage_from(context).list_schedules() if not schedules: await update.message.reply_text( f"{ICON_CLOCK} No schedules yet\nSet auto refresh from menu:", parse_mode="HTML", reply_markup=schedule_menu_for(context), ) return scheduler = scheduler_from_app(context.application) await update.message.reply_text( "\n\n".join(schedule_line(item, scheduler) for item in schedules), parse_mode="HTML", reply_markup=schedule_menu_for(context), ) async def delschedule(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await guard(update, context): return if not context.args or not context.args[0].isdigit(): await update.message.reply_text(f"{ICON_DELETE} Usage: /delschedule <id>", parse_mode="HTML") return schedule_id = int(context.args[0]) ok = storage_from(context).delete_schedule(schedule_id) if ok: scheduler: BotScheduler = context.application.bot_data["scheduler"] scheduler.remove_runtime_job(schedule_id) await update.message.reply_text(f"{ICON_DELETE} Deleted." if ok else f"{ICON_FAIL} Schedule not found.") async def run_schedule(context: dict[str, Any], application: Application) -> None: settings: Settings = application.bot_data["settings"] storage: Storage = application.bot_data["storage"] server = str(context["server"]) purpose = str(context["purpose"]) chat_id = int(context["chat_id"]) targets: list[tuple[AccountFile, dict[str, Any] | None]] = [] if server == "all": file_list = storage.list_files() if purpose == "all" else storage.list_files(purpose=purpose) for file in file_list: if file.status == "enabled": targets.append((file, storage.queue_metadata_for_file(file, settings.queue_mode))) else: queue_result = storage.next_file_with_meta(server, purpose, settings.queue_mode) if queue_result: targets.append((queue_result["file"], queue_result)) if not targets: await application.bot.send_message(chat_id=chat_id, text=f"{ICON_FAIL} Scheduled generation found no enabled files.") return fake_context = type("ContextProxy", (), {"application": application, "bot": application.bot})() for file, queue_meta in targets: await generate_file( context=fake_context, chat_id=chat_id, file=file, success_header=command_success_text(settings, file.server, file.purpose), queue_meta=queue_meta, ) async def post_init(application: Application) -> None: async def callback(schedule: dict[str, Any]) -> None: try: await run_schedule(schedule, application) except Exception: logger.exception("scheduled generation failed") storage: Storage = application.bot_data["storage"] scheduler = BotScheduler(storage, callback) application.bot_data["scheduler"] = scheduler scheduler.start() async def post_shutdown(application: Application) -> None: scheduler = application.bot_data.get("scheduler") if scheduler: scheduler.shutdown() def build_app() -> Application: settings = load_settings() ensure_data_dirs(settings) storage = Storage( db_path=settings.db_path, uploads_dir=settings.uploads_dir, generated_dir=settings.generated_dir, failed_dir=settings.failed_dir, ) storage.init_db() app = ( Application.builder() .token(settings.bot_token) .post_init(post_init) .post_shutdown(post_shutdown) .build() ) app.bot_data["settings"] = settings app.bot_data["storage"] = storage app.bot_data["started_monotonic"] = time.monotonic() upload_conv = ConversationHandler( entry_points=[ CommandHandler("upload", upload_start), MessageHandler(filters.Regex("^" + re.escape(BTN_UPLOAD) + "$"), upload_start), ], states={ UPLOAD_FILE: [MessageHandler(filters.Document.ALL, upload_file)], UPLOAD_SERVER: [CallbackQueryHandler(upload_server, pattern=r"^upload_server:")], UPLOAD_PURPOSE: [CallbackQueryHandler(upload_purpose, pattern=r"^upload_purpose:")], UPLOAD_META: [MessageHandler(filters.TEXT & ~filters.COMMAND, upload_meta)], }, fallbacks=[CommandHandler("cancel", upload_cancel)], ) app.add_handler(upload_conv) app.add_handler(CommandHandler("start", start)) app.add_handler(CommandHandler("help", help_command)) app.add_handler(CommandHandler("status", status)) app.add_handler(CommandHandler("lastgen", lastgen)) app.add_handler(CommandHandler("files", files_command)) app.add_handler(CommandHandler("gennext", gennext)) app.add_handler(CommandHandler("generate", generate_menu)) app.add_handler(CommandHandler("genfile", genfile)) app.add_handler(CommandHandler("genall", genall)) app.add_handler(CommandHandler("resetqueue", resetqueue)) app.add_handler(CommandHandler("enablefile", enablefile)) app.add_handler(CommandHandler("disablefile", disablefile)) app.add_handler(CommandHandler("deletefile", deletefile)) app.add_handler(CommandHandler("movefile", movefile)) app.add_handler(CommandHandler("schedule", schedule_command)) app.add_handler(CommandHandler("schedules", schedules_command)) app.add_handler(CommandHandler("delschedule", delschedule)) app.add_handler(CallbackQueryHandler(help_topic_callback, pattern=r"^help:")) app.add_handler(CallbackQueryHandler(schedule_quick_callback, pattern=r"^schedquick:")) app.add_handler(CallbackQueryHandler(generate_callback, pattern=r"^gen_")) app.add_handler(CallbackQueryHandler(delete_callback, pattern=r"^delete:")) app.add_handler( MessageHandler( filters.Regex( "^(" + "|".join( re.escape(item) for item in ( BTN_UPLOAD, BTN_GENERATE, BTN_FILES, BTN_STATUS, BTN_SCHEDULES, BTN_HELP, ) ) + ")$" ), premium_button_router, ) ) return app def main() -> None: try: asyncio.get_running_loop() except RuntimeError: asyncio.set_event_loop(asyncio.new_event_loop()) app = build_app() print("Bot is running...") try: app.run_polling(allowed_updates=Update.ALL_TYPES) except (KeyboardInterrupt, asyncio.CancelledError): print("Bot stopped.") except RuntimeError as exc: if "Event loop stopped before Future completed" in str(exc): print("Bot stopped.") else: raise if __name__ == "__main__": try: main() except KeyboardInterrupt: print("Bot stopped.")