From e7699df9da24230212179556ba0058ba7840db3c Mon Sep 17 00:00:00 2001 From: arsam Date: Fri, 18 Sep 2026 22:41:17 +0330 Subject: [PATCH] Add bot administration commands --- handlers/admin.py | 737 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 737 insertions(+) create mode 100644 handlers/admin.py diff --git a/handlers/admin.py b/handlers/admin.py new file mode 100644 index 0000000..6fefed7 --- /dev/null +++ b/handlers/admin.py @@ -0,0 +1,737 @@ +""" +Admin & user commands. +Owner-only commands are guarded by is_owner(); auth-user commands by is_authorized(). +""" +from html import escape +from telegram import Update +from telegram.constants import ChatAction, ParseMode +from telegram.ext import ContextTypes +from services import auth, ai_client +from utils import context, chunker, formatter +from config import ( + BOT_VERSION, MAX_TOKENS_DEFAULT, MAX_TOKENS_MIN, MAX_TOKENS_MAX, + DEFAULT_SYSTEM_PROMPT, IMAGE_DAILY_LIMIT, +) + + +# ──────────────────────────────────────────────────────────── +# Helpers +# ──────────────────────────────────────────────────────────── + +def _owner_required(func): + """Decorator: silently skip if caller is not the owner.""" + async def wrapper(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + u = update.effective_user + if not auth.is_owner(u.id, u.username or ""): + return + return await func(update, ctx) + wrapper.__name__ = func.__name__ + return wrapper + + +def _auth_required(func): + """Decorator: silently skip if caller is not authorized (respects public mode).""" + async def wrapper(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + u = update.effective_user + if not auth.is_authorized(u.id, u.username or ""): + return + auth.register_known(u.id) + return await func(update, ctx) + wrapper.__name__ = func.__name__ + return wrapper + + +async def _send_text(update: Update, ctx: ContextTypes.DEFAULT_TYPE, text: str): + await ctx.bot.send_message(chat_id=update.effective_chat.id, text=text) + + +async def _send_html(update: Update, ctx: ContextTypes.DEFAULT_TYPE, text: str): + await ctx.bot.send_message( + chat_id=update.effective_chat.id, + text=text, + parse_mode=ParseMode.HTML, + ) + + +# ──────────────────────────────────────────────────────────── +# General / all-auth commands (12 total) +# ──────────────────────────────────────────────────────────── + +@_auth_required +async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + user = update.effective_user + name = escape(user.first_name or "friend") + await update.message.reply_html( + f"👋 Hey {name}! I'm Puppy (پاپی) — your AI companion.\n\n" + "• Chat with me directly in private\n" + "• In groups: say my name (Puppy / پاپی), @mention me, or reply to me\n" + "• Send photos and I'll analyze them\n" + "Use /help to see all commands. Let's go! 🐾" + ) + + +@_auth_required +async def cmd_ping(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + await update.message.reply_text("🟢 Pong! Puppy is alive and wagging. 🐾") + + +@_auth_required +async def cmd_clear(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + context.clear_history(update.effective_chat.id) + await update.message.reply_text("🧹 Your conversation history has been cleared.") + + +@_auth_required +async def cmd_myhistory(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + chat_id = update.effective_chat.id + summary = context.history_summary(chat_id, n=8) + await update.message.reply_html( + f"📜 Your recent conversation:\n\n
{escape(summary)}
" + ) + + +@_auth_required +async def cmd_setmaxtoken(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args: + await update.message.reply_html( + f"Usage: /setmaxtoken <{MAX_TOKENS_MIN}–{MAX_TOKENS_MAX}>" + ) + return + + try: + n = int(ctx.args[0]) + except ValueError: + await update.message.reply_text("❌ Please provide a number.") + return + + n = max(MAX_TOKENS_MIN, min(MAX_TOKENS_MAX, n)) + context.set_user_setting(update.effective_user.id, "max_tokens", n) + await update.message.reply_html(f"✅ Max reply tokens set to {n}.") + + +@_auth_required +async def cmd_getmaxtoken(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + settings = context.get_user_settings(update.effective_user.id) + await update.message.reply_html( + f"Max tokens: {settings['max_tokens']} (default: {MAX_TOKENS_DEFAULT})" + ) + + +@_auth_required +async def cmd_resettoken(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + context.set_user_setting(update.effective_user.id, "max_tokens", MAX_TOKENS_DEFAULT) + await update.message.reply_html(f"✅ Max tokens reset to {MAX_TOKENS_DEFAULT}.") + + +@_auth_required +async def cmd_setlang(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args or ctx.args[0] not in ("auto", "fa", "en"): + await update.message.reply_html( + "Usage: /setlang <auto|fa|en>" + ) + return + lang = ctx.args[0] + context.set_user_setting(update.effective_user.id, "lang", lang) + label = {"auto": "Auto-detect", "fa": "Persian (فارسی)", "en": "English"}[lang] + await update.message.reply_html(f"✅ Language set to {label}.") + + +@_auth_required +async def cmd_status(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + user_id = update.effective_user.id + settings = context.get_user_settings(user_id) + search_icon = "✅" if settings["search_enabled"] else "❌" + lang_map = {"auto": "Auto-detect", "fa": "Persian", "en": "English"} + history_len = len(context.get_history(update.effective_chat.id)) + + provider = ai_client.provider_status() + await update.message.reply_html( + f"⚙️ Your settings\n\n" + f"Max tokens: {settings['max_tokens']}\n" + f"Web search: {search_icon}\n" + f"Language: {lang_map.get(settings['lang'], settings['lang'])}\n" + f"History messages: {history_len}\n" + f"AI provider: {provider['provider']}\n" + f"AI model: {provider['model']}" + ) + + +@_auth_required +async def cmd_togglesearch(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + user_id = update.effective_user.id + settings = context.get_user_settings(user_id) + new_val = not settings["search_enabled"] + context.set_user_setting(user_id, "search_enabled", new_val) + icon = "✅ enabled" if new_val else "❌ disabled" + await update.message.reply_text(f"Web search is now {icon}.") + + +@_auth_required +async def cmd_search(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args: + await update.message.reply_text("Usage: /search ") + return + + query = " ".join(ctx.args) + await ctx.bot.send_chat_action(update.effective_chat.id, action="typing") + + try: + result = await ai_client.search_web(query) + header = f"🔍 Results for: {escape(query)}\n\n" + for chunk in chunker.chunk_text(header + result): + await update.message.reply_html(chunk) + except Exception as exc: + context.record_error(f"Search failed: {exc}") + await update.message.reply_text(f"❌ Search failed: {exc}") + + +@_auth_required +async def cmd_ocr(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + target = update.message.reply_to_message if update.message else None + if not target: + await update.message.reply_text("Reply to a photo or image file with /ocr.") + return + + if target.photo: + file_id = target.photo[-1].file_id + mime_type = "image/jpeg" + elif target.document and (target.document.mime_type or "").startswith("image/"): + file_id = target.document.file_id + mime_type = target.document.mime_type or "image/jpeg" + else: + await update.message.reply_text("Reply to a photo or image file with /ocr.") + return + + await ctx.bot.send_chat_action(update.effective_chat.id, ChatAction.TYPING) + + try: + image_file = await ctx.bot.get_file(file_id) + image_bytes = bytes(await image_file.download_as_bytearray()) + settings = context.get_user_settings(update.effective_user.id) + caption = " ".join(ctx.args).strip() or target.caption or "" + sys_prompt = context.get_system_prompt() + context.reset_chat_if_input_too_large(update.effective_chat.id, sys_prompt, keep_last=False) + reply = await ai_client.chat_with_image( + messages=context.get_history(update.effective_chat.id), + system_prompt=sys_prompt, + max_tokens=settings["max_tokens"], + image_bytes=image_bytes, + mime_type=mime_type, + caption=caption, + analysis_mode="ocr", + ) + + context.add_message(update.effective_chat.id, "user", "[OCR image]" + (f": {caption}" if caption else "")) + context.add_message(update.effective_chat.id, "assistant", reply) + context.increment_msg() + context.record_event("ocr_requests") + + formatted, mode = formatter.safe_send(reply) + for part in chunker.chunk_text(formatted): + await update.message.reply_text(part, parse_mode=mode) + except Exception as exc: + context.record_error(f"OCR failed: {exc}") + await update.message.reply_text(f"❌ OCR failed: {exc}") + + +# ──────────────────────────────────────────────────────────── +# Owner-only commands (15 total) +# ──────────────────────────────────────────────────────────── + +@_owner_required +async def cmd_stats(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + stats = context.get_stats() + events = stats["events"] + provider = ai_client.provider_status() + known = len(auth.get_known_users()) + authed = len(auth.get_authorized()) + banned = len(auth.get_banned()) + public = "ON" if auth.is_public() else "OFF" + + await update.message.reply_html( + f"📊 Admin Dashboard\n\n" + f"Uptime: {stats['uptime']}\n" + f"Messages processed: {stats['messages_processed']}\n" + f"Active chats: {stats['active_chats']}\n" + f"Stored history messages: {stats['stored_history_messages']}\n" + f"Known users: {known}\n" + f"Authorized users: {authed}\n" + f"Banned users: {banned}\n" + f"Public mode: {public}\n\n" + f"Usage\n" + f"Text chats: {events.get('text_chats', 0)}\n" + f"Images analyzed: {events.get('images_analyzed', 0)}\n" + f"Images generated: {events.get('images_generated', 0)}\n" + f"Web searches: {events.get('web_searches', 0)}\n" + f"Image searches: {events.get('image_searches', 0)}\n" + f"OCR requests: {events.get('ocr_requests', 0)}\n" + f"Multi-image analyses: {events.get('multi_image_analyses', 0)}\n" + f"Provider fallbacks: {events.get('provider_fallbacks', 0)}\n" + f"History auto-resets: {events.get('history_auto_resets', 0)}\n" + f"Model-change resets: {events.get('history_resets_on_model_change', 0)}\n" + f"Errors: {events.get('errors', 0)}\n\n" + f"AI\n" + f"Provider: {provider['provider']}\n" + f"Model: {provider['model']}\n" + f"Google key: {'✅' if provider['google_configured'] else '❌'}\n" + f"OpenRouter key: {'✅' if provider['openrouter_configured'] else '❌'}\n" + f"Vision: {provider['vision_provider']}\n" + f"Last fallback: {escape(provider['last_fallback'])}\n" + f"Last error: {escape(stats['last_error'] or 'none')}" + ) + + +@_owner_required +async def cmd_uptime(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + stats = context.get_stats() + await update.message.reply_html(f"⏱️ Uptime: {stats['uptime']}") + + +@_owner_required +async def cmd_version(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + provider = ai_client.provider_status() + await update.message.reply_html( + f"🐾 Puppy Bot v{BOT_VERSION}\n" + f"Provider: {provider['provider']}\n" + f"Model: {provider['model']}" + ) + + +@_owner_required +async def cmd_addauth(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args: + await update.message.reply_text("Usage: /addauth ") + return + try: + uid = int(ctx.args[0]) + except ValueError: + await update.message.reply_text("❌ Invalid user ID.") + return + auth.add_user(uid) + await update.message.reply_html(f"✅ User {uid} is now authorized.") + + +@_owner_required +async def cmd_removeauth(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args: + await update.message.reply_text("Usage: /removeauth ") + return + try: + uid = int(ctx.args[0]) + except ValueError: + await update.message.reply_text("❌ Invalid user ID.") + return + auth.remove_user(uid) + await update.message.reply_html(f"✅ User {uid} removed from authorized list.") + + +@_owner_required +async def cmd_listauth(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + users = auth.get_authorized() + if not users: + await update.message.reply_text("No authorized users.") + return + lines = "\n".join(f"• {uid}" for uid in users) + await update.message.reply_html(f"Authorized users ({len(users)}):\n{lines}") + + +@_owner_required +async def cmd_ban(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args: + await update.message.reply_text("Usage: /ban ") + return + try: + uid = int(ctx.args[0]) + except ValueError: + await update.message.reply_text("❌ Invalid user ID.") + return + auth.ban_user(uid) + await update.message.reply_html(f"🚫 User {uid} has been banned.") + + +@_owner_required +async def cmd_unban(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args: + await update.message.reply_text("Usage: /unban ") + return + try: + uid = int(ctx.args[0]) + except ValueError: + await update.message.reply_text("❌ Invalid user ID.") + return + auth.unban_user(uid) + await update.message.reply_html(f"✅ User {uid} has been unbanned.") + + +@_owner_required +async def cmd_listban(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + users = auth.get_banned() + if not users: + await update.message.reply_text("No banned users.") + return + lines = "\n".join(f"• {uid}" for uid in users) + await update.message.reply_html(f"Banned users ({len(users)}):\n{lines}") + + +@_owner_required +async def cmd_userinfo(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args: + await update.message.reply_text("Usage: /userinfo ") + return + try: + uid = int(ctx.args[0]) + except ValueError: + await update.message.reply_text("❌ Invalid user ID.") + return + + is_auth = auth.is_authorized(uid) + is_ban = uid in auth.get_banned() + is_known = uid in auth.get_known_users() + settings = context.get_user_settings(uid) + + await update.message.reply_html( + f"👤 User info for {uid}\n\n" + f"Authorized: {'✅' if is_auth else '❌'}\n" + f"Banned: {'🚫' if is_ban else '—'}\n" + f"Known (interacted): {'✅' if is_known else '—'}\n" + f"Max tokens: {settings['max_tokens']}\n" + f"Web search: {'✅' if settings['search_enabled'] else '❌'}\n" + f"Language: {settings['lang']}" + ) + + +@_owner_required +async def cmd_broadcast(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args: + await update.message.reply_text("Usage: /broadcast ") + return + + text = " ".join(ctx.args) + known = auth.get_known_users() + if not known: + await update.message.reply_text("No known users to broadcast to.") + return + + sent = failed = 0 + for uid in known: + try: + await ctx.bot.send_message( + uid, + f"📢 Broadcast from Puppy:\n\n{escape(text)}", + parse_mode=ParseMode.HTML, + ) + sent += 1 + except Exception: + failed += 1 + + await update.message.reply_html( + f"✅ Broadcast sent: {sent} delivered, {failed} failed." + ) + + +@_owner_required +async def cmd_clearall(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + context.clear_all_histories() + await update.message.reply_text("🧹 All conversation histories cleared.") + + +@_owner_required +async def cmd_setpersonality(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + text = update.message.text or "" + # Strip the /setpersonality command prefix + prompt = text.partition(" ")[2].strip() + if not prompt: + current = context.get_system_prompt() + await update.message.reply_html( + f"Usage: /setpersonality <new system prompt>\n\n" + f"Current prompt:\n
{escape(current[:800])}
" + ) + return + context.set_system_prompt(prompt) + await update.message.reply_html( + f"✅ System prompt updated.\n\n
{escape(prompt[:300])}
" + ) + + +@_owner_required +async def cmd_resetpersonality(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + context.reset_system_prompt() + await update.message.reply_text("✅ Personality reset to default.") + + +@_owner_required +async def cmd_setmodel(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args: + provider = ai_client.provider_status() + await _send_html( + update, + ctx, + f"Provider: {provider['provider']}\n" + f"Model: {provider['model']}\n\n" + "Usage: /setmodel [google|openrouter] <model_name>\n" + "Examples:\n" + "/setmodel google gemini-2.5-flash-lite\n" + "/setmodel openrouter openai/gpt-4o-mini" + ) + return + + provider = ai_client.get_provider() + model_args = ctx.args + explicit_provider = False + if ctx.args[0].lower() in ai_client.get_available_providers(): + provider = ctx.args[0].lower() + model_args = ctx.args[1:] + explicit_provider = True + if not model_args: + await _send_text(update, ctx, "Usage: /setmodel [google|openrouter] ") + return + + model = model_args[0] + if not explicit_provider and provider == "google" and "/" in model: + provider = "openrouter" + explicit_provider = True + + ok, note = await ai_client.validate_model(model, provider=provider) + if not ok: + await _send_html(update, ctx, f"❌ {escape(note)}") + return + + ai_client.set_model(model, provider=provider) + if provider != ai_client.get_provider(): + ai_client.set_provider(provider) + context.clear_all_histories() + context.record_event("history_resets_on_model_change") + + await _send_html( + update, + ctx, + f"✅ Model switched to {model} " + f"(provider: {provider}).\n" + "🧹 Conversation memory reset for the new model." + + (f"\n\nℹ️ {escape(note)}" if note else "") + ) + + +@_owner_required +async def cmd_models(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + """List available models. Usage: /models [google|openrouter] [free] [filter]""" + args = ctx.args or [] + provider = ai_client.get_provider() + if args and args[0].lower() in ai_client.get_available_providers(): + provider = args[0].lower() + args = args[1:] + + if provider == "google": + lines = [ + f"• {escape(model)}" + (" ✅" if model == ai_client.get_model_for_provider("google") else "") + for model in ai_client.GOOGLE_TEXT_MODELS + ] + await update.message.reply_html( + "Google Models\n\n" + + "\n".join(lines) + + "\n\nUse /setmodel google <model_id> to switch." + ) + return + + free_only = "free" in [a.lower() for a in args] + search_term = " ".join(a for a in args if a.lower() != "free").strip().lower() + + status = await update.message.reply_text("🔍 Fetching models…") + models = await ai_client.fetch_or_models(free_only=free_only) + + if not models: + await status.edit_text("❌ Could not fetch model list from OpenRouter.") + return + + if search_term: + models = [m for m in models if search_term in m.get("id", "").lower() + or search_term in m.get("name", "").lower()] + + if not models: + await status.edit_text(f"No models found matching {escape(search_term)}.", parse_mode=ParseMode.HTML) + return + + # Cap at 30 results + total = len(models) + models = models[:30] + + lines = [] + for m in models: + mid = m.get("id", "") + ctx_k = m.get("context_length", 0) + ctx_str = f"{ctx_k // 1000}k" if ctx_k else "?" + p_prompt = m.get("pricing", {}).get("prompt", "?") + free_tag = " 🆓" if str(p_prompt) == "0" else "" + lines.append(f"• {escape(mid)} [{ctx_str}]{free_tag}") + + header = ( + f"OpenRouter Models ({total} total" + + (f", showing {len(models)}" if total > len(models) else "") + + (", free only" if free_only else "") + + (f", filter: {escape(search_term)}" if search_term else "") + + ")\n\n" + ) + footer = "\n\nUse /setmodel openrouter <model_id> to switch.\nTip: /models openrouter free — show only free models\n/models openrouter <keyword> — filter by name" + body = "\n".join(lines) + + from utils import chunker + full_text = header + body + footer + await status.delete() + for chunk in chunker.chunk_text(full_text, max_len=3800): + await update.message.reply_html(chunk) + + +@_owner_required +async def cmd_setprovider(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not ctx.args or ctx.args[0].lower() not in ("google", "openrouter"): + provider = ai_client.provider_status() + await update.message.reply_html( + f"Current provider: {provider['provider']}\n" + f"Current model: {provider['model']}\n" + f"Google model: {provider['google_model']}\n" + f"OpenRouter model: {provider['openrouter_model']}\n\n" + "Usage: /setprovider <google|openrouter> [model]\n\n" + "google — Gemini models via Google AI Studio\n" + "openrouter — 300+ models (GPT-4o, Claude, Llama, etc.) via OpenRouter\n\n" + "Note: image generation and vision always use Google regardless of provider." + ) + return + provider = ctx.args[0].lower() + try: + if len(ctx.args) > 1: + model = ctx.args[1] + ok, note = await ai_client.validate_model(model, provider=provider) + if not ok: + await update.message.reply_html(f"❌ {escape(note)}") + return + ai_client.set_model(model, provider=provider) + else: + note = "" + ai_client.set_provider(provider) + context.clear_all_histories() + context.record_event("history_resets_on_model_change") + await update.message.reply_html( + f"✅ Provider switched to {provider}.\n" + f"Active model: {ai_client.get_model()}\n\n" + f"Use /setmodel to change the model for this provider.\n" + "🧹 Conversation memory reset for the new provider." + + (f"\n\nℹ️ {escape(note)}" if note else "") + ) + except ValueError as exc: + await update.message.reply_text(f"❌ {exc}") + + +@_auth_required +async def cmd_model(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + provider = ai_client.provider_status() + await update.message.reply_html( + f"AI Provider Status\n\n" + f"Active provider: {provider['provider']}\n" + f"Active model: {provider['model']}\n" + f"Google model: {provider['google_model']}\n" + f"OpenRouter model: {provider['openrouter_model']}\n" + f"Vision provider: {provider['vision_provider']}\n" + f"Image generation provider: {provider['image_provider']}\n\n" + f"Last fallback: {escape(provider['last_fallback'])}\n\n" + "Commands:\n" + "/setprovider google|openrouter [model]\n" + "/setmodel [google|openrouter] <model>\n" + "/models [google|openrouter] [free] [keyword]" + ) + + +@_owner_required +async def cmd_reload(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + auth.load() + await update.message.reply_text("🔄 authorized_users.json reloaded from disk.") + + +# ──────────────────────────────────────────────────────────── +# Image generation (all authorized users) +# ──────────────────────────────────────────────────────────── + +@_auth_required +async def cmd_imagine(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + user_id = update.effective_user.id + + if not ctx.args: + used = context.images_used_today(user_id) + await update.message.reply_html( + f"Usage: /imagine <description>\n" + f"Images used today: {used}/{IMAGE_DAILY_LIMIT}" + ) + return + + if not context.can_generate_image(user_id): + used = context.images_used_today(user_id) + await update.message.reply_html( + f"Daily limit reached ({used}/{IMAGE_DAILY_LIMIT}). " + "Try again in 24 hours." + ) + return + + prompt = " ".join(ctx.args) + status = await update.message.reply_text("🎨 Generating…") + + try: + image_bytes = await ai_client.generate_image(prompt) + context.record_image(user_id) + await status.delete() + await update.message.reply_photo( + photo=image_bytes, + caption=f"🎨 {prompt[:900]}", + ) + except Exception as exc: + context.record_error(f"Image generation failed: {exc}") + await status.edit_text(f"❌ Image generation failed: {exc}") + + +# ──────────────────────────────────────────────────────────── +# Utility: /myid (anyone — useful for first-time setup) +# ──────────────────────────────────────────────────────────── + +async def cmd_myid(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + user = update.effective_user + chat = update.effective_chat + await update.message.reply_html( + f"Your Telegram ID: {user.id}\n" + f"Username: @{escape(user.username or 'none')}\n" + f"Chat ID: {chat.id}" + ) + + +# ──────────────────────────────────────────────────────────── +# Public mode unlock / lock — owner only +# ──────────────────────────────────────────────────────────── + +@_owner_required +async def cmd_unlock(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if auth.is_public(): + await update.message.reply_html( + "ℹ️ Bot is already public — everyone can use it.\n" + "Use /lock to restrict back to authorized users only." + ) + return + auth.set_public(True) + await update.message.reply_html( + "🔓 Public mode ON\n\n" + "Anyone can now chat with Puppy.\n" + "• Existing per-user settings are preserved\n" + "• Banned users are still blocked\n" + "Use /lock to re-restrict." + ) + + +@_owner_required +async def cmd_lock(update: Update, ctx: ContextTypes.DEFAULT_TYPE): + if not auth.is_public(): + await update.message.reply_html( + "ℹ️ Bot is already locked — only authorized users can use it." + ) + return + auth.set_public(False) + await update.message.reply_html( + "🔒 Public mode OFF\n\n" + "Only authorized users can chat with Puppy again.\n" + "Use /unlock to open to everyone." + )