sevenhill/QMAX
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Show senders in MAX group chats

Browse Source
This commit is contained in:
Курнат Андрей
2026-08-08 14:43:14 +03:00
parent 43a857afb5
commit 5845584933
5 changed files with 171 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pymax-worker/src/chat_identity.py
+31
View File
@@ -0,0 +1,31 @@
from __future__ import annotations
from typing import Any, Mapping
def normalize_chat_kind_value(value: Any) -> str:
raw = getattr(value, "value", value)
text = str(raw or "").strip().upper()
if text.startswith("CHATTYPE."):
text = text.removeprefix("CHATTYPE.")
if text == "CHANNEL":
return "Channel"
if text in {"CHAT", "GROUP"}:
return "Group"
return "MaxDialog"
def resolve_sender_name(
sender_id: int | None,
*,
is_outgoing: bool,
phone_contact_names: Mapping[str, str] | None,
profile_name: str | None,
) -> str | None:
if is_outgoing:
return "You"
if sender_id is not None:
phone_name = (phone_contact_names or {}).get(str(sender_id))
if phone_name and phone_name.strip():
return phone_name.strip()
return profile_name.strip() if profile_name and profile_name.strip() else None
pymax-worker/src/server.py
+53 -10
View File
@@ -22,6 +22,7 @@ from aiohttp import web
from pymax import Client, ExtraConfig, File, Photo, SyncOverrides, Video
from pymax.types import ContactInfo
from .chat_identity import normalize_chat_kind_value, resolve_sender_name
from .message_labels import INCOMING_CALL_TEXT, is_call_media_label
@@ -572,10 +573,7 @@ def attachment_preview(attaches: list[Any]) -> str | None:
def normalize_chat_kind(chat: Any) -> str:
chat_type = value_name(getattr(chat, "type", "")).upper()
if "CHANNEL" in chat_type:
return "Channel"
return "MaxDialog"
return normalize_chat_kind_value(getattr(chat, "type", ""))
def normalize_attachment_kind(att: Any) -> str:
@@ -735,6 +733,7 @@ async def normalize_message(
message: Any,
user_map: dict[str, Any] | None = None,
me_id: int | None = None,
phone_contact_names: dict[str, str] | 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"]
@@ -753,7 +752,12 @@ async def normalize_message(
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)
sender_name = resolve_sender_name(
sender_id,
is_outgoing=is_outgoing,
phone_contact_names=phone_contact_names,
profile_name=user_display_name(user),
)
return {
"externalId": clean_id(getattr(message, "id", None)),
"senderExternalId": clean_id(sender) or None,
@@ -772,6 +776,7 @@ async def normalize_chat_update(
include_history: bool = False,
user_map: dict[str, Any] | None = None,
me_id: int | None = None,
phone_contact_names: dict[str, str] | None = None,
) -> dict[str, Any]:
last_message = getattr(chat, "last_message", None)
raw_messages: list[Any] = []
@@ -786,7 +791,12 @@ async def normalize_chat_update(
me_id = get_me_user_id(client)
if user_map is None:
user_map = await build_user_map(client, [chat], raw_messages)
messages = [await normalize_message(client, chat, m, user_map, me_id) for m in raw_messages]
if phone_contact_names is None:
phone_contact_names = load_phone_contact_names()
messages = [
await normalize_message(client, chat, m, user_map, me_id, phone_contact_names)
for m in raw_messages
]
last_preview = None
last_at = getattr(chat, "last_event_time", None)
@@ -1038,6 +1048,30 @@ def save_phone_contact_ids(contact_ids: set[int]) -> None:
part_path.replace(path)
def load_phone_contact_names() -> dict[str, str]:
path = runtime.session_dir / "phone-contact-names.json"
try:
values = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError, TypeError):
return {}
if not isinstance(values, dict):
return {}
return {
str(contact_id): str(name).strip()
for contact_id, name in values.items()
if coerce_int(contact_id) is not None and str(name).strip()
}
def save_phone_contact_names(contact_names: dict[int, str]) -> None:
runtime.session_dir.mkdir(parents=True, exist_ok=True)
path = runtime.session_dir / "phone-contact-names.json"
part_path = path.with_suffix(".json.part")
values = {str(contact_id): name.strip() for contact_id, name in contact_names.items() if name.strip()}
part_path.write_text(json.dumps(values, ensure_ascii=False, sort_keys=True), encoding="utf-8")
part_path.replace(path)
def append_saved_contacts(client: Client, contacts: list[Any]) -> None:
saved = client.contacts
saved_ids = {
@@ -1062,7 +1096,7 @@ async def contacts_import(request: web.Request) -> web.Response:
return json_response({"success": False, "error": "contacts limit is 5000"}, status=400)
contacts_to_import: list[ContactInfo] = []
requested_phone_keys: set[str] = set()
requested_names_by_phone: dict[str, str] = {}
for item in raw_contacts:
if not isinstance(item, dict):
continue
@@ -1070,8 +1104,8 @@ async def contacts_import(request: web.Request) -> web.Response:
first_name = str(item.get("firstName") or item.get("name") or phone).strip()
last_name = str(item.get("lastName") or "").strip() or None
key = phone_key(phone)
if phone and key and key not in requested_phone_keys:
requested_phone_keys.add(key)
if phone and key and key not in requested_names_by_phone:
requested_names_by_phone[key] = " ".join(part for part in (first_name, last_name) if part).strip()
contacts_to_import.append(ContactInfo(phone=phone, first_name=first_name or phone, last_name=last_name))
if not contacts_to_import:
return json_response({"success": False, "error": "contacts contain no phone numbers"}, status=400)
@@ -1094,7 +1128,7 @@ async def contacts_import(request: web.Request) -> web.Response:
if (
contact_id is not None
and contact_id > 0
and phone_key(contact_data.get("phone") or getattr(contact, "phone", None)) in requested_phone_keys
and phone_key(contact_data.get("phone") or getattr(contact, "phone", None)) in requested_names_by_phone
):
matched_by_id[contact_id] = contact
@@ -1106,6 +1140,15 @@ async def contacts_import(request: web.Request) -> web.Response:
and contact_id > 0
}
save_phone_contact_ids(imported_ids)
phone_names_by_id = {
contact_id: requested_names_by_phone[contact_phone_key]
for contact in matched
if (contact_data := dump_model(contact))
if (contact_id := coerce_int(contact_data.get("id") or getattr(contact, "id", None))) is not None
if (contact_phone_key := phone_key(contact_data.get("phone") or getattr(contact, "phone", None)))
in requested_names_by_phone
}
save_phone_contact_names(phone_names_by_id)
me_id = get_me_user_id(client)
result = [
normalized
pymax-worker/tests/test_chat_identity.py
+62
View File
@@ -0,0 +1,62 @@
import unittest
from enum import Enum
from src.chat_identity import normalize_chat_kind_value, resolve_sender_name
class SampleChatType(str, Enum):
DIALOG = "DIALOG"
CHAT = "CHAT"
CHANNEL = "CHANNEL"
class ChatIdentityTests(unittest.TestCase):
def test_chat_kind_is_normalized(self) -> None:
cases = {
SampleChatType.DIALOG: "MaxDialog",
SampleChatType.CHAT: "Group",
SampleChatType.CHANNEL: "Channel",
"ChatType.CHAT": "Group",
"GROUP": "Group",
None: "MaxDialog",
}
for value, expected in cases.items():
with self.subTest(value=value):
self.assertEqual(expected, normalize_chat_kind_value(value))
def test_phone_book_name_has_priority(self) -> None:
self.assertEqual(
"Мама",
resolve_sender_name(
42,
is_outgoing=False,
phone_contact_names={"42": " Мама "},
profile_name="MAX Profile",
),
)
def test_profile_name_is_used_when_phone_name_is_missing(self) -> None:
self.assertEqual(
"MAX Profile",
resolve_sender_name(
42,
is_outgoing=False,
phone_contact_names={},
profile_name="MAX Profile",
),
)
def test_outgoing_message_uses_you(self) -> None:
self.assertEqual(
"You",
resolve_sender_name(
42,
is_outgoing=True,
phone_contact_names={"42": "Мама"},
profile_name="MAX Profile",
),
)
if __name__ == "__main__":
unittest.main()