Compare commits

..
10 Commits
13 changed files with 2695 additions and 0 deletions
+39
View File
@@ -0,0 +1,39 @@
# Repository Guidelines
## Project Structure & Module Organization
This repository contains a Python Telegram bot. `bot.py` is the entry point and registers command, text, photo, image-document, and inline handlers. `handlers/` contains Telegram-facing command and update handlers, grouped by feature (`admin.py`, `chat.py`, `inline.py`, `help.py`). `services/` contains external integrations and shared business logic such as AI calls, auth persistence, and saved runtime settings. `utils/` contains reusable helpers for context, formatting, chunking, and Persian-language handling. Runtime state and credentials live in `.env`, `settings.json`, `authorized_users.json`, and `*.session` files; do not treat these as source assets.
## Build, Test, and Development Commands
Create an isolated environment before installing dependencies:
```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```
Run the bot locally with:
```bash
python bot.py
```
Use `.env.example` as the template for required environment variables, then set `TELEGRAM_TOKEN`, `GOOGLE_API_KEY`, and `OWNER_ID` in `.env`. Runtime choices such as active provider, model, public mode, system prompt, and per-user settings are saved in `settings.json`.
## Coding Style & Naming Conventions
Follow idiomatic Python with 4-space indentation, clear function names, and small modules organized by responsibility. Handler functions use the `cmd_<command>` pattern for commands, such as `cmd_ping` or `cmd_setmodel`; keep new command names consistent with the registered Telegram command. Prefer async functions for Telegram callbacks and service calls. Keep configuration values centralized in `config.py` rather than scattering `os.getenv` calls.
## Testing Guidelines
No test suite is currently present. For new behavior, add focused `pytest` tests under a future `tests/` directory and name files `test_<module>.py`. Prioritize pure helpers in `utils/` and service logic that can be tested without Telegram network access. Until tests exist, validate changes by running `python bot.py` with a development bot token and exercising the affected command or message path.
## Commit & Pull Request Guidelines
The local `.git` directory has no usable history, so no existing commit convention is available. Use concise imperative commit messages, for example `Add image OCR command` or `Fix auth file reload handling`. Pull requests should include a short summary, configuration changes, manual test steps, and screenshots or chat transcripts when user-visible Telegram behavior changes.
## Security & Configuration Tips
Never commit `.env`, session files, API keys, bot tokens, or real user authorization data. Keep `.env.example` updated when adding configuration. Treat `authorized_users.json` as local runtime state unless a sanitized fixture is intentionally needed for tests.
+111
View File
@@ -0,0 +1,111 @@
"""Puppy Bot entry point. Run with: python bot.py"""
import logging
import sys
from telegram.ext import (
ApplicationBuilder,
CommandHandler,
InlineQueryHandler,
MessageHandler,
filters,
)
from config import TELEGRAM_TOKEN
import services.auth as auth
from services import settings as settings_store
from handlers import chat, inline
from handlers import admin
from handlers.help import cmd_help
from utils import context as runtime_context
logging.basicConfig(
format="%(asctime)s | %(name)s | %(levelname)s | %(message)s",
level=logging.INFO,
stream=sys.stdout,
)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("telegram").setLevel(logging.WARNING)
logger = logging.getLogger("puppy")
async def _error_handler(update, ctx):
err = getattr(ctx, "error", None)
logger.exception("Unhandled bot error: %s", err)
runtime_context.record_error(f"Unhandled bot error: {err}")
def _register_handlers(app):
# ── General / all-auth ─────────────────────────────────
app.add_handler(CommandHandler("start", admin.cmd_start))
app.add_handler(CommandHandler("help", cmd_help))
app.add_handler(CommandHandler("ping", admin.cmd_ping))
app.add_handler(CommandHandler("clear", admin.cmd_clear))
app.add_handler(CommandHandler("myhistory", admin.cmd_myhistory))
app.add_handler(CommandHandler("setmaxtoken", admin.cmd_setmaxtoken))
app.add_handler(CommandHandler("getmaxtoken", admin.cmd_getmaxtoken))
app.add_handler(CommandHandler("resettoken", admin.cmd_resettoken))
app.add_handler(CommandHandler("setlang", admin.cmd_setlang))
app.add_handler(CommandHandler("status", admin.cmd_status))
app.add_handler(CommandHandler("model", admin.cmd_model))
app.add_handler(CommandHandler("togglesearch", admin.cmd_togglesearch))
app.add_handler(CommandHandler("search", admin.cmd_search))
app.add_handler(CommandHandler("ocr", admin.cmd_ocr))
app.add_handler(CommandHandler("imagine", admin.cmd_imagine))
app.add_handler(CommandHandler("myid", admin.cmd_myid))
# ── Owner-only ─────────────────────────────────────────
app.add_handler(CommandHandler("stats", admin.cmd_stats))
app.add_handler(CommandHandler("dashboard", admin.cmd_stats))
app.add_handler(CommandHandler("uptime", admin.cmd_uptime))
app.add_handler(CommandHandler("version", admin.cmd_version))
app.add_handler(CommandHandler("addauth", admin.cmd_addauth))
app.add_handler(CommandHandler("removeauth", admin.cmd_removeauth))
app.add_handler(CommandHandler("listauth", admin.cmd_listauth))
app.add_handler(CommandHandler("ban", admin.cmd_ban))
app.add_handler(CommandHandler("unban", admin.cmd_unban))
app.add_handler(CommandHandler("listban", admin.cmd_listban))
app.add_handler(CommandHandler("userinfo", admin.cmd_userinfo))
app.add_handler(CommandHandler("broadcast", admin.cmd_broadcast))
app.add_handler(CommandHandler("clearall", admin.cmd_clearall))
app.add_handler(CommandHandler("setpersonality", admin.cmd_setpersonality))
app.add_handler(CommandHandler("resetpersonality",admin.cmd_resetpersonality))
app.add_handler(CommandHandler("setmodel", admin.cmd_setmodel))
app.add_handler(CommandHandler("setprovider", admin.cmd_setprovider))
app.add_handler(CommandHandler("models", admin.cmd_models))
app.add_handler(CommandHandler("unlock", admin.cmd_unlock))
app.add_handler(CommandHandler("lock", admin.cmd_lock))
app.add_handler(CommandHandler("reload", admin.cmd_reload))
# ── Message handlers ───────────────────────────────────
app.add_handler(MessageHandler(filters.PHOTO & ~filters.COMMAND, chat.message_handler))
app.add_handler(MessageHandler(filters.Document.IMAGE & ~filters.COMMAND, chat.message_handler))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, chat.message_handler))
# ── Inline mode ────────────────────────────────────────
app.add_handler(InlineQueryHandler(inline.inline_handler))
def main():
if not TELEGRAM_TOKEN:
logger.error("TELEGRAM_TOKEN is not set.")
sys.exit(1)
auth.load()
settings_store.load()
logger.info("Auth loaded — %d authorized, %d banned",
len(auth.get_authorized()), len(auth.get_banned()))
app = (
ApplicationBuilder()
.token(TELEGRAM_TOKEN)
.build()
)
_register_handlers(app)
app.add_error_handler(_error_handler)
logger.info("Puppy is starting…")
app.run_polling(drop_pending_updates=True)
if __name__ == "__main__":
main()
+737
View File
@@ -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 <b>{name}</b>! I'm <b>Puppy</b> (پاپی) — your AI companion.\n\n"
"• Chat with me directly in private\n"
"• In groups: say my name (<code>Puppy</code> / <code>پاپی</code>), @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"<b>📜 Your recent conversation:</b>\n\n<pre>{escape(summary)}</pre>"
)
@_auth_required
async def cmd_setmaxtoken(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not ctx.args:
await update.message.reply_html(
f"Usage: <code>/setmaxtoken &lt;{MAX_TOKENS_MIN}–{MAX_TOKENS_MAX}&gt;</code>"
)
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 <code>{n}</code>.")
@_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: <code>{settings['max_tokens']}</code> (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 <code>{MAX_TOKENS_DEFAULT}</code>.")
@_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: <code>/setlang &lt;auto|fa|en&gt;</code>"
)
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 <b>{label}</b>.")
@_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"<b>⚙️ Your settings</b>\n\n"
f"Max tokens: <code>{settings['max_tokens']}</code>\n"
f"Web search: {search_icon}\n"
f"Language: <code>{lang_map.get(settings['lang'], settings['lang'])}</code>\n"
f"History messages: <code>{history_len}</code>\n"
f"AI provider: <code>{provider['provider']}</code>\n"
f"AI model: <code>{provider['model']}</code>"
)
@_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 <query>")
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"🔍 <b>Results for:</b> {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"<b>📊 Admin Dashboard</b>\n\n"
f"Uptime: <code>{stats['uptime']}</code>\n"
f"Messages processed: <code>{stats['messages_processed']}</code>\n"
f"Active chats: <code>{stats['active_chats']}</code>\n"
f"Stored history messages: <code>{stats['stored_history_messages']}</code>\n"
f"Known users: <code>{known}</code>\n"
f"Authorized users: <code>{authed}</code>\n"
f"Banned users: <code>{banned}</code>\n"
f"Public mode: <code>{public}</code>\n\n"
f"<b>Usage</b>\n"
f"Text chats: <code>{events.get('text_chats', 0)}</code>\n"
f"Images analyzed: <code>{events.get('images_analyzed', 0)}</code>\n"
f"Images generated: <code>{events.get('images_generated', 0)}</code>\n"
f"Web searches: <code>{events.get('web_searches', 0)}</code>\n"
f"Image searches: <code>{events.get('image_searches', 0)}</code>\n"
f"OCR requests: <code>{events.get('ocr_requests', 0)}</code>\n"
f"Multi-image analyses: <code>{events.get('multi_image_analyses', 0)}</code>\n"
f"Provider fallbacks: <code>{events.get('provider_fallbacks', 0)}</code>\n"
f"History auto-resets: <code>{events.get('history_auto_resets', 0)}</code>\n"
f"Model-change resets: <code>{events.get('history_resets_on_model_change', 0)}</code>\n"
f"Errors: <code>{events.get('errors', 0)}</code>\n\n"
f"<b>AI</b>\n"
f"Provider: <code>{provider['provider']}</code>\n"
f"Model: <code>{provider['model']}</code>\n"
f"Google key: {'✅' if provider['google_configured'] else '❌'}\n"
f"OpenRouter key: {'✅' if provider['openrouter_configured'] else '❌'}\n"
f"Vision: <code>{provider['vision_provider']}</code>\n"
f"Last fallback: <code>{escape(provider['last_fallback'])}</code>\n"
f"Last error: <code>{escape(stats['last_error'] or 'none')}</code>"
)
@_owner_required
async def cmd_uptime(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
stats = context.get_stats()
await update.message.reply_html(f"⏱️ Uptime: <code>{stats['uptime']}</code>")
@_owner_required
async def cmd_version(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
provider = ai_client.provider_status()
await update.message.reply_html(
f"🐾 <b>Puppy Bot</b> v<code>{BOT_VERSION}</code>\n"
f"Provider: <code>{provider['provider']}</code>\n"
f"Model: <code>{provider['model']}</code>"
)
@_owner_required
async def cmd_addauth(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not ctx.args:
await update.message.reply_text("Usage: /addauth <user_id>")
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 <code>{uid}</code> 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 <user_id>")
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 <code>{uid}</code> 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"• <code>{uid}</code>" for uid in users)
await update.message.reply_html(f"<b>Authorized users ({len(users)}):</b>\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 <user_id>")
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 <code>{uid}</code> 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 <user_id>")
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 <code>{uid}</code> 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"• <code>{uid}</code>" for uid in users)
await update.message.reply_html(f"<b>Banned users ({len(users)}):</b>\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 <user_id>")
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"<b>👤 User info for <code>{uid}</code></b>\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: <code>{settings['max_tokens']}</code>\n"
f"Web search: {'✅' if settings['search_enabled'] else '❌'}\n"
f"Language: <code>{settings['lang']}</code>"
)
@_owner_required
async def cmd_broadcast(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
if not ctx.args:
await update.message.reply_text("Usage: /broadcast <message>")
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"📢 <b>Broadcast from Puppy:</b>\n\n{escape(text)}",
parse_mode=ParseMode.HTML,
)
sent += 1
except Exception:
failed += 1
await update.message.reply_html(
f"✅ Broadcast sent: <code>{sent}</code> delivered, <code>{failed}</code> 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: <code>/setpersonality &lt;new system prompt&gt;</code>\n\n"
f"<b>Current prompt:</b>\n<pre>{escape(current[:800])}</pre>"
)
return
context.set_system_prompt(prompt)
await update.message.reply_html(
f"✅ System prompt updated.\n\n<pre>{escape(prompt[:300])}</pre>"
)
@_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: <code>{provider['provider']}</code>\n"
f"Model: <code>{provider['model']}</code>\n\n"
"Usage: <code>/setmodel [google|openrouter] &lt;model_name&gt;</code>\n"
"Examples:\n"
"<code>/setmodel google gemini-2.5-flash-lite</code>\n"
"<code>/setmodel openrouter openai/gpt-4o-mini</code>"
)
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] <model_name>")
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 <code>{model}</code> "
f"(provider: <code>{provider}</code>).\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"• <code>{escape(model)}</code>" + (" ✅" if model == ai_client.get_model_for_provider("google") else "")
for model in ai_client.GOOGLE_TEXT_MODELS
]
await update.message.reply_html(
"<b>Google Models</b>\n\n"
+ "\n".join(lines)
+ "\n\n<i>Use /setmodel google &lt;model_id&gt; to switch.</i>"
)
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 <code>{escape(search_term)}</code>.", 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"• <code>{escape(mid)}</code> [{ctx_str}]{free_tag}")
header = (
f"<b>OpenRouter Models</b> ({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\n<i>Use /setmodel openrouter &lt;model_id&gt; to switch.\nTip: /models openrouter free — show only free models\n/models openrouter &lt;keyword&gt; — filter by name</i>"
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: <code>{provider['provider']}</code>\n"
f"Current model: <code>{provider['model']}</code>\n"
f"Google model: <code>{provider['google_model']}</code>\n"
f"OpenRouter model: <code>{provider['openrouter_model']}</code>\n\n"
"Usage: <code>/setprovider &lt;google|openrouter&gt; [model]</code>\n\n"
"<b>google</b> — Gemini models via Google AI Studio\n"
"<b>openrouter</b> — 300+ models (GPT-4o, Claude, Llama, etc.) via OpenRouter\n\n"
"<i>Note: image generation and vision always use Google regardless of provider.</i>"
)
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 <b>{provider}</b>.\n"
f"Active model: <code>{ai_client.get_model()}</code>\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"<b>AI Provider Status</b>\n\n"
f"Active provider: <code>{provider['provider']}</code>\n"
f"Active model: <code>{provider['model']}</code>\n"
f"Google model: <code>{provider['google_model']}</code>\n"
f"OpenRouter model: <code>{provider['openrouter_model']}</code>\n"
f"Vision provider: <code>{provider['vision_provider']}</code>\n"
f"Image generation provider: <code>{provider['image_provider']}</code>\n\n"
f"Last fallback: <code>{escape(provider['last_fallback'])}</code>\n\n"
"Commands:\n"
"<code>/setprovider google|openrouter [model]</code>\n"
"<code>/setmodel [google|openrouter] &lt;model&gt;</code>\n"
"<code>/models [google|openrouter] [free] [keyword]</code>"
)
@_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: <code>/imagine &lt;description&gt;</code>\n"
f"Images used today: <code>{used}/{IMAGE_DAILY_LIMIT}</code>"
)
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 (<code>{used}/{IMAGE_DAILY_LIMIT}</code>). "
"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"<b>Your Telegram ID:</b> <code>{user.id}</code>\n"
f"<b>Username:</b> @{escape(user.username or 'none')}\n"
f"<b>Chat ID:</b> <code>{chat.id}</code>"
)
# ────────────────────────────────────────────────────────────
# 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 <b>public</b> — everyone can use it.\n"
"Use /lock to restrict back to authorized users only."
)
return
auth.set_public(True)
await update.message.reply_html(
"🔓 <b>Public mode ON</b>\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 <b>locked</b> — only authorized users can use it."
)
return
auth.set_public(False)
await update.message.reply_html(
"🔒 <b>Public mode OFF</b>\n\n"
"Only authorized users can chat with Puppy again.\n"
"Use /unlock to open to everyone."
)
+290
View File
@@ -0,0 +1,290 @@
import re
import asyncio
import logging
import time
from telegram import Update, Message
from telegram.constants import ChatAction, ChatType
from telegram.ext import ContextTypes
from services import auth, ai_client
from utils import context, chunker, persian, formatter
logger = logging.getLogger(__name__)
_MEDIA_GROUP_WAIT = 1.2
# ── Sending helpers ───────────────────────────────────────────
async def _send(message: Message, text: str):
"""Send text with HTML formatting; fall back to plain on parse error."""
formatted, mode = formatter.safe_send(text)
for chunk in chunker.chunk_text(formatted):
try:
await message.reply_text(chunk, parse_mode=mode)
except Exception:
plain = re.sub(r"<[^>]+>", "", chunk)
await message.reply_text(plain)
async def _send_images(message: Message, urls: list[str]):
"""Upload up to 5 image URLs to Telegram."""
sent = 0
for url in urls[:5]:
try:
await message.reply_photo(url)
sent += 1
except Exception as exc:
logger.warning("Could not send image %s: %s", url[:60], exc)
return sent
# ── Group-mention / reply detection ──────────────────────────
def _is_addressed(message: Message, bot_username: str, bot_id: int) -> bool:
text = message.text or message.caption or ""
# Reply to bot's own message
rt = message.reply_to_message
if rt and rt.from_user:
if rt.from_user.id == bot_id:
return True
if rt.from_user.username and rt.from_user.username.lower() == bot_username.lower():
return True
# @mention in entities
for entity in list(message.entities or []) + list(message.caption_entities or []):
if entity.type == "mention":
mention = text[entity.offset: entity.offset + entity.length]
if mention.lower() == f"@{bot_username.lower()}":
return True
# Name mention (Puppy / پاپی)
return persian.is_puppy_mentioned(text)
def _clean_text(text: str, bot_username: str) -> str:
text = re.sub(rf"@{re.escape(bot_username)}\s*", "", text, flags=re.IGNORECASE)
text = persian.strip_puppy_prefix(text)
return text.strip()
def _get_quoted_context(message: Message, bot_id: int) -> str:
"""If user replied to another user's message, return a quoted prefix for context."""
rt = message.reply_to_message
if not rt or not rt.from_user:
return ""
if rt.from_user.id == bot_id:
return "" # replying to bot is handled naturally via history
quoted = rt.text or rt.caption or ""
if not quoted:
return ""
name = rt.from_user.first_name or "someone"
return f'[Replying to {name}: "{quoted[:300]}"]\n'
def _image_analysis_mode(caption: str) -> str:
text = caption.lower()
if any(word in text for word in ("ocr", "text", "read", "extract", "متن", "بخوان")):
return "ocr"
if any(word in text for word in ("describe", "description", "توصیف", "شرح")):
return "describe"
if any(word in text for word in ("objects", "things", "items", "اشیا", "چیزها")):
return "objects"
if "?" in caption or any(word in text for word in ("what", "why", "how", "where", "who", "چی", "چرا", "کجا")):
return "question"
return "auto"
async def _download_image(message: Message, ctx: ContextTypes.DEFAULT_TYPE) -> tuple[bytes, str] | None:
if message.photo:
image_file = await ctx.bot.get_file(message.photo[-1].file_id)
return bytes(await image_file.download_as_bytearray()), "image/jpeg"
if message.document and (message.document.mime_type or "").startswith("image/"):
image_file = await ctx.bot.get_file(message.document.file_id)
return bytes(await image_file.download_as_bytearray()), message.document.mime_type or "image/jpeg"
return None
def _is_image_message(message: Message) -> bool:
return bool(message.photo or (message.document and (message.document.mime_type or "").startswith("image/")))
# ── Image handler ─────────────────────────────────────────────
async def _handle_photo(update: Update, ctx: ContextTypes.DEFAULT_TYPE, force_mode: str | None = None):
message = update.effective_message
user = update.effective_user
chat = update.effective_chat
caption = _clean_text(message.caption or "", ctx.bot.username or "")
settings = context.get_user_settings(user.id)
history = context.get_history(message.chat_id)
sys_prompt = context.get_system_prompt()
if context.reset_chat_if_input_too_large(message.chat_id, sys_prompt, keep_last=False):
history = []
await ctx.bot.send_chat_action(chat.id, ChatAction.TYPING)
try:
downloaded = await _download_image(message, ctx)
if not downloaded:
return
image_bytes, mime_type = downloaded
analysis_mode = force_mode or _image_analysis_mode(caption)
reply = await ai_client.chat_with_image(
messages=history,
system_prompt=sys_prompt,
max_tokens=settings["max_tokens"],
image_bytes=image_bytes,
mime_type=mime_type,
caption=caption,
analysis_mode=analysis_mode,
)
context.add_message(message.chat_id, "user", f"[Photo]{': ' + caption if caption else ''}")
context.add_message(message.chat_id, "assistant", reply)
context.increment_msg()
context.record_event("images_analyzed")
await _send(message, reply)
except Exception as exc:
context.record_error(f"Image error: {exc}")
await message.reply_text(f"❌ Image error: {exc}")
async def _handle_media_group(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
message = update.effective_message
group_id = message.media_group_id
if not group_id:
await _handle_photo(update, ctx)
return
groups = ctx.application.bot_data.setdefault("media_groups", {})
bucket = groups.setdefault(group_id, {"messages": [], "last_seen": 0.0})
bucket["messages"].append(message)
bucket["last_seen"] = time.time()
if len(bucket["messages"]) > 1:
return
await asyncio.sleep(_MEDIA_GROUP_WAIT)
bucket = groups.pop(group_id, bucket)
messages = bucket["messages"][:10]
first = messages[0]
user = first.from_user
settings = context.get_user_settings(user.id)
history = context.get_history(first.chat_id)
sys_prompt = context.get_system_prompt()
if context.reset_chat_if_input_too_large(first.chat_id, sys_prompt, keep_last=False):
history = []
caption = ""
images: list[tuple[bytes, str]] = []
await ctx.bot.send_chat_action(first.chat_id, ChatAction.TYPING)
try:
for msg in messages:
if not caption and msg.caption:
caption = _clean_text(msg.caption, ctx.bot.username or "")
downloaded = await _download_image(msg, ctx)
if downloaded:
images.append(downloaded)
if not images:
return
reply = await ai_client.chat_with_images(
messages=history,
system_prompt=sys_prompt,
max_tokens=settings["max_tokens"],
images=images,
caption=caption,
analysis_mode=_image_analysis_mode(caption),
)
context.add_message(first.chat_id, "user", f"[{len(images)} photos]{': ' + caption if caption else ''}")
context.add_message(first.chat_id, "assistant", reply)
context.increment_msg()
context.record_event("multi_image_analyses")
await _send(first, reply)
except Exception as exc:
context.record_error(f"Multi-image error: {exc}")
await first.reply_text(f"❌ Multi-image error: {exc}")
# ── Main message handler ──────────────────────────────────────
async def message_handler(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
message = update.effective_message
if not message:
return
user = update.effective_user
chat = update.effective_chat
if not auth.is_authorized(user.id, user.username or ""):
return
auth.register_known(user.id, username=user.username or "")
# Group gate: must be addressed
if chat.type in (ChatType.GROUP, ChatType.SUPERGROUP):
if not _is_addressed(message, ctx.bot.username or "", ctx.bot.id):
return
# Image branch: Telegram photos and image files/documents.
if _is_image_message(message):
if message.media_group_id:
await _handle_media_group(update, ctx)
else:
await _handle_photo(update, ctx)
return
text = message.text or ""
if not text:
return
text = _clean_text(text, ctx.bot.username or "")
if not text:
return
# Include quoted message from another user as context prefix
quoted = _get_quoted_context(message, ctx.bot.id)
full_text = quoted + text
settings = context.get_user_settings(user.id)
sys_prompt = context.get_system_prompt()
# Add to history first so AI sees the current message
context.add_message(message.chat_id, "user", full_text)
context.reset_chat_if_input_too_large(message.chat_id, sys_prompt, keep_last=True)
history = context.get_history(message.chat_id)
await ctx.bot.send_chat_action(chat.id, ChatAction.TYPING)
try:
reply, image_urls = await ai_client.chat(
messages=history,
system_prompt=sys_prompt,
max_tokens=settings["max_tokens"],
search_enabled=settings["search_enabled"],
)
context.add_message(message.chat_id, "assistant", reply)
context.increment_msg()
context.record_event("text_chats")
# Send images first (appear above text)
if image_urls:
await _send_images(message, image_urls)
await _send(message, reply)
except Exception as exc:
context.pop_last_message(message.chat_id)
context.record_error(f"Chat error: {exc}")
logger.error("Chat error: %s", exc)
await message.reply_text(f"❌ Error: {exc}")
+87
View File
@@ -0,0 +1,87 @@
from telegram import Update
from telegram.constants import ParseMode
from telegram.ext import ContextTypes
from services import auth, ai_client
from config import BOT_VERSION
HELP_TEXT = """
<b>🐾 Puppy Bot v{version} — Help</b>
Puppy is an AI assistant powered by Google Gemini (or OpenRouter models).
<b>Groups:</b> mention me by name (<code>Puppy</code> / <code>پاپی</code>), reply to my messages, or @mention me.
<b>Private:</b> just write directly.
━━━━━━━━━━━━━━━━━━━━
<b>💬 Chat</b>
Send a message to chat.
Send a <b>photo</b> or image album → I'll analyze it. Caption with OCR/read/text, describe, objects, or a question for focused image understanding.
Reply to <b>someone else's message</b> and call my name → I'll respond in context.
━━━━━━━━━━━━━━━━━━━━
<b>🔍 Search & Images</b>
I auto-search the web for real-time info (news, weather, date, prices…).
Ask for images → I'll search and send up to 5 results.
<code>/search &lt;query&gt;</code> — Direct web search
<code>/ocr</code> — Reply to an image and extract readable text
<code>/imagine &lt;description&gt;</code> — Generate an image with Imagen 4 Fast
━━━━━━━━━━━━━━━━━━━━
<b>⚙️ Your Settings</b>
<code>/status</code> — View your settings
<code>/setmaxtoken &lt;n&gt;</code> — Set max reply tokens (50–4000, default 300)
<code>/getmaxtoken</code> — Show current max tokens
<code>/resettoken</code> — Reset to default (300)
<code>/setlang &lt;auto|fa|en&gt;</code> — Override response language
<code>/togglesearch</code> — Enable / disable auto web search
<code>/model</code> — Show active AI provider and model
<code>/clear</code> — Clear your conversation history
<code>/myhistory</code> — View your recent conversation
━━━━━━━━━━━━━━━━━━━━
<b>🔁 Inline Mode</b>
Type <code>@{username} your question</code> in any chat.
━━━━━━━━━━━━━━━━━━━━
<b>🛠️ Admin Commands</b> <i>(owner only)</i>
<code>/unlock</code> — Open bot to everyone (public mode)
<code>/lock</code> — Restrict back to authorized users only
<code>/setprovider &lt;google|openrouter&gt;</code> — Switch AI provider
<code>/setmodel [google|openrouter] &lt;model&gt;</code> — Change model with validation
<code>/models [google|openrouter] [free] [keyword]</code> — Browse available models
<code>/stats</code> / <code>/dashboard</code> — Admin dashboard
<code>/addauth &lt;id&gt;</code> — Authorize a user
<code>/removeauth &lt;id&gt;</code> — Remove authorization
<code>/listauth</code> — List authorized users
<code>/ban &lt;id&gt;</code> — Ban a user
<code>/unban &lt;id&gt;</code> — Unban a user
<code>/listban</code> — List banned users
<code>/userinfo &lt;id&gt;</code> — Show user info
<code>/broadcast &lt;text&gt;</code> — Message all known users
<code>/clearall</code> — Clear all histories
<code>/setpersonality &lt;text&gt;</code> — Set custom system prompt
<code>/resetpersonality</code> — Restore default personality
<code>/reload</code> — Reload authorized_users.json from disk
<code>/uptime</code> — Show bot uptime
<code>/version</code> — Show bot version
<code>/ping</code> — Health check
<code>/myid</code> — Show your Telegram user ID
━━━━━━━━━━━━━━━━━━━━
<i>Provider: {provider} | Model: {model}</i>
""".strip()
async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE):
user = update.effective_user
if not auth.is_authorized(user.id, user.username or ""):
return
bot_username = context.bot.username or "PuppyBot"
text = HELP_TEXT.format(
version=BOT_VERSION,
username=bot_username,
provider=ai_client.get_provider(),
model=ai_client.get_model(),
)
await update.message.reply_text(text, parse_mode=ParseMode.HTML)
+36
View File
@@ -0,0 +1,36 @@
import uuid
from telegram import Update, InlineQueryResultArticle, InputTextMessageContent
from telegram.ext import ContextTypes
from services import auth, ai_client
from utils.context import get_system_prompt, get_user_settings
async def inline_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
query = update.inline_query.query.strip()
user_id = update.inline_query.from_user.id
if not auth.is_authorized(user_id):
await update.inline_query.answer([], cache_time=0)
return
if not query:
await update.inline_query.answer([], cache_time=0)
return
try:
settings = get_user_settings(user_id)
max_tokens = min(settings["max_tokens"], 300)
answer = await ai_client.quick_reply(query, get_system_prompt(), max_tokens=max_tokens)
except Exception as e:
answer = f"Error: {e}"
result = InlineQueryResultArticle(
id=str(uuid.uuid4()),
title="Puppy's answer",
description=answer[:120] + ("…" if len(answer) > 120 else ""),
input_message_content=InputTextMessageContent(
message_text=answer,
),
)
await update.inline_query.answer([result], cache_time=0)
+805
View File
@@ -0,0 +1,805 @@
"""
AI client:
• Text + tool calls → native google-genai (Google) OR OpenAI client (OpenRouter)
• Vision → google-genai native always
• Image generation → google-genai Imagen 4 Fast always
"""
import asyncio
import json
import logging
import time
from openai import AsyncOpenAI
from openai import APIStatusError
from google import genai
from google.genai import types as gt
from config import (
GOOGLE_API_KEY, GOOGLE_AI_BASE_URL,
OPENROUTER_API_KEY, OPENROUTER_BASE_URL,
DEFAULT_MODEL, DEFAULT_OR_MODEL, IMAGEN_MODEL,
MAX_API_CONTEXT,
)
logger = logging.getLogger(__name__)
# ── Clients ───────────────────────────────────────────────────
_or_client = AsyncOpenAI(
api_key=OPENROUTER_API_KEY or "sk-no-key",
base_url=OPENROUTER_BASE_URL,
default_headers={
"HTTP-Referer": "https://github.com/puppy-bot",
"X-Title": "Puppy Bot",
},
)
_gc = genai.Client(api_key=GOOGLE_API_KEY)
_current_provider: str = "google"
_google_model: str = DEFAULT_MODEL
_or_model: str = DEFAULT_OR_MODEL
_last_fallback: str = ""
GOOGLE_TEXT_MODELS = [
"gemini-2.5-flash-lite",
"gemini-2.5-flash",
"gemini-2.5-pro",
"gemini-2.0-flash",
"gemini-2.0-flash-lite",
]
CHEAP_IMAGEN_MODEL = "imagen-4.0-fast-generate-001"
IMAGEN_FALLBACK_MODELS = [CHEAP_IMAGEN_MODEL]
# ── Provider / model control ──────────────────────────────────
def set_provider(provider: str):
global _current_provider
if provider not in ("google", "openrouter"):
raise ValueError(f"Unknown provider: {provider}")
_current_provider = provider
_autosave()
def get_provider() -> str:
return _current_provider
def get_available_providers() -> list[str]:
return ["google", "openrouter"]
def set_model(model: str, provider: str | None = None):
global _google_model, _or_model
p = provider or _current_provider
if p == "google":
_google_model = model
else:
_or_model = model
_autosave()
def get_model() -> str:
return _google_model if _current_provider == "google" else _or_model
def get_model_for_provider(provider: str) -> str:
if provider == "google":
return _google_model
if provider == "openrouter":
return _or_model
raise ValueError(f"Unknown provider: {provider}")
def provider_status() -> dict:
return {
"provider": _current_provider,
"model": get_model(),
"google_model": _google_model,
"openrouter_model": _or_model,
"google_configured": bool(GOOGLE_API_KEY),
"openrouter_configured": bool(OPENROUTER_API_KEY),
"vision_provider": "google",
"image_provider": "google",
"last_fallback": _last_fallback or "none",
}
def import_provider_settings(data: dict):
global _current_provider, _google_model, _or_model
provider = data.get("provider")
if provider in ("google", "openrouter"):
_current_provider = provider
if isinstance(data.get("google_model"), str):
_google_model = data["google_model"]
if isinstance(data.get("openrouter_model"), str):
_or_model = data["openrouter_model"]
def export_provider_settings() -> dict:
return {
"provider": _current_provider,
"google_model": _google_model,
"openrouter_model": _or_model,
}
def _autosave():
try:
from services import settings as settings_store
settings_store.save()
except Exception:
pass
async def validate_model(model: str, provider: str | None = None) -> tuple[bool, str]:
p = provider or _current_provider
if p == "google":
if model in GOOGLE_TEXT_MODELS:
return True, ""
return False, "Unknown Google model. Use /models google to see supported examples."
if p == "openrouter":
models = await fetch_or_models()
if not models:
return True, "Could not verify OpenRouter model list; saved model anyway."
ids = {m.get("id") for m in models}
if model in ids:
return True, ""
return False, "Model was not found on OpenRouter. Use /models openrouter <keyword>."
return False, f"Unknown provider: {p}"
# ── Tool definitions ──────────────────────────────────────────
# Native google-genai tool declarations
_GC_TOOLS = [
gt.Tool(function_declarations=[
gt.FunctionDeclaration(
name="search_web",
description=(
"Search the internet for current information. "
"Use for today's date/time, news, prices, weather, recent events, "
"or any fact that might have changed."
),
parameters=gt.Schema(
type="object",
properties={"query": gt.Schema(type="string", description="Search query")},
required=["query"],
),
),
gt.FunctionDeclaration(
name="image_search",
description="Search the web for images. Use when user asks to find, show, or search for images/pictures.",
parameters=gt.Schema(
type="object",
properties={"query": gt.Schema(type="string", description="Image search query")},
required=["query"],
),
),
])
]
# OpenAI-format tool definitions (for OpenRouter)
_OAI_TOOLS = [
{
"type": "function",
"function": {
"name": "search_web",
"description": (
"Search the internet for current information. "
"Use for today's date/time, news, prices, weather, recent events."
),
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "image_search",
"description": "Search the web for images. Use when user asks to find or show images/pictures.",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
},
},
]
# ── Search functions ──────────────────────────────────────────
async def search_web(query: str, max_results: int = 5) -> str:
from utils import context
context.record_event("web_searches")
def _sync():
try:
from ddgs import DDGS
results = list(DDGS().text(query, max_results=max_results))
if not results:
return "No results found."
lines = ["Sources:"]
for idx, r in enumerate(results, start=1):
title = r.get("title", "Untitled")
href = r.get("href", "")
body = (r.get("body", "") or "")[:350]
lines.append(f"[{idx}] {title}\nURL: {href}\nSnippet: {body}")
lines.append("When answering, cite sources inline like [1] or [2].")
return "\n\n".join(lines)
except Exception as exc:
return f"Search error: {exc}"
return await asyncio.to_thread(_sync)
async def image_search(query: str, max_results: int = 6) -> tuple[list[str], str]:
from utils import context
context.record_event("image_searches")
def _sync():
try:
from ddgs import DDGS
results = list(DDGS().images(query, max_results=max_results))
urls = [r["image"] for r in results if r.get("image")]
return urls, f"Found {len(urls)} images for '{query}'."
except Exception as exc:
return [], f"Image search error: {exc}"
return await asyncio.to_thread(_sync)
# ── OpenRouter model listing ──────────────────────────────────
_or_models_cache: list[dict] = []
_or_models_fetched_at: float = 0.0
_OR_CACHE_TTL = 3600 # 1 hour
async def fetch_or_models(free_only: bool = False) -> list[dict]:
"""Fetch OpenRouter model list, cached for 1 hour. Returns list of model dicts."""
global _or_models_cache, _or_models_fetched_at
if _or_models_cache and (time.time() - _or_models_fetched_at) < _OR_CACHE_TTL:
models = _or_models_cache
else:
import httpx
try:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(
"https://openrouter.ai/api/v1/models",
headers={"Authorization": f"Bearer {OPENROUTER_API_KEY}"},
)
resp.raise_for_status()
models = resp.json().get("data", [])
_or_models_cache = models
_or_models_fetched_at = time.time()
except Exception as exc:
logger.warning("Failed to fetch OR models: %s", exc)
return []
if free_only:
models = [
m for m in models
if str(m.get("pricing", {}).get("prompt", "1")) == "0"
]
# Sort by name
models.sort(key=lambda m: m.get("id", ""))
return models
# ── Core chat: Google (native genai) ─────────────────────────
def _oai_to_gc_contents(messages: list[dict]) -> list[gt.Content]:
"""Convert OpenAI-format message history to google-genai Content list."""
contents = []
for m in messages:
role = m["role"]
if role == "system":
continue # handled via system_instruction
gc_role = "model" if role == "assistant" else "user"
content = m["content"]
if isinstance(content, str) and content:
contents.append(gt.Content(role=gc_role, parts=[gt.Part.from_text(text=content)]))
return contents
async def _chat_google(
messages: list[dict],
system_prompt: str,
max_tokens: int,
search_enabled: bool,
) -> tuple[str, list[str]]:
trimmed = messages[-MAX_API_CONTEXT:]
contents = _oai_to_gc_contents(trimmed)
config = gt.GenerateContentConfig(
system_instruction=system_prompt,
tools=_GC_TOOLS if search_enabled else None,
max_output_tokens=max_tokens,
)
found_image_urls: list[str] = []
while True:
response = await asyncio.to_thread(
lambda c=contents: _gc.models.generate_content(
model=_google_model,
contents=c,
config=config,
)
)
candidate = response.candidates[0]
func_calls = [p.function_call for p in candidate.content.parts if p.function_call]
if not func_calls:
return response.text or "", found_image_urls
# Append model's turn (with function calls) to history
contents.append(candidate.content)
# Execute all tool calls and collect responses
tool_response_parts: list[gt.Part] = []
for fc in func_calls:
args = dict(fc.args)
name = fc.name
if name == "search_web":
result = await search_web(args.get("query", ""))
logger.info("search_web: %s", args.get("query"))
elif name == "image_search":
urls, summary = await image_search(args.get("query", ""))
found_image_urls.extend(urls)
result = summary
logger.info("image_search: %s → %d urls", args.get("query"), len(urls))
else:
result = f"Unknown tool: {name}"
tool_response_parts.append(
gt.Part.from_function_response(name=name, response={"result": result})
)
contents.append(gt.Content(role="user", parts=tool_response_parts))
# ── Core chat: OpenRouter (OpenAI client) ─────────────────────
def _or_friendly_error(exc: Exception) -> str:
"""Convert OpenRouter API errors to actionable user-facing messages."""
msg = str(exc)
if isinstance(exc, APIStatusError):
code = exc.status_code
try:
body = exc.response.json()
err = body.get("error", {})
raw = err.get("metadata", {}).get("raw", "")
provider = err.get("metadata", {}).get("provider_name", "")
except Exception:
err, raw, provider = {}, "", ""
if code == 502:
hint = f" (provider: {provider})" if provider else ""
return (
f"❌ OpenRouter provider error{hint} — the upstream model is down or unavailable.\n"
f"Try a different model with /setmodel or switch provider with /setprovider google."
)
if code == 404 or "not found" in msg.lower():
return f"❌ Model not found on OpenRouter: `{_or_model}`\nCheck available models with /models."
if code == 429:
return "❌ OpenRouter rate limit reached. Try again in a moment."
if code == 402:
return "❌ OpenRouter credits exhausted. Top up at openrouter.ai."
if raw:
return f"❌ OpenRouter error ({code}): {raw[:200]}"
return f"❌ OpenRouter error: {msg[:300]}"
async def _chat_openrouter(
messages: list[dict],
system_prompt: str,
max_tokens: int,
search_enabled: bool,
) -> tuple[str, list[str]]:
trimmed = messages[-MAX_API_CONTEXT:]
full = [{"role": "system", "content": system_prompt}] + trimmed
tools = _OAI_TOOLS if search_enabled else []
kwargs: dict = {"model": _or_model, "messages": full, "max_tokens": max_tokens}
if tools:
kwargs["tools"] = tools
kwargs["tool_choice"] = "auto"
try:
response = await _or_client.chat.completions.create(**kwargs)
except APIStatusError as exc:
# Retry without tools if the model doesn't support them
if tools and exc.status_code in (400, 422):
logger.warning("OpenRouter rejected tools (%s), retrying without", exc.status_code)
kwargs.pop("tools", None)
kwargs.pop("tool_choice", None)
try:
response = await _or_client.chat.completions.create(**kwargs)
except APIStatusError as exc2:
raise RuntimeError(_or_friendly_error(exc2)) from exc2
else:
raise RuntimeError(_or_friendly_error(exc)) from exc
except Exception as exc:
raise RuntimeError(f"❌ OpenRouter error: {exc}") from exc
found_image_urls: list[str] = []
while response.choices[0].finish_reason == "tool_calls":
choice = response.choices[0]
tool_calls = choice.message.tool_calls
full.append({
"role": "assistant",
"content": choice.message.content,
"tool_calls": [
{"id": tc.id, "type": "function",
"function": {"name": tc.function.name, "arguments": tc.function.arguments}}
for tc in tool_calls
],
})
for tc in tool_calls:
args = json.loads(tc.function.arguments)
if tc.function.name == "search_web":
result = await search_web(args.get("query", ""))
logger.info("search_web: %s", args.get("query"))
elif tc.function.name == "image_search":
urls, summary = await image_search(args.get("query", ""))
found_image_urls.extend(urls)
result = summary
logger.info("image_search: %s → %d urls", args.get("query"), len(urls))
else:
result = f"Unknown tool: {tc.function.name}"
full.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result,
"name": tc.function.name,
})
try:
response = await _or_client.chat.completions.create(
model=_or_model,
messages=full,
max_tokens=max_tokens,
)
except APIStatusError as exc:
raise RuntimeError(_or_friendly_error(exc)) from exc
return response.choices[0].message.content or "", found_image_urls
# ── Public chat dispatcher ────────────────────────────────────
def _sanitize(text: str) -> str:
from utils.formatter import strip_raw_tool_calls
return strip_raw_tool_calls(text)
def _limit_reply(text: str, max_tokens: int) -> str:
"""Silent hard cap after generation; providers can overshoot max_tokens."""
if not text:
return text
# Conservative approximation: one token is usually around 3-4 chars in English
# and varies by language. This keeps Telegram output visibly inside the user's cap.
max_chars = max(120, int(max_tokens * 3.5))
if len(text) <= max_chars:
return text
clipped = text[:max_chars].rstrip()
split_at = max(clipped.rfind("\n"), clipped.rfind(". "), clipped.rfind("؟ "), clipped.rfind("! "), clipped.rfind(" "))
if split_at > max_chars * 0.65:
clipped = clipped[:split_at].rstrip()
return clipped
def _with_limit_instruction(system_prompt: str, max_tokens: int) -> str:
return _with_runtime_instruction(
system_prompt,
max_tokens,
_current_provider,
get_model(),
)
def _with_runtime_instruction(
system_prompt: str,
max_tokens: int,
provider: str,
model: str,
) -> str:
creator_hint = _model_creator_hint(model)
return (
f"{system_prompt}\n\n"
f"RUNTIME METADATA: This bot is currently routing your replies through provider `{provider}` "
f"with model ID `{model}`. "
f"{creator_hint}"
"If the user asks what model/provider is being used, include this exact provider and model ID. "
"If the user asks who created/trained you, answer from the model family/developer when you know it; "
"do not infer creator from the hosting provider alone. If uncertain, say you only know the active model ID. "
"You may still answer naturally as the underlying model; this metadata only corrects runtime identity.\n\n"
f"REPLY LIMIT: You have at most {max_tokens} output tokens for the final answer. "
f"Plan briefly inside that budget and finish naturally before token {max_tokens}. "
"Do not mention the token limit, truncation, or that the answer was cut."
)
def _model_creator_hint(model: str) -> str:
lower = model.lower()
hints = [
("meta-llama/", "Model family/developer hint: Meta AI. "),
("llama", "Model family/developer hint: Meta AI. "),
("anthropic/", "Model family/developer hint: Anthropic. "),
("claude", "Model family/developer hint: Anthropic. "),
("openai/", "Model family/developer hint: OpenAI. "),
("gpt-", "Model family/developer hint: OpenAI. "),
("google/", "Model family/developer hint: Google. "),
("gemini", "Model family/developer hint: Google. "),
("mistralai/", "Model family/developer hint: Mistral AI. "),
("deepseek/", "Model family/developer hint: DeepSeek. "),
("qwen/", "Model family/developer hint: Alibaba/Qwen. "),
("x-ai/", "Model family/developer hint: xAI. "),
]
for needle, hint in hints:
if needle in lower:
return hint
return ""
async def chat(
messages: list[dict],
system_prompt: str,
max_tokens: int,
search_enabled: bool = True,
) -> tuple[str, list[str]]:
global _last_fallback
primary = _current_provider
fallback = "openrouter" if primary == "google" else "google"
primary_prompt = _with_runtime_instruction(
system_prompt,
max_tokens,
primary,
get_model_for_provider(primary),
)
fallback_prompt = _with_runtime_instruction(
system_prompt,
max_tokens,
fallback,
get_model_for_provider(fallback),
)
try:
if primary == "google":
reply, urls = await _chat_google(messages, primary_prompt, max_tokens, search_enabled)
else:
reply, urls = await _chat_openrouter(messages, primary_prompt, max_tokens, search_enabled)
except Exception as primary_exc:
if fallback == "openrouter" and not OPENROUTER_API_KEY:
raise
if fallback == "google" and not GOOGLE_API_KEY:
raise
from utils import context
context.record_event("provider_fallbacks")
logger.warning("Primary provider %s failed; trying %s: %s", primary, fallback, primary_exc)
try:
if fallback == "google":
reply, urls = await _chat_google(messages, fallback_prompt, max_tokens, search_enabled)
else:
reply, urls = await _chat_openrouter(messages, fallback_prompt, max_tokens, search_enabled)
_last_fallback = f"{primary} -> {fallback}"
except Exception as fallback_exc:
raise RuntimeError(
f"Primary provider failed: {primary_exc}\nFallback provider failed: {fallback_exc}"
) from fallback_exc
return _limit_reply(_sanitize(reply), max_tokens), urls
# ── Vision (native google-genai) ─────────────────────────────
async def chat_with_image(
messages: list[dict],
system_prompt: str,
max_tokens: int,
image_bytes: bytes,
mime_type: str = "image/jpeg",
caption: str = "",
analysis_mode: str = "auto",
) -> str:
def _sync():
history_lines = []
for m in messages[-6:]:
c = m["content"]
history_lines.append(
f"{m['role'].capitalize()}: {c if isinstance(c, str) else '[media]'}"
)
ctx = "\n".join(history_lines)
prompt = _with_limit_instruction(system_prompt, max_tokens)
if ctx:
prompt += f"\n\nRecent conversation:\n{ctx}"
mode_prompts = {
"auto": (
"Analyze the image carefully. Identify visible subjects, important details, "
"text/OCR, context, and answer the user's caption if provided."
),
"describe": "Describe the image clearly and include the most important visual details.",
"ocr": "Extract all readable text from the image, preserving line breaks when possible, then summarize it.",
"objects": "List the main objects, people, places, and notable visual attributes in the image.",
"question": "Answer the user's question about the image using only visible evidence.",
}
instruction = mode_prompts.get(analysis_mode, mode_prompts["auto"])
if caption:
prompt += f"\n\nImage task: {instruction}\nUser caption/question: {caption}"
else:
prompt += f"\n\nImage task: {instruction}"
resp = _gc.models.generate_content(
model=_google_model,
contents=[
gt.Content(role="user", parts=[
gt.Part.from_bytes(data=image_bytes, mime_type=mime_type),
gt.Part.from_text(text=prompt),
])
],
config=gt.GenerateContentConfig(max_output_tokens=max_tokens),
)
return resp.text or ""
return _limit_reply(_sanitize(await asyncio.to_thread(_sync)), max_tokens)
async def chat_with_images(
messages: list[dict],
system_prompt: str,
max_tokens: int,
images: list[tuple[bytes, str]],
caption: str = "",
analysis_mode: str = "auto",
) -> str:
def _sync():
history_lines = []
for m in messages[-6:]:
c = m["content"]
history_lines.append(
f"{m['role'].capitalize()}: {c if isinstance(c, str) else '[media]'}"
)
ctx = "\n".join(history_lines)
prompt = _with_limit_instruction(system_prompt, max_tokens)
if ctx:
prompt += f"\n\nRecent conversation:\n{ctx}"
prompt += (
f"\n\nAnalyze these {len(images)} images together. Compare them when useful, "
"identify relationships, extract visible text, and answer the user's caption/question."
)
if analysis_mode == "ocr":
prompt += " Focus on OCR and preserve readable text by image."
if caption:
prompt += f"\nUser caption/question: {caption}"
parts = []
for idx, (image_bytes, mime_type) in enumerate(images, start=1):
parts.append(gt.Part.from_text(text=f"Image {idx}:"))
parts.append(gt.Part.from_bytes(data=image_bytes, mime_type=mime_type))
parts.append(gt.Part.from_text(text=prompt))
resp = _gc.models.generate_content(
model=_google_model,
contents=[gt.Content(role="user", parts=parts)],
config=gt.GenerateContentConfig(max_output_tokens=max_tokens),
)
return resp.text or ""
return _limit_reply(_sanitize(await asyncio.to_thread(_sync)), max_tokens)
# ── Image generation (cheap Imagen 4 Fast — always Google) ───
async def generate_image(prompt: str) -> bytes:
# Keep image generation on the cheapest Imagen path. Do not fall back to
# standard/ultra Imagen here; those cost more per image.
models = list(dict.fromkeys([CHEAP_IMAGEN_MODEL, *IMAGEN_FALLBACK_MODELS]))
errors: list[str] = []
def _empty_reason(resp) -> str:
for attr in ("filtered_reason", "prompt_feedback", "safety_ratings"):
value = getattr(resp, attr, None)
if value:
return f"{attr}: {value}"
try:
dumped = resp.model_dump(exclude_none=True)
if dumped:
return str(dumped)[:500]
except Exception:
pass
return "empty generated_images"
def _sync():
for model in models:
try:
resp = _gc.models.generate_images(
model=model,
prompt=prompt,
config=gt.GenerateImagesConfig(number_of_images=1),
)
imgs = resp.generated_images or []
for item in imgs:
image = getattr(item, "image", None)
image_bytes = getattr(image, "image_bytes", None)
if image_bytes:
return image_bytes
errors.append(f"{model}: {_empty_reason(resp)}")
except Exception as exc:
errors.append(f"{model}: {exc}")
detail = "; ".join(errors[-3:]) if errors else "unknown empty response"
raise RuntimeError(
"No image returned by cheap Imagen 4 Fast. The prompt may have been "
f"filtered or the model returned empty output. Details: {detail}"
)
return await asyncio.to_thread(_sync)
# ── Inline quick reply ────────────────────────────────────────
async def quick_reply(query: str, system_prompt: str, max_tokens: int = 250) -> str:
google_prompt = _with_runtime_instruction(
system_prompt,
max_tokens,
"google",
get_model_for_provider("google"),
)
openrouter_prompt = _with_runtime_instruction(
system_prompt,
max_tokens,
"openrouter",
get_model_for_provider("openrouter"),
)
async def _quick_google() -> str:
def _sync():
resp = _gc.models.generate_content(
model=_google_model,
contents=[gt.Content(role="user", parts=[gt.Part.from_text(text=query)])],
config=gt.GenerateContentConfig(
system_instruction=google_prompt,
max_output_tokens=max_tokens,
),
)
return resp.text or ""
return await asyncio.to_thread(_sync)
async def _quick_openrouter() -> str:
response = await _or_client.chat.completions.create(
model=_or_model,
messages=[
{"role": "system", "content": openrouter_prompt},
{"role": "user", "content": query},
],
max_tokens=max_tokens,
)
return response.choices[0].message.content or ""
try:
reply = await (_quick_google() if _current_provider == "google" else _quick_openrouter())
except Exception:
if _current_provider == "google" and OPENROUTER_API_KEY:
reply = await _quick_openrouter()
elif _current_provider == "openrouter" and GOOGLE_API_KEY:
reply = await _quick_google()
else:
raise
return _limit_reply(_sanitize(reply), max_tokens)
+105
View File
@@ -0,0 +1,105 @@
import json
import os
from config import AUTH_FILE, OWNER_ID, OWNER_USERNAME
_authorized: set[int] = set()
_banned: set[int] = set()
_known_users: set[int] = set()
_public_mode: bool = False
def load():
global _authorized, _banned
if os.path.exists(AUTH_FILE):
with open(AUTH_FILE) as f:
data = json.load(f)
_authorized = {int(x) for x in data.get("authorized", [])}
_banned = {int(x) for x in data.get("banned", [])}
if OWNER_ID:
_authorized.add(OWNER_ID)
def save():
with open(AUTH_FILE, "w") as f:
json.dump(
{"authorized": sorted(_authorized), "banned": sorted(_banned)},
f,
indent=2,
)
def is_owner(user_id: int, username: str = "") -> bool:
if OWNER_ID > 0 and user_id == OWNER_ID:
return True
if OWNER_USERNAME and username and username.lower() == OWNER_USERNAME.lower():
return True
return False
def is_strictly_authorized(user_id: int, username: str = "") -> bool:
"""Always checks the actual auth list — ignores public mode. Used for sensitive commands (VC)."""
if is_owner(user_id, username):
return True
return user_id in _authorized and user_id not in _banned
def is_authorized(user_id: int, username: str = "") -> bool:
if is_owner(user_id, username):
return True
if user_id in _banned:
return False
if _public_mode:
return True
return user_id in _authorized
def set_public(enabled: bool):
global _public_mode
_public_mode = enabled
try:
from services import settings as settings_store
settings_store.save()
except Exception:
pass
def is_public() -> bool:
return _public_mode
def add_user(user_id: int):
_authorized.add(user_id)
_banned.discard(user_id)
save()
def remove_user(user_id: int):
_authorized.discard(user_id)
save()
def ban_user(user_id: int):
_banned.add(user_id)
_authorized.discard(user_id)
save()
def unban_user(user_id: int):
_banned.discard(user_id)
save()
def register_known(user_id: int, username: str = ""):
_known_users.add(user_id)
def get_authorized() -> list[int]:
return sorted(_authorized)
def get_banned() -> list[int]:
return sorted(_banned)
def get_known_users() -> list[int]:
return sorted(_known_users)
+41
View File
@@ -0,0 +1,41 @@
import json
import os
from config import SETTINGS_FILE
from services import ai_client, auth
from utils import context
_loading = False
def load():
global _loading
if not os.path.exists(SETTINGS_FILE):
save()
return
_loading = True
try:
with open(SETTINGS_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, dict):
ai_client.import_provider_settings(data.get("ai", {}))
context.import_settings(data.get("context", {}))
auth.set_public(bool(data.get("public_mode", False)))
finally:
_loading = False
def save():
if _loading:
return
data = {
"ai": ai_client.export_provider_settings(),
"public_mode": auth.is_public(),
"context": context.export_settings(),
}
tmp = f"{SETTINGS_FILE}.tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
f.write("\n")
os.replace(tmp, SETTINGS_FILE)
+26
View File
@@ -0,0 +1,26 @@
TG_LIMIT = 4096
def chunk_text(text: str, max_length: int = TG_LIMIT) -> list[str]:
"""Split text into ≤max_length chunks, preferring newline then space boundaries."""
if len(text) <= max_length:
return [text]
chunks: list[str] = []
while text:
if len(text) <= max_length:
chunks.append(text)
break
split_at = text.rfind('\n', 0, max_length)
if split_at == -1:
split_at = text.rfind(' ', 0, max_length)
if split_at == -1:
split_at = max_length
chunk = text[:split_at].strip()
if chunk:
chunks.append(chunk)
text = text[split_at:].strip()
return chunks
+216
View File
@@ -0,0 +1,216 @@
import time
from collections import defaultdict
from config import (
MAX_HISTORY_MESSAGES,
MAX_TOKENS_DEFAULT,
DEFAULT_SYSTEM_PROMPT,
IMAGE_DAILY_LIMIT,
MAX_INPUT_TOKENS,
)
# chat_id → list of {"role": ..., "content": ...}
_histories: dict[int, list] = defaultdict(list)
# user_id → settings dict
_user_settings: dict[int, dict] = {}
_system_prompt: str = DEFAULT_SYSTEM_PROMPT
_start_time: float = time.time()
_msg_count: int = 0
_event_counts: dict[str, int] = defaultdict(int)
_last_error: str = ""
# ── History ──────────────────────────────────────────────────
def get_history(chat_id: int) -> list:
return list(_histories[chat_id])
def add_message(chat_id: int, role: str, content):
_histories[chat_id].append({"role": role, "content": content})
if len(_histories[chat_id]) > MAX_HISTORY_MESSAGES:
_histories[chat_id] = _histories[chat_id][-MAX_HISTORY_MESSAGES:]
def clear_history(chat_id: int):
_histories[chat_id].clear()
def clear_all_histories():
_histories.clear()
def pop_last_message(chat_id: int):
"""Remove the most recent message from the history (for error rollback)."""
if _histories[chat_id]:
_histories[chat_id].pop()
def history_summary(chat_id: int, n: int = 6) -> str:
msgs = _histories[chat_id][-n:]
lines = []
for m in msgs:
content = m["content"]
if isinstance(content, list):
text = "[image/media]"
else:
text = str(content)[:200]
lines.append(f"{m['role'].capitalize()}: {text}")
return "\n".join(lines) if lines else "(empty)"
def estimate_tokens(text) -> int:
if text is None:
return 0
if not isinstance(text, str):
text = str(text)
# Conservative enough for cost guarding without provider-specific tokenizers.
return max(1, len(text) // 3)
def estimate_input_tokens(messages: list[dict], system_prompt: str = "") -> int:
total = estimate_tokens(system_prompt)
for msg in messages:
total += estimate_tokens(msg.get("role", ""))
total += estimate_tokens(msg.get("content", ""))
return total
def reset_chat_if_input_too_large(chat_id: int, system_prompt: str = "", keep_last: bool = True) -> bool:
messages = get_history(chat_id)
if estimate_input_tokens(messages, system_prompt) <= MAX_INPUT_TOKENS:
return False
last = messages[-1:] if keep_last else []
clear_history(chat_id)
for msg in last:
add_message(chat_id, msg.get("role", "user"), msg.get("content", ""))
record_event("history_auto_resets")
return True
# ── User settings ─────────────────────────────────────────────
def get_user_settings(user_id: int) -> dict:
if user_id not in _user_settings:
_user_settings[user_id] = {
"max_tokens": MAX_TOKENS_DEFAULT,
"search_enabled": True,
"lang": "auto",
}
return _user_settings[user_id]
def set_user_setting(user_id: int, key: str, value):
get_user_settings(user_id)[key] = value
_autosave()
# ── System prompt ─────────────────────────────────────────────
def get_system_prompt() -> str:
return _system_prompt
def set_system_prompt(prompt: str):
global _system_prompt
_system_prompt = prompt
_autosave()
def reset_system_prompt():
global _system_prompt
_system_prompt = DEFAULT_SYSTEM_PROMPT
_autosave()
def export_settings() -> dict:
return {
"system_prompt": _system_prompt,
"user_settings": {
str(uid): dict(settings)
for uid, settings in _user_settings.items()
},
}
def import_settings(data: dict):
global _system_prompt, _user_settings
if isinstance(data.get("system_prompt"), str):
_system_prompt = data["system_prompt"]
raw_settings = data.get("user_settings", {})
if isinstance(raw_settings, dict):
loaded: dict[int, dict] = {}
for uid, settings in raw_settings.items():
try:
user_id = int(uid)
except (TypeError, ValueError):
continue
if isinstance(settings, dict):
loaded[user_id] = dict(settings)
_user_settings = loaded
# ── Stats ─────────────────────────────────────────────────────
def increment_msg():
global _msg_count
_msg_count += 1
def record_event(name: str, amount: int = 1):
_event_counts[name] += amount
def record_error(message: str):
global _last_error
_event_counts["errors"] += 1
_last_error = str(message)[:300]
def _autosave():
try:
from services import settings as settings_store
settings_store.save()
except Exception:
pass
# ── Image rate limiting (cost guard) ─────────────────────────
_image_log: dict[int, list[float]] = defaultdict(list) # user_id → timestamps
def can_generate_image(user_id: int) -> bool:
cutoff = time.time() - 86400
_image_log[user_id] = [t for t in _image_log[user_id] if t > cutoff]
return len(_image_log[user_id]) < IMAGE_DAILY_LIMIT
def record_image(user_id: int):
_image_log[user_id].append(time.time())
record_event("images_generated")
def images_used_today(user_id: int) -> int:
cutoff = time.time() - 86400
return len([t for t in _image_log[user_id] if t > cutoff])
# ── Stats ─────────────────────────────────────────────────────
def get_stats() -> dict:
elapsed = int(time.time() - _start_time)
h, remainder = divmod(elapsed, 3600)
m, s = divmod(remainder, 60)
return {
"uptime": f"{h}h {m}m {s}s",
"uptime_seconds": elapsed,
"messages_processed": _msg_count,
"active_chats": len([k for k, v in _histories.items() if v]),
"stored_history_messages": sum(len(v) for v in _histories.values()),
"events": dict(_event_counts),
"last_error": _last_error,
}
+157
View File
@@ -0,0 +1,157 @@
"""
Convert AI markdown output → Telegram HTML.
Supports: **bold**, *italic*, `code`, ```blocks```, ~~strike~~, ||spoiler||,
[links](url), # headers, bullet lists, horizontal rules, markdown tables.
"""
import re
from html import escape as _esc
def _split_table_row(line: str) -> list[str]:
return [_plain_table_cell(cell.strip()) for cell in line.strip().strip("|").split("|")]
def _plain_table_cell(cell: str) -> str:
cell = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r"\1 (\2)", cell)
cell = re.sub(r"(```|`|\*\*|__|\*|_|~~|\|\|)", "", cell)
return cell.strip()
def _is_table_separator(line: str) -> bool:
cells = _split_table_row(line)
return bool(cells) and all(re.fullmatch(r":?-{3,}:?", cell.strip()) for cell in cells)
def _render_table(lines: list[str]) -> str:
rows = [_split_table_row(line) for line in lines if not _is_table_separator(line)]
if not rows:
return "\n".join(lines)
width = max(len(row) for row in rows)
normalized = [row + [""] * (width - len(row)) for row in rows]
col_widths = [max(len(row[i]) for row in normalized) for i in range(width)]
rendered = []
for idx, row in enumerate(normalized):
rendered.append(" ".join(row[i].ljust(col_widths[i]) for i in range(width)).rstrip())
if idx == 0 and len(normalized) > 1:
rendered.append(" ".join("-" * col_widths[i] for i in range(width)).rstrip())
return "```\n" + "\n".join(rendered) + "\n```"
def _normalize_tables(text: str) -> str:
lines = text.splitlines()
out: list[str] = []
i = 0
while i < len(lines):
if (
"|" in lines[i]
and i + 1 < len(lines)
and "|" in lines[i + 1]
and _is_table_separator(lines[i + 1])
):
block = [lines[i], lines[i + 1]]
i += 2
while i < len(lines) and "|" in lines[i] and lines[i].strip():
block.append(lines[i])
i += 1
out.append(_render_table(block))
continue
out.append(lines[i])
i += 1
return "\n".join(out)
def _protect(text: str) -> tuple[str, list[str]]:
"""Extract code blocks/inline code and replace with placeholders."""
saved: list[str] = []
def _save_block(m: re.Match) -> str:
lang = (m.group(1) or "").strip()
code = _esc(m.group(2))
tag = f'<pre><code class="language-{lang}">{code}</code></pre>' if lang else f"<pre>{code}</pre>"
saved.append(tag)
return f"\x00BLK{len(saved)-1}\x00"
def _save_inline(m: re.Match) -> str:
saved.append(f"<code>{_esc(m.group(1))}</code>")
return f"\x00INL{len(saved)-1}\x00"
text = re.sub(r"```(\w*)\n?(.*?)```", _save_block, text, flags=re.DOTALL)
text = re.sub(r"`([^`\n]+)`", _save_inline, text)
return text, saved
def _restore(text: str, saved: list[str]) -> str:
for i, block in enumerate(saved):
text = text.replace(f"\x00BLK{i}\x00", block)
text = text.replace(f"\x00INL{i}\x00", block)
return text
def md_to_tg(text: str) -> str:
"""Convert AI markdown response to Telegram-safe HTML."""
text = _normalize_tables(text)
text, saved = _protect(text)
# Escape HTML special chars in non-code content
text = _esc(text)
# ── Block-level ──────────────────────────────────────────
# Headers → bold line
text = re.sub(r"^#{1,6}\s+(.+)$", r"<b>\1</b>", text, flags=re.MULTILINE)
# Horizontal rule
text = re.sub(r"^[-*_]{3,}$", "────────────────", text, flags=re.MULTILINE)
# ── Inline ───────────────────────────────────────────────
# Bold **text** or __text__
text = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", text, flags=re.DOTALL)
text = re.sub(r"__(.+?)__", r"<b>\1</b>", text)
# Italic *text* or _text_ (don't match ** or __)
text = re.sub(r"(?<!\*)\*(?!\*)([^*\n]+?)(?<!\*)\*(?!\*)", r"<i>\1</i>", text)
text = re.sub(r"(?<!_)_(?!_)([^_\n]+?)(?<!_)_(?!_)", r"<i>\1</i>", text)
# Strikethrough ~~text~~
text = re.sub(r"~~(.+?)~~", r"<s>\1</s>", text, flags=re.DOTALL)
# Spoiler ||text||
text = re.sub(r"\|\|(.+?)\|\|", r"<tg-spoiler>\1</tg-spoiler>", text, flags=re.DOTALL)
# Links [text](url) — must use HTML escape for the URL too
def _link(m: re.Match) -> str:
label = m.group(1)
url = m.group(2).replace('"', "&quot;")
return f'<a href="{url}">{label}</a>'
text = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", _link, text)
text = _restore(text, saved)
return text
def safe_send(text: str) -> tuple[str, str | None]:
"""Return (formatted_text, parse_mode). Falls back to plain if formatting looks empty."""
formatted = md_to_tg(text)
from telegram.constants import ParseMode
return formatted, ParseMode.HTML
# Matches raw XML-style tool call blocks that some models leak into text output.
# Covers: <tool_call>...</tool_call>, <longcattoolcall>...</longcattoolcall>,
# <function_calls>...</function_calls>, <function=name>...</function>, etc.
_RAW_TOOL_CALL_RE = re.compile(
r"<\w*tool\w*>.*?</\w*tool\w*>"
r"|<function[^>]*>.*?</function[^>]*>"
r"|<function_calls>.*?</function_calls>"
r"|<\w*toolcall\w*>.*?</\w*toolcall\w*>",
re.DOTALL | re.IGNORECASE,
)
def strip_raw_tool_calls(text: str) -> str:
"""Remove raw XML tool-call blocks that models sometimes leak into responses."""
cleaned = _RAW_TOOL_CALL_RE.sub("", text).strip()
# Collapse any runs of blank lines left behind
cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
return cleaned
+45
View File
@@ -0,0 +1,45 @@
import re
import unicodedata
# Normalize Arabic lookalikes to Persian
_ARABIC_TO_PERSIAN = str.maketrans({
'ي': 'ی', # Arabic ya → Persian ya
'ك': 'ک', # Arabic kaf → Persian kaf
'ى': 'ی', # Alef maqsura → Persian ya
})
_PUPPY_RE = re.compile(r'puppy|پاپی|پاپي', re.IGNORECASE | re.UNICODE)
# Rough threshold: if >15% of chars are Arabic/Persian script → Persian
_PERSIAN_BLOCK = re.compile(r'[؀-ۿ]')
def normalize(text: str) -> str:
text = unicodedata.normalize('NFC', text)
text = text.translate(_ARABIC_TO_PERSIAN)
text = text.replace('‌', '').replace('‍', '') # ZWNJ / ZWJ
return text
def is_puppy_mentioned(text: str) -> bool:
if not text:
return False
return bool(_PUPPY_RE.search(normalize(text)))
def detect_language(text: str) -> str:
if not text:
return 'en'
persian_count = len(_PERSIAN_BLOCK.findall(text))
return 'fa' if persian_count > len(text) * 0.15 else 'en'
def strip_puppy_prefix(text: str) -> str:
"""Remove 'Puppy/پاپی' name prefix from a command-style message."""
cleaned = re.sub(
r'^[\s]*(puppy|پاپی|پاپي)[،,:\s]+',
'',
normalize(text),
flags=re.IGNORECASE | re.UNICODE,
)
return cleaned.strip() or text.strip()