Handle incoming MAX call events

pymax-worker/src/message_labels.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+INCOMING_CALL_TEXT = "\u0412\u0445\u043e\u0434\u044f\u0449\u0438\u0439 \u0437\u0432\u043e\u043d\u043e\u043a"
+
+
+def is_call_media_label(value: Any) -> bool:
+    if value is None:
+        return False
+    text = str(value).strip().lower()
+    return text == "call" or text.startswith("call-")

pymax-worker/src/server.py

@@ -22,6 +22,8 @@ from aiohttp import web
 from pymax import Client, ExtraConfig, File, Photo, SyncOverrides, Video
 from pymax.types import ContactInfo
 
+from .message_labels import INCOMING_CALL_TEXT, is_call_media_label
+
 
 PORT = int(os.environ.get("PORT", "3002"))
 PHONE_NUMBER = os.environ.get("PYMAX_PHONE_NUMBER") or os.environ.get("QMAX_MAX_PHONE_NUMBER") or ""
@@ -581,6 +583,8 @@ def normalize_attachment_kind(att: Any) -> str:
     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 is_call_media_label(raw_type) or "callattachment" in class_name:
+        return "call"
     if "photo" in merged:
         return "photo"
     if "video" in merged:
@@ -600,6 +604,8 @@ def attachment_preview(attaches: list[Any]) -> str | None:
     if not attaches:
         return None
     kinds = [normalize_attachment_kind(att) for att in attaches]
+    if any(is_call_media_label(kind) for kind in kinds):
+        return INCOMING_CALL_TEXT
     if any(kind == "photo" for kind in kinds):
         return "\u0424\u043e\u0442\u043e"
     if any(kind == "video" for kind in kinds):
@@ -731,16 +737,20 @@ async def normalize_message(
     me_id: int | None = None,
 ) -> dict[str, Any]:
     attaches = list(getattr(message, "attaches", None) or [])
+    call_attaches = [att for att in attaches if normalize_attachment_kind(att) == "call"]
+    downloadable_attaches = [att for att in attaches if normalize_attachment_kind(att) != "call"]
     attachments = [
         await normalize_attachment(client, message, att, index)
-        for index, att in enumerate(attaches)
+        for index, att in enumerate(downloadable_attaches)
     ]
     sender = getattr(message, "sender", None)
     sender_id = coerce_int(sender)
     if me_id is None:
         me_id = get_me_user_id(client)
     is_outgoing = sender_id is not None and me_id is not None and sender_id == me_id
-    text = getattr(message, "text", None) or attachment_preview(attaches)
+    raw_text = getattr(message, "text", None)
+    is_call_text = is_call_media_label(raw_text)
+    text = INCOMING_CALL_TEXT if call_attaches or is_call_text else raw_text or attachment_preview(attaches)
     status = value_name(getattr(message, "status", None)).lower() or None
     user = (user_map or {}).get(str(sender_id)) if sender_id is not None else None
     sender_name = "You" if is_outgoing else user_display_name(user)

pymax-worker/tests/test_message_labels.py

@@ -0,0 +1,22 @@
+import unittest
+
+from src.message_labels import INCOMING_CALL_TEXT, is_call_media_label
+
+
+class MessageLabelsTests(unittest.TestCase):
+    def test_call_media_labels_are_detected(self) -> None:
+        for value in ("call", "call-start", "CALL-MISSED", " call-ended "):
+            with self.subTest(value=value):
+                self.assertTrue(is_call_media_label(value))
+
+    def test_unrelated_values_are_not_detected(self) -> None:
+        for value in (None, "", "callback", "video-call", "phone"):
+            with self.subTest(value=value):
+                self.assertFalse(is_call_media_label(value))
+
+    def test_incoming_call_text_is_stable(self) -> None:
+        self.assertEqual("\u0412\u0445\u043e\u0434\u044f\u0449\u0438\u0439 \u0437\u0432\u043e\u043d\u043e\u043a", INCOMING_CALL_TEXT)
+
+
+if __name__ == "__main__":
+    unittest.main()