commit 8f6d2c07d6ed0b6d61660210f7ce851e8acf47b9 Author: Artemii Peretiachenko Date: Mon Jul 20 18:19:53 2026 +0200 Initial commit: 428th exchange Telegram bot. Co-authored-by: Cursor diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..139f7ad --- /dev/null +++ b/.env.example @@ -0,0 +1,13 @@ +BOT_TOKEN=123456:ABC-DEF +MANAGER_GROUP_ID=-1001234567890 +DATABASE_PATH=data/bot.db +# Optional text overrides (per language). Prefer _RU / _UK. +# Bare WELCOME_TEXT / ACK_TEXT / COOLDOWN_TEXT apply only to RU +# (so UK keeps catalog strings and the services button stays in sync). +# WELCOME_TEXT_RU=Здравствуйте! Напишите ваш запрос. +# WELCOME_TEXT_UK=Вітаємо! Напишіть ваш запит. +# ACK_TEXT_RU=Спасибо, подождите ответ. +# ACK_TEXT_UK=Дякуємо, зачекайте на відповідь. +# COOLDOWN_TEXT_RU=Подождите немного, не спамьте. +# COOLDOWN_TEXT_UK=Зачекайте трохи, не спамте. +CLIENT_COOLDOWN_SECONDS=5 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ff86d91 --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +.env +.venv/ +venv/ +__pycache__/ +*.py[cod] +*.egg-info/ +.pytest_cache/ +.mypy_cache/ +data/ +*.db +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..bd48ae4 --- /dev/null +++ b/README.md @@ -0,0 +1,72 @@ +# 428th Exchange Bot + +Telegram-бот — прослойка между клиентом и менеджером. + +Клиент пишет боту в личку. Менеджер отвечает в супергруппе с **Topics**: у каждого клиента своя тема. Клиент получает ответы от имени бота. + +## Требования + +- Python 3.9+ +- Бот от [@BotFather](https://t.me/BotFather) +- Супергруппа с включёнными темами (Topics) + +## Быстрый старт + +```bash +python3 -m venv .venv +source .venv/bin/activate +pip install -r requirements.txt +cp .env.example .env +# отредактируйте .env +python -m bot +``` + +## Настройка Telegram + +1. Создайте бота у @BotFather и скопируйте token в `BOT_TOKEN`. +2. Создайте супергруппу, включите **Topics** (настройки группы → Topics). +3. Добавьте бота в группу и сделайте **администратором** с правами: + - Manage topics + - Post messages + - Delete messages (нужно для `/del`) +4. Узнайте `chat_id` группы и пропишите в `MANAGER_GROUP_ID` (обычно отрицательный, вида `-100…`). + - Запустите бота, добавьте его в группу — в лог напишется id чата. + - Либо перешлите любое сообщение из группы боту [@userinfobot](https://t.me/userinfobot) / аналогу. +5. Напишите боту `/start` с тестового аккаунта и отправьте сообщение — в группе появится новая тема. + +## Переменные окружения + +| Переменная | Описание | +|---|---| +| `BOT_TOKEN` | Token бота | +| `MANAGER_GROUP_ID` | Id супергруппы с Topics | +| `DATABASE_PATH` | Путь к SQLite (по умолчанию `data/bot.db`) | +| `WELCOME_TEXT_RU` / `WELCOME_TEXT_UK` | Опционально: приветствие для языка | +| `ACK_TEXT_RU` / `ACK_TEXT_UK` | Опционально: текст после сообщения клиента | +| `COOLDOWN_TEXT_RU` / `COOLDOWN_TEXT_UK` | Опционально: ответ при кулдауне | +| `WELCOME_TEXT` / `ACK_TEXT` / `COOLDOWN_TEXT` | Legacy: только для RU (UK не перезаписывается) | +| `CLIENT_COOLDOWN_SECONDS` | Кулдаун клиента в секундах (по умолчанию `5`) | + +## Как это работает + +1. Клиент: `/start` → выбор языка (RU/UK) → приветствие, запись в SQLite. +2. Клиент пишет сообщение → бот создаёт тему (если ещё нет) и копирует сообщение менеджерам → «Спасибо, подождите ответ». +3. Менеджер пишет в теме клиента → бот копирует содержимое клиенту в личку от своего имени. +4. Правки сообщений синхронизируются в обе стороны (текст и подписи). +5. Реакции синхронизируются в обе стороны. +6. Удаление: ответьте на сообщение командой `/del` — копия удалится и у клиента, и у менеджера. +7. Язык можно сменить командой `/lang`. Список услуг — кнопка под приветствием. + +Сообщения бота в теме (копии от клиента) обратно клиенту не уходят. + +## Запуск + +```bash +source .venv/bin/activate +python -m bot +``` + +Данные клиентов хранятся в SQLite (`telegram_id` ↔ `topic_id`). +Рядом с `bot.db` могут появляться файлы `-wal` / `-shm` (режим WAL) — их тоже не нужно коммитить (`data/` в `.gitignore`). + +При старте бот проверяет `MANAGER_GROUP_ID`: супергруппа с Topics, бот — админ с правами Manage topics и Delete messages. При ошибке процесс завершится с понятным текстом в логе и stderr. diff --git a/bot/__init__.py b/bot/__init__.py new file mode 100644 index 0000000..a4f5843 --- /dev/null +++ b/bot/__init__.py @@ -0,0 +1 @@ +"""Telegram relay bot: client DM ↔ manager forum topics.""" diff --git a/bot/__main__.py b/bot/__main__.py new file mode 100644 index 0000000..c76de54 --- /dev/null +++ b/bot/__main__.py @@ -0,0 +1,4 @@ +from bot.main import main + +if __name__ == "__main__": + main() diff --git a/bot/config.py b/bot/config.py new file mode 100644 index 0000000..5c7cf0a --- /dev/null +++ b/bot/config.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path +from typing import Dict, Optional + +from dotenv import load_dotenv + +load_dotenv() + +DEFAULT_CLIENT_COOLDOWN_SECONDS = 5.0 + + +@dataclass(frozen=True) +class Config: + bot_token: str + manager_group_id: int + database_path: Path + # Optional per-language overrides for i18n keys (lang -> text). + welcome_text: Dict[str, str] + ack_text: Dict[str, str] + cooldown_text: Dict[str, str] + client_cooldown_seconds: float + + +def _optional_env(name: str) -> Optional[str]: + raw = os.getenv(name) + if raw is None: + return None + value = raw.strip() + return value or None + + +def _text_overrides_for(key: str) -> Dict[str, str]: + """Load per-language text overrides. + + Supported env vars (example for welcome): + WELCOME_TEXT_RU / WELCOME_TEXT_UK — language-specific + WELCOME_TEXT — legacy; applies only to default language (ru), so UK + keeps its catalog string and the services button stays consistent. + """ + from bot.i18n import DEFAULT_LANGUAGE, SUPPORTED_LANGUAGES + + overrides: Dict[str, str] = {} + legacy = _optional_env(key) + if legacy is not None: + overrides[DEFAULT_LANGUAGE] = legacy + + for lang in SUPPORTED_LANGUAGES: + specific = _optional_env(f"{key}_{lang.upper()}") + if specific is not None: + overrides[lang] = specific + return overrides + + +def load_config() -> Config: + token = os.getenv("BOT_TOKEN", "").strip() + if not token: + raise RuntimeError("BOT_TOKEN is not set") + + group_raw = os.getenv("MANAGER_GROUP_ID", "").strip() + if not group_raw: + raise RuntimeError("MANAGER_GROUP_ID is not set") + + try: + manager_group_id = int(group_raw) + except ValueError as exc: + raise RuntimeError("MANAGER_GROUP_ID must be an integer") from exc + + db_path = Path(os.getenv("DATABASE_PATH", "data/bot.db")).expanduser() + + cooldown_raw = os.getenv("CLIENT_COOLDOWN_SECONDS", str(DEFAULT_CLIENT_COOLDOWN_SECONDS)) + try: + client_cooldown_seconds = float(cooldown_raw) + except ValueError as exc: + raise RuntimeError("CLIENT_COOLDOWN_SECONDS must be a number") from exc + if client_cooldown_seconds < 0: + raise RuntimeError("CLIENT_COOLDOWN_SECONDS must be >= 0") + + return Config( + bot_token=token, + manager_group_id=manager_group_id, + database_path=db_path, + welcome_text=_text_overrides_for("WELCOME_TEXT"), + ack_text=_text_overrides_for("ACK_TEXT"), + cooldown_text=_text_overrides_for("COOLDOWN_TEXT"), + client_cooldown_seconds=client_cooldown_seconds, + ) + + +def apply_text_overrides(config: Config) -> None: + """Apply optional env text overrides into the i18n catalog (per language).""" + from bot.i18n import TEXTS + + for lang, text in config.welcome_text.items(): + if lang in TEXTS: + TEXTS[lang]["welcome"] = text + for lang, text in config.ack_text.items(): + if lang in TEXTS: + TEXTS[lang]["ack"] = text + for lang, text in config.cooldown_text.items(): + if lang in TEXTS: + TEXTS[lang]["cooldown"] = text diff --git a/bot/db/__init__.py b/bot/db/__init__.py new file mode 100644 index 0000000..3106c32 --- /dev/null +++ b/bot/db/__init__.py @@ -0,0 +1,3 @@ +from bot.db.repository import Client, ClientRepository + +__all__ = ["Client", "ClientRepository"] diff --git a/bot/db/repository.py b/bot/db/repository.py new file mode 100644 index 0000000..76c0795 --- /dev/null +++ b/bot/db/repository.py @@ -0,0 +1,281 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + +import aiosqlite + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS clients ( + telegram_id INTEGER PRIMARY KEY, + username TEXT, + first_name TEXT, + last_name TEXT, + topic_id INTEGER UNIQUE, + language TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_clients_topic_id ON clients(topic_id); + +CREATE TABLE IF NOT EXISTS message_links ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + client_telegram_id INTEGER NOT NULL, + client_message_id INTEGER NOT NULL, + manager_message_id INTEGER NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(client_telegram_id, client_message_id), + UNIQUE(manager_message_id) +); + +CREATE INDEX IF NOT EXISTS idx_message_links_manager + ON message_links(manager_message_id); +""" + + +@dataclass +class Client: + telegram_id: int + username: Optional[str] + first_name: Optional[str] + last_name: Optional[str] + topic_id: Optional[int] + language: Optional[str] + created_at: str + updated_at: str + + +@dataclass +class MessageLink: + client_telegram_id: int + client_message_id: int + manager_message_id: int + + +def _utcnow() -> str: + return datetime.now(timezone.utc).isoformat() + + +class ClientRepository: + def __init__(self, db_path: Path) -> None: + self._db_path = db_path + self._conn: Optional[aiosqlite.Connection] = None + + async def connect(self) -> None: + self._db_path.parent.mkdir(parents=True, exist_ok=True) + self._conn = await aiosqlite.connect(self._db_path) + self._conn.row_factory = aiosqlite.Row + await self._conn.execute("PRAGMA journal_mode=WAL") + await self._conn.execute("PRAGMA busy_timeout=5000") + await self._conn.executescript(SCHEMA) + await self._ensure_columns() + await self._conn.commit() + + async def _ensure_columns(self) -> None: + conn = self._require_conn() + async with conn.execute("PRAGMA table_info(clients)") as cursor: + rows = await cursor.fetchall() + columns = {row[1] for row in rows} + if "language" not in columns: + await conn.execute("ALTER TABLE clients ADD COLUMN language TEXT") + + async def close(self) -> None: + if self._conn is not None: + await self._conn.close() + self._conn = None + + def _require_conn(self) -> aiosqlite.Connection: + if self._conn is None: + raise RuntimeError("Database is not connected") + return self._conn + + @staticmethod + def _row_to_client(row: aiosqlite.Row) -> Client: + return Client( + telegram_id=row["telegram_id"], + username=row["username"], + first_name=row["first_name"], + last_name=row["last_name"], + topic_id=row["topic_id"], + language=row["language"], + created_at=row["created_at"], + updated_at=row["updated_at"], + ) + + async def upsert_client( + self, + telegram_id: int, + username: Optional[str], + first_name: Optional[str], + last_name: Optional[str], + ) -> Client: + conn = self._require_conn() + now = _utcnow() + await conn.execute( + """ + INSERT INTO clients ( + telegram_id, username, first_name, last_name, topic_id, language, + created_at, updated_at + ) VALUES (?, ?, ?, ?, NULL, NULL, ?, ?) + ON CONFLICT(telegram_id) DO UPDATE SET + username = excluded.username, + first_name = excluded.first_name, + last_name = excluded.last_name, + updated_at = excluded.updated_at + """, + (telegram_id, username, first_name, last_name, now, now), + ) + await conn.commit() + client = await self.get_by_telegram_id(telegram_id) + if client is None: + raise RuntimeError(f"Failed to upsert client {telegram_id}") + return client + + async def get_by_telegram_id(self, telegram_id: int) -> Optional[Client]: + conn = self._require_conn() + async with conn.execute( + "SELECT * FROM clients WHERE telegram_id = ?", + (telegram_id,), + ) as cursor: + row = await cursor.fetchone() + return self._row_to_client(row) if row else None + + async def get_by_topic_id(self, topic_id: int) -> Optional[Client]: + conn = self._require_conn() + async with conn.execute( + "SELECT * FROM clients WHERE topic_id = ?", + (topic_id,), + ) as cursor: + row = await cursor.fetchone() + return self._row_to_client(row) if row else None + + async def set_topic_id(self, telegram_id: int, topic_id: int) -> None: + conn = self._require_conn() + await conn.execute( + """ + UPDATE clients + SET topic_id = ?, updated_at = ? + WHERE telegram_id = ? + """, + (topic_id, _utcnow(), telegram_id), + ) + await conn.commit() + + async def clear_topic_id(self, telegram_id: int) -> None: + conn = self._require_conn() + await conn.execute( + """ + UPDATE clients + SET topic_id = NULL, updated_at = ? + WHERE telegram_id = ? + """, + (_utcnow(), telegram_id), + ) + await conn.commit() + + async def set_language(self, telegram_id: int, language: str) -> None: + conn = self._require_conn() + await conn.execute( + """ + UPDATE clients + SET language = ?, updated_at = ? + WHERE telegram_id = ? + """, + (language, _utcnow(), telegram_id), + ) + await conn.commit() + + async def save_message_link( + self, + client_telegram_id: int, + client_message_id: int, + manager_message_id: int, + ) -> None: + conn = self._require_conn() + await conn.execute( + "DELETE FROM message_links WHERE manager_message_id = ?", + (manager_message_id,), + ) + await conn.execute( + """ + DELETE FROM message_links + WHERE client_telegram_id = ? AND client_message_id = ? + """, + (client_telegram_id, client_message_id), + ) + await conn.execute( + """ + INSERT INTO message_links ( + client_telegram_id, client_message_id, manager_message_id, created_at + ) VALUES (?, ?, ?, ?) + """, + (client_telegram_id, client_message_id, manager_message_id, _utcnow()), + ) + await conn.commit() + + async def get_link_by_client_message( + self, + client_telegram_id: int, + client_message_id: int, + ) -> Optional[MessageLink]: + conn = self._require_conn() + async with conn.execute( + """ + SELECT client_telegram_id, client_message_id, manager_message_id + FROM message_links + WHERE client_telegram_id = ? AND client_message_id = ? + """, + (client_telegram_id, client_message_id), + ) as cursor: + row = await cursor.fetchone() + if row is None: + return None + return MessageLink( + client_telegram_id=row["client_telegram_id"], + client_message_id=row["client_message_id"], + manager_message_id=row["manager_message_id"], + ) + + async def get_link_by_manager_message( + self, + manager_message_id: int, + ) -> Optional[MessageLink]: + conn = self._require_conn() + async with conn.execute( + """ + SELECT client_telegram_id, client_message_id, manager_message_id + FROM message_links + WHERE manager_message_id = ? + """, + (manager_message_id,), + ) as cursor: + row = await cursor.fetchone() + if row is None: + return None + return MessageLink( + client_telegram_id=row["client_telegram_id"], + client_message_id=row["client_message_id"], + manager_message_id=row["manager_message_id"], + ) + + async def delete_message_link( + self, + *, + client_telegram_id: int, + client_message_id: int, + manager_message_id: int, + ) -> None: + conn = self._require_conn() + await conn.execute( + """ + DELETE FROM message_links + WHERE client_telegram_id = ? + AND client_message_id = ? + AND manager_message_id = ? + """, + (client_telegram_id, client_message_id, manager_message_id), + ) + await conn.commit() diff --git a/bot/handlers/__init__.py b/bot/handlers/__init__.py new file mode 100644 index 0000000..0792b86 --- /dev/null +++ b/bot/handlers/__init__.py @@ -0,0 +1,4 @@ +from bot.handlers.client import router as client_router +from bot.handlers.manager import build_manager_router + +__all__ = ["client_router", "build_manager_router"] diff --git a/bot/handlers/client.py b/bot/handlers/client.py new file mode 100644 index 0000000..adcd42b --- /dev/null +++ b/bot/handlers/client.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import logging + +from aiogram import F, Router +from aiogram.enums import ChatType, ParseMode +from aiogram.filters import Command, CommandStart +from aiogram.types import CallbackQuery, Message, MessageReactionUpdated + +from bot.db.repository import ClientRepository +from bot.i18n import ( + format_bot_html, + format_bot_message, + language_keyboard, + normalize_language, + services_keyboard, + t, +) +from bot.services.cooldown import CooldownService +from bot.services.relay import RelayService, is_relayable_content + +logger = logging.getLogger(__name__) + +router = Router(name="client") + + +async def _answer_bot(message: Message, language: str | None, key: str) -> None: + await message.answer( + format_bot_message(language, t(language, key)), + parse_mode=ParseMode.HTML, + ) + + +async def _answer_welcome(message: Message, language: str) -> None: + await message.answer( + format_bot_message(language, t(language, "welcome")), + reply_markup=services_keyboard(language), + parse_mode=ParseMode.HTML, + ) + + +async def _prompt_language(message: Message) -> None: + await message.answer( + t(None, "choose_language"), + reply_markup=language_keyboard(), + parse_mode=None, + ) + + +@router.message(CommandStart(), F.chat.type == ChatType.PRIVATE) +async def cmd_start(message: Message, relay: RelayService) -> None: + if message.from_user is None: + return + client = await relay.ensure_client(message.from_user) + if client.language is None: + await _prompt_language(message) + return + await _answer_welcome(message, client.language) + + +@router.message(Command("lang"), F.chat.type == ChatType.PRIVATE) +async def cmd_lang(message: Message, relay: RelayService) -> None: + if message.from_user is None: + return + await relay.ensure_client(message.from_user) + await _prompt_language(message) + + +@router.message(Command("del"), F.chat.type == ChatType.PRIVATE) +async def cmd_del(message: Message, relay: RelayService) -> None: + """Reply to a relayed message with /del to remove it on both sides.""" + if message.from_user is None: + return + client = await relay.ensure_client(message.from_user) + if message.reply_to_message is None: + return + try: + deleted = await relay.delete_paired_from_client(message.bot, message) + except Exception: + logger.exception("Failed to delete paired message for client %s", message.from_user.id) + return + if deleted: + try: + await message.delete() + except Exception: + pass + await _answer_bot(message, client.language, "deleted_notice") + + +@router.callback_query(F.data.startswith("lang:")) +async def on_language_chosen( + callback: CallbackQuery, + relay: RelayService, + repo: ClientRepository, +) -> None: + if callback.from_user is None or callback.data is None: + return + if callback.message is None or callback.message.chat.type != ChatType.PRIVATE: + await callback.answer() + return + + language = normalize_language(callback.data.split(":", 1)[1]) + if language is None: + await callback.answer() + return + + await relay.ensure_client(callback.from_user) + await repo.set_language(callback.from_user.id, language) + await callback.answer() + + body = f"{t(language, 'language_set')}\n\n{t(language, 'welcome')}" + await callback.message.edit_text( + format_bot_message(language, body), + reply_markup=services_keyboard(language), + parse_mode=ParseMode.HTML, + ) + + +@router.callback_query(F.data == "services") +async def on_services( + callback: CallbackQuery, + relay: RelayService, + cooldown: CooldownService, +) -> None: + if callback.from_user is None or callback.message is None: + return + if callback.message.chat.type != ChatType.PRIVATE: + await callback.answer() + return + + client = await relay.ensure_client(callback.from_user) + if client.language is None: + await callback.answer() + await _prompt_language(callback.message) + return + + if not cooldown.try_take_services(callback.from_user.id): + await callback.answer() + return + + await callback.answer() + await callback.message.answer( + format_bot_html(client.language, t(client.language, "services")), + parse_mode=ParseMode.HTML, + ) + + +@router.message_reaction(F.chat.type == ChatType.PRIVATE) +async def client_reaction(event: MessageReactionUpdated, relay: RelayService) -> None: + try: + await relay.relay_reaction(event.bot, event) + except Exception: + logger.exception( + "Failed to relay client reaction on message %s", + event.message_id, + ) + + +@router.edited_message(F.chat.type == ChatType.PRIVATE) +async def client_edited(message: Message, relay: RelayService) -> None: + if message.from_user is None or not is_relayable_content(message): + return + try: + await relay.relay_client_edit(message.bot, message) + except Exception: + logger.exception( + "Failed to relay client edit from %s", + message.from_user.id, + ) + + +@router.message(F.chat.type == ChatType.PRIVATE) +async def client_message( + message: Message, + relay: RelayService, + cooldown: CooldownService, +) -> None: + if message.from_user is None or not is_relayable_content(message): + return + + client = await relay.ensure_client(message.from_user) + if client.language is None: + await _prompt_language(message) + return + + decision = cooldown.check(message.from_user.id, message.media_group_id) + if not decision.allowed: + await _answer_bot(message, client.language, "cooldown") + return + + try: + await relay.relay_client_to_manager(message.bot, message) + except Exception: + logger.exception("Failed to relay client message from %s", message.from_user.id) + await _answer_bot(message, client.language, "send_failed") + return + + cooldown.commit(message.from_user.id, message.media_group_id) + if decision.should_ack: + await _answer_bot(message, client.language, "ack") diff --git a/bot/handlers/manager.py b/bot/handlers/manager.py new file mode 100644 index 0000000..fee92a4 --- /dev/null +++ b/bot/handlers/manager.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import logging + +from aiogram import F, Router +from aiogram.enums import ChatType +from aiogram.filters import Command +from aiogram.types import Message, MessageReactionUpdated + +from bot.services.relay import GENERAL_TOPIC_ID, RelayService, is_relayable_content + +logger = logging.getLogger(__name__) + + +def build_manager_router(manager_group_id: int) -> Router: + router = Router(name="manager") + + @router.message( + Command("del"), + F.chat.id == manager_group_id, + F.chat.type == ChatType.SUPERGROUP, + ) + async def manager_del(message: Message, relay: RelayService) -> None: + """Reply to a relayed message with /del to remove it on both sides.""" + if relay.is_self(message.from_user.id if message.from_user else None): + return + topic_id = message.message_thread_id + if topic_id is None or topic_id == GENERAL_TOPIC_ID: + return + if message.reply_to_message is None: + return + try: + deleted = await relay.delete_paired_from_manager(message.bot, message) + except Exception: + logger.exception("Failed to delete paired message in topic %s", topic_id) + return + if deleted: + try: + await message.delete() + except Exception: + pass + + @router.message( + F.chat.id == manager_group_id, + F.chat.type == ChatType.SUPERGROUP, + ) + async def manager_message(message: Message, relay: RelayService) -> None: + if relay.is_self(message.from_user.id if message.from_user else None): + return + + topic_id = message.message_thread_id + if topic_id is None or topic_id == GENERAL_TOPIC_ID: + return + if not is_relayable_content(message): + return + + try: + await relay.relay_manager_to_client(message.bot, message) + except Exception: + logger.exception("Failed to relay manager message in topic %s", topic_id) + + @router.edited_message( + F.chat.id == manager_group_id, + F.chat.type == ChatType.SUPERGROUP, + ) + async def manager_edited(message: Message, relay: RelayService) -> None: + if relay.is_self(message.from_user.id if message.from_user else None): + return + + topic_id = message.message_thread_id + if topic_id is None or topic_id == GENERAL_TOPIC_ID: + return + if not is_relayable_content(message): + return + + try: + await relay.relay_manager_edit(message.bot, message) + except Exception: + logger.exception("Failed to relay manager edit in topic %s", topic_id) + + @router.message_reaction(F.chat.id == manager_group_id) + async def manager_reaction(event: MessageReactionUpdated, relay: RelayService) -> None: + try: + await relay.relay_reaction(event.bot, event) + except Exception: + logger.exception( + "Failed to relay manager reaction on message %s", + event.message_id, + ) + + return router diff --git a/bot/i18n.py b/bot/i18n.py new file mode 100644 index 0000000..1a05552 --- /dev/null +++ b/bot/i18n.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +from html import escape +from typing import Dict, Optional + +from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup + +DEFAULT_LANGUAGE = "ru" +SUPPORTED_LANGUAGES = ("ru", "uk") + +Language = str + +SERVICES_RU = ( + "🌐 1. Покупка/продажа криптовалюты за наличные — лучшие условия " + "во всех странах и крупных городах.\n\n" + "💸 2. Перестановка наличных — выдача день в день.\n\n" + "💳 3. Оплата на карты\n\n" + "• Покупка/продажа криптовалюты.\n\n" + "• Оплата на карты EUR, TRY, KZT, PLN, THB.\n\n" + "• Оплата физ.лиц (со счетов в ЕС): USD, EUR, GBP через Wise, Paysera, Revolut, ZEN.\n\n" + "• Прием/оплата UAH: ТОВ, ФОП, физлица.\n\n" + "• Оплата по ссылкам: Visa, Mastercard, American Express.\n\n" + "• Оплата Инвойсов (USD, EUR): товар, логистика, маркетинг, IT, Консалтинг, " + "тур. услуги, авто (Copart, IAAI), оплата авто в ЕС.\n\n" + "🔒 4. Электронные деньги и платежные системы\n\n" + "• Payeer, Capitalist, Perfect Money, Skrill, AdvCash, Payoneer.\n\n" + "• Оплата Юань на карты: AliPay, WeChat Pay." +) + +SERVICES_UK = ( + "🌐 1. Купівля/продаж криптовалюти за готівку — найкращі умови " + "в усіх країнах і великих містах.\n\n" + "💸 2. Перестановка готівки — видача в той же день.\n\n" + "💳 3. Оплата на картки\n\n" + "• Купівля/продаж криптовалюти.\n\n" + "• Оплата на картки EUR, TRY, KZT, PLN, THB.\n\n" + "• Оплата фіз.осіб (з рахунків в ЄС): USD, EUR, GBP через Wise, Paysera, Revolut, ZEN.\n\n" + "• Прийом/оплата UAH: ТОВ, ФОП, фізособи.\n\n" + "• Оплата за посиланнями: Visa, Mastercard, American Express.\n\n" + "• Оплата інвойсів (USD, EUR): товар, логістика, маркетинг, IT, консалтинг, " + "тур. послуги, авто (Copart, IAAI), оплата авто в ЄС.\n\n" + "🔒 4. Електронні гроші та платіжні системи\n\n" + "• Payeer, Capitalist, Perfect Money, Skrill, AdvCash, Payoneer.\n\n" + "• Оплата юаня на картки: AliPay, WeChat Pay." +) + +TEXTS: Dict[str, Dict[str, str]] = { + "ru": { + "welcome": ( + "Здравствуйте!\n\n" + "Напишите ваш запрос — менеджер ответит здесь.\n\n" + "Укажите что конкретно вам нужно сделать.\n\n" + "🔶 Все наши услуги посмотрите по кнопке ниже." + ), + "ack": "Спасибо, подождите ответ.", + "cooldown": "Подождите немного, не спамьте.", + "send_failed": ( + "Не удалось отправить сообщение менеджеру. Попробуйте позже." + ), + "choose_language": "Выберите язык / Оберіть мову:", + "language_set": "Язык сохранён: русский.", + "prefix_bot": "Бот:", + "prefix_manager": "Менеджер:", + "lang_button": "Ru", + "services_button": "Список услуг", + "services": SERVICES_RU, + "client_blocked": ( + "⚠️ Клиент заблокировал бота. Сообщение не доставлено." + ), + "deliver_failed": "⚠️ Не удалось доставить сообщение клиенту.", + "edit_failed": "⚠️ Не удалось обновить сообщение у клиента.", + "deleted_notice": "🗑 Сообщение удалено.", + }, + "uk": { + "welcome": ( + "Вітаємо!\n\n" + "Напишіть ваш запит — менеджер відповість тут.\n\n" + "Вкажіть, що саме вам потрібно зробити.\n\n" + "🔶 Всі наші послуги подивіться по кнопці нижче." + ), + "ack": "Дякуємо, зачекайте на відповідь.", + "cooldown": "Зачекайте трохи, не спамте.", + "send_failed": ( + "Не вдалося надіслати повідомлення менеджеру. Спробуйте пізніше." + ), + "choose_language": "Оберіть мову / Выберите язык:", + "language_set": "Мову збережено: українська.", + "prefix_bot": "Бот:", + "prefix_manager": "Менеджер:", + "lang_button": "Ukr", + "services_button": "Список послуг", + "services": SERVICES_UK, + "client_blocked": ( + "⚠️ Клієнт заблокував бота. Повідомлення не доставлено." + ), + "deliver_failed": "⚠️ Не вдалося доставити повідомлення клієнту.", + "edit_failed": "⚠️ Не вдалося оновити повідомлення у клієнта.", + "deleted_notice": "🗑 Повідомлення видалено.", + }, +} + + +def normalize_language(language: Optional[str]) -> Optional[str]: + if language is None: + return None + code = language.strip().lower() + if code in SUPPORTED_LANGUAGES: + return code + return None + + +def resolve_language(language: Optional[str]) -> str: + return normalize_language(language) or DEFAULT_LANGUAGE + + +def t(language: Optional[str], key: str) -> str: + lang = resolve_language(language) + return TEXTS[lang][key] + + +def format_prefix(language: Optional[str], key: str) -> str: + return f"{escape(t(language, key))}" + + +def format_labeled_message(label: str, text: str) -> str: + """Label + plain text body (body is escaped).""" + return f"{escape(label)}\n\n{escape(text)}" + + +def format_labeled_html(label: str, html_body: str) -> str: + """Label + pre-formatted HTML body (body is not escaped).""" + return f"{escape(label)}\n\n{html_body}" + + +def format_bot_message(language: Optional[str], text: str) -> str: + return format_labeled_message(t(language, "prefix_bot"), text) + + +def format_bot_html(language: Optional[str], html_body: str) -> str: + """Bot prefix + pre-formatted HTML body (not escaped).""" + return f"{format_prefix(language, 'prefix_bot')}\n\n{html_body}" + + +def format_manager_message(language: Optional[str], text: str) -> str: + return format_labeled_message(t(language, "prefix_manager"), text) + + +def format_manager_html(language: Optional[str], html_body: str) -> str: + return format_labeled_html(t(language, "prefix_manager"), html_body) + + +def language_keyboard() -> InlineKeyboardMarkup: + return InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text=TEXTS[code]["lang_button"], + callback_data=f"lang:{code}", + ) + ] + for code in SUPPORTED_LANGUAGES + ] + ) + + +def services_keyboard(language: Optional[str]) -> InlineKeyboardMarkup: + return InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text=t(language, "services_button"), + callback_data="services", + ) + ] + ] + ) diff --git a/bot/main.py b/bot/main.py new file mode 100644 index 0000000..1ec3bcf --- /dev/null +++ b/bot/main.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +import asyncio +import logging +import sys + +from aiogram import Bot, Dispatcher +from aiogram.client.default import DefaultBotProperties +from aiogram.enums import ChatMemberStatus, ChatType, ParseMode +from aiogram.exceptions import TelegramAPIError +from aiogram.types import ChatMemberUpdated + +from bot.config import Config, apply_text_overrides, load_config +from bot.db.repository import ClientRepository +from bot.handlers.client import router as client_router +from bot.handlers.manager import build_manager_router +from bot.services.cooldown import CooldownService +from bot.services.relay import RelayService + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s [%(name)s] %(message)s", + stream=sys.stdout, +) +logger = logging.getLogger(__name__) + + +async def on_bot_added(event: ChatMemberUpdated, config: Config, relay: RelayService) -> None: + """Log chat id when the bot is added to a group — helps set MANAGER_GROUP_ID.""" + new = event.new_chat_member + if not relay.is_self(new.user.id): + return + if new.status in {ChatMemberStatus.ADMINISTRATOR, ChatMemberStatus.MEMBER}: + logger.info( + "Bot added to chat %s (%s). Use this as MANAGER_GROUP_ID if it is the forum.", + event.chat.id, + event.chat.title, + ) + + +async def validate_manager_group(bot: Bot, manager_group_id: int) -> None: + """Fail fast if MANAGER_GROUP_ID is wrong or the bot lacks required rights.""" + try: + chat = await bot.get_chat(manager_group_id) + except TelegramAPIError as exc: + raise RuntimeError( + f"Не удалось получить чат MANAGER_GROUP_ID={manager_group_id}: {exc}. " + "Проверьте id и что бот добавлен в группу." + ) from exc + + if chat.type != ChatType.SUPERGROUP: + raise RuntimeError( + f"MANAGER_GROUP_ID должен быть супергруппой, сейчас type={chat.type!r} " + f"(chat_id={chat.id}, title={chat.title!r})." + ) + + if not bool(getattr(chat, "is_forum", False)): + raise RuntimeError( + f"В чате {chat.id} ({chat.title!r}) не включены Topics. " + "Включите темы в настройках группы." + ) + + me = await bot.get_me() + try: + member = await bot.get_chat_member(manager_group_id, me.id) + except TelegramAPIError as exc: + raise RuntimeError( + f"Не удалось проверить права бота в группе {chat.id}: {exc}." + ) from exc + + if member.status == ChatMemberStatus.CREATOR: + logger.info("Manager group OK: %s (%s), bot is creator", chat.id, chat.title) + return + + if member.status != ChatMemberStatus.ADMINISTRATOR: + raise RuntimeError( + f"Бот должен быть администратором группы {chat.title!r} (id={chat.id}). " + f"Текущий статус: {member.status!r}." + ) + + missing: list[str] = [] + if not bool(getattr(member, "can_manage_topics", True)): + missing.append("Manage topics") + if not bool(getattr(member, "can_delete_messages", True)): + missing.append("Delete messages") + can_post = getattr(member, "can_post_messages", None) + # In groups can_post_messages may be None; treat explicit False as missing. + if can_post is False: + missing.append("Post messages") + + if missing: + raise RuntimeError( + f"Бот — админ группы {chat.title!r}, но не хватает прав: {', '.join(missing)}. " + "Выдайте их в настройках администраторов." + ) + + logger.info("Manager group OK: %s (%s), bot is admin", chat.id, chat.title) + + +async def run() -> None: + config = load_config() + apply_text_overrides(config) + repo = ClientRepository(config.database_path) + await repo.connect() + + relay = RelayService(repo, config.manager_group_id) + cooldown = CooldownService(config.client_cooldown_seconds) + bot = Bot( + token=config.bot_token, + default=DefaultBotProperties(parse_mode=ParseMode.HTML), + ) + + me = await bot.get_me() + relay.set_bot_id(me.id) + + try: + await validate_manager_group(bot, config.manager_group_id) + except RuntimeError as exc: + logger.error("%s", exc) + print(f"ERROR: {exc}", file=sys.stderr) + await repo.close() + await bot.session.close() + raise SystemExit(1) from exc + + dp = Dispatcher() + dp["config"] = config + dp["repo"] = repo + dp["relay"] = relay + dp["cooldown"] = cooldown + + dp.include_router(client_router) + dp.include_router(build_manager_router(config.manager_group_id)) + dp.my_chat_member.register(on_bot_added) + + logger.info( + "Starting bot @%s (manager group %s, client cooldown %.1fs)", + me.username, + config.manager_group_id, + config.client_cooldown_seconds, + ) + try: + await dp.start_polling( + bot, + allowed_updates=dp.resolve_used_update_types(), + ) + finally: + await repo.close() + await bot.session.close() + + +def main() -> None: + asyncio.run(run()) + + +if __name__ == "__main__": + main() diff --git a/bot/services/__init__.py b/bot/services/__init__.py new file mode 100644 index 0000000..a4f3fe2 --- /dev/null +++ b/bot/services/__init__.py @@ -0,0 +1,4 @@ +from bot.services.cooldown import CooldownService +from bot.services.relay import RelayService, build_topic_name, is_relayable_content + +__all__ = ["CooldownService", "RelayService", "build_topic_name", "is_relayable_content"] diff --git a/bot/services/cooldown.py b/bot/services/cooldown.py new file mode 100644 index 0000000..fbe9853 --- /dev/null +++ b/bot/services/cooldown.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +import time +from dataclasses import dataclass +from threading import Lock +from typing import Optional + + +@dataclass(frozen=True) +class CooldownDecision: + """Result of a cooldown check before relay.""" + + allowed: bool + should_ack: bool + + +class CooldownService: + """Per-user cooldown. In-memory; enough for a single bot process. + + Cooldown is committed only after a successful relay via :meth:`commit`. + Media albums share one ack and one cooldown window. + Expired entries are pruned opportunistically so the maps stay bounded. + """ + + def __init__(self, seconds: float) -> None: + self._seconds = max(0.0, seconds) + self._last_allowed: dict[int, float] = {} + # user_id -> (media_group_id, monotonic_ts when first item was accepted) + self._active_albums: dict[int, tuple[str, float]] = {} + self._services_last: dict[int, float] = {} + self._lock = Lock() + # Album items can arrive a few seconds apart. + self._album_window = max(self._seconds, 3.0) + self._services_seconds = max(self._seconds, 5.0) + + def _retention(self) -> float: + return max(self._album_window, self._services_seconds, self._seconds, 1.0) * 2 + + def _prune_unlocked(self, now: float) -> None: + retain = self._retention() + stale_users = [uid for uid, ts in self._last_allowed.items() if now - ts > retain] + for uid in stale_users: + del self._last_allowed[uid] + + stale_albums = [ + uid for uid, (_gid, ts) in self._active_albums.items() if now - ts > retain + ] + for uid in stale_albums: + del self._active_albums[uid] + + stale_services = [ + uid for uid, ts in self._services_last.items() if now - ts > retain + ] + for uid in stale_services: + del self._services_last[uid] + + def check(self, user_id: int, media_group_id: Optional[str] = None) -> CooldownDecision: + """Return whether the user may relay, without consuming the cooldown.""" + if self._seconds <= 0 and media_group_id is None: + return CooldownDecision(allowed=True, should_ack=True) + + now = time.monotonic() + with self._lock: + self._prune_unlocked(now) + + if media_group_id is not None: + active = self._active_albums.get(user_id) + if ( + active is not None + and active[0] == media_group_id + and now - active[1] < self._album_window + ): + return CooldownDecision(allowed=True, should_ack=False) + + last = self._last_allowed.get(user_id) + if last is not None and now - last < self._seconds: + return CooldownDecision(allowed=False, should_ack=False) + + return CooldownDecision(allowed=True, should_ack=True) + + def commit(self, user_id: int, media_group_id: Optional[str] = None) -> None: + """Mark cooldown after a successful relay.""" + if self._seconds <= 0 and media_group_id is None: + return + + now = time.monotonic() + with self._lock: + self._prune_unlocked(now) + + if media_group_id is not None: + active = self._active_albums.get(user_id) + if ( + active is not None + and active[0] == media_group_id + and now - active[1] < self._album_window + ): + # Continuation of the same album — cooldown already stamped. + return + self._active_albums[user_id] = (media_group_id, now) + + self._last_allowed[user_id] = now + + def try_take_services(self, user_id: int) -> bool: + """Return True once; silently False if tapped again within the services window.""" + if self._services_seconds <= 0: + return True + + now = time.monotonic() + with self._lock: + self._prune_unlocked(now) + last = self._services_last.get(user_id) + if last is not None and now - last < self._services_seconds: + return False + self._services_last[user_id] = now + return True diff --git a/bot/services/relay.py b/bot/services/relay.py new file mode 100644 index 0000000..1b0a635 --- /dev/null +++ b/bot/services/relay.py @@ -0,0 +1,736 @@ +from __future__ import annotations + +import asyncio +import logging +from html import escape +from typing import Any, Dict, Optional + +from aiogram import Bot +from aiogram.enums import ChatType, ParseMode +from aiogram.exceptions import TelegramAPIError, TelegramBadRequest, TelegramForbiddenError +from aiogram.types import Message, MessageReactionUpdated, ReactionTypeCustomEmoji, ReactionTypeEmoji, User +from aiogram.utils.text_decorations import html_decoration + +from bot.db.repository import Client, ClientRepository, MessageLink +from bot.i18n import ( + DEFAULT_LANGUAGE, + format_labeled_html, + format_manager_html, + format_prefix, + t, +) + +logger = logging.getLogger(__name__) + +# General topic in forum groups uses thread id 1 (or None in some clients). +GENERAL_TOPIC_ID = 1 + +TOPIC_NAME_MAX_LEN = 128 +CAPTION_MAX_LEN = 1024 + + +def build_topic_name(user: User) -> str: + name = (user.full_name or "Client").strip() or "Client" + if user.username: + label = f"{name} (@{user.username}) · {user.id}" + else: + label = f"{name} · {user.id}" + if len(label) <= TOPIC_NAME_MAX_LEN: + return label + # Keep trailing id; trim the name part. + suffix = f" · {user.id}" + if user.username: + suffix = f" (@{user.username}){suffix}" + keep = TOPIC_NAME_MAX_LEN - len(suffix) + return f"{name[: max(1, keep)].rstrip()}{suffix}" + + +def _is_captionable(message: Message) -> bool: + return bool( + message.photo + or message.document + or message.video + or message.audio + or message.animation + or message.voice + ) + + +def _html_body(text: Optional[str], entities) -> str: + if not text: + return "" + return html_decoration.unparse(text, entities or []) + + +def _safe_prefixed_caption( + prefix_html: str, + caption: Optional[str], + entities, +) -> str: + """Build prefix + caption HTML without chopping mid-entity / mid-escape.""" + if len(prefix_html) > CAPTION_MAX_LEN: + return prefix_html[: CAPTION_MAX_LEN - 1] + "…" + + if not caption: + return prefix_html + + rich = f"{prefix_html}\n\n{_html_body(caption, entities)}" + if len(rich) <= CAPTION_MAX_LEN: + return rich + + # Overflow: fall back to escaped plain text and shrink until it fits. + plain = caption + while plain: + candidate = f"{prefix_html}\n\n{escape(plain)}" + if len(candidate) <= CAPTION_MAX_LEN: + if plain != caption: + with_ellipsis = f"{prefix_html}\n\n{escape(plain + '…')}" + if len(with_ellipsis) <= CAPTION_MAX_LEN: + return with_ellipsis + return candidate + plain = plain[:-1] + return prefix_html + + +def client_display_name(user: User) -> str: + if user.username: + return f"@{user.username}" + name = (user.full_name or "").strip() + return name or "Client" + + +def client_prefix_html(user: User) -> str: + return f"{escape(client_display_name(user))}:" + + +def bot_reactions_from_update(event: MessageReactionUpdated) -> list: + """Bots may set at most one reaction; skip paid types.""" + for reaction in event.new_reaction: + if isinstance(reaction, ReactionTypeEmoji): + return [ReactionTypeEmoji(emoji=reaction.emoji)] + if isinstance(reaction, ReactionTypeCustomEmoji): + return [ReactionTypeCustomEmoji(custom_emoji_id=reaction.custom_emoji_id)] + return [] + + +def _is_message_gone_error(exc: BaseException) -> bool: + text = str(exc).lower() + return any( + marker in text + for marker in ( + "message to edit not found", + "message to delete not found", + "message can't be edited", + "message not found", + "message_id_invalid", + ) + ) + + +def _is_reply_error(exc: BaseException) -> bool: + text = str(exc).lower() + return "reply" in text and ( + "not found" in text or "message to be replied" in text or "replied message" in text + ) + + +def _is_topic_missing_error(exc: BaseException) -> bool: + text = str(exc).lower() + upper = str(exc).upper() + return any( + marker in text + for marker in ( + "message thread not found", + "thread not found", + "topic not found", + "topic_deleted", + ) + ) or "TOPIC_DELETED" in upper + + +def _thread_matches(expected: int, actual: Optional[int]) -> bool: + """False when Telegram dropped the thread (e.g. deleted topic → General).""" + if actual is None or actual == GENERAL_TOPIC_ID: + return False + return actual == expected + + +class RelayService: + def __init__(self, repo: ClientRepository, manager_group_id: int) -> None: + self._repo = repo + self._manager_group_id = manager_group_id + self._bot_id: Optional[int] = None + self._topic_locks: Dict[int, asyncio.Lock] = {} + self._topic_locks_guard = asyncio.Lock() + + def set_bot_id(self, bot_id: int) -> None: + self._bot_id = bot_id + + def is_self(self, user_id: Optional[int]) -> bool: + return self._bot_id is not None and user_id == self._bot_id + + async def _lock_for_user(self, user_id: int) -> asyncio.Lock: + async with self._topic_locks_guard: + lock = self._topic_locks.get(user_id) + if lock is None: + lock = asyncio.Lock() + self._topic_locks[user_id] = lock + return lock + + async def ensure_client(self, user: User) -> Client: + return await self._repo.upsert_client( + telegram_id=user.id, + username=user.username, + first_name=user.first_name, + last_name=user.last_name, + ) + + async def _topic_exists(self, bot: Bot, topic_id: int) -> bool: + """True if the forum topic still exists. + + Uses send_chat_action (no visible side effects). Closed topics still + accept chat actions; deleted ones raise a topic-missing error. + """ + try: + await bot.send_chat_action( + chat_id=self._manager_group_id, + action="typing", + message_thread_id=topic_id, + ) + return True + except TelegramBadRequest as exc: + if _is_topic_missing_error(exc): + return False + logger.warning("Unexpected error verifying topic %s: %s", topic_id, exc) + return True + + async def ensure_topic(self, bot: Bot, user: User, client: Client) -> int: + if client.topic_id is not None: + if await self._topic_exists(bot, client.topic_id): + return client.topic_id + logger.warning( + "Stored topic %s for client %s is gone; will recreate", + client.topic_id, + user.id, + ) + await self._repo.clear_topic_id(user.id) + + lock = await self._lock_for_user(user.id) + async with lock: + # Re-read under lock — another coroutine may have created the topic. + fresh = await self._repo.get_by_telegram_id(user.id) + if fresh is not None and fresh.topic_id is not None: + if await self._topic_exists(bot, fresh.topic_id): + return fresh.topic_id + await self._repo.clear_topic_id(user.id) + + topic = await bot.create_forum_topic( + chat_id=self._manager_group_id, + name=build_topic_name(user), + ) + await self._repo.set_topic_id(user.id, topic.message_thread_id) + logger.info( + "Created topic %s for client %s", + topic.message_thread_id, + user.id, + ) + return topic.message_thread_id + + async def _link_messages( + self, + client_telegram_id: int, + client_message_id: int, + manager_message_id: int, + ) -> None: + try: + await self._repo.save_message_link( + client_telegram_id=client_telegram_id, + client_message_id=client_message_id, + manager_message_id=manager_message_id, + ) + except Exception: + logger.exception( + "Failed to save message link client=%s/%s manager=%s", + client_telegram_id, + client_message_id, + manager_message_id, + ) + + async def _reply_to_manager_id( + self, + client_telegram_id: int, + message: Message, + ) -> Optional[int]: + reply = message.reply_to_message + if reply is None: + return None + link = await self._repo.get_link_by_client_message(client_telegram_id, reply.message_id) + return link.manager_message_id if link else None + + async def _reply_to_client_id(self, message: Message) -> Optional[int]: + reply = message.reply_to_message + if reply is None: + return None + link = await self._repo.get_link_by_manager_message(reply.message_id) + return link.client_message_id if link else None + + async def _call_with_optional_reply(self, api_call, kwargs: Dict[str, Any]): + """Call Telegram API; retry once without reply_to if the reply target is gone.""" + try: + return await api_call(**kwargs) + except TelegramBadRequest as exc: + if kwargs.get("reply_to_message_id") is not None and _is_reply_error(exc): + retry = dict(kwargs) + retry.pop("reply_to_message_id", None) + return await api_call(**retry) + raise + + async def _copy_client_message( + self, + bot: Bot, + message: Message, + topic_id: int, + user: User, + ) -> tuple[Optional[int], Optional[int]]: + """Copy/send client content into the topic. + + Returns (manager_message_id, actual_message_thread_id). + actual_message_thread_id is set for send_message; None for copy_message + (Telegram only returns MessageId there). + """ + prefix = client_prefix_html(user) + reply_to = await self._reply_to_manager_id(user.id, message) + + if message.text is not None and not _is_captionable(message): + body = _html_body(message.text, message.entities) + kwargs: Dict[str, Any] = { + "chat_id": self._manager_group_id, + "message_thread_id": topic_id, + "text": format_labeled_html(f"{client_display_name(user)}:", body), + "parse_mode": ParseMode.HTML, + } + if reply_to is not None: + kwargs["reply_to_message_id"] = reply_to + sent = await self._call_with_optional_reply(bot.send_message, kwargs) + return sent.message_id, sent.message_thread_id + + if _is_captionable(message): + caption = _safe_prefixed_caption( + prefix, + message.caption, + message.caption_entities, + ) + kwargs = { + "chat_id": self._manager_group_id, + "from_chat_id": message.chat.id, + "message_id": message.message_id, + "message_thread_id": topic_id, + "caption": caption, + "parse_mode": ParseMode.HTML, + } + if reply_to is not None: + kwargs["reply_to_message_id"] = reply_to + result = await self._call_with_optional_reply(bot.copy_message, kwargs) + return result.message_id, None + + # Stickers, video notes, etc. have no caption — topic already identifies the client. + kwargs = { + "chat_id": self._manager_group_id, + "from_chat_id": message.chat.id, + "message_id": message.message_id, + "message_thread_id": topic_id, + } + if reply_to is not None: + kwargs["reply_to_message_id"] = reply_to + result = await self._call_with_optional_reply(bot.copy_message, kwargs) + return result.message_id, None + + async def _recreate_topic_for_user(self, bot: Bot, user: User, stale_topic_id: int) -> int: + logger.warning("Topic %s missing for client %s; recreating", stale_topic_id, user.id) + await self._repo.clear_topic_id(user.id) + client = await self._repo.get_by_telegram_id(user.id) + if client is None: + client = await self.ensure_client(user) + else: + # Ensure ensure_topic does not trust the cleared id from a stale object. + client.topic_id = None + return await self.ensure_topic(bot, user, client) + + async def relay_client_to_manager(self, bot: Bot, message: Message) -> None: + if message.from_user is None: + return + + user = message.from_user + client = await self.ensure_client(user) + topic_id = await self.ensure_topic(bot, user, client) + + try: + manager_message_id, actual_thread = await self._copy_client_message( + bot, message, topic_id, user + ) + except TelegramBadRequest as exc: + # Topic may have been deleted manually — recreate once. + if not _is_topic_missing_error(exc): + logger.exception("Failed to copy client message to topic %s", topic_id) + raise + + topic_id = await self._recreate_topic_for_user(bot, user, topic_id) + manager_message_id, actual_thread = await self._copy_client_message( + bot, message, topic_id, user + ) + + # Text sends can silently land in General when the topic is gone. + if ( + manager_message_id is not None + and actual_thread is not None + and not _thread_matches(topic_id, actual_thread) + ): + logger.warning( + "Message for client %s landed in thread %s instead of %s; recreating topic", + user.id, + actual_thread, + topic_id, + ) + await self._safe_delete(bot, self._manager_group_id, manager_message_id) + topic_id = await self._recreate_topic_for_user(bot, user, topic_id) + manager_message_id, actual_thread = await self._copy_client_message( + bot, message, topic_id, user + ) + if ( + manager_message_id is not None + and actual_thread is not None + and not _thread_matches(topic_id, actual_thread) + ): + await self._safe_delete(bot, self._manager_group_id, manager_message_id) + raise RuntimeError( + f"Failed to deliver client {user.id} message into topic {topic_id}" + ) + + if manager_message_id is not None: + await self._link_messages(user.id, message.message_id, manager_message_id) + + async def _deliver_manager_message( + self, + bot: Bot, + client: Client, + message: Message, + ) -> Optional[int]: + prefix = format_prefix(client.language, "prefix_manager") + reply_to = await self._reply_to_client_id(message) + + if message.text is not None and not _is_captionable(message): + body = _html_body(message.text, message.entities) + kwargs: Dict[str, Any] = { + "chat_id": client.telegram_id, + "text": format_manager_html(client.language, body), + "parse_mode": ParseMode.HTML, + } + if reply_to is not None: + kwargs["reply_to_message_id"] = reply_to + sent = await self._call_with_optional_reply(bot.send_message, kwargs) + return sent.message_id + + if _is_captionable(message): + caption = _safe_prefixed_caption( + prefix, + message.caption, + message.caption_entities, + ) + kwargs = { + "chat_id": client.telegram_id, + "from_chat_id": message.chat.id, + "message_id": message.message_id, + "caption": caption, + "parse_mode": ParseMode.HTML, + } + if reply_to is not None: + kwargs["reply_to_message_id"] = reply_to + result = await self._call_with_optional_reply(bot.copy_message, kwargs) + return result.message_id + + kwargs = { + "chat_id": client.telegram_id, + "from_chat_id": message.chat.id, + "message_id": message.message_id, + } + if reply_to is not None: + kwargs["reply_to_message_id"] = reply_to + result = await self._call_with_optional_reply(bot.copy_message, kwargs) + return result.message_id + + async def _notify_managers(self, bot: Bot, topic_id: int, key: str) -> None: + try: + await bot.send_message( + chat_id=self._manager_group_id, + message_thread_id=topic_id, + text=t(DEFAULT_LANGUAGE, key), + ) + except TelegramAPIError: + logger.exception("Failed to notify managers (%s) in topic %s", key, topic_id) + + async def relay_manager_to_client(self, bot: Bot, message: Message) -> None: + topic_id = message.message_thread_id + if topic_id is None or topic_id == GENERAL_TOPIC_ID: + return + + client = await self._repo.get_by_topic_id(topic_id) + if client is None: + logger.warning("No client mapped to topic %s", topic_id) + return + + try: + client_message_id = await self._deliver_manager_message(bot, client, message) + if client_message_id is not None: + await self._link_messages( + client.telegram_id, + client_message_id, + message.message_id, + ) + except TelegramForbiddenError: + logger.warning("Client %s blocked the bot", client.telegram_id) + await self._notify_managers(bot, topic_id, "client_blocked") + except TelegramBadRequest as exc: + logger.exception( + "Failed to deliver manager message to client %s: %s", + client.telegram_id, + exc, + ) + await self._notify_managers(bot, topic_id, "deliver_failed") + + async def relay_client_edit(self, bot: Bot, message: Message) -> None: + """Mirror a client message edit into the manager topic.""" + if message.from_user is None: + return + + link = await self._repo.get_link_by_client_message( + message.from_user.id, + message.message_id, + ) + if link is None: + return + + user = message.from_user + try: + if message.text is not None and not _is_captionable(message): + body = _html_body(message.text, message.entities) + await bot.edit_message_text( + chat_id=self._manager_group_id, + message_id=link.manager_message_id, + text=format_labeled_html(f"{client_display_name(user)}:", body), + parse_mode=ParseMode.HTML, + ) + return + + if _is_captionable(message): + caption = _safe_prefixed_caption( + client_prefix_html(user), + message.caption, + message.caption_entities, + ) + await bot.edit_message_caption( + chat_id=self._manager_group_id, + message_id=link.manager_message_id, + caption=caption, + parse_mode=ParseMode.HTML, + ) + except TelegramBadRequest as exc: + if _is_message_gone_error(exc): + await self._delete_paired_after_gone( + bot, + client_telegram_id=link.client_telegram_id, + client_message_id=link.client_message_id, + manager_message_id=link.manager_message_id, + delete_manager=False, + delete_client=True, + ) + else: + logger.exception( + "Failed to sync client edit to manager message %s", + link.manager_message_id, + ) + + async def relay_manager_edit(self, bot: Bot, message: Message) -> None: + """Mirror a manager message edit into the client DM.""" + topic_id = message.message_thread_id + if topic_id is None or topic_id == GENERAL_TOPIC_ID: + return + + link = await self._repo.get_link_by_manager_message(message.message_id) + if link is None: + return + + client = await self._repo.get_by_telegram_id(link.client_telegram_id) + if client is None: + return + + try: + if message.text is not None and not _is_captionable(message): + body = _html_body(message.text, message.entities) + await bot.edit_message_text( + chat_id=client.telegram_id, + message_id=link.client_message_id, + text=format_manager_html(client.language, body), + parse_mode=ParseMode.HTML, + ) + return + + if _is_captionable(message): + caption = _safe_prefixed_caption( + format_prefix(client.language, "prefix_manager"), + message.caption, + message.caption_entities, + ) + await bot.edit_message_caption( + chat_id=client.telegram_id, + message_id=link.client_message_id, + caption=caption, + parse_mode=ParseMode.HTML, + ) + except TelegramForbiddenError: + await self._notify_managers(bot, topic_id, "client_blocked") + except TelegramBadRequest as exc: + if _is_message_gone_error(exc): + await self._delete_paired_after_gone( + bot, + client_telegram_id=link.client_telegram_id, + client_message_id=link.client_message_id, + manager_message_id=link.manager_message_id, + delete_manager=True, + delete_client=False, + ) + else: + logger.exception( + "Failed to sync manager edit to client message %s", + link.client_message_id, + ) + await self._notify_managers(bot, topic_id, "edit_failed") + + async def delete_paired_from_client(self, bot: Bot, message: Message) -> bool: + """Delete the manager-side copy for a client message (reply+/del).""" + if message.from_user is None or message.reply_to_message is None: + return False + link = await self._repo.get_link_by_client_message( + message.from_user.id, + message.reply_to_message.message_id, + ) + if link is None: + return False + await self._delete_both_sides(bot, link.client_telegram_id, link) + return True + + async def delete_paired_from_manager(self, bot: Bot, message: Message) -> bool: + """Delete the client-side copy for a manager message (reply+/del).""" + if message.reply_to_message is None: + return False + link = await self._repo.get_link_by_manager_message( + message.reply_to_message.message_id, + ) + if link is None: + return False + await self._delete_both_sides(bot, link.client_telegram_id, link) + return True + + async def _delete_both_sides( + self, + bot: Bot, + client_telegram_id: int, + link: MessageLink, + ) -> None: + await self._safe_delete(bot, client_telegram_id, link.client_message_id) + await self._safe_delete(bot, self._manager_group_id, link.manager_message_id) + await self._repo.delete_message_link( + client_telegram_id=link.client_telegram_id, + client_message_id=link.client_message_id, + manager_message_id=link.manager_message_id, + ) + + async def _delete_paired_after_gone( + self, + bot: Bot, + *, + client_telegram_id: int, + client_message_id: int, + manager_message_id: int, + delete_manager: bool, + delete_client: bool, + ) -> None: + if delete_client: + await self._safe_delete(bot, client_telegram_id, client_message_id) + if delete_manager: + await self._safe_delete(bot, self._manager_group_id, manager_message_id) + await self._repo.delete_message_link( + client_telegram_id=client_telegram_id, + client_message_id=client_message_id, + manager_message_id=manager_message_id, + ) + + async def _safe_delete(self, bot: Bot, chat_id: int, message_id: int) -> None: + try: + await bot.delete_message(chat_id=chat_id, message_id=message_id) + except TelegramAPIError: + logger.info("Could not delete message %s in chat %s", message_id, chat_id) + + async def relay_reaction(self, bot: Bot, event: MessageReactionUpdated) -> None: + if event.user is not None and self.is_self(event.user.id): + return + + reactions = bot_reactions_from_update(event) + + if event.chat.type == ChatType.PRIVATE: + link = await self._repo.get_link_by_client_message( + event.chat.id, + event.message_id, + ) + if link is None: + return + try: + await bot.set_message_reaction( + chat_id=self._manager_group_id, + message_id=link.manager_message_id, + reaction=reactions, + ) + except TelegramAPIError: + logger.exception( + "Failed to sync client reaction to manager message %s", + link.manager_message_id, + ) + return + + if event.chat.id != self._manager_group_id: + return + + link = await self._repo.get_link_by_manager_message(event.message_id) + if link is None: + return + try: + await bot.set_message_reaction( + chat_id=link.client_telegram_id, + message_id=link.client_message_id, + reaction=reactions, + ) + except TelegramAPIError: + logger.exception( + "Failed to sync manager reaction to client message %s", + link.client_message_id, + ) + + +def is_relayable_content(message: Message) -> bool: + """True for user content we can copy (not service messages).""" + return bool( + message.text + or message.caption + or message.photo + or message.document + or message.video + or message.audio + or message.voice + or message.video_note + or message.sticker + or message.animation + or message.contact + or message.location + or message.venue + or message.poll + or message.dice + ) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..445ceaa --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +aiogram>=3.15,<4 +aiosqlite>=0.20 +python-dotenv>=1.0