Add PyMax worker POC

pymax-worker/Dockerfile

@@ -0,0 +1,23 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+COPY requirements.txt ./
+RUN pip install --no-cache-dir --upgrade pip \
+    && pip install --no-cache-dir \
+        "aiofiles>=25.1.0" \
+        "aiohttp>=3.9.0" \
+        "aiosqlite>=0.22.0" \
+        "msgpack>=1.1.0" \
+        "pydantic>=2.10.0" \
+        "python-socks[asyncio]>=2.8.0" \
+        "qrcode>=8.2" \
+        "websockets>=16.0" \
+        "zstandard>=0.25.0" \
+    && pip install --no-cache-dir --no-deps "maxapi-python==2.3.1"
+COPY src ./src
+
+ENV PORT=3002
+ENV PYMAX_SESSION_DIR=/data/pymax-session
+EXPOSE 3002
+
+CMD ["python", "-m", "src.server"]

pymax-worker/requirements.txt

@@ -0,0 +1,9 @@
+aiofiles>=25.1.0
+aiohttp>=3.9.0
+aiosqlite>=0.22.0
+msgpack>=1.1.0
+pydantic>=2.10.0
+python-socks[asyncio]>=2.8.0
+qrcode>=8.2
+websockets>=16.0
+zstandard>=0.25.0

pymax-worker/src/__init__.py

@@ -0,0 +1 @@
+

pymax-worker/src/server.py

@@ -0,0 +1,635 @@
+from __future__ import annotations
+
+import asyncio
+import base64
+import mimetypes
+import os
+import pathlib
+import stat
+import time
+import traceback
+from collections.abc import Awaitable, Callable
+from datetime import UTC, datetime
+from typing import Any
+
+import aiohttp
+from aiohttp import web
+from pymax import Client, File, Photo, Video
+
+
+PORT = int(os.environ.get("PORT", "3002"))
+PHONE_NUMBER = os.environ.get("PYMAX_PHONE_NUMBER") or os.environ.get("QMAX_MAX_PHONE_NUMBER") or ""
+SESSION_DIR = pathlib.Path(os.environ.get("PYMAX_SESSION_DIR", "/data/pymax-session"))
+SESSION_NAME = os.environ.get("PYMAX_SESSION_NAME", "session.db")
+CHAT_FETCH_LIMIT = int(os.environ.get("PYMAX_CHAT_FETCH_LIMIT", "350"))
+HISTORY_LIMIT = int(os.environ.get("PYMAX_HISTORY_LIMIT", "80"))
+SEND_ROOTS = [pathlib.Path(p) for p in os.environ.get("PYMAX_SEND_ROOTS", "/qmax-data:/tmp").split(":") if p]
+
+
+def utc_now() -> str:
+    return datetime.now(UTC).isoformat()
+
+
+def millis_to_iso(value: int | None) -> str:
+    if not value:
+        return utc_now()
+    return datetime.fromtimestamp(value / 1000, UTC).isoformat()
+
+
+def clean_id(value: Any) -> str:
+    if value is None:
+        return ""
+    return str(value)
+
+
+def normalize_phone(phone: str) -> str:
+    digits = "".join(ch for ch in phone if ch.isdigit())
+    if len(digits) == 11 and digits.startswith("8"):
+        return "+7" + digits[1:]
+    if len(digits) == 11 and digits.startswith("7"):
+        return "+" + digits
+    if len(digits) == 10:
+        return "+7" + digits
+    return phone.strip()
+
+
+def dump_model(obj: Any) -> dict[str, Any]:
+    if obj is None:
+        return {}
+    model_dump = getattr(obj, "model_dump", None)
+    if callable(model_dump):
+        try:
+            return sanitize_value(model_dump(mode="json", by_alias=False, exclude_none=True))
+        except UnicodeDecodeError:
+            return sanitize_value(model_dump(mode="python", by_alias=False, exclude_none=True))
+    if isinstance(obj, dict):
+        return sanitize_value(obj)
+    return sanitize_value({name: getattr(obj, name) for name in dir(obj) if not name.startswith("_")})
+
+
+def sanitize_value(value: Any) -> Any:
+    if isinstance(value, bytes):
+        return base64.b64encode(value).decode("ascii")
+    if isinstance(value, dict):
+        return {str(k): sanitize_value(v) for k, v in value.items()}
+    if isinstance(value, list):
+        return [sanitize_value(item) for item in value]
+    if isinstance(value, tuple):
+        return [sanitize_value(item) for item in value]
+    return value
+
+
+def value_name(value: Any) -> str:
+    raw = getattr(value, "value", value)
+    return str(raw) if raw is not None else ""
+
+
+class DeferredCodeProvider:
+    def __init__(self) -> None:
+        self._future: asyncio.Future[str] | None = None
+        self.waiting_for_code = False
+        self.requested_phone: str | None = None
+
+    async def get_code(self, phone: str) -> str:
+        self.requested_phone = phone
+        self.waiting_for_code = True
+        loop = asyncio.get_running_loop()
+        self._future = loop.create_future()
+        try:
+            return await self._future
+        finally:
+            self.waiting_for_code = False
+            self._future = None
+
+    def submit(self, code: str) -> bool:
+        if self._future is None or self._future.done():
+            return False
+        self._future.set_result(code)
+        return True
+
+
+class DeferredPasswordProvider:
+    def __init__(self) -> None:
+        self._future: asyncio.Future[str] | None = None
+        self.waiting_for_password = False
+        self.hint: str | None = None
+
+    async def get_password(self, hint: str | None = None) -> str:
+        self.hint = hint
+        self.waiting_for_password = True
+        loop = asyncio.get_running_loop()
+        self._future = loop.create_future()
+        try:
+            return await self._future
+        finally:
+            self.waiting_for_password = False
+            self._future = None
+
+    def submit(self, password: str) -> bool:
+        if self._future is None or self._future.done():
+            return False
+        self._future.set_result(password)
+        return True
+
+
+class PyMaxRuntime:
+    def __init__(self) -> None:
+        self.client: Client | None = None
+        self.task: asyncio.Task[None] | None = None
+        self.ready = asyncio.Event()
+        self.lock = asyncio.Lock()
+        self.code_provider = DeferredCodeProvider()
+        self.password_provider = DeferredPasswordProvider()
+        self.last_error: str | None = None
+        self.last_started_at: str | None = None
+        self.last_title: str | None = None
+
+    @property
+    def is_authorized(self) -> bool:
+        return self.ready.is_set() and self.client is not None
+
+    def login_stage(self) -> str:
+        if self.is_authorized:
+            return "Authorized"
+        if self.password_provider.waiting_for_password:
+            return "Password"
+        if self.code_provider.waiting_for_code:
+            return "Code"
+        if self.task is not None and not self.task.done():
+            return "Starting"
+        return "Idle"
+
+    async def ensure_started(self, phone: str | None = None) -> None:
+        async with self.lock:
+            if self.task is not None and not self.task.done():
+                return
+            self.ready.clear()
+            self.last_error = None
+            use_phone = normalize_phone(phone or PHONE_NUMBER)
+            if not use_phone:
+                self.last_error = "PYMAX_PHONE_NUMBER is not configured."
+                return
+            SESSION_DIR.mkdir(parents=True, exist_ok=True)
+            SESSION_DIR.chmod(0o700)
+            client = Client(
+                phone=use_phone,
+                work_dir=str(SESSION_DIR),
+                session_name=SESSION_NAME,
+                sms_code_provider=self.code_provider,
+                password_provider=self.password_provider,
+            )
+
+            @client.on_start()
+            async def on_start(c: Client) -> None:
+                self.client = c
+                me = getattr(c, "me", None)
+                self.last_title = str(getattr(me, "first_name", "") or getattr(me, "name", "") or "PyMax")
+                self.last_started_at = utc_now()
+                self.ready.set()
+
+            self.client = client
+            self.task = asyncio.create_task(self._run_client(client))
+
+    async def _run_client(self, client: Client) -> None:
+        try:
+            await client.start()
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:  # noqa: BLE001
+            self.last_error = f"{type(exc).__name__}: {exc}"
+            traceback.print_exc()
+        finally:
+            self.ready.clear()
+
+    async def wait_ready(self, seconds: float = 30.0) -> bool:
+        if self.ready.is_set():
+            return True
+        try:
+            await asyncio.wait_for(self.ready.wait(), timeout=seconds)
+            return True
+        except TimeoutError:
+            return False
+
+    async def get_client(self) -> Client:
+        await self.ensure_started()
+        if not await self.wait_ready(15):
+            raise RuntimeError(self.last_error or f"PyMax is not ready; stage={self.login_stage()}")
+        assert self.client is not None
+        return self.client
+
+    async def stop(self) -> None:
+        if self.client is not None:
+            await self.client.stop()
+        if self.task is not None:
+            self.task.cancel()
+
+
+runtime = PyMaxRuntime()
+
+
+def json_response(data: Any, status: int = 200) -> web.Response:
+    return web.json_response(data, status=status)
+
+
+def status_payload() -> dict[str, Any]:
+    return {
+        "mode": "PyMax",
+        "isAuthorized": runtime.is_authorized,
+        "loginStage": runtime.login_stage(),
+        "status": "Ready" if runtime.is_authorized else runtime.login_stage(),
+        "url": None,
+        "title": runtime.last_title,
+        "lastError": runtime.last_error,
+        "updatedAt": utc_now(),
+    }
+
+
+def attachment_preview(attaches: list[Any]) -> str | None:
+    if not attaches:
+        return None
+    kinds = [normalize_attachment_kind(att) for att in attaches]
+    if any(kind == "photo" for kind in kinds):
+        return "Фото"
+    if any(kind == "video" for kind in kinds):
+        return "Видео"
+    if any(kind == "audio" for kind in kinds):
+        return "Голосовое сообщение"
+    if any(kind == "sticker" for kind in kinds):
+        return "Стикер"
+    if any(kind == "contact" for kind in kinds):
+        return "Контакт"
+    return "Файл"
+
+
+def normalize_chat_kind(chat: Any) -> str:
+    chat_type = value_name(getattr(chat, "type", "")).upper()
+    if "CHANNEL" in chat_type:
+        return "Channel"
+    return "MaxDialog"
+
+
+def normalize_attachment_kind(att: Any) -> str:
+    data = dump_model(att)
+    raw_type = value_name(data.get("type") or data.get("_type") or getattr(att, "type", "")).lower()
+    class_name = type(att).__name__.lower()
+    merged = f"{raw_type} {class_name}"
+    if "photo" in merged:
+        return "photo"
+    if "video" in merged:
+        return "video"
+    if "audio" in merged or "voice" in merged:
+        return "audio"
+    if "sticker" in merged:
+        return "sticker"
+    if "contact" in merged:
+        return "contact"
+    if "file" in merged:
+        return "file"
+    return raw_type or class_name or "attachment"
+
+
+async def normalize_attachment(client: Client, message: Any, att: Any, sort_order: int) -> dict[str, Any]:
+    data = dump_model(att)
+    kind = normalize_attachment_kind(att)
+    external_id = (
+        data.get("photo_id")
+        or data.get("video_id")
+        or data.get("file_id")
+        or data.get("id")
+        or data.get("token")
+        or f"{clean_id(getattr(message, 'id', None))}:{sort_order}"
+    )
+    file_name = (
+        data.get("name")
+        or data.get("filename")
+        or data.get("file_name")
+        or f"{kind}-{external_id}"
+    )
+    remote_url = data.get("base_url") or data.get("url") or data.get("thumbnail")
+    content_type = None
+    file_size = data.get("size") or data.get("file_size") or data.get("fileSize")
+
+    chat_id = getattr(message, "chat_id", None)
+    message_id = getattr(message, "id", None)
+    try:
+        if kind == "file" and chat_id is not None and message_id is not None and data.get("file_id"):
+            req = await client.get_file_by_id(int(chat_id), int(message_id), int(data["file_id"]))
+            req_data = dump_model(req)
+            remote_url = req_data.get("url") or remote_url
+        elif kind == "video" and chat_id is not None and message_id is not None and data.get("video_id"):
+            req = await client.get_video_by_id(int(chat_id), int(message_id), int(data["video_id"]))
+            req_data = dump_model(req)
+            remote_url = req_data.get("url") or remote_url
+    except Exception:  # noqa: BLE001
+        pass
+
+    if kind == "photo":
+        content_type = "image/jpeg"
+    elif kind == "video":
+        content_type = "video/mp4"
+    elif kind == "audio":
+        content_type = "audio/ogg"
+    elif file_name:
+        content_type = mimetypes.guess_type(str(file_name))[0]
+
+    text_content = None
+    if kind in {"sticker", "contact"}:
+        text_content = str(data)
+
+    return {
+        "externalId": clean_id(external_id),
+        "fileName": str(file_name),
+        "contentType": content_type,
+        "fileSizeBytes": file_size,
+        "remoteUrl": remote_url,
+        "kind": kind,
+        "sortOrder": sort_order,
+        "textContent": text_content,
+    }
+
+
+async def normalize_message(client: Client, chat: Any, message: Any) -> dict[str, Any]:
+    attaches = list(getattr(message, "attaches", None) or [])
+    attachments = [
+        await normalize_attachment(client, message, att, index)
+        for index, att in enumerate(attaches)
+    ]
+    sender = getattr(message, "sender", None)
+    me_id = None
+    me = getattr(client, "me", None)
+    contact = getattr(me, "contact", None)
+    if contact is not None:
+        me_id = getattr(contact, "id", None)
+    if me_id is None and me is not None:
+        me_id = getattr(me, "id", None)
+    is_outgoing = sender is not None and me_id is not None and str(sender) == str(me_id)
+    text = getattr(message, "text", None) or attachment_preview(attaches)
+    status = value_name(getattr(message, "status", None)).lower() or None
+    return {
+        "externalId": clean_id(getattr(message, "id", None)),
+        "senderExternalId": clean_id(sender) or None,
+        "senderName": None,
+        "isOutgoing": is_outgoing,
+        "text": text,
+        "sentAt": millis_to_iso(getattr(message, "time", None)),
+        "attachments": attachments,
+        "deliveryState": status,
+    }
+
+
+async def normalize_chat_update(client: Client, chat: Any, include_history: bool = False) -> dict[str, Any]:
+    last_message = getattr(chat, "last_message", None)
+    messages: list[dict[str, Any]] = []
+    if include_history:
+        history = await client.fetch_history(chat_id=int(chat.id), backward=HISTORY_LIMIT)
+        messages = [await normalize_message(client, chat, m) for m in (history or [])]
+    elif last_message is not None:
+        messages = [await normalize_message(client, chat, last_message)]
+
+    last_preview = None
+    last_at = getattr(chat, "last_event_time", None)
+    last_is_outgoing = None
+    if messages:
+        last = messages[-1]
+        last_preview = last.get("text")
+        last_at = int(datetime.fromisoformat(last["sentAt"]).timestamp() * 1000)
+        last_is_outgoing = last.get("isOutgoing")
+
+    title = getattr(chat, "title", None) or f"MAX {chat.id}"
+    return {
+        "externalId": clean_id(getattr(chat, "id", None)),
+        "title": title,
+        "avatarUrl": getattr(chat, "base_icon_url", None) or getattr(chat, "base_raw_icon_url", None),
+        "updatedAt": millis_to_iso(last_at),
+        "messages": messages,
+        "lastMessagePreview": last_preview,
+        "lastMessageIsOutgoing": last_is_outgoing,
+        "lastMessageAt": millis_to_iso(last_at) if last_at else None,
+        "lastMessageTimeText": None,
+        "chatUrl": f"pymax://chat/{chat.id}",
+        "kind": normalize_chat_kind(chat),
+    }
+
+
+async def fetch_chat_pages(client: Client, limit: int) -> list[Any]:
+    chats: list[Any] = []
+    marker: int | None = None
+    while len(chats) < limit:
+        page = await client.fetch_chats(marker=marker)
+        page = page or []
+        if not page:
+            break
+        chats.extend(page)
+        times = [getattr(chat, "last_event_time", None) for chat in page if getattr(chat, "last_event_time", None)]
+        if not times:
+            break
+        marker = min(times) - 1
+        if len(page) < 50:
+            break
+    return chats[:limit]
+
+
+def parse_chat_id(value: Any) -> int:
+    text = clean_id(value).strip()
+    if text.startswith("pymax://chat/"):
+        text = text.rsplit("/", 1)[-1]
+    return int(text)
+
+
+def validate_send_path(raw: str) -> pathlib.Path:
+    path = pathlib.Path(raw)
+    resolved = path.resolve(strict=True)
+    if path.is_symlink() or stat.S_ISLNK(path.lstat().st_mode):
+        raise PermissionError("Symlink upload paths are not allowed.")
+    if not resolved.is_file():
+        raise ValueError("Upload path is not a regular file.")
+    roots = [root.resolve() for root in SEND_ROOTS if root.exists()]
+    if not roots:
+        raise PermissionError("No PYMAX_SEND_ROOTS paths exist.")
+    if not any(resolved.is_relative_to(root) for root in roots):
+        raise PermissionError("Upload path is outside PYMAX_SEND_ROOTS.")
+    return resolved
+
+
+def send_attachment_for(path: pathlib.Path) -> Any:
+    ext = path.suffix.lower()
+    if ext in {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"}:
+        return Photo(path=str(path))
+    if ext in {".mp4", ".mov", ".webm", ".mkv"}:
+        return Video(path=str(path))
+    return File(path=str(path))
+
+
+async def read_json(request: web.Request) -> dict[str, Any]:
+    if request.can_read_body:
+        return await request.json()
+    return {}
+
+
+def route_errors(handler: Callable[[web.Request], Awaitable[web.StreamResponse]]) -> Callable[[web.Request], Awaitable[web.StreamResponse]]:
+    async def wrapped(request: web.Request) -> web.StreamResponse:
+        try:
+            return await handler(request)
+        except Exception as exc:  # noqa: BLE001
+            runtime.last_error = f"{type(exc).__name__}: {exc}"
+            traceback.print_exc()
+            return json_response({"success": False, "error": runtime.last_error}, status=500)
+
+    return wrapped
+
+
+@route_errors
+async def health(_request: web.Request) -> web.Response:
+    return json_response({"ok": True, "mode": "PyMax", "updatedAt": utc_now()})
+
+
+@route_errors
+async def status(_request: web.Request) -> web.Response:
+    await runtime.ensure_started()
+    return json_response(status_payload())
+
+
+@route_errors
+async def login_start(request: web.Request) -> web.Response:
+    data = await read_json(request)
+    phone = str(data.get("phoneNumber") or PHONE_NUMBER).strip()
+    await runtime.ensure_started(phone)
+    return json_response(status_payload())
+
+
+@route_errors
+async def login_code(request: web.Request) -> web.Response:
+    data = await read_json(request)
+    code = str(data.get("code") or "").strip()
+    password = str(data.get("password") or "").strip()
+    if password and runtime.password_provider.submit(password):
+        pass
+    elif code and runtime.code_provider.submit(code):
+        pass
+    else:
+        return json_response({
+            **status_payload(),
+            "lastError": "PyMax is not waiting for a code or password.",
+        })
+    wait_ms = int(data.get("waitMs") or 0)
+    if wait_ms > 0:
+        await runtime.wait_ready(wait_ms / 1000)
+    return json_response(status_payload())
+
+
+@route_errors
+async def snapshot(_request: web.Request) -> web.Response:
+    return json_response({
+        "url": None,
+        "title": runtime.last_title or "PyMax",
+        "bodyText": str(status_payload()),
+        "screenshotPngBase64": "",
+        "capturedAt": utc_now(),
+    })
+
+
+@route_errors
+async def updates(_request: web.Request) -> web.Response:
+    client = await runtime.get_client()
+    chats = await fetch_chat_pages(client, CHAT_FETCH_LIMIT)
+    result = [await normalize_chat_update(client, chat, include_history=False) for chat in chats]
+    return json_response(result)
+
+
+@route_errors
+async def chat_history(request: web.Request) -> web.Response:
+    data = await read_json(request)
+    client = await runtime.get_client()
+    chat_id = parse_chat_id(data.get("externalChatId") or data.get("chatUrl"))
+    chat = await client.get_chat(chat_id)
+    result = await normalize_chat_update(client, chat, include_history=True)
+    return json_response(result)
+
+
+@route_errors
+async def resolve_chat_url(request: web.Request) -> web.Response:
+    data = await read_json(request)
+    chat_id = parse_chat_id(data.get("externalChatId"))
+    return json_response({"success": True, "chatUrl": f"pymax://chat/{chat_id}", "error": None})
+
+
+@route_errors
+async def chat_presence(_request: web.Request) -> web.Response:
+    return json_response({"isTyping": False, "statusText": None, "updatedAt": utc_now()})
+
+
+@route_errors
+async def disabled_action(_request: web.Request) -> web.Response:
+    return json_response({"success": False, "error": "Not implemented in PyMax POC worker."})
+
+
+@route_errors
+async def send_text(request: web.Request) -> web.Response:
+    data = await read_json(request)
+    client = await runtime.get_client()
+    chat_id = parse_chat_id(data.get("externalChatId") or data.get("chatUrl"))
+    text = str(data.get("text") or "")
+    sent = await client.send_message(chat_id=chat_id, text=text)
+    external_id = clean_id(getattr(sent, "id", None))
+    return json_response({"success": bool(external_id), "externalMessageId": external_id or None, "error": None if external_id else "PyMax returned no message id.", "chatUrl": f"pymax://chat/{chat_id}"})
+
+
+@route_errors
+async def send_attachment(request: web.Request) -> web.Response:
+    data = await read_json(request)
+    client = await runtime.get_client()
+    chat_id = parse_chat_id(data.get("externalChatId") or data.get("chatUrl"))
+    path = validate_send_path(str(data.get("path") or ""))
+    caption = str(data.get("caption") or "")
+    sent = await client.send_message(chat_id=chat_id, text=caption, attachments=[send_attachment_for(path)])
+    external_id = clean_id(getattr(sent, "id", None))
+    return json_response({"success": bool(external_id), "externalMessageId": external_id or None, "error": None if external_id else "PyMax returned no message id.", "chatUrl": f"pymax://chat/{chat_id}"})
+
+
+@route_errors
+async def media_fetch(request: web.Request) -> web.StreamResponse:
+    remote_url = request.query.get("url")
+    if not remote_url:
+        return json_response({"error": "url is required"}, status=400)
+    async with aiohttp.ClientSession() as session:
+        async with session.get(remote_url) as response:
+            if response.status >= 400:
+                return json_response({"error": f"remote returned {response.status}"}, status=502)
+            body = await response.read()
+            return web.Response(
+                body=body,
+                content_type=response.headers.get("Content-Type", "application/octet-stream"),
+            )
+
+
+def create_app() -> web.Application:
+    app = web.Application(client_max_size=1024 * 1024)
+    app.router.add_get("/health", health)
+    app.router.add_get("/status", status)
+    app.router.add_post("/login/start", login_start)
+    app.router.add_post("/login/code", login_code)
+    app.router.add_get("/snapshot", snapshot)
+    app.router.add_get("/updates", updates)
+    app.router.add_post("/chat/history", chat_history)
+    app.router.add_post("/chat/resolve-url", resolve_chat_url)
+    app.router.add_post("/chat/presence", chat_presence)
+    app.router.add_post("/chat/clear-history", disabled_action)
+    app.router.add_post("/chat/delete", disabled_action)
+    app.router.add_post("/send/text", send_text)
+    app.router.add_post("/send/attachment", send_attachment)
+    app.router.add_post("/message/edit", disabled_action)
+    app.router.add_post("/message/delete", disabled_action)
+    app.router.add_post("/message/reaction", disabled_action)
+    app.router.add_delete("/message/reaction", disabled_action)
+    app.router.add_get("/media/fetch", media_fetch)
+    return app
+
+
+async def on_cleanup(_app: web.Application) -> None:
+    await runtime.stop()
+
+
+if __name__ == "__main__":
+    app = create_app()
+    app.on_cleanup.append(on_cleanup)
+    web.run_app(app, host="0.0.0.0", port=PORT)