"""Đăng nhập Zalo cá nhân bằng QR (port độc lập từ thansa-os `server/zalo_login.py`). Luồng: chạy `zalo-agent login --json` với HOME là một thư mục RIÊNG cho tài khoản -> CLI in từng dòng JSON (`{"event":"qr","dataUrl":...}` rồi một event thành công) -> đưa ảnh QR lên UI -> người dùng quét bằng app Zalo -> thành công thì gọi `on_success(home, info)` để bạn lưu "kết nối" (tối thiểu là đường dẫn `home`). QUAN TRỌNG: tài khoản đang hoạt động của zalo-agent-cli là TOÀN CỤC theo HOME. Hai tài khoản chung một HOME = tài khoản sau đè tài khoản trước. Nên mỗi tài khoản một thư mục, và tên thư mục phải có phần ngẫu nhiên. """ from __future__ import annotations import base64 import json import os import re import subprocess import threading import time import uuid from typing import Callable, Dict, Optional from .cli import ZaloCli, kill_tree, no_window_kwargs QR_TIMEOUT = 180 # QR Zalo sống ngắn SESSION_TTL = 900 _SUCCESS = {"login", "login_success", "success", "ready", "logged_in", "authenticated"} def _slug(s: str) -> str: return re.sub(r"[^a-z0-9]+", "-", (s or "zalo").lower()).strip("-")[:30] or "zalo" class QrLogin: def __init__(self, homes_root: str, cli: ZaloCli, on_success: Callable[[str, dict], Optional[str]]): """`homes_root`: nơi tạo thư mục phiên. `on_success(home, info)` trả None nếu lưu được, hoặc chuỗi lỗi.""" self.homes_root = homes_root self.cli = cli self.on_success = on_success self._s: Dict[str, dict] = {} def start(self, label: str = "") -> dict: self._sweep() base = self.cli.base_command() if not base: return {"ok": False, "error": "Cần Node.js 20+ (lệnh npx)"} sid = uuid.uuid4().hex[:10] home = os.path.join(self.homes_root, f"zalo-{_slug(label)}-{sid[:6]}") os.makedirs(home, exist_ok=True) env = dict(os.environ, HOME=home, USERPROFILE=home) try: proc = subprocess.Popen(base + ["login", "--json"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL, text=True, encoding="utf-8", errors="replace", env=env, **no_window_kwargs()) except OSError as e: return {"ok": False, "error": f"Không chạy được CLI: {e}"} self._s[sid] = {"state": "starting", "qr": "", "label": label, "error": "", "info": {}, "proc": proc, "home": home, "ts": time.time()} threading.Thread(target=self._reader, args=(sid,), daemon=True).start() threading.Thread(target=self._watchdog, args=(sid,), daemon=True).start() return {"ok": True, "sid": sid} def status(self, sid: str) -> dict: s = self._s.get(sid) if not s: return {"state": "error", "error": "Phiên đăng nhập không tồn tại (hết hạn?)"} return {k: s[k] for k in ("state", "qr", "label", "error", "home", "info")} def cancel(self, sid: str) -> dict: s = self._s.get(sid) if s and s["proc"].poll() is None: kill_tree(s["proc"]) if s and s["state"] != "done": s.update(state="error", error=s["error"] or "Đã huỷ") return {"ok": True} # ---- nội bộ ---------------------------------------------------------------------------- def _sweep(self): now = time.time() for sid in [k for k, v in self._s.items() if now - v["ts"] > SESSION_TTL]: self.cancel(sid) self._s.pop(sid, None) @staticmethod def _qr(obj: dict, home: str) -> str: for k in ("dataUrl", "data_url", "image"): v = obj.get(k) if isinstance(v, str) and v.startswith("data:image"): return v for f in (obj.get("file"), os.path.join(home, ".zalo-agent-cli", "qr.png")): if isinstance(f, str) and os.path.isfile(f): with open(f, "rb") as fh: return "data:image/png;base64," + base64.b64encode(fh.read()).decode("ascii") return "" def _ok(self, s: dict, obj: dict): label = (obj.get("displayName") or obj.get("display_name") or obj.get("name") or obj.get("ownId") or obj.get("own_id") or s["label"] or "Zalo") info = {"label": str(label)[:60], "own_id": str(obj.get("ownId") or obj.get("own_id") or "")} err = self.on_success(s["home"], info) if err: s.update(state="error", error=f"Đăng nhập được nhưng không lưu kết nối: {err}") else: s.update(state="done", label=info["label"], info=info) def _reader(self, sid: str): s = self._s[sid] proc, got_qr = s["proc"], False try: for raw in iter(proc.stdout.readline, ""): if s["state"] in ("done", "error"): break try: obj = json.loads(raw.strip()) except ValueError: continue # log của npx if not isinstance(obj, dict): continue ev = str(obj.get("event") or "").lower() if ev == "qr": qr = self._qr(obj, s["home"]) if qr: got_qr = True s.update(state="qr", qr=qr) elif ev in _SUCCESS or (not ev and (obj.get("ownId") or obj.get("own_id"))): self._ok(s, obj) break elif ev in ("error", "failed"): s.update(state="error", error=str(obj.get("message") or obj.get("error") or "Đăng nhập thất bại")) break except Exception as e: if s["state"] not in ("done", "error"): s.update(state="error", error=f"{type(e).__name__}: {e}") finally: try: rc = proc.wait(timeout=5) except Exception: rc = None if s["state"] not in ("done", "error"): if rc == 0 and got_qr: # CLI thoát êm sau khi đã hiện QR mà không in event thành công self._ok(s, {}) else: tail = "" try: tail = (proc.stderr.read() or "")[-300:] except Exception: pass s.update(state="error", error="Đăng nhập chưa hoàn tất" + (f" (exit {rc}): {tail}" if rc else "")) def _watchdog(self, sid: str): time.sleep(QR_TIMEOUT) s = self._s.get(sid) if s and s["state"] in ("starting", "qr"): s.update(state="error", error="Mã QR hết hạn, bấm thử lại") if s["proc"].poll() is None: kill_tree(s["proc"])