"""Kênh Zalo CÁ NHÂN (tài khoản thật, API không chính thức qua zca-js) - port độc lập từ thansa-os `server/zalo_personal_channel.py` + `server/channels/zalo_personal.py`. ĐỌC : vòng lặp gọi tool MCP `zalo_get_messages` theo con trỏ `since` mỗi `poll_interval` giây. GỬI : tool MCP `zalo_send_message` (chỉ chữ). Có tag người trong nhóm thì đi CLI `msg send --mention`. Bạn cấp `on_event(ev)` (sync hoặc async) để nhận từng tin đã chuẩn hoá (khuôn ở `normalize`). Mọi chính sách bot (trả lời ai, nhóm nào được phép, hạn mức) là việc của nơi đấu nối - module này chỉ cung cấp các tín hiệu cần để không nhắn bậy dưới tên chủ: `is_echo`, `owner_typed_recently`, `self_identity`, `mentioned`. """ from __future__ import annotations import asyncio import re import sys import time import unicodedata from typing import Any, Awaitable, Callable, Dict, Optional, Union from .cli import ZaloCli, is_timeout from .mcp_stdio import StdioMcp POLL_INTERVAL = 20 # giây giữa hai lần đọc ERROR_BACKOFF = 90 # nghỉ sau khi một lượt đọc lỗi (MCP chết, chưa đăng nhập) LIMIT = 100 # MCP cho phép 1..100 MAX_PAGES = 5 # đọc tối đa từng này lô liên tiếp khi hasMore THREADS_TTL = 300 # giây giữ bảng threadId -> {name, type} THREADS_FORCE_GAP = 15 # tối thiểu giữa hai lần làm mới cưỡng bức (gặp cuộc chat lạ) ECHO_TTL = 900 ECHO_MIN_LEN = 25 # tin "khách" phải dài từng này mới bị coi là tiếng vọng của câu bot vừa gửi OWNER_QUIET = 600 # giây coi là "chủ đang tự nhắn cuộc chat này" TAG_TIMEOUT = 40 EventCb = Callable[[dict], Union[None, Awaitable[None]]] def _get(d: dict, *keys, default=None): for k in keys: v = d.get(k) if v not in (None, ""): return v return default def _ts(v: Any) -> float: try: f = float(v or 0) except (TypeError, ValueError): return time.time() if f <= 0: return time.time() return f / 1000.0 if f > 1e12 else f # Zalo trả mili-giây def _int_cursor(v: Any) -> Optional[int]: if isinstance(v, bool): return None if isinstance(v, (int, float)) and v >= 0: return int(v) if isinstance(v, str) and v.strip().isdigit(): return int(v.strip()) return None def _norm_name(s: Any) -> str: s = unicodedata.normalize("NFD", str(s or "")).replace("đ", "d").replace("Đ", "D") s = "".join(c for c in s if unicodedata.category(c) != "Mn") return " ".join(s.lower().split()) def _utf16_len(s: str) -> int: """Zalo đo vị trí mention theo UTF-16 (emoji = 2), khác len() của Python.""" return len(s.encode("utf-16-le")) // 2 def tagged_text(text: str, name: str) -> tuple: tok = "@" + name m = re.search(re.escape(tok) + r"(?!\w)", text, re.IGNORECASE) if m: return text, _utf16_len(text[:m.start()]), _utf16_len(text[m.start():m.end()]) return tok + " " + text, 0, _utf16_len(tok) def message_type(msg: dict) -> str: t = str(_get(msg, "type", "msgType", "message_type", default="text") or "text").lower() for key, out in (("image", "image"), ("photo", "image"), ("voice", "audio"), ("audio", "audio"), ("video", "video"), ("sticker", "sticker"), ("file", "file"), ("doc", "file")): if key in t: return out return "text" if t in ("text", "chat", "webchat") else "other" class ZaloPersonal: """Một tài khoản Zalo cá nhân đã đăng nhập (thư mục phiên = `home`).""" def __init__(self, account_id: str, home: str, cli: ZaloCli, label: str = "Zalo", on_event: Optional[EventCb] = None, poll_interval: int = POLL_INTERVAL): self.id, self.home, self.label = str(account_id), home, label self.cli = cli self.on_event = on_event self.poll_interval = poll_interval base = cli.base_command() if not base: raise RuntimeError("Máy chưa có Node.js 20+ (npx)") # MCP chạy bằng chính đầu lệnh của CLI (bản ghim đã cài, hoặc npx -y zalo-agent-cli@1.6.2). self.mcp = StdioMcp(base + ["mcp", "start"], env={"HOME": home, "USERPROFILE": home}) self._since = 0 self._since_epoch = 0 self._threads: Dict[str, dict] = {} self._threads_ts = 0.0 self._sent: Dict[str, list] = {} # thread -> [(ts, text đã chuẩn hoá)] self._owner: Dict[str, float] = {} # thread -> giờ chủ tự nhắn gần nhất self._self_ids: set = set() self.self_name = "" self.last_error = "" self.last_read = 0.0 self._task: Optional[asyncio.Task] = None # ---- danh tính chính nick này ------------------------------------------------------------- @property def self_uid(self) -> str: """Id Zalo của chính nick, học từ tin nó gửi. Chỉ tin khi MỌI tin của nick cho đúng một giá trị.""" return next(iter(self._self_ids)) if len(self._self_ids) == 1 else "" def _learn_self(self, msg: dict): uid = str(_get(msg, "from", "senderId", "sender_id", "uidFrom", default="") or "").strip() thread = str(_get(msg, "threadId", "thread_id", "chatId", default="") or "").strip() if uid and uid != thread: self._self_ids.add(uid) name = str(_get(msg, "senderName", "sender_name", "fromName", "dName", default="") or "").strip() if name: self.self_name = name # ---- bảng cuộc chat ------------------------------------------------------------------------ async def threads(self, force: bool = False) -> Dict[str, dict]: age = time.time() - self._threads_ts if self._threads and age < THREADS_TTL and not (force and age >= THREADS_FORCE_GAP): return self._threads try: d = await self.mcp.call_tool("zalo_list_threads", {"limit": 200}) for t in ((d.get("threads") if isinstance(d, dict) else d) or []): if isinstance(t, dict): tid = str(_get(t, "threadId", "id", default="") or "") if tid: # Khoá THẬT của MCP 1.6.2 là `threadType` ("group"|"dm"|"unknown"); tài liệu ghi `type`. self._threads[tid] = {"name": str(t.get("name") or "")[:120], "type": str(t.get("threadType") or t.get("type") or "")} except Exception as e: print(f"[zalo {self.label}] list_threads: {e}", file=sys.stderr) self._threads_ts = time.time() return self._threads def _is_group(self, msg: dict) -> Optional[bool]: t = _get(msg, "threadType", "thread_type", default=None) if t is not None: s = str(t).strip().lower() if s in ("1", "group", "true"): return True if s in ("0", "dm", "user", "private", "false"): return False if isinstance(msg.get("isGroup"), bool): return msg["isGroup"] th = self._threads.get(str(_get(msg, "threadId", "thread_id", "chatId", default=""))) if th: return str(th.get("type") or "").lower() == "group" return None # KHÔNG RÕ - đừng đoán là chat riêng # ---- chuẩn hoá ------------------------------------------------------------------------------ def normalize(self, msg: dict) -> Optional[dict]: thread = str(_get(msg, "threadId", "thread_id", "chatId", default="") or "").strip() if not thread: return None group = self._is_group(msg) th = self._threads.get(thread) or {} is_self = next((msg[k] for k in ("isSelf", "fromMe", "isOwn", "self", "is_self", "from_me") if isinstance(msg.get(k), bool)), False) sender_id = str(_get(msg, "from", "senderId", "sender_id", "uidFrom", default="") or "").strip() sender_name = str(_get(msg, "senderName", "sender_name", "fromName", "dName", default="") or "").strip() if not sender_name and group is False: sender_name = th.get("name") or "" text = _get(msg, "text", "content", "msg", "message", default="") if isinstance(text, dict): text = _get(text, "text", "title", "description", "href", default="") return { "channel": "zalo_personal", "account_id": self.id, "chat_id": thread, "chat_type": "group" if group else ("private" if group is False else "unknown"), "chat_title": th.get("name", "") if group else "", "from_self": bool(is_self) or (bool(self.self_uid) and sender_id == self.self_uid), "sender_id": sender_id or ("" if group else thread), "sender_name": sender_name, "message_type": message_type(msg), "text": str(text or ""), "message_id": str(_get(msg, "id", "msgId", "message_id", default="") or ""), "created_at": _ts(_get(msg, "ts", "timestamp", "time", default=0)), "mentions": msg.get("mentions"), "reply_to": msg.get("replyTo"), "raw": msg, } # ---- đọc ----------------------------------------------------------------------------------- async def read_once(self) -> int: """Một lượt đọc. Trả số tin đã đẩy cho `on_event`.""" since = self._since if (self._since and self._since_epoch == self.mcp.epoch) else 0 n = 0 for _ in range(MAX_PAGES): args: Dict[str, Any] = {"limit": LIMIT} if since: args["since"] = since # KHÔNG phải `cursor` - MCP bỏ qua khoá lạ mà không báo ep = self.mcp.epoch d = await self.mcp.call_tool("zalo_get_messages", args) if since and self.mcp.epoch != ep: # MCP vừa bị dựng lại -> số thứ tự đếm lại từ 1 since = 0 d = await self.mcp.call_tool("zalo_get_messages", {"limit": LIMIT}) if not isinstance(d, dict): d = {"messages": d if isinstance(d, list) else []} msgs = [m for m in (d.get("messages") or []) if isinstance(m, dict)] if msgs: await self.threads(force=any(str(_get(m, "threadId", default="")) not in self._threads for m in msgs)) for m in msgs: ev = self.normalize(m) if ev and await self._accept(ev, m): n += 1 nc = _int_cursor(d.get("cursor")) if nc is None: nc = _int_cursor(d.get("nextCursor")) if nc is None: self._since, self._since_epoch = 0, 0 break self._since, self._since_epoch = nc, self.mcp.epoch if not (d.get("hasMore") and msgs and nc > since): break since = nc self.last_read, self.last_error = time.time(), "" return n async def _accept(self, ev: dict, raw: dict) -> bool: if ev["from_self"]: self._learn_self(raw) if self.is_echo(ev["chat_id"], ev["text"]): return False # câu bot vừa gửi quay về self._owner[ev["chat_id"]] = ev["created_at"] elif self.is_echo(ev["chat_id"], ev["text"], ECHO_MIN_LEN): return False # MCP không gắn cờ "của mình" -> câu bot quay về như tin khách if self.on_event: r = self.on_event(ev) if asyncio.iscoroutine(r): await r return True async def _loop(self): while True: try: await self.read_once() await asyncio.sleep(self.poll_interval) except asyncio.CancelledError: raise except Exception as e: self.last_error = f"{type(e).__name__}: {e}"[:300] print(f"[zalo {self.label}] {self.last_error}", file=sys.stderr) await asyncio.sleep(ERROR_BACKOFF) def start(self): if not self._task or self._task.done(): self._task = asyncio.get_running_loop().create_task(self._loop()) async def stop(self): if self._task: self._task.cancel() self._task = None await self.mcp.close() # ---- gửi ----------------------------------------------------------------------------------- def _remember_sent(self, thread: str, text: str): now = time.time() lst = [x for x in self._sent.get(thread, []) if now - x[0] < ECHO_TTL] lst.append((now, " ".join(text.split()))) self._sent[thread] = lst[-20:] def is_echo(self, thread: str, text: str, min_len: int = 0) -> bool: t = " ".join(str(text or "").split()) if len(t) < min_len or not t: return False now = time.time() lst = [x for x in self._sent.get(thread, []) if now - x[0] < ECHO_TTL] for i, (_, s) in enumerate(lst): if t == s or t.startswith(s[:80]) or s.startswith(t[:80]): del lst[i] self._sent[thread] = lst return True self._sent[thread] = lst return False def owner_typed_recently(self, thread: str, window: int = OWNER_QUIET) -> bool: t = self._owner.get(str(thread)) return bool(t) and time.time() - t < window def mentioned(self, ev: dict, extra_names=()) -> tuple: """(được tag, được reply) theo nick này. Chưa kiểm trên mọi khuôn nhóm thật - xem README.""" names = {_norm_name(x) for x in (*extra_names, self.label, self.self_name) if x} - {""} text = _norm_name(ev.get("text")) tag = any(re.search("@" + re.escape(n) + r"(?!\w)", text) for n in names) uid = self.self_uid if not tag and uid: ms = ev.get("mentions") or [] for m in (ms if isinstance(ms, list) else [ms]): x = _get(m, "uid", "id", "userId", default="") if isinstance(m, dict) else m if str(x).strip() == uid: tag = True rep, rt = False, ev.get("reply_to") if isinstance(rt, dict): if uid and any(str(rt.get(k) or "") == uid for k in ("uid", "uidFrom", "senderId", "from", "userId")): rep = True else: rep = _norm_name(_get(rt, "senderName", "dName", "name", "fromName", default="")) in names return tag, rep async def send_text(self, chat_id: str, text: str, group: bool) -> tuple: """Gửi chữ qua MCP. Trả (ok, lỗi). Khoá kiểu chat mà MCP 1.6.2 THẬT SỰ đọc là `threadType` (0 riêng / 1 nhóm), không phải `type` như tài liệu; thiếu nó thì tin nhóm đi như chat riêng và KHÔNG tới nơi mà không báo lỗi. Gửi cả hai khoá. """ loai = 1 if group else 0 self._remember_sent(chat_id, text) try: d = await self.mcp.call_tool("zalo_send_message", {"threadId": str(chat_id), "text": text, "threadType": loai, "type": loai}) except Exception as e: return False, str(e)[:300] if isinstance(d, dict) and d.get("success") is False: return False, str(d.get("error") or d.get("message") or "Zalo từ chối")[:300] return True, "" async def send_mention(self, group_id: str, text: str, uid: str, name: str) -> Optional[tuple]: """Gửi tin nhóm CÓ TAG qua CLI. None = chưa gửi gì (người gọi nên gửi thường); (ok, lỗi) = đã có kết cục. Hết giờ trả lỗi, KHÔNG rơi về gửi thường (tin có thể đã đi -> thành hai tin dưới tên chủ).""" if not re.fullmatch(r"\d{3,25}", str(uid or "")) or not name: return None final, pos, ln = tagged_text(text, name) self._remember_sent(group_id, final) ok, _d, err = await self.cli.run(self.home, ["msg", "send"], [group_id, final], ["-t", "1", "--mention", f"{pos}:{uid}:{ln}"], timeout=TAG_TIMEOUT) if ok: return True, "" if is_timeout(err): return False, f"không rõ tin có tag đã đi chưa ({err})" return None async def send(self, chat_id: str, text: str, group: bool = False, mention: Optional[dict] = None) -> tuple: if group and mention: r = await self.send_mention(chat_id, text, str(mention.get("uid") or ""), str(mention.get("name") or "")) if r is not None: return r return await self.send_text(chat_id, text, group) # ---- các lệnh CLI MCP không có (ảnh/file, ghi chú, nhắc hẹn, poll) -------------------------- async def send_files(self, chat_id: str, paths: list, group: bool = False, caption: str = "", timeout: int = 120) -> tuple: """Gửi ảnh (`msg send-image`) hoặc file (`msg send-file`). Một lượt chỉ một loại. Trả (ok, lỗi). Tool MCP `zalo_send_message` chỉ nhận chữ, nên ảnh/file phải đi CLI.""" img = (".jpg", ".jpeg", ".png", ".webp", ".gif") cmd = "send-image" if all(str(p).lower().endswith(img) for p in paths) else "send-file" opts = ["-t", "1" if group else "0"] + (["-m", caption] if caption else []) ok, _d, err = await self.cli.run(self.home, ["msg", cmd], [chat_id, *map(str, paths)], opts, timeout=timeout) return ok, err async def run_cli(self, command: list, positionals=None, options=None, timeout: int = 120): """Lệnh CLI bất kỳ trong phiên này, ví dụ `["group", "note-create"]`, `["poll", "create"]`, `["reminder", "create"]`, `["group", "members-info"]` (cờ: xem original/plugins/zalo-group/plugin.py).""" return await self.cli.run(self.home, command, positionals, options, timeout=timeout)