from __future__ import annotations import json import logging import re from datetime import UTC, datetime from pathlib import Path from threading import Thread from typing import Any import telebot from telebot import types from app.config import Settings from app.db import Database, TransactionInput, amount_to_cents logger = logging.getLogger(__name__) CATEGORY_PATTERN = re.compile(r"^[^\x00-\x1f]{1,40}$") def webapp_url_for_user(settings: Settings, user_id: int) -> str: return f"{settings.webapp_url}?user_id={user_id}&mode=keyboard" def decode_payload(raw: str) -> dict[str, Any]: try: payload = json.loads(raw) except (json.JSONDecodeError, TypeError) as exc: raise ValueError("Не удалось прочитать данные") from exc if not isinstance(payload, dict): raise ValueError("Неизвестный формат данных") return payload def parse_category_name(value: Any) -> str: category = str(value or "").strip() if not CATEGORY_PATTERN.fullmatch(category): raise ValueError("Категория должна содержать от 1 до 40 символов") return category def parse_transaction(raw: str) -> TransactionInput: payload = decode_payload(raw) if payload.get("type") != "transaction" or payload.get("v") != 1: raise ValueError("Неизвестный формат данных") kind = payload.get("kind") if kind not in {"income", "expense"}: raise ValueError("Некорректный тип операции") category = parse_category_name(payload.get("category")) note = str(payload.get("note", "")).strip() or None if note and len(note) > 160: raise ValueError("Комментарий слишком длинный") occurred_at_raw = str(payload.get("occurred_at", "")).strip() try: occurred_at = datetime.fromisoformat(occurred_at_raw.replace("Z", "+00:00")) except ValueError: occurred_at = datetime.now(UTC) if occurred_at.tzinfo is None: occurred_at = occurred_at.replace(tzinfo=UTC) return TransactionInput( kind=kind, amount_cents=amount_to_cents(payload.get("amount")), category=category, note=note, occurred_at=occurred_at.astimezone(UTC).isoformat(), ) def parse_category(raw: str) -> tuple[str, str]: payload = decode_payload(raw) if payload.get("type") != "category_create" or payload.get("v") != 1: raise ValueError("Неизвестный формат данных") kind = payload.get("kind") if kind not in {"income", "expense"}: raise ValueError("Некорректный тип категории") return kind, parse_category_name(payload.get("category")) def parse_cancellation(raw: str) -> int: payload = decode_payload(raw) if payload.get("type") != "transaction_cancel" or payload.get("v") != 1: raise ValueError("Неизвестный формат данных") transaction_id = payload.get("transaction_id") if not isinstance(transaction_id, int) or transaction_id <= 0: raise ValueError("Некорректный ID записи") return transaction_id class TelegramBotService: def __init__(self, settings: Settings, database: Database): self.settings = settings self.database = database self.bot = telebot.TeleBot(settings.telegram_token, threaded=True) self.thread: Thread | None = None self._register_handlers() def _keyboard(self, user_id: int) -> types.ReplyKeyboardMarkup: keyboard = types.ReplyKeyboardMarkup(resize_keyboard=True) keyboard.add( types.KeyboardButton( text="Открыть Пожитки", web_app=types.WebAppInfo( url=webapp_url_for_user(self.settings, user_id) ), ) ) return keyboard def _register_handlers(self) -> None: @self.bot.message_handler(commands=["start"]) def start(message: types.Message) -> None: user = message.from_user if user is None: return self._save_user(user) parts = (message.text or "").split(maxsplit=1) friend_added = False if len(parts) == 2 and parts[1].startswith("friend_"): try: inviter_id = int(parts[1].removeprefix("friend_")) except ValueError: inviter_id = 0 if inviter_id > 0: friend_added = self.database.add_friendship(user.id, inviter_id) prefix = "Теперь вы друзья!\n\n" if friend_added else "" self.bot.send_message( message.chat.id, prefix + "Пожитки считают доходы, расходы и процент накоплений. " "Нажмите кнопку ниже, чтобы открыть приложение.", reply_markup=self._keyboard(user.id), ) @self.bot.message_handler(content_types=["web_app_data"]) def web_app_data(message: types.Message) -> None: user = message.from_user if user is None or message.web_app_data is None: return self._save_user(user) try: raw = message.web_app_data.data action = decode_payload(raw).get("type") if action == "transaction": item = parse_transaction(raw) self.database.add_transaction(user.id, item) operation = "Доход" if item.kind == "income" else "Расход" response = ( f"{operation} сохранён в категории «{item.category}». " "Откройте приложение снова, чтобы увидеть обновлённую статистику." ) elif action == "category_create": kind, category = parse_category(raw) created = self.database.add_category(user.id, kind, category) response = ( f"Категория «{category}» добавлена." if created else f"Категория «{category}» уже существует." ) elif action == "transaction_cancel": transaction_id = parse_cancellation(raw) if not self.database.cancel_transaction(user.id, transaction_id): raise ValueError("Запись не найдена или уже отменена") response = "Запись отменена и больше не участвует в расчётах." else: raise ValueError("Неизвестный формат данных") except ValueError as exc: self.bot.send_message( message.chat.id, f"Не получилось сохранить операцию: {exc}", reply_markup=self._keyboard(user.id), ) return self.bot.send_message( message.chat.id, response, reply_markup=self._keyboard(user.id), ) @self.bot.message_handler(commands=["help"]) def help_message(message: types.Message) -> None: user = message.from_user if user is None: return self._save_user(user) self.bot.send_message( message.chat.id, "Добавляйте операции через WebApp. Ссылка «Стать друзьями» находится на вкладке рейтинга.", reply_markup=self._keyboard(user.id), ) def _save_user(self, user: types.User) -> None: self.database.upsert_user( user_id=user.id, first_name=user.first_name or "Пользователь", last_name=user.last_name, username=user.username, ) try: photos = self.bot.get_user_profile_photos(user.id, limit=1) if not photos.photos: return photo = photos.photos[0][-1] if self.database.get_avatar_file_id(user.id) == photo.file_id: return file_info = self.bot.get_file(photo.file_id) content = self.bot.download_file(file_info.file_path) suffix = Path(file_info.file_path).suffix or ".jpg" filename = f"{user.id}{suffix}" target = self.settings.avatars_dir / filename self.settings.avatars_dir.mkdir(parents=True, exist_ok=True) target.write_bytes(content) self.database.upsert_user( user_id=user.id, first_name=user.first_name or "Пользователь", last_name=user.last_name, username=user.username, avatar_file_id=photo.file_id, avatar_path=filename, ) except Exception: logger.exception("Could not update avatar for Telegram user %s", user.id) def start(self) -> None: me = self.bot.get_me() if me.username: self.database.set_meta("bot_username", me.username) self.bot.set_my_commands( [ types.BotCommand("start", "Открыть приложение"), types.BotCommand("help", "Как пользоваться"), ] ) self.thread = Thread( target=self.bot.infinity_polling, kwargs={"skip_pending": True, "timeout": 30, "long_polling_timeout": 30}, name="telegram-polling", daemon=True, ) self.thread.start() logger.info("Telegram bot @%s started", me.username) def stop(self) -> None: self.bot.stop_polling()