diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index efc763d..53c38e5 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -49,9 +49,29 @@ services: networks: - qmax + qmax-pymax-worker: + build: + context: ../pymax-worker + dockerfile: Dockerfile + container_name: qmax-pymax-worker + restart: unless-stopped + environment: + PORT: 3002 + PYMAX_PHONE_NUMBER: ${QMAX_MAX_PHONE_NUMBER} + PYMAX_SESSION_DIR: /data/pymax-session + PYMAX_SEND_ROOTS: /qmax-data:/tmp + PYMAX_CHAT_FETCH_LIMIT: ${QMAX_PYMAX_CHAT_FETCH_LIMIT:-350} + PYMAX_HISTORY_LIMIT: ${QMAX_PYMAX_HISTORY_LIMIT:-80} + volumes: + - qmax-pymax-session:/data + - qmax-data:/qmax-data + networks: + - qmax + volumes: qmax-data: qmax-max-profile: + qmax-pymax-session: networks: qmax: diff --git a/pymax-worker/Dockerfile b/pymax-worker/Dockerfile new file mode 100644 index 0000000..e145969 --- /dev/null +++ b/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"] diff --git a/pymax-worker/requirements.txt b/pymax-worker/requirements.txt new file mode 100644 index 0000000..cdb65ee --- /dev/null +++ b/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 diff --git a/pymax-worker/src/__init__.py b/pymax-worker/src/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/pymax-worker/src/__init__.py @@ -0,0 +1 @@ + diff --git a/pymax-worker/src/server.py b/pymax-worker/src/server.py new file mode 100644 index 0000000..0bc1218 --- /dev/null +++ b/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)