diff --git a/bot/db/repository.py b/bot/db/repository.py index 4690340..9509604 100644 --- a/bot/db/repository.py +++ b/bot/db/repository.py @@ -27,6 +27,7 @@ CREATE TABLE IF NOT EXISTS message_links ( client_telegram_id INTEGER NOT NULL, client_message_id INTEGER NOT NULL, manager_message_id INTEGER NOT NULL, + content_fingerprint TEXT, created_at TEXT NOT NULL, UNIQUE(client_telegram_id, client_message_id), UNIQUE(manager_message_id) @@ -55,6 +56,7 @@ class MessageLink: client_telegram_id: int client_message_id: int manager_message_id: int + content_fingerprint: Optional[str] = None def _utcnow() -> str: @@ -97,6 +99,14 @@ class ClientRepository: """ ) + async with conn.execute("PRAGMA table_info(message_links)") as cursor: + link_rows = await cursor.fetchall() + link_columns = {row[1] for row in link_rows} + if "content_fingerprint" not in link_columns: + await conn.execute( + "ALTER TABLE message_links ADD COLUMN content_fingerprint TEXT" + ) + async def close(self) -> None: if self._conn is not None: await self._conn.close() @@ -216,11 +226,26 @@ class ClientRepository: ) await conn.commit() + @staticmethod + def _row_to_link(row: aiosqlite.Row) -> MessageLink: + fingerprint = None + try: + fingerprint = row["content_fingerprint"] + except (IndexError, KeyError): + fingerprint = None + return MessageLink( + client_telegram_id=row["client_telegram_id"], + client_message_id=row["client_message_id"], + manager_message_id=row["manager_message_id"], + content_fingerprint=fingerprint, + ) + async def save_message_link( self, client_telegram_id: int, client_message_id: int, manager_message_id: int, + content_fingerprint: Optional[str] = None, ) -> None: conn = self._require_conn() await conn.execute( @@ -237,10 +262,43 @@ class ClientRepository: 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, + content_fingerprint, created_at + ) VALUES (?, ?, ?, ?, ?) """, - (client_telegram_id, client_message_id, manager_message_id, _utcnow()), + ( + client_telegram_id, + client_message_id, + manager_message_id, + content_fingerprint, + _utcnow(), + ), + ) + await conn.commit() + + async def set_link_fingerprint( + self, + *, + client_telegram_id: int, + client_message_id: int, + manager_message_id: int, + content_fingerprint: str, + ) -> None: + conn = self._require_conn() + await conn.execute( + """ + UPDATE message_links + SET content_fingerprint = ? + WHERE client_telegram_id = ? + AND client_message_id = ? + AND manager_message_id = ? + """, + ( + content_fingerprint, + client_telegram_id, + client_message_id, + manager_message_id, + ), ) await conn.commit() @@ -252,20 +310,15 @@ class ClientRepository: conn = self._require_conn() async with conn.execute( """ - SELECT client_telegram_id, client_message_id, manager_message_id + SELECT client_telegram_id, client_message_id, manager_message_id, + content_fingerprint 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"], - ) + return self._row_to_link(row) if row else None async def get_link_by_manager_message( self, @@ -274,20 +327,15 @@ class ClientRepository: conn = self._require_conn() async with conn.execute( """ - SELECT client_telegram_id, client_message_id, manager_message_id + SELECT client_telegram_id, client_message_id, manager_message_id, + content_fingerprint 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"], - ) + return self._row_to_link(row) if row else None async def delete_message_link( self, diff --git a/bot/services/relay.py b/bot/services/relay.py index a087398..fbe2520 100644 --- a/bot/services/relay.py +++ b/bot/services/relay.py @@ -1,9 +1,11 @@ from __future__ import annotations import asyncio +import hashlib import logging +import time from html import escape -from typing import Any, Dict, Optional +from typing import Any, Dict, Optional, Tuple from aiogram import Bot from aiogram.enums import ChatType, ParseMode @@ -34,6 +36,8 @@ TOPIC_NAME_MAX_LEN = 128 CAPTION_MAX_LEN = 1024 # How many times to wait out Telegram flood control before giving up. FLOOD_RETRY_ATTEMPTS = 5 +# Collapse identical manager warnings so a tag-change flood cannot spam topics. +NOTIFY_COOLDOWN_SECONDS = 120.0 async def _with_flood_retry(api_call, *args, **kwargs): @@ -146,6 +150,43 @@ def bot_reactions_from_update(event: MessageReactionUpdated) -> list: return [] +def _entities_key(entities) -> str: + if not entities: + return "" + chunks = [] + for entity in entities: + kind = getattr(entity.type, "value", entity.type) + url = getattr(entity, "url", None) or "" + custom_emoji_id = getattr(entity, "custom_emoji_id", None) or "" + chunks.append(f"{kind}:{entity.offset}:{entity.length}:{url}:{custom_emoji_id}") + return "|".join(chunks) + + +def _media_key(message: Message) -> str: + if message.photo: + return f"photo:{message.photo[-1].file_unique_id}" + for attr in ("document", "video", "audio", "animation", "voice", "video_note", "sticker"): + obj = getattr(message, attr, None) + if obj is not None: + uid = getattr(obj, "file_unique_id", "") or "" + return f"{attr}:{uid}" + return "" + + +def message_content_fingerprint(message: Message) -> str: + """Hash of fields we relay. Ignores sender_tag, edit_date, and similar.""" + payload = "\n".join( + ( + message.text or "", + message.caption or "", + _entities_key(message.entities), + _entities_key(message.caption_entities), + _media_key(message), + ) + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + def _is_message_gone_error(exc: BaseException) -> bool: text = str(exc).lower() return any( @@ -153,13 +194,34 @@ def _is_message_gone_error(exc: BaseException) -> bool: 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_not_modified_error(exc: BaseException) -> bool: + return "message is not modified" in str(exc).lower() + + +def _is_uneditable_error(exc: BaseException) -> bool: + text = str(exc).lower() + return any( + marker in text + for marker in ( + "message can't be edited", + "message can not be edited", + "message_too_old", + "message is too old", + ) + ) + + +def _is_benign_edit_error(exc: BaseException) -> bool: + """True for errors that do not mean the paired message is gone.""" + return _is_not_modified_error(exc) or _is_uneditable_error(exc) + + def _is_reply_error(exc: BaseException) -> bool: text = str(exc).lower() # Telegram uses "replied" (y→i), so "reply" is not a substring of "replied". @@ -201,6 +263,7 @@ class RelayService: self._bot_id: Optional[int] = None self._topic_locks: Dict[int, asyncio.Lock] = {} self._topic_locks_guard = asyncio.Lock() + self._notify_last: Dict[Tuple[int, str], float] = {} def set_bot_id(self, bot_id: int) -> None: self._bot_id = bot_id @@ -281,12 +344,14 @@ class RelayService: client_telegram_id: int, client_message_id: int, manager_message_id: int, + source: Message, ) -> 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, + content_fingerprint=message_content_fingerprint(source), ) except Exception: logger.exception( @@ -296,6 +361,20 @@ class RelayService: manager_message_id, ) + async def _remember_fingerprint(self, link: MessageLink, fingerprint: str) -> None: + try: + await self._repo.set_link_fingerprint( + client_telegram_id=link.client_telegram_id, + client_message_id=link.client_message_id, + manager_message_id=link.manager_message_id, + content_fingerprint=fingerprint, + ) + except Exception: + logger.exception( + "Failed to store content fingerprint for manager message %s", + link.manager_message_id, + ) + async def _reply_to_manager_id( self, client_telegram_id: int, @@ -447,7 +526,9 @@ class RelayService: ) if manager_message_id is not None: - await self._link_messages(user.id, message.message_id, manager_message_id) + await self._link_messages( + user.id, message.message_id, manager_message_id, message + ) async def _deliver_manager_message( self, @@ -499,6 +580,17 @@ class RelayService: return result.message_id async def _notify_managers(self, bot: Bot, topic_id: int, key: str) -> None: + now = time.monotonic() + stamp_key = (topic_id, key) + last = self._notify_last.get(stamp_key) + if last is not None and now - last < NOTIFY_COOLDOWN_SECONDS: + logger.info( + "Skipping duplicate manager notify %s in topic %s", + key, + topic_id, + ) + return + self._notify_last[stamp_key] = now try: await _with_flood_retry( bot.send_message, @@ -526,6 +618,7 @@ class RelayService: client.telegram_id, client_message_id, message.message_id, + message, ) except TelegramForbiddenError: logger.warning("Client %s blocked the bot", client.telegram_id) @@ -550,6 +643,10 @@ class RelayService: if link is None: return + fingerprint = message_content_fingerprint(message) + if link.content_fingerprint == fingerprint: + return + user = message.from_user try: if message.text is not None and not _is_captionable(message): @@ -561,9 +658,7 @@ class RelayService: text=format_labeled_html(f"{client_display_name(user)}:", body), parse_mode=ParseMode.HTML, ) - return - - if _is_captionable(message): + elif _is_captionable(message): caption = _safe_prefixed_caption( client_prefix_html(user), message.caption, @@ -576,7 +671,11 @@ class RelayService: caption=caption, parse_mode=ParseMode.HTML, ) + await self._remember_fingerprint(link, fingerprint) except TelegramBadRequest as exc: + if _is_benign_edit_error(exc): + await self._remember_fingerprint(link, fingerprint) + return if _is_message_gone_error(exc): await self._delete_paired_after_gone( bot, @@ -602,6 +701,10 @@ class RelayService: if link is None: return + fingerprint = message_content_fingerprint(message) + if link.content_fingerprint == fingerprint: + return + client = await self._repo.get_by_telegram_id(link.client_telegram_id) if client is None: return @@ -616,9 +719,7 @@ class RelayService: text=format_manager_html(client.language, body), parse_mode=ParseMode.HTML, ) - return - - if _is_captionable(message): + elif _is_captionable(message): caption = _safe_prefixed_caption( format_prefix(client.language, "prefix_manager"), message.caption, @@ -631,9 +732,13 @@ class RelayService: caption=caption, parse_mode=ParseMode.HTML, ) + await self._remember_fingerprint(link, fingerprint) except TelegramForbiddenError: await self._notify_managers(bot, topic_id, "client_blocked") except TelegramBadRequest as exc: + if _is_benign_edit_error(exc): + await self._remember_fingerprint(link, fingerprint) + return if _is_message_gone_error(exc): await self._delete_paired_after_gone( bot,