reverse bridge

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:

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)
+

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])

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: