import os
import logging
from dotenv import load_dotenv
import html
import io
import asyncio
from PIL import Image, ImageDraw, ImageFont
import re
import base64 # encode images to base64 for multimodal API
import mimetypes # determine MIME type when possible
import json
from playwright.async_api import async_playwright
from llm_providers.openrouter_provider import call_openrouter_stream
from multi_agent_system import get_orchestrator, AgentState
from telegram import Update, InputFile, ReplyKeyboardMarkup, Message, LabeledPrice, BotCommand, BotCommandScopeDefault, InlineKeyboardButton, InlineKeyboardMarkup, CallbackQuery
from telegram.constants import ChatAction
from telegram.ext import Application, MessageHandler, CommandHandler, ContextTypes, filters, PreCheckoutQueryHandler, CallbackQueryHandler
# Gemini provider
from llm_providers.gemini_provider import call_gemini_stream
# Subject prompts
from prompts import get_prompts
from prompts.tg_suffix import TG_PLAIN_TEXT_SUFFIX
from db_config import connection_pool
from datetime import datetime, timedelta, date
import mysql.connector
# --- USER CONFIG (edit here) ---
# Comments in English by request
USER_TG_BOT_TOKEN = "8552679246:AAEkaPPhgIYdNZJ5dgk2N6eyqFPyiAYs5mQ" # Insert your Telegram Bot token here
USER_GEMINI_API_KEY = "AIzaSyAEhkP212iADaJt7UmMi0ETvFffpNDEMpA" # Insert your Gemini API key here
# Optional: set a fixed system instruction for the model (leave empty to use default)
USER_SYSTEM_INSTRUCTION = "ОШИБКА 1"
# Default system instruction used if none provided above or via env
DEFAULT_SYSTEM_INSTRUCTION = "ОШИБКА 2"
# --- END USER CONFIG ---
load_dotenv()
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
# Configuration
TG_BOT_TOKEN = (USER_TG_BOT_TOKEN or os.getenv("TG_BOT_TOKEN", "")).strip() or "8552679246:AAEkaPPhgIYdNZJ5dgk2N6eyqFPyiAYs5mQ"
GEMINI_API_KEY = (USER_GEMINI_API_KEY or os.getenv("GEMINI_API_KEY", "")).strip()
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash").strip()
GEMINI_ROUTER_MODEL = os.getenv("GEMINI_ROUTER_MODEL", "").strip() or GEMINI_MODEL
SYSTEM_INSTRUCTION = (
(USER_SYSTEM_INSTRUCTION or os.getenv("GEMINI_SYSTEM_INSTRUCTION", "")).strip()
or DEFAULT_SYSTEM_INSTRUCTION
)
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY", "").strip()
OPENROUTER_MODEL = os.getenv("OPENROUTER_MODEL", "google/gemini-2.0-flash-lite").strip()
OPENROUTER_URL = os.getenv("OPENROUTER_URL", "https://openrouter.ai/api/v1/chat/completions").strip()
# Admin IDs with unlimited access
ADMIN_IDS = {5245006576, 181581865, 2007688483, 5475206825}
# Short ad message shown while the model is thinking
AD_MESSAGE_TEXT = "Пока готовлю ответ… Загляните на наш сайт ClassGPT: https://classgpt.ru"
AD_IMAGE_PATH = os.getenv("AD_IMAGE_PATH", "static/images/classgpt_ad.jpg")
CLASSIFIER_SYSTEM_PROMPT = (
'''
Твоя задача — определить, должен ли итоговый ответ пользователю Telegram-бота быть выведен текстом или изображением. Telegram некорректно отображает сложные математические формулы, поэтому формат ответа должен выбирается строго по правилам ниже.
Выводить ответ изображением:
- Если ответ содержит формулы или выражения, которые нельзя корректно передать обычным текстом.
Выводить ответ текстом:
- Если ответ содержит текст и формулы или выражения, которые можно корректно передать обычным текстом.
Логика принятия решения:
Продумать, как будет выглядеть ответ пользователю.
Определить, содержит ли ответ сложные формулы или и будет ли он корректно отображаться в Telegram как текст.
В случае неопределенности выбирать вариант "IMAGE".
Формат ответа:
Модель должна вернуть одно слово (БЕЗ КАВЫЧЕК):
"IMAGE" — если итоговый ответ пользователю требуется выводить изображением
"TEXT" — если ответ можно вывести текстом
'''
)
TELEGRAM_MESSAGE_LIMIT = 3500
MATH_KEYWORDS = [
]
# --- Playwright Browser Global ---
PLAYWRIGHT_MANAGER = None
BROWSER_INSTANCE = None
BROWSER_LOCK = asyncio.Lock()
async def get_browser():
"""Get or create a global Playwright browser instance."""
global BROWSER_INSTANCE, PLAYWRIGHT_MANAGER, BROWSER_LOCK
async with BROWSER_LOCK:
if BROWSER_INSTANCE is None:
try:
PLAYWRIGHT_MANAGER = await async_playwright().start()
BROWSER_INSTANCE = await PLAYWRIGHT_MANAGER.chromium.launch(
headless=True,
args=["--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage", "--disable-gpu"]
)
logger.info("Playwright Chromium launched successfully.")
except Exception as err:
logger.error("Failed to launch Playwright: %s", err, exc_info=True)
if "Executable doesn't exist" in str(err):
logger.error("CRITICAL: Chromium not installed. Run 'playwright install chromium'")
raise
return BROWSER_INSTANCE
# --- DATABASE HELPERS ---
def setup_tg_db():
"""Ensure the telegram user table exists."""
# Comments in English by request
conn = connection_pool.get_connection()
cursor = conn.cursor()
try:
cursor.execute("""
CREATE TABLE IF NOT EXISTS tg_user (
tg_user_id BIGINT PRIMARY KEY,
plus_until DATETIME DEFAULT NULL,
daily_messages_count INT DEFAULT 0,
last_message_date DATE DEFAULT NULL
)
""")
conn.commit()
except mysql.connector.Error as err:
logger.error("Failed to setup tg_user table: %s", err)
finally:
cursor.close()
conn.close()
def get_user_status(tg_user_id: int):
"""Check if user has plus and how many messages they sent today."""
# Comments in English by request
conn = connection_pool.get_connection()
cursor = conn.cursor(dictionary=True)
try:
cursor.execute("SELECT plus_until, daily_messages_count, last_message_date FROM tg_user WHERE tg_user_id = %s", (tg_user_id,))
user = cursor.fetchone()
now = datetime.now()
today = date.today()
if not user:
# Create user if not exists
cursor.execute("INSERT INTO tg_user (tg_user_id, last_message_date) VALUES (%s, %s)", (tg_user_id, today))
conn.commit()
return False, 0 # is_plus=False, daily_count=0
is_plus = False
if user['plus_until'] and user['plus_until'] > now:
is_plus = True
daily_count = user['daily_messages_count']
if user['last_message_date'] != today:
daily_count = 0
# Reset daily count in DB
cursor.execute("UPDATE tg_user SET daily_messages_count = 0, last_message_date = %s WHERE tg_user_id = %s", (today, tg_user_id))
conn.commit()
return is_plus, daily_count
except mysql.connector.Error as err:
logger.error("Database error in get_user_status: %s", err)
return False, 0
finally:
cursor.close()
conn.close()
def increment_message_count(tg_user_id: int):
"""Increment daily message count for the user."""
# Comments in English by request
conn = connection_pool.get_connection()
cursor = conn.cursor()
try:
cursor.execute("UPDATE tg_user SET daily_messages_count = daily_messages_count + 1 WHERE tg_user_id = %s", (tg_user_id,))
conn.commit()
except mysql.connector.Error as err:
logger.error("Database error in increment_message_count: %s", err)
finally:
cursor.close()
conn.close()
def set_user_plus(tg_user_id: int, days: int = 30):
"""Grant plus status to a user."""
# Comments in English by request
conn = connection_pool.get_connection()
cursor = conn.cursor()
try:
plus_until = datetime.now() + timedelta(days=days)
cursor.execute("""
INSERT INTO tg_user (tg_user_id, plus_until)
VALUES (%s, %s)
ON DUPLICATE KEY UPDATE plus_until = %s
""", (tg_user_id, plus_until, plus_until))
conn.commit()
logger.info("User %s granted plus until %s", tg_user_id, plus_until)
except mysql.connector.Error as err:
logger.error("Database error in set_user_plus: %s", err)
finally:
cursor.close()
conn.close()
# --- END DATABASE HELPERS ---
async def _text_to_png(text: str) -> bytes:
"""Render Markdown+LaTeX to PNG using Playwright (HTML-based)."""
try:
browser = await get_browser()
page = await browser.new_page(viewport={"width": 900, "height": 800})
# HTML template with KaTeX and Marked.js
html_template = rf"""
"""
await page.set_content(html_template, wait_until="domcontentloaded")
# Give some time for KaTeX and fonts to load/render
await asyncio.sleep(0.5)
# Wait for KaTeX to be present if any math exists
if "$" in text or "\\" in text:
try:
# We use a shorter timeout for wait_for_selector
await page.wait_for_selector(".katex", timeout=3000)
except Exception as e:
logger.debug("KaTeX rendering timeout or not found: %s", e)
# Adjust height to content
try:
height = await page.evaluate("document.documentElement.scrollHeight")
# Ensure a minimum height
height = max(height, 100)
await page.set_viewport_size({"width": 900, "height": height})
except Exception as e:
logger.error("Failed to evaluate scrollHeight: %s", e)
# fallback height
await page.set_viewport_size({"width": 900, "height": 1200})
# Take screenshot
png_bytes = await page.screenshot(full_page=True, animations="disabled", timeout=5000)
await page.close()
return png_bytes
except Exception as e:
logger.error("Error in _text_to_png: %s", e, exc_info=True)
raise
def _main_keyboard() -> ReplyKeyboardMarkup:
"""Return persistent keyboard with New Chat button."""
# Comments in English by request
rows = [
["🆕 Новый чат"],
["Алгебра", "Геометрия"],
["Физика", "Химия"],
["Биология", "География"],
["История", "Обществознание"],
["Информатика", "Английский язык"],
["Вероятность и статистика"],
]
return ReplyKeyboardMarkup(rows, resize_keyboard=True, one_time_keyboard=False)
def _looks_math_heavy(text: str) -> bool:
"""Heuristic fallback to detect math-heavy prompts when classifier is unavailable."""
# Comments in English by request
lowered = text.lower()
if any(keyword in lowered for keyword in MATH_KEYWORDS):
return True
if re.search(r"[=<>±≈∑∫√∞]", text):
return True
if re.search(r"(sin|cos|tan|tg|log|ln|sqrt)", lowered):
return True
if re.search(r"[a-zа-я]\s*[\+\-\*/]\s*\d", lowered):
return True
if re.search(r"\d+\s*[a-zа-я]", lowered):
return True
return False
def _split_text_for_telegram(text: str, limit: int = TELEGRAM_MESSAGE_LIMIT) -> list[str]:
"""Split long responses into Telegram-safe chunks."""
# Comments in English by request
if len(text) <= limit:
return [text]
chunks: list[str] = []
start = 0
length = len(text)
while start < length:
end = min(length, start + limit)
if end < length:
newline = text.rfind("\n", start, end)
if newline != -1 and newline > start + 200:
end = newline + 1
chunk = text[start:end].strip()
if chunk:
chunks.append(chunk)
start = end
return chunks or [text]
def _build_classifier_payload(
user_text: str,
context: ContextTypes.DEFAULT_TYPE,
max_items: int = 6,
assistant_text: str | None = None,
) -> str:
"""Construct payload with recent chat history plus latest response for the classifier."""
# Comments in English by request
subject = context.chat_data.get("subject") or "не указан"
history = _get_chat_history(context)
if history:
tail = history[-max_items:]
formatted: list[str] = []
for entry in tail:
role = entry.get("role", "user")
content = entry.get("content", "")
if not isinstance(content, str):
content = str(content)
content = content.strip()
if len(content) > 400:
content = content[-400:]
formatted.append(f"{role.upper()}: {content}")
history_text = "\n".join(formatted)
else:
history_text = "(нет предыдущих сообщений)"
payload_parts = [
f"Контекст предмета: {subject}",
f"Недавний диалог:\n{history_text}",
f"Текущее сообщение пользователя:\nUSER: {user_text.strip()}",
]
if assistant_text:
payload_parts.append(f"Черновик ответа ассистента:\nASSISTANT: {assistant_text.strip()}")
return "\n".join(payload_parts)
def _run_classifier_request(prompt_payload: str) -> str:
"""Blocking call to Gemini classifier."""
# Comments in English by request
response_chunks: list[str] = []
try:
for chunk in call_gemini_stream(
chat_history=[{"role": "user", "content": prompt_payload}],
model_name=GEMINI_ROUTER_MODEL,
api_key=GEMINI_API_KEY,
system_instruction=CLASSIFIER_SYSTEM_PROMPT,
subject="classifier",
reasoner_mode=0,
):
if chunk:
response_chunks.append(chunk)
except Exception as err:
logger.warning("Classifier request failed: %s", err, exc_info=True)
raise
return "".join(response_chunks)
async def _should_render_response_as_image(
user_text: str,
assistant_text: str,
context: ContextTypes.DEFAULT_TYPE,
) -> bool:
"""Decide whether the final answer must be rendered as an image based on the response content."""
# Comments in English by request
reference_text = assistant_text.strip() or user_text.strip()
if not reference_text:
return False
if not GEMINI_API_KEY:
return _looks_math_heavy(reference_text)
try:
payload = _build_classifier_payload(user_text, context, assistant_text=assistant_text)
decision = await asyncio.to_thread(_run_classifier_request, payload)
normalized = decision.strip().lower()
if normalized.startswith("image"):
return True
if normalized.startswith("text"):
return False
except Exception:
return _looks_math_heavy(reference_text)
return _looks_math_heavy(reference_text)
async def _send_text_reply(update: Update, context: ContextTypes.DEFAULT_TYPE, text: str) -> None:
"""Send long responses as plain Telegram messages."""
# Comments in English by request
chunks = _split_text_for_telegram(text)
total = len(chunks)
for idx, chunk in enumerate(chunks):
msg = await update.effective_chat.send_message(
chunk,
reply_markup=_main_keyboard() if idx == total - 1 else None,
)
_track_message_id(context, msg.message_id)
def _get_context_system_prompt(context: ContextTypes.DEFAULT_TYPE, prefer_plain: bool = False) -> str:
"""Select system prompt variant stored in chat_data."""
# Comments in English by request
if prefer_plain:
return (
context.chat_data.get("system_prompt_plain")
or context.chat_data.get("system_prompt_rich")
or SYSTEM_INSTRUCTION
)
return context.chat_data.get("system_prompt_rich") or SYSTEM_INSTRUCTION
def _get_chat_history(context: ContextTypes.DEFAULT_TYPE) -> list[dict]:
"""Return per-chat history list stored in chat_data."""
# Comments in English by request
history = context.chat_data.get("history")
if history is None:
history = []
context.chat_data["history"] = history
return history
def _append_history(context: ContextTypes.DEFAULT_TYPE, role: str, content: str) -> None:
"""Append message to per-chat history with a cap."""
# Comments in English by request
history = _get_chat_history(context)
history.append({"role": role, "content": content})
try:
max_messages = int(os.getenv("BOT_HISTORY_MAX_MESSAGES", "24"))
except Exception:
max_messages = 24
if len(history) > max_messages:
del history[0 : len(history) - max_messages]
def _reset_history(context: ContextTypes.DEFAULT_TYPE) -> None:
"""Clear per-chat history."""
# Comments in English by request
context.chat_data["history"] = []
def _get_tracked_message_ids(context: ContextTypes.DEFAULT_TYPE) -> list[int]:
"""Return per-chat list of Telegram message IDs scheduled for purge."""
tracked = context.chat_data.get("tracked_message_ids")
if tracked is None:
tracked = []
context.chat_data["tracked_message_ids"] = tracked
return tracked
def _track_message_id(context: ContextTypes.DEFAULT_TYPE, message_id: int | None) -> None:
"""Remember message id so we can delete it when chat resets."""
if not message_id:
return
tracked = _get_tracked_message_ids(context)
tracked.append(message_id)
try:
max_tracked = int(os.getenv("BOT_TRACKED_MESSAGES_MAX", "400"))
except Exception:
max_tracked = 400
if len(tracked) > max_tracked:
del tracked[0 : len(tracked) - max_tracked]
def _remove_tracked_message_id(context: ContextTypes.DEFAULT_TYPE, message_id: int | None) -> None:
"""Forget message id if it was deleted earlier."""
if not message_id:
return
tracked = context.chat_data.get("tracked_message_ids")
if not tracked:
return
try:
tracked.remove(message_id)
except ValueError:
pass
async def _purge_tracked_messages(context: ContextTypes.DEFAULT_TYPE, chat_id: int | None) -> None:
"""Delete all tracked messages belonging to previous chat context."""
if not chat_id:
return
tracked_snapshot = list(_get_tracked_message_ids(context))
if not tracked_snapshot:
return
context.chat_data["tracked_message_ids"] = []
for message_id in tracked_snapshot:
try:
await context.bot.delete_message(chat_id=chat_id, message_id=message_id)
except Exception:
continue
def _track_incoming_update_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Store current Telegram message id if available."""
if update.message:
_track_message_id(context, update.message.message_id)
# --- Ad helpers ---
def _ensure_ad_image(path: str = AD_IMAGE_PATH) -> str | None:
"""Ensure local ad image exists; generate lightweight version if missing."""
# Comments in English by request
try:
os.makedirs(os.path.dirname(path), exist_ok=True)
if os.path.exists(path):
return path
# Basic purple background
width, height = 1080, 1920
bg_top = (82, 47, 132)
bg_bottom = (55, 25, 95)
img = Image.new("RGB", (width, height), bg_bottom)
draw = ImageDraw.Draw(img)
# Create simple vertical gradient
for y in range(height):
blend = y / height
r = int(bg_top[0] * (1 - blend) + bg_bottom[0] * blend)
g = int(bg_top[1] * (1 - blend) + bg_bottom[1] * blend)
b = int(bg_top[2] * (1 - blend) + bg_bottom[2] * blend)
draw.line([(0, y), (width, y)], fill=(r, g, b))
title_font = _get_bold_font(80)
accent_font = _get_bold_font(96)
text_font = _get_font(48)
small_font = _get_font(40)
accent_color = (255, 215, 64)
white = (245, 240, 250)
# Title
draw.text((72, 120), "Переходи на сайт", font=title_font, fill=white)
draw.text((72, 210), "ClassGPT!", font=accent_font, fill=accent_color)
bullets = [
"Выбор учебника по любому предмету",
"Прохождение параграфов вместе с репетитором ClassGPT",
"Современный и удобный интерфейс",
"Доступен на мобильных устройствах и на компьютере",
"Решение задач со школьным оформлением",
]
y = 380
check = "✓ "
for item in bullets:
draw.text((96, y), check, font=text_font, fill=accent_color)
draw.text((96 + 36, y), item, font=text_font, fill=white)
y += 90
# Website footer
draw.text((72, height - 160), "classgpt.ru", font=small_font, fill=accent_color)
img.save(path, format="JPEG", quality=92)
return path
except Exception as err:
logger.warning("Failed to prepare ad image: %s", err)
return None
async def _send_ad_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Message | None:
"""Send short ad message while the model is thinking."""
# Comments in English by request
try:
ad_path = _ensure_ad_image()
if ad_path:
with open(ad_path, "rb") as photo:
msg = await update.effective_chat.send_photo(
photo=photo,
caption=AD_MESSAGE_TEXT,
)
else:
msg = await update.effective_chat.send_message(
AD_MESSAGE_TEXT,
disable_web_page_preview=True,
)
_track_message_id(context, msg.message_id)
return msg
except Exception as err:
logger.warning("Failed to send ad message: %s", err)
return None
async def _safe_delete_message(context: ContextTypes.DEFAULT_TYPE, chat_id: int, message_id: int) -> None:
"""Safely delete a message; ignore errors (e.g., already deleted)."""
# Comments in English by request
try:
await context.bot.delete_message(chat_id=chat_id, message_id=message_id)
_remove_tracked_message_id(context, message_id)
except Exception:
# Silently ignore deletion errors to avoid disrupting user flow
pass
# Subject name to subject_id mapping aligned with website backend
SUBJECT_ID_MAP: dict[str, int] = {
"алгебра": 2,
"геометрия": 3,
"физика": 1,
"химия": 4,
"биология": 5,
"история": 6,
"география": 10,
"информатика": 12,
"обществознание": 8,
"английский язык": 11,
"вероятность и статистика": 13,
}
# Subjects that must always reply with plain Telegram-safe text
FORCE_TEXT_SUBJECTS: set[str] = {
"биология",
"география",
"история",
"обществознание",
"английский язык",
}
# Subjects mapping (button -> prompts)
def _build_subject_prompt_maps() -> tuple[dict[str, str], dict[str, str]]:
"""Collect both rich and Telegram-safe prompts per subject."""
system_map: dict[str, str] = {}
plain_map: dict[str, str] = {}
for name_lower, subject_id in SUBJECT_ID_MAP.items():
try:
system_prompt, _, plain_prompt = get_prompts(subject_id)
except Exception as err:
logger.warning("Failed to load prompts for subject %s: %s", name_lower, err)
system_prompt = ""
plain_prompt = ""
system_map[name_lower] = (system_prompt or "").strip()
plain_map[name_lower] = (plain_prompt or system_prompt or "").strip()
return system_map, plain_map
SUBJECT_SYSTEM_PROMPTS, SUBJECT_TEXT_PROMPTS = _build_subject_prompt_maps()
def _set_subject_chat(context: ContextTypes.DEFAULT_TYPE, subject_display: str) -> None:
"""Reset history and set per-chat subject and system prompt."""
# Comments in English by request
_reset_history(context)
lower_name = subject_display.lower()
context.chat_data["subject"] = subject_display
rich_prompt = SUBJECT_SYSTEM_PROMPTS.get(lower_name, "")
plain_prompt = SUBJECT_TEXT_PROMPTS.get(lower_name, rich_prompt)
force_text = lower_name in FORCE_TEXT_SUBJECTS
# Always append Telegram suffix for forced text subjects
if force_text:
if TG_PLAIN_TEXT_SUFFIX not in plain_prompt:
plain_prompt += TG_PLAIN_TEXT_SUFFIX
if TG_PLAIN_TEXT_SUFFIX not in rich_prompt:
rich_prompt += TG_PLAIN_TEXT_SUFFIX
if force_text and plain_prompt:
rich_prompt = plain_prompt # Always reuse Telegram-safe prompt when forced
context.chat_data["system_prompt_rich"] = rich_prompt
context.chat_data["system_prompt_plain"] = plain_prompt
context.chat_data["system_prompt"] = rich_prompt or plain_prompt
# Store mapped subject_id for backend agents
context.chat_data["subject_id"] = SUBJECT_ID_MAP.get(lower_name, 1)
context.chat_data["force_text_subject"] = force_text
async def start_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Send a short help message."""
# Comments in English by request
_track_incoming_update_message(update, context)
chat = update.effective_chat
if chat is None:
return
sent = await chat.send_message(
"Привет! Я готов помочь тебе с учебой. Выбирай предмет, кидай задачу или вопрос, и я отвечу тебе на него!",
reply_markup=_main_keyboard(),
)
_track_message_id(context, sent.message_id)
async def new_chat_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Reset chat history and confirm."""
# Comments in English by request
chat = update.effective_chat
chat_id = chat.id if chat else None
if chat is None:
return
await _purge_tracked_messages(context, chat_id)
_track_incoming_update_message(update, context)
_reset_history(context)
subject_name = context.chat_data.get("subject") or "предмет не выбран"
sent = await chat.send_message(
f"Создан новый чат по предмету: {subject_name}.",
reply_markup=_main_keyboard(),
)
_track_message_id(context, sent.message_id)
async def tex_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Solve user's math query and return rendered PNG photo (text-based)."""
# Comments in English by request
_track_incoming_update_message(update, context)
query_text = " ".join(context.args).strip() if context.args else ""
if not query_text:
sent = await update.effective_chat.send_message(
"Использование: /tex <задача>. Я верну решение в виде SVG-файла."
)
_track_message_id(context, sent.message_id)
return
try:
await context.bot.send_chat_action(chat_id=update.effective_chat.id, action=ChatAction.TYPING)
ad_msg: Message | None = await _send_ad_message(update, context)
try:
# Use multi-agent orchestrator, same as website backend
subject_id = int(context.chat_data.get("subject_id") or 1)
tg_user_id = update.effective_user.id if update.effective_user else None
state = AgentState(user_id=tg_user_id, subject_id=subject_id, grade=8, chat_id=None)
# Pass prior Telegram chat history to orchestrator (so agents see context)
state.context["external_history"] = list(_get_chat_history(context))
orchestrator = get_orchestrator()
chunks: list[str] = []
try:
agent_response = orchestrator.handle_message(query_text, state)
if agent_response and agent_response.content:
chunks.append(agent_response.content)
elif agent_response and agent_response.stream:
for chunk in agent_response.stream:
if isinstance(chunk, str) and chunk:
chunks.append(chunk)
elif hasattr(chunk, "text") and chunk.text:
chunks.append(chunk.text)
elif isinstance(chunk, dict):
text_part = chunk.get("content") or chunk.get("text")
if text_part:
chunks.append(text_part)
else:
chunks.append("Не удалось получить ответ от агента.")
except Exception as orch_err:
logger.exception("Orchestrator failed in /tex: %s", orch_err)
# Optional fallback to OpenRouter if configured
if OPENROUTER_API_KEY:
sent = await update.effective_chat.send_message("Бэкенд агентов недоступен, пробую OpenRouter…")
_track_message_id(context, sent.message_id)
prior = list(_get_chat_history(context))
fallback_prompt = _get_context_system_prompt(context, prefer_plain=False)
chat_history = [{"role": "system", "content": fallback_prompt}] + prior + [{"role": "user", "content": query_text}]
for chunk in call_openrouter_stream(
chat_history=chat_history,
model_name=OPENROUTER_MODEL,
api_key=OPENROUTER_API_KEY,
api_url=OPENROUTER_URL,
):
if chunk:
chunks.append(chunk)
else:
raise
answer_text = ("".join(chunks)).strip()
if not answer_text:
sent = await update.effective_chat.send_message("Не удалось сгенерировать решение.")
_track_message_id(context, sent.message_id)
return
# Render to PNG with Markdown+LaTeX using Playwright and send as photo
png_bytes = await _text_to_png(answer_text)
photo_msg = await update.message.reply_photo(
photo=InputFile(io.BytesIO(png_bytes), filename="solution.png"),
caption="Готово ✅",
reply_markup=_main_keyboard(),
)
_track_message_id(context, photo_msg.message_id)
_append_history(context, "user", query_text)
_append_history(context, "assistant", answer_text)
finally:
if ad_msg:
await _safe_delete_message(context, update.effective_chat.id, ad_msg.message_id)
except Exception as exc:
logger.exception("tex_cmd error: %s", exc)
error_msg = f"Ошибка при подготовке решения: {exc}"
if update.effective_user and update.effective_user.id in ADMIN_IDS:
await update.effective_chat.send_message(f"⚠️ ADMIN DEBUG (TEX):\n{error_msg}")
else:
await update.effective_chat.send_message("Ошибка при подготовке решения.")
_track_message_id(context, sent.message_id if 'sent' in locals() else None)
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Handle user text messages and reply as a PNG photo."""
# Comments in English by request
if update.message is None or not getattr(update.message, "text", None):
return
user_text = update.message.text.strip()
# Handle subject selection
lower_text = user_text.lower()
chat = update.effective_chat
chat_id = chat.id if chat else None
if lower_text in SUBJECT_SYSTEM_PROMPTS:
await _purge_tracked_messages(context, chat_id)
_track_incoming_update_message(update, context)
_set_subject_chat(context, user_text)
sent = await update.effective_chat.send_message(
f"Создан новый чат по предмету: {user_text}. Пишите ваш запрос.",
reply_markup=_main_keyboard(),
)
_track_message_id(context, sent.message_id)
return
# Handle New Chat button/text
if user_text.lower() in {"новый чат", "🆕 новый чат", "/new", "/newchat"}:
await _purge_tracked_messages(context, chat_id)
_track_incoming_update_message(update, context)
_reset_history(context)
subject_name = context.chat_data.get("subject") or "предмет не выбран"
sent = await update.effective_chat.send_message(
f"Создан новый чат по предмету: {subject_name}.",
reply_markup=_main_keyboard(),
)
_track_message_id(context, sent.message_id)
return
_track_incoming_update_message(update, context)
# Check subscription and daily limit
tg_user_id = update.effective_user.id if update.effective_user else None
is_plus, is_admin = False, False
if tg_user_id:
is_plus, daily_count = get_user_status(tg_user_id)
is_admin = tg_user_id in ADMIN_IDS
if not is_plus and not is_admin and daily_count >= 5:
await update.message.reply_text(
"Бесплатный дневной лимит (5 сообщений) закончился. "
"Чтобы продолжить обучение без ограничений, приобретите подписку Plus.\n\n"
"подробнее - /plus",
reply_markup=_main_keyboard()
)
return
try:
force_text_subject = bool(context.chat_data.get("force_text_subject"))
prefer_plain_reply = force_text_subject or not _looks_math_heavy(user_text)
# Show typing action while processing
await context.bot.send_chat_action(chat_id=update.effective_chat.id, action=ChatAction.TYPING)
ad_msg: Message | None = await _send_ad_message(update, context)
try:
# Ask the model for Markdown+LaTeX (bold, lists, formulas)
response_text_chunks: list[str] = []
try:
# Use the same agent-based backend as the website
subject_id = int(context.chat_data.get("subject_id") or 1)
tg_user_id = update.effective_user.id if update.effective_user else None
state = AgentState(
user_id=tg_user_id,
subject_id=subject_id,
grade=8,
chat_id=None,
plain_text_mode=prefer_plain_reply,
)
# Pass prior Telegram chat history so agents have context
state.context["external_history"] = list(_get_chat_history(context))
orchestrator = get_orchestrator()
agent_response = orchestrator.handle_message(user_text, state)
if agent_response and agent_response.content:
response_text_chunks.append(agent_response.content)
elif agent_response and agent_response.stream:
for chunk in agent_response.stream:
text_content = None
if isinstance(chunk, str):
text_content = chunk
elif hasattr(chunk, "text"):
text_content = chunk.text
elif isinstance(chunk, dict):
text_content = chunk.get("content") or chunk.get("text")
if text_content:
response_text_chunks.append(text_content)
else:
response_text_chunks.append("Пустой ответ от системы.")
except Exception as orch_err:
logger.warning("Agent orchestrator failed: %s", orch_err, exc_info=True)
if OPENROUTER_API_KEY:
sent = await update.effective_chat.send_message("Бэкенд агентов недоступен, пробую OpenRouter…")
_track_message_id(context, sent.message_id)
system_prompt = _get_context_system_prompt(context, prefer_plain=prefer_plain_reply)
prior = list(_get_chat_history(context))
chat_history = [{"role": "system", "content": system_prompt}] + prior + [{"role": "user", "content": user_text}]
for chunk in call_openrouter_stream(
chat_history=chat_history,
model_name=OPENROUTER_MODEL,
api_key=OPENROUTER_API_KEY,
api_url=OPENROUTER_URL,
):
if chunk:
response_text_chunks.append(chunk)
else:
raise
answer_text = ("".join(response_text_chunks)).strip()
if not answer_text:
sent = await update.effective_chat.send_message("(пустой ответ)")
_track_message_id(context, sent.message_id)
return
needs_image = False
if not force_text_subject:
needs_image = await _should_render_response_as_image(user_text, answer_text, context)
if needs_image:
# Indicate upload while preparing PNG
await context.bot.send_chat_action(chat_id=update.effective_chat.id, action=ChatAction.UPLOAD_PHOTO)
# Render text to PNG using Playwright and send as photo
png_bytes = await _text_to_png(answer_text)
photo_msg = await update.message.reply_photo(
photo=InputFile(io.BytesIO(png_bytes), filename="answer.png"),
reply_markup=_main_keyboard(),
)
_track_message_id(context, photo_msg.message_id)
else:
await _send_text_reply(update, context, answer_text)
# Increment message count for non-plus and non-admin users
if tg_user_id and not is_plus and not is_admin:
increment_message_count(tg_user_id)
_append_history(context, "user", user_text)
_append_history(context, "assistant", answer_text)
finally:
if ad_msg:
await _safe_delete_message(context, update.effective_chat.id, ad_msg.message_id)
except Exception as exc:
logger.exception("Failed to process message: %s", exc)
error_msg = f"Ошибка при обработке запроса: {exc}"
if update.effective_user and update.effective_user.id in ADMIN_IDS:
await update.effective_chat.send_message(f"⚠️ ADMIN DEBUG:\n{error_msg}")
else:
await update.effective_chat.send_message("Ошибка при обработке запроса. Попробуйте позже.")
_track_message_id(context, sent.message_id if 'sent' in locals() else None)
async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Handle user photo messages: send image (and caption) to backend API and reply with PNG."""
# Comments in English by request
if update.message is None or not getattr(update.message, "photo", None):
return
_track_incoming_update_message(update, context)
# Check subscription and daily limit
tg_user_id = update.effective_user.id if update.effective_user else None
is_plus, is_admin = False, False
if tg_user_id:
is_plus, daily_count = get_user_status(tg_user_id)
is_admin = tg_user_id in ADMIN_IDS
if not is_plus and not is_admin and daily_count >= 5:
await update.message.reply_text(
"Бесплатный дневной лимит (5 сообщений) закончился. "
"Чтобы продолжить обучение без ограничений, приобретите подписку Plus.\n\n"
"подробнее - /plus",
reply_markup=_main_keyboard()
)
return
try:
# Indicate upload/processing
await context.bot.send_chat_action(chat_id=update.effective_chat.id, action=ChatAction.UPLOAD_PHOTO)
ad_msg: Message | None = await _send_ad_message(update, context)
try:
# Get the highest resolution photo
photo_sizes = update.message.photo
tg_file = await context.bot.get_file(photo_sizes[-1].file_id)
# Download bytes into memory (async, non-blocking)
buf = io.BytesIO()
await tg_file.download_to_memory(out=buf)
image_bytes = buf.getvalue()
# Prepare base64 and MIME
b64 = base64.b64encode(image_bytes).decode("utf-8")
mime_type = "image/jpeg" # Telegram photos are JPEG by default
# Optional user caption as text part
caption = (update.message.caption or "").strip()
# Build multimodal content parts for orchestrator/Gemini
# parts format is aligned with convert_to_gemini_parts()
content_parts: list[dict] = []
if caption:
content_parts.append({"type": "text", "text": caption})
content_parts.append({
"type": "image",
"source": {
"type": "base64",
"media_type": mime_type,
"data": b64,
},
})
# Use the same agent-based backend as the website
subject_id = int(context.chat_data.get("subject_id") or 1)
force_text_subject = bool(context.chat_data.get("force_text_subject"))
tg_user_id = update.effective_user.id if update.effective_user else None
state = AgentState(
user_id=tg_user_id,
subject_id=subject_id,
grade=8,
chat_id=None,
plain_text_mode=force_text_subject,
)
# Provide prior Telegram chat history for context
state.context["external_history"] = list(_get_chat_history(context))
orchestrator = get_orchestrator()
response_text_chunks: list[str] = []
try:
# Pass multimodal parts directly as the message (supported downstream)
agent_response = orchestrator.handle_message(content_parts, state)
if agent_response and agent_response.content:
response_text_chunks.append(agent_response.content)
elif agent_response and agent_response.stream:
for chunk in agent_response.stream:
text_content = None
if isinstance(chunk, str):
text_content = chunk
elif hasattr(chunk, "text"):
text_content = chunk.text
elif isinstance(chunk, dict):
text_content = chunk.get("content") or chunk.get("text")
if text_content:
response_text_chunks.append(text_content)
else:
response_text_chunks.append("Пустой ответ от системы.")
except Exception as orch_err:
# For images, we avoid OpenRouter fallback due to varying multimodal support
logging.exception("Agent orchestrator failed on photo: %s", orch_err)
sent = await update.effective_chat.send_message("Бэкенд агентов недоступен. Попробуйте позже.")
_track_message_id(context, sent.message_id)
return
answer_text = ("".join(response_text_chunks)).strip()
if not answer_text:
sent = await update.effective_chat.send_message("(пустой ответ)")
_track_message_id(context, sent.message_id)
return
if force_text_subject:
await _send_text_reply(update, context, answer_text)
else:
# Render text reply to PNG using Playwright and send as photo
png_bytes = await _text_to_png(answer_text)
photo_msg = await update.message.reply_photo(
photo=InputFile(io.BytesIO(png_bytes), filename="answer.png"),
reply_markup=_main_keyboard(),
)
_track_message_id(context, photo_msg.message_id)
# Save simplified history entry (without storing raw image)
history_note = "Фото" + (f": {caption}" if caption else "")
# Increment message count for non-plus and non-admin users
if tg_user_id and not is_plus and not is_admin:
increment_message_count(tg_user_id)
_append_history(context, "user", history_note)
_append_history(context, "assistant", answer_text)
finally:
if ad_msg:
await _safe_delete_message(context, update.effective_chat.id, ad_msg.message_id)
except Exception as exc:
logger.exception("Failed to process photo: %s", exc)
error_msg = f"Ошибка при обработке изображения: {exc}"
if update.effective_user and update.effective_user.id in ADMIN_IDS:
await update.effective_chat.send_message(f"⚠️ ADMIN DEBUG (PHOTO):\n{error_msg}")
else:
await update.effective_chat.send_message("Ошибка при обработке изображения. Попробуйте позже.")
_track_message_id(context, sent.message_id if 'sent' in locals() else None)
async def plus_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Show plus subscription options."""
# Comments in English by request
text = (
"💎 **ClassGPT Plus**\n\n"
"Получите максимум от обучения с Plus-подпиской:\n"
"✅ Безлимитные сообщения\n"
"Выберите подходящий тариф:"
)
keyboard = [
[InlineKeyboardButton("1 месяц — 990 ⭐", callback_data="buy_plus:1")],
[InlineKeyboardButton("12 месяцев — 7490 ⭐", callback_data="buy_plus:12")],
]
reply_markup = InlineKeyboardMarkup(keyboard)
if update.message:
await update.message.reply_text(text, reply_markup=reply_markup, parse_mode="Markdown")
elif update.callback_query:
await update.callback_query.message.edit_text(text, reply_markup=reply_markup, parse_mode="Markdown")
async def buy_plus_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Handle tariff selection from inline buttons."""
# Comments in English by request
query = update.callback_query
await query.answer()
data = query.data
if data.startswith("buy_plus:"):
months = int(data.split(":")[1])
await pay_stars_cmd(update, context, months=months)
async def pay_stars_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE, months: int = 1) -> None:
"""Send an invoice for Telegram Stars."""
# Comments in English by request
chat_id = update.effective_chat.id
# Prices based on duration
if months == 12:
price = 7490
else:
price = 990
title = f"Подписка ClassGPT Plus ({months} мес.)"
description = f"Подписка на {months} мес.: безлимитные сообщения"
# Payload used to identify the duration in months
payload = f"plus-subscription:{months}"
# For Telegram Stars, currency must be "XTR"
currency = "XTR"
label = f"{months} Мес. Plus"
prices = [LabeledPrice(label, price)]
try:
await context.bot.send_invoice(
chat_id,
title,
description,
payload,
"", # provider_token must be empty for Telegram Stars
currency,
prices
)
except Exception as e:
logger.error("Failed to send invoice: %s", e)
if update.message:
await update.message.reply_text("Не удалось отправить счёт на оплату.")
elif update.callback_query:
await update.callback_query.message.reply_text("Не удалось отправить счёт на оплату.")
async def precheckout_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Answer the PreCheckoutQuery."""
# Comments in English by request
query = update.pre_checkout_query
# Check if the payload matches our payload prefix
if not query.invoice_payload.startswith("plus-subscription:"):
await query.answer(ok=False, error_message="Что-то пошло не так...")
else:
await query.answer(ok=True)
async def successful_payment_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Confirm successful payment and grant plus status."""
# Comments in English by request
tg_user_id = update.effective_user.id if update.effective_user else None
payload = update.message.successful_payment.invoice_payload
try:
months = int(payload.split(":")[1])
except (ValueError, IndexError):
months = 1
if tg_user_id:
set_user_plus(tg_user_id, days=months * 30)
await update.message.reply_text(f"Спасибо! Оплата прошла успешно. ✅ Теперь у вас есть доступ к ClassGPT Plus на {months} мес. без ограничений!")
else:
await update.message.reply_text("Спасибо! Оплата прошла успешно. ✅")
async def post_init(application: Application) -> None:
"""Set bot commands for the menu button."""
# Comments in English by request
commands = [
BotCommand("start", "👋 Что умеет бот"),
BotCommand("new", "🆕 Новый чат"),
BotCommand("plus", "⭐ Подписка Plus"),
]
await application.bot.set_my_commands(commands, scope=BotCommandScopeDefault())
def build_application() -> Application:
"""Build PTB application with polling handlers."""
# Comments in English by request
if not TG_BOT_TOKEN:
raise RuntimeError("TG_BOT_TOKEN не задан. Задайте переменную окружения или укажите токен в коде.")
app = (
Application.builder()
.token(TG_BOT_TOKEN)
.post_init(post_init)
.build()
)
app.add_handler(CommandHandler("start", start_cmd))
app.add_handler(CommandHandler("new", new_chat_cmd))
app.add_handler(CommandHandler("plus", plus_cmd))
app.add_handler(CommandHandler("tex", tex_cmd))
# Use MessageHandler for Cyrillic "command" because CommandHandler only supports ASCII
app.add_handler(MessageHandler(filters.Regex(r"^/оплата(\s+\d+)?$"), pay_stars_cmd))
app.add_handler(CommandHandler("pay", pay_stars_cmd)) # Also add ASCII version for convenience
# Callback query handlers
app.add_handler(CallbackQueryHandler(buy_plus_callback, pattern="^buy_plus:"))
# Payment handlers
app.add_handler(PreCheckoutQueryHandler(precheckout_callback))
app.add_handler(MessageHandler(filters.SUCCESSFUL_PAYMENT, successful_payment_callback))
# Photo handler should be before text handler
app.add_handler(MessageHandler(filters.PHOTO, handle_photo))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
return app
def main() -> None:
setup_tg_db()
app = build_application()
logger.info("Telegram bot is running in polling mode...")
app.run_polling(allowed_updates=Update.ALL_TYPES)
if __name__ == "__main__":
main()