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