diff --git a/src/main.py b/src/main.py index 52b738c..69a25b1 100644 --- a/src/main.py +++ b/src/main.py @@ -6,6 +6,7 @@ from dotenv import load_dotenv from bridge import MaxToTelegramBridge from config import load_settings +from reverse_bridge import TelegramToMaxBridge from storage import BridgeStorage from telegram_api import TelegramClient @@ -35,12 +36,21 @@ def build_client() -> tuple[MaxClient, MaxToTelegramBridge]: def main() -> None: load_dotenv() _setup_logging() - max_client, bridge = build_client() + settings = load_settings() + max_client = MaxClient(phone=settings.max_phone, work_dir=settings.max_work_dir) + telegram_client = TelegramClient( + bot_token=settings.telegram_bot_token, + fallback_user_id=settings.telegram_fallback_user_id, + ) + storage = BridgeStorage(settings.sqlite_path) + bridge = MaxToTelegramBridge(max_client=max_client, telegram=telegram_client, storage=storage) + reverse_bridge = TelegramToMaxBridge(max_client=max_client, telegram=telegram_client, storage=storage) logger = logging.getLogger("max2telegram") @max_client.on_start async def on_start() -> None: logger.info("Max client started as %s", max_client.me.id) + asyncio.create_task(reverse_bridge.start()) @max_client.on_message() async def on_message(message: Message) -> None: diff --git a/src/reverse_bridge.py b/src/reverse_bridge.py new file mode 100644 index 0000000..f3c18a3 --- /dev/null +++ b/src/reverse_bridge.py @@ -0,0 +1,323 @@ +import asyncio +import logging +import time +from dataclasses import dataclass, field +from typing import Any + +from pymax import MaxClient +from pymax.files import Photo, Video + +from storage import BridgeStorage +from telegram_api import TelegramClient + +logger = logging.getLogger(__name__) + + +def _normalize_title(value: str) -> str: + return str(value or "").strip().casefold() + + +def _telegram_chat_title(chat: dict[str, Any]) -> str: + # для каналов обычно есть title; для групп тоже; в крайнем случае — username + return str(chat.get("title") or chat.get("username") or "").strip() + + +@dataclass +class _MediaGroupBuffer: + first_seen_monotonic: float + updates: list[dict[str, Any]] = field(default_factory=list) + + +class TelegramToMaxBridge: + def __init__(self, *, max_client: MaxClient, telegram: TelegramClient, storage: BridgeStorage) -> None: + self._max_client = max_client + self._telegram = telegram + self._storage = storage + self._max_title_to_id: dict[str, int] = {} + + self._bot_id: str | None = None + self._offset: int | None = None + + self._media_groups: dict[tuple[str, str], _MediaGroupBuffer] = {} + self._media_group_grace_sec = 1.2 + + async def start(self) -> None: + me = await self._telegram.get_me() + self._bot_id = str(me.get("id") or "") + if not self._bot_id: + raise RuntimeError("Cannot resolve Telegram bot id (getMe)") + + self._refresh_max_chat_cache() + logger.info("Telegram->Max bridge started (bot_id=%s)", self._bot_id) + + while True: + try: + updates = await self._telegram.get_updates(offset=self._offset, timeout=25, limit=100) + await self._handle_updates(updates) + except Exception: + logger.exception("Telegram polling loop error") + await asyncio.sleep(2) + + async def _handle_updates(self, updates: list[dict[str, Any]]) -> None: + max_update_id = None + for upd in updates: + upd_id = upd.get("update_id") + if isinstance(upd_id, int): + max_update_id = upd_id if max_update_id is None else max(max_update_id, upd_id) + + message = None + for container in ("message", "edited_message", "channel_post", "edited_channel_post"): + candidate = upd.get(container) + if isinstance(candidate, dict): + message = candidate + break + if not message: + continue + + if self._is_own_telegram_message(message): + continue + + await self._handle_message(message) + + if max_update_id is not None: + self._offset = max_update_id + 1 + + await self._flush_ready_media_groups() + + def _is_own_telegram_message(self, message: dict[str, Any]) -> bool: + sender = message.get("from") + if isinstance(sender, dict): + if sender.get("is_bot") is True: + # важно: не уйти в цикл на собственных постах бота + return True + if self._bot_id and str(sender.get("id") or "") == self._bot_id: + return True + return False + + async def _handle_message(self, message: dict[str, Any]) -> None: + chat = message.get("chat") + if not isinstance(chat, dict): + return + + chat_title = _telegram_chat_title(chat) + normalized = _normalize_title(chat_title) + if not normalized: + logger.error("Telegram chat without title/username, skip (chat=%s)", chat) + return + + max_chat_id = self._resolve_max_chat_id_by_title(normalized) + if max_chat_id is None: + # требование: если в MAX нет канала/группы — ошибка и не пересылать + logger.error("MAX чат с названием '%s' не найден — сообщение не пересылаю", chat_title) + return + + telegram_chat_id = str(chat.get("id")) + telegram_message_id = str(message.get("message_id")) + + media_group_id = message.get("media_group_id") + if media_group_id is not None: + key = (telegram_chat_id, str(media_group_id)) + buf = self._media_groups.get(key) + if buf is None: + buf = _MediaGroupBuffer(first_seen_monotonic=time.monotonic()) + self._media_groups[key] = buf + buf.updates.append(message) + return + + await self._forward_single_message( + max_chat_id=max_chat_id, + telegram_chat_id=telegram_chat_id, + telegram_message_id=telegram_message_id, + message=message, + media_group_id=None, + ) + + async def _flush_ready_media_groups(self) -> None: + now = time.monotonic() + ready: list[tuple[tuple[str, str], _MediaGroupBuffer]] = [] + for key, buf in self._media_groups.items(): + if (now - buf.first_seen_monotonic) >= self._media_group_grace_sec: + ready.append((key, buf)) + + for key, buf in ready: + self._media_groups.pop(key, None) + telegram_chat_id, media_group_id = key + # сообщения альбома приходят отдельно; отправляем в MAX одним сообщением с несколькими attachments + await self._forward_media_group( + telegram_chat_id=telegram_chat_id, + media_group_id=media_group_id, + messages=buf.updates, + ) + + async def _forward_media_group(self, *, telegram_chat_id: str, media_group_id: str, messages: list[dict[str, Any]]) -> None: + if not messages: + return + + chat = messages[0].get("chat") + if not isinstance(chat, dict): + return + chat_title = _telegram_chat_title(chat) + normalized = _normalize_title(chat_title) + max_chat_id = self._resolve_max_chat_id_by_title(normalized) + if max_chat_id is None: + logger.error("MAX чат с названием '%s' не найден — media group не пересылаю", chat_title) + return + + # Telegram может прислать несколько элементов с caption только на первом. Берём text/caption с первого, где он есть. + text = "" + for m in messages: + cand = str(m.get("text") or m.get("caption") or "").strip() + if cand: + text = cand + break + + reply_to = self._resolve_reply_to_max_id(max_chat_id=max_chat_id, message=messages[0]) + + attachments: list[Any] = [] + for m in messages: + attachments.extend(await self._extract_attachments(m)) + + if not text and not attachments: + return + + sent = await self._max_client.send_message( + chat_id=max_chat_id, + text=text, + attachments=attachments if attachments else None, + reply_to=reply_to, + ) + if not sent: + logger.error("MAX send_message returned empty for media group (chat_id=%s)", max_chat_id) + return + + max_message_id = str(getattr(sent, "id", "") or "") + if not max_message_id: + logger.error("Cannot resolve MAX message id after sending media group (chat_id=%s)", max_chat_id) + return + + for m in messages: + tid = str(m.get("message_id")) + if tid: + self._storage.save_mapping( + telegram_chat_id=telegram_chat_id, + telegram_message_id=tid, + max_chat_id=str(max_chat_id), + max_message_id=max_message_id, + media_group_id=media_group_id, + ) + + logger.info( + "Forwarded Telegram media group %s (count=%s) -> MAX %s/%s", + media_group_id, + len(messages), + max_chat_id, + max_message_id, + ) + + async def _forward_single_message( + self, + *, + max_chat_id: int, + telegram_chat_id: str, + telegram_message_id: str, + message: dict[str, Any], + media_group_id: str | None, + ) -> None: + text = str(message.get("text") or message.get("caption") or "").strip() + attachments = await self._extract_attachments(message) + if not text and not attachments: + return + + reply_to = self._resolve_reply_to_max_id(max_chat_id=max_chat_id, message=message) + + sent = await self._max_client.send_message( + chat_id=max_chat_id, + text=text, + attachments=attachments if attachments else None, + reply_to=reply_to, + ) + if not sent: + logger.error("MAX send_message returned empty (chat_id=%s)", max_chat_id) + return + + max_message_id = str(getattr(sent, "id", "") or "") + if not max_message_id: + logger.error("Cannot resolve MAX message id after sending (chat_id=%s)", max_chat_id) + return + + self._storage.save_mapping( + telegram_chat_id=telegram_chat_id, + telegram_message_id=telegram_message_id, + max_chat_id=str(max_chat_id), + max_message_id=max_message_id, + media_group_id=media_group_id, + ) + logger.info("Forwarded Telegram %s/%s -> MAX %s/%s", telegram_chat_id, telegram_message_id, max_chat_id, max_message_id) + + def _resolve_reply_to_max_id(self, *, max_chat_id: int, message: dict[str, Any]) -> str | None: + reply = message.get("reply_to_message") + if not isinstance(reply, dict): + return None + reply_mid = reply.get("message_id") + if reply_mid is None: + return None + + chat = message.get("chat") + if not isinstance(chat, dict): + return None + telegram_chat_id = str(chat.get("id")) + mapped = self._storage.get_max_message_id_for_telegram( + telegram_chat_id=telegram_chat_id, + telegram_message_id=str(reply_mid), + ) + # reply_to в MAX — это id сообщения; если не нашли, просто отправляем без reply + return mapped + + async def _extract_attachments(self, message: dict[str, Any]) -> list[Any]: + attachments: list[Any] = [] + + # photo: массив размеров, берём последний (самый большой) + photos = message.get("photo") + if isinstance(photos, list) and photos: + best = None + for p in photos: + if isinstance(p, dict) and p.get("file_id"): + best = p + if best and isinstance(best, dict): + file_id = str(best.get("file_id")) + if file_id: + try: + url = await self._telegram.get_file_url(file_id) + attachments.append(Photo(url=url)) + except Exception: + logger.exception("Cannot fetch Telegram photo URL") + + video = message.get("video") + if isinstance(video, dict) and video.get("file_id"): + file_id = str(video.get("file_id")) + if file_id: + try: + url = await self._telegram.get_file_url(file_id) + attachments.append(Video(url=url)) + except Exception: + logger.exception("Cannot fetch Telegram video URL") + + return attachments + + def _refresh_max_chat_cache(self) -> None: + title_to_id: dict[str, int] = {} + for chat in getattr(self._max_client, "chats", []) or []: + title = getattr(chat, "title", None) + chat_id = getattr(chat, "id", None) + if title and chat_id is not None: + title_to_id[_normalize_title(str(title))] = int(chat_id) + self._max_title_to_id = title_to_id + + def _resolve_max_chat_id_by_title(self, normalized_title: str) -> int | None: + chat_id = self._max_title_to_id.get(normalized_title) + if chat_id is not None: + return chat_id + # на всякий случай обновим кэш (например, если добавили чат во время работы) + self._refresh_max_chat_cache() + return self._max_title_to_id.get(normalized_title) + diff --git a/src/storage.py b/src/storage.py index 34a65c5..b55a077 100644 --- a/src/storage.py +++ b/src/storage.py @@ -22,6 +22,25 @@ class BridgeStorage: ) """ ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS message_mapping ( + telegram_chat_id TEXT NOT NULL, + telegram_message_id TEXT NOT NULL, + max_chat_id TEXT NOT NULL, + max_message_id TEXT NOT NULL, + media_group_id TEXT, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (telegram_chat_id, telegram_message_id) + ) + """ + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_message_mapping_max ON message_mapping (max_chat_id, max_message_id)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_message_mapping_media_group ON message_mapping (telegram_chat_id, media_group_id)" + ) conn.commit() def was_forwarded(self, message_id: str, chat_id: str) -> bool: @@ -39,3 +58,39 @@ class BridgeStorage: (message_id, chat_id), ) conn.commit() + + def save_mapping( + self, + *, + telegram_chat_id: str, + telegram_message_id: str, + max_chat_id: str, + max_message_id: str, + media_group_id: str | None = None, + ) -> None: + with closing(self._connect()) as conn: + conn.execute( + """ + INSERT OR REPLACE INTO message_mapping + (telegram_chat_id, telegram_message_id, max_chat_id, max_message_id, media_group_id) + VALUES (?, ?, ?, ?, ?) + """, + (telegram_chat_id, telegram_message_id, max_chat_id, max_message_id, media_group_id), + ) + conn.commit() + + def get_max_message_id_for_telegram( + self, *, telegram_chat_id: str, telegram_message_id: str + ) -> str | None: + with closing(self._connect()) as conn: + row = conn.execute( + """ + SELECT max_message_id + FROM message_mapping + WHERE telegram_chat_id = ? AND telegram_message_id = ? + """, + (telegram_chat_id, telegram_message_id), + ).fetchone() + if not row: + return None + return str(row[0]) diff --git a/src/telegram_api.py b/src/telegram_api.py index f5a17d6..86aabec 100644 --- a/src/telegram_api.py +++ b/src/telegram_api.py @@ -11,9 +11,11 @@ class TelegramApiError(RuntimeError): class TelegramClient: def __init__(self, bot_token: str, fallback_user_id: str, timeout: int = 30) -> None: self._base_url = f"https://api.telegram.org/bot{bot_token}" + self._file_base_url = f"https://api.telegram.org/file/bot{bot_token}" self._fallback_user_id = fallback_user_id self._timeout = timeout self._chat_title_to_id: dict[str, str] = {} + self._me: dict[str, Any] | None = None async def resolve_target_chat_id(self, max_chat_name: str) -> tuple[str, bool]: chat_id = await self._find_chat_id_by_title(max_chat_name) @@ -79,6 +81,41 @@ class TelegramClient: }, ) + async def get_me(self) -> dict[str, Any]: + if self._me is not None: + return self._me + data = await self._request("getMe", {}) + me = data.get("result") + if not isinstance(me, dict): + raise TelegramApiError(f"Telegram getMe: unexpected payload {data}") + self._me = me + return me + + async def get_updates(self, *, offset: int | None, timeout: int = 25, limit: int = 100) -> list[dict[str, Any]]: + payload: dict[str, Any] = { + "timeout": timeout, + "limit": limit, + # чтобы получать посты из каналов (channel_post) и обычные сообщения + "allowed_updates": ["message", "edited_message", "channel_post", "edited_channel_post"], + } + if offset is not None: + payload["offset"] = offset + data = await self._request("getUpdates", payload) + result = data.get("result", []) + if not isinstance(result, list): + return [] + return [u for u in result if isinstance(u, dict)] + + async def get_file_url(self, file_id: str) -> str: + data = await self._request("getFile", {"file_id": file_id}) + result = data.get("result") + if not isinstance(result, dict): + raise TelegramApiError(f"Telegram getFile: unexpected payload {data}") + file_path = result.get("file_path") + if not isinstance(file_path, str) or not file_path.strip(): + raise TelegramApiError(f"Telegram getFile: missing file_path {data}") + return f"{self._file_base_url}/{file_path}" + async def _find_chat_id_by_title(self, chat_title: str) -> str | None: normalized = self._normalize_title(chat_title) if not normalized: