tg2max-relay: peresylka otchyotov report-shot v MAX (lichnyy akkaunt, vkmax)

2026-09-17 17:17:18 +03:00
commit 36cf1060d7
11 changed files with 542 additions and 0 deletions
@@ -0,0 +1,125 @@
+"""Разовая активация MAX-сессии (запускать на своём ПК, не на сервере):
+
+    python -m app.login --phone +79991234567
+    (попросит код из SMS/пуша MAX, можно через --code-file для автоматики)
+    python -m app.login --chats        # список чатов с chatId (для MAX_TARGET_CHAT)
+
+Результат: max_session.json (device_id, login_token, my_id) — значения едут в
+переменные Coolify: MAX_DEVICE_ID, MAX_LOGIN_TOKEN, MAX_TARGET_CHAT."""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+from pathlib import Path
+
+from vkmax.client import MaxClient
+from vkmax.functions.messages import send_message
+
+SESSION_FILE = Path(
+    os.environ.get("MAX_SESSION_FILE")
+    or Path(__file__).resolve().parent.parent / "max_session.json"
+)
+
+
+def _load() -> dict:
+    return json.loads(SESSION_FILE.read_text("utf-8"))
+
+
+def _find_keys(node, wanted: set[str], out: dict | None = None) -> dict:
+    if out is None:
+        out = {}
+    if isinstance(node, dict):
+        for k, v in node.items():
+            if k in wanted and isinstance(v, (int, str)) and k not in out:
+                out[k] = v
+            _find_keys(v, wanted, out)
+    elif isinstance(node, list):
+        for v in node[:60]:
+            _find_keys(v, wanted, out)
+    return out
+
+
+async def _wait_code(code_file: str | None) -> str:
+    if code_file:
+        p = Path(code_file)
+        print(f"RELAY-AWAIT-CODE: запишите код в {p.resolve()}", flush=True)
+        while not p.exists():
+            await asyncio.sleep(2)
+        await asyncio.sleep(1)
+        code = p.read_text(errors="replace").strip()
+        p.unlink(missing_ok=True)
+        return code
+    return input("Код подтверждения из MAX: ").strip()
+
+
+async def login(phone: str, code_file: str | None) -> None:
+    client = MaxClient()
+    await client.connect()
+    sms_token = await client.send_code(phone)
+    code = await _wait_code(code_file)
+    resp = await client.sign_in(sms_token, int(code))
+    payload = resp["payload"]
+    login_token = payload["tokenAttrs"]["LOGIN"]["token"]
+    ids = _find_keys(payload, {"myId", "userId"})
+    session = {
+        "device_id": client.device_id,
+        "login_token": login_token,
+        "my_id": ids.get("myId") or ids.get("userId"),
+    }
+    SESSION_FILE.write_text(json.dumps(session, indent=2), "utf-8")
+    await client.disconnect()
+    print("RELAY-SESSION-DONE", flush=True)
+    print(json.dumps(session, ensure_ascii=False, indent=1))
+
+
+async def chats() -> None:
+    s = _load()
+    client = MaxClient()
+    await client.connect()
+    resp = await client.login_by_token(s["login_token"], s["device_id"])
+    payload = resp["payload"]
+    print("payload keys:", sorted(payload.keys()))
+    chat_list = payload.get("chats")
+    if isinstance(chat_list, dict):
+        chat_list = chat_list.get("chats") or list(chat_list.values())
+    if isinstance(chat_list, list):
+        for c in chat_list[:60]:
+            cid = c.get("chatId") or c.get("id")
+            info = c.get("chat") or c.get("lastChat") or c
+            print(f"- chatId={cid} | title={info.get('title')!r} | type={info.get('type')}")
+    else:
+        print(json.dumps(payload, ensure_ascii=False)[:4000])
+    await client.disconnect()
+
+
+async def test(chat_id: int) -> None:
+    s = _load()
+    client = MaxClient()
+    await client.connect()
+    await client.login_by_token(s["login_token"], s["device_id"])
+    await send_message(client, chat_id, "🟢 tg2max-relay: проверка сессии (локальный запуск)")
+    print("TEST-SENT to", chat_id, flush=True)
+    await client.disconnect()
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--phone")
+    ap.add_argument("--code-file")
+    ap.add_argument("--chats", action="store_true")
+    ap.add_argument("--test", type=int, metavar="CHAT_ID")
+    a = ap.parse_args()
+    if a.phone:
+        asyncio.run(login(a.phone, a.code_file))
+    elif a.chats:
+        asyncio.run(chats())
+    elif a.test:
+        asyncio.run(test(a.test))
+    else:
+        ap.print_help()
+
+
+if __name__ == "__main__":
+    main()