diff --git a/src/reverse_bridge.py b/src/reverse_bridge.py
index 94c40b7..da51b21 100644
--- a/src/reverse_bridge.py
+++ b/src/reverse_bridge.py
@@ -10,6 +10,7 @@ from pymax.files import Photo, Video
from storage import BridgeStorage
from telegram_api import TelegramClient
from health import HealthState
+from telegram_max_commands import handle_control_command
logger = logging.getLogger(__name__)
@@ -138,6 +139,22 @@ class TelegramToMaxBridge:
if not isinstance(chat, dict):
return
+ # Управление MAX через Telegram: только личка боту и только от fallback_user_id.
+ # В этом случае команду не пересылаем в MAX.
+ try:
+ reply = await handle_control_command(message, max_client=self._max_client, telegram=self._telegram)
+ except Exception:
+ logger.exception("Telegram MAX control command handler failed")
+ reply = None
+ if reply is not None:
+ chat_id = str(chat.get("id") or "")
+ if chat_id:
+ try:
+ await self._telegram.send_text(chat_id=chat_id, text=reply)
+ except Exception:
+ logger.exception("Cannot send Telegram reply for control command")
+ return
+
# Команда привязки чата Telegram к названию чата в MAX (для Max->Telegram маршрутизации).
# Работает даже при privacy mode, т.к. команды приходят боту.
text = str(message.get("text") or "").strip()
diff --git a/src/telegram_api.py b/src/telegram_api.py
index f13a065..f86e7ab 100644
--- a/src/telegram_api.py
+++ b/src/telegram_api.py
@@ -17,6 +17,10 @@ class TelegramClient:
self._chat_title_to_id: dict[str, str] = {}
self._me: dict[str, Any] | None = None
+ @property
+ def fallback_user_id(self) -> str:
+ return str(self._fallback_user_id or "")
+
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)
if chat_id:
diff --git a/src/telegram_max_commands.py b/src/telegram_max_commands.py
new file mode 100644
index 0000000..b632810
--- /dev/null
+++ b/src/telegram_max_commands.py
@@ -0,0 +1,166 @@
+import logging
+from typing import Any
+
+from pymax import MaxClient
+
+from telegram_api import TelegramClient
+
+logger = logging.getLogger(__name__)
+
+
+def _is_private_chat(message: dict[str, Any]) -> bool:
+ chat = message.get("chat")
+ if not isinstance(chat, dict):
+ return False
+ return str(chat.get("type") or "").strip().casefold() == "private"
+
+
+def _sender_id(message: dict[str, Any]) -> str:
+ sender = message.get("from")
+ if not isinstance(sender, dict):
+ return ""
+ return str(sender.get("id") or "")
+
+
+def _chat_id(message: dict[str, Any]) -> str:
+ chat = message.get("chat")
+ if not isinstance(chat, dict):
+ return ""
+ return str(chat.get("id") or "")
+
+
+def _infer_max_chat_type(chat_obj: Any) -> str:
+ """
+ Возвращает один из: direct / группа / канал
+ Best-effort: структура объектов pymax может отличаться между версиями.
+ """
+ # 1) Явные признаки по полям.
+ for name in ("type", "chat_type", "kind"):
+ value = getattr(chat_obj, name, None)
+ if value is None:
+ continue
+ v = str(value).strip().casefold()
+ if v in {"channel", "канал"}:
+ return "канал"
+ if v in {"group", "supergroup", "группа"}:
+ return "группа"
+ if v in {"direct", "dm", "private"}:
+ return "direct"
+
+ # 2) Флаги.
+ for true_names, mapped in (
+ (("is_channel", "channel"), "канал"),
+ (("is_group", "group"), "группа"),
+ (("is_direct", "direct"), "direct"),
+ ):
+ for n in true_names:
+ try:
+ if getattr(chat_obj, n, False) is True:
+ return mapped
+ except Exception:
+ continue
+
+ # 3) Эвристика по количеству участников.
+ for n in ("members_count", "participants", "participants_count", "member_count"):
+ try:
+ value = getattr(chat_obj, n, None)
+ except Exception:
+ value = None
+ if isinstance(value, int):
+ return "группа" if value > 2 else "direct"
+
+ return "direct"
+
+
+def _max_chat_title(chat_obj: Any) -> str:
+ for n in ("title", "name", "chat_title"):
+ try:
+ value = getattr(chat_obj, n, None)
+ except Exception:
+ value = None
+ if isinstance(value, str) and value.strip():
+ return value.strip()
+ # запасной вариант
+ cid = getattr(chat_obj, "id", None)
+ return f"chat {cid}" if cid is not None else "chat"
+
+
+async def _join_by_link(max_client: MaxClient, link: str) -> str:
+ """
+ PyMax: вступление в группу по ссылке — join_group(link).
+ См. исходники: https://github.com/MaxApiTeam/PyMax/blob/041dedeb9f9461b3360e2881a8a18a767de74871/src/pymax/mixins/group.py#L260
+ """
+ method = getattr(max_client, "join_group", None)
+ if method is None:
+ return "Не смог присоединиться: у клиента MAX нет метода join_group(link)."
+
+ try:
+ await method(link)
+ return "Ок: вступил по ссылке."
+ except Exception as e:
+ logger.exception("MAX join_group failed")
+ return f"Ошибка при присоединении: {e}"
+
+
+async def handle_control_command(
+ message: dict[str, Any],
+ *,
+ max_client: MaxClient,
+ telegram: TelegramClient,
+) -> str | None:
+ """
+ Возвращает текст ответа, если команда обработана.
+ Если None — не команда/не наш случай.
+ """
+ text = str(message.get("text") or "").strip()
+ if not text.startswith("/"):
+ return None
+
+ if not _is_private_chat(message):
+ return None
+
+ if _sender_id(message) != str(telegram.fallback_user_id):
+ return None
+
+ cmd, *rest = text.split(maxsplit=1)
+ cmd = cmd.split("@", 1)[0].strip().casefold()
+ arg = rest[0].strip() if rest else ""
+
+ if cmd == "/help":
+ return (
+ "Команды управления MAX (только для fallback_user_id в личке):\n"
+ "/help — справка\n"
+ "/list — список активных чатов MAX\n"
+ "/join — присоединиться к группе/каналу по ссылке"
+ )
+
+ if cmd == "/list":
+ # Обновим список чатов, чтобы /list был полезнее сразу после запуска.
+ try:
+ fetch = getattr(max_client, "fetch_chats", None)
+ if fetch is not None:
+ await fetch()
+ except Exception:
+ logger.debug("MAX fetch_chats failed (non-fatal)", exc_info=True)
+
+ chats = list(getattr(max_client, "chats", []) or [])
+ if not chats:
+ return "Список чатов пуст (или клиент MAX ещё не успел их загрузить)."
+
+ lines: list[str] = []
+ for c in chats:
+ title = _max_chat_title(c)
+ ctype = _infer_max_chat_type(c)
+ lines.append(f"- {title} ({ctype})")
+ return "Активные чаты MAX:\n" + "\n".join(lines)
+
+ if cmd == "/join":
+ if not arg:
+ return "Использование: /join "
+ return await _join_by_link(max_client, arg)
+
+ return (
+ "Неизвестная команда.\n"
+ "Набери /help чтобы увидеть доступные команды."
+ )
+