285 lines
11 KiB
Python
285 lines
11 KiB
Python
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
|
||
from urllib.parse import urlencode
|
||
|
||
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,
|
||
summary: dict[str, Any] | None = None,
|
||
) -> str:
|
||
params: dict[str, str | int] = {"user_id": user_id, "mode": "keyboard"}
|
||
if summary:
|
||
params.update(
|
||
nickname=str(summary["nickname"]),
|
||
balance=int(summary["balance_cents"]),
|
||
income=int(summary["income_cents"]),
|
||
expense=int(summary["expense_cents"]),
|
||
)
|
||
return f"{settings.webapp_url}?{urlencode(params)}"
|
||
|
||
|
||
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
|
||
|
||
|
||
def parse_friend_removal(raw: str) -> int:
|
||
payload = decode_payload(raw)
|
||
if payload.get("type") != "friend_remove" or payload.get("v") != 1:
|
||
raise ValueError("Неизвестный формат данных")
|
||
friend_user_id = payload.get("friend_user_id")
|
||
if not isinstance(friend_user_id, int) or friend_user_id <= 0:
|
||
raise ValueError("Некорректный ID друга")
|
||
return friend_user_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,
|
||
self.database.summary(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 = "Запись отменена и больше не участвует в расчётах."
|
||
elif action == "friend_remove":
|
||
friend_user_id = parse_friend_removal(raw)
|
||
if not self.database.remove_friendship(user.id, friend_user_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()
|