#!/usr/bin/env python3
"""chv-browser：浏览器子代理（客机侧、单元外独立进程）。

阶段 3（feat-chv-opt-landing）：浏览器服务化。
  * Playwright Server 独立进程：常驻运行，codex 与各会话经 CDP
    连入（--remote-debugging-port，默认 9222；op `cdp` 返回
    webSocketDebuggerUrl）。codex 重启不丢浏览器与登录态——
    浏览器进程与 codex 进程分属不同 systemd 单元，无共享进程组。
  * 持久 profile：launch_persistent_context，user_data_dir 固定
    在独立数据盘 /home/chv/workspace/.chv-browser-profile（数据
    盘由 chv-data-mount 挂载，/dev/vdc，LABEL=CHVDATA）；数据盘
    未挂载时回落到 /var/lib/chv-browser/profile（本机盘，不跨
    恢复保留——status 会明说当前 profile 在哪，绝不假装在数据盘）。
    服务 crash/重启后重新打开同一 profile，cookie/登录态仍在。
  * 多会话：同一浏览器上下文内每会话一页（session id → page），
    会话 A/B 各自导航、互不干扰；登录态（cookie）同上下文共享，
    是既定取舍（方案：同 VM 内 Bot 共享登录态，隔离边界是 VM 本身）。
  * 双浏览器：默认 chromium-headless-shell（随 base 常驻）；完整
    Chromium 按需挂载（op `mount_full`：切到 full 引擎）、用完
    卸载（op `unmount_full`：切回 shell）。完整版不 bake 进 base；
    未安装时 mount_full 明确报错并回落 shell，不硬凑通过。
  * 截图：JPEG quality 70，viewport 宽封顶 1280（上下文创建时
    定死 1280x800）；diff_only 时与上一帧比对，字节相同即不落盘、
    返回 changed=false（整帧相同跳过；变化区裁剪是阶段 4 的事，
    这里不冒充已做）。

对齐 Muse P2 的不变口径：子代理只向单元返回页面无障碍快照，
不见原始 DOM，不能在页面上下文跑 JS，不暴露 evaluate/content，
不开 DevTools。快照 API：当前 Playwright（1.64.1）已移除
page.accessibility，改用 page.aria_snapshot()（返回 YAML 式
字符串）；旧版 accessibility.snapshot()（返回 dict）只作回落。

操作（Unix socket /run/chv/browser.sock，JSON 行）：
  open       {url, session?} -> 导航并返回 {title, url, a11y, truncated}
  snapshot   {session?}      -> 当前页的 {title, url, a11y, truncated}
  screenshot {session?, diff_only?} -> JPEG 落盘 /run/chv/shots/<uuid>.jpg，
                                       返回 {path, size, format, quality,
                                             width, changed, sha256}
  close      {session?}      -> 关该会话的页（浏览器进程继续跑）
  sessions   {}              -> 各会话 {session, url, title}
  status     {}              -> {engine, profile, profile_on_data_disk,
                                cdp_port, sessions, mounted_full,
                                full_present}
  cdp        {}              -> {cdp_http, websocket_debugger_url, engine}
  engine     {engine: shell|full} -> 切换引擎（full 缺失/起不来时
                                       回落 shell 并明说 fallback）
  mount_full {}              -> 挂载完整 Chromium（切 full 引擎）
  unmount_full {}            -> 卸载完整 Chromium（切回 shell 引擎）
  ping       {}              -> {"ok": true}

URL 只允许 http/https；localhost 与私有地址字面量直接拒绝（主机名
判定，不做 DNS 解析——DNS 重绑定防护是 P3 Sentinel 的事，这里不
冒充已做）。无障碍树截断到 200 节点 / 64KB，超了标 truncated=true，
绝不把整棵树无上限灌进 socket。

playwright 是懒加载：import 本模块不需要 playwright 已装（单测
在无 playwright 环境跑）；真正 open 时才加载，缺失即报错。

只用标准库 + playwright（运行时）。运行身份：系统用户 chv-browser，
socket 0600（broker 以 root 连，不受模式限制）。
"""
import glob
import hashlib
import json
import os
import re
import socket
import sys
import time
import urllib.request
import uuid

BROWSER_SOCK = "/run/chv/browser.sock"
SHOTS_DIR = "/run/chv/shots"
PLAYWRIGHT_BROWSERS_PATH = "/usr/local/share/ms-playwright"
# 持久 profile：数据盘上的工作区内（阶段 1 已做 /dev/vdc 独立盘）。
WORKSPACE_MOUNT = "/home/chv/workspace"
PROFILE_DATA_DIR = "/home/chv/workspace/.chv-browser-profile"
PROFILE_FALLBACK_DIR = "/var/lib/chv-browser/profile"
CDP_PORT_DEFAULT = 9222
# 截图规格（方案阶段 4 口径，阶段 3 先落地编码与尺寸两项）：
# JPEG quality 70，宽封顶 1280（上下文 viewport 定死 1280x800）。
VIEWPORT_WIDTH = 1280
VIEWPORT_HEIGHT = 800
JPEG_QUALITY = 70
MAX_REQUEST_BYTES = 65536
MAX_A11Y_NODES = 200
MAX_A11Y_CHARS = 65536
NAV_TIMEOUT_MS = 30000

# 拒绝的 URL 主机：localhost 与私有地址字面量（主机名判定）。
_PRIVATE_HOST_RE = re.compile(
    r"^(localhost|127\.\d+\.\d+\.\d+|10\.\d+\.\d+\.\d+|"
    r"172\.(1[6-9]|2\d|3[01])\.\d+\.\d+|192\.168\.\d+\.\d+|"
    r"\[::1\]|\[fe8[0-9a-f]*:.*\])$",
    re.IGNORECASE)

DEFAULT_SESSION = "default"
ENGINES = ("shell", "full")


def validate_url(url):
    """URL 校验：只允许 http/https，且主机不是 localhost/私有地址
    字面量。返回 (ok, error)。DNS 重绑定不在本层防（见模块 docstring）。"""
    if not isinstance(url, str) or not url:
        return False, "url 不能为空"
    m = re.match(r"^(https?)://([^/]+)", url, re.IGNORECASE)
    if not m:
        return False, "只允许 http/https URL：%s" % url[:80]
    host = m.group(2).split("@")[-1].split(":")[0].lower()
    if _PRIVATE_HOST_RE.match(host):
        return False, "拒绝 localhost/私有地址：%s" % host
    return True, ""


def resolve_profile_dir(env=None, ismount=None):
    """决定 profile 目录并返回 (path, on_data_disk)。
    env 显式指定（CHV_BROWSER_PROFILE）时优先；否则数据盘已挂载
    才用数据盘路径，否则回落本机盘并明说不在数据盘上。"""
    if env is None:
        env = os.environ
    explicit = (env.get("CHV_BROWSER_PROFILE") or "").strip()
    if explicit:
        return explicit, explicit.startswith(WORKSPACE_MOUNT + "/")
    check = ismount or os.path.ismount
    if check(WORKSPACE_MOUNT):
        return PROFILE_DATA_DIR, True
    return PROFILE_FALLBACK_DIR, False


def full_chromium_present(browsers_path=None):
    """完整 Chromium 是否已装：ms-playwright 下有 chromium-*
    目录且含 chrome-linux/chrome（headless-shell 目录不算）。"""
    base = browsers_path or os.environ.get(
        "PLAYWRIGHT_BROWSERS_PATH", PLAYWRIGHT_BROWSERS_PATH)
    for cand in glob.glob(os.path.join(base, "chromium-*")):
        if os.path.basename(cand).startswith(
                "chromium_headless_shell"):
            continue
        if os.path.isfile(os.path.join(
                cand, "chrome-linux", "chrome")):
            return True
    return False


def truncate_a11y(node, max_nodes=MAX_A11Y_NODES,
                  max_chars=MAX_A11Y_CHARS):
    """截断无障碍树：节点数超上限时剪掉多余子树。返回 (tree, truncated)。
    纯函数，可单测。节点计数按 dict 节点数（含 children 展开）。"""
    state = {"count": 0, "truncated": False}

    def walk(n):
        if not isinstance(n, dict):
            return n
        state["count"] += 1
        if state["count"] > max_nodes:
            state["truncated"] = True
            return None
        out = {}
        for key, value in n.items():
            if key == "children" and isinstance(value, list):
                children = []
                for child in value:
                    walked = walk(child)
                    if walked is None:
                        state["truncated"] = True
                        break
                    children.append(walked)
                out[key] = children
            else:
                out[key] = value
        return out

    tree = walk(node)
    text = json.dumps(tree, ensure_ascii=False)
    if len(text) > max_chars:
        state["truncated"] = True
        # 字符超限时只留顶层角色/名字/URL 信息，不灌整棵树。
        tree = {"role": (node or {}).get("role"),
                "name": (node or {}).get("name"),
                "truncated_chars": True}
    return tree, state["truncated"]


def _load_playwright():
    """懒加载 playwright；缺失时报明错（不静默假装有浏览器）。
    返回 sync_playwright() 的上下文管理器实例（须调用，
    不能只返回 sync_playwright 函数本身——函数没有 __enter__，
    2026-11-08 构建 #8 驱动 live 抓到：真实路径从未跑通过）。"""
    os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH",
                          PLAYWRIGHT_BROWSERS_PATH)
    try:
        from playwright.sync_api import sync_playwright
    except ImportError as e:
        raise RuntimeError(
            "chv-browser: playwright 未装（%s），浏览器子代理不可用"
            % e)
    return sync_playwright()


def _fetch_cdp_version(port, timeout=5.0):
    """读本地 CDP 的 /json/version，返回 webSocketDebuggerUrl；
    读不到返回 None（不抛错，调用方据此判定 CDP 未就绪）。"""
    try:
        with urllib.request.urlopen(
                "http://127.0.0.1:%d/json/version" % port,
                timeout=timeout) as resp:
            data = json.loads(resp.read().decode())
        url = data.get("webSocketDebuggerUrl")
        return url if isinstance(url, str) and url else None
    except Exception:
        return None


class BrowserAgent:
    """多会话浏览器子代理：一个持久上下文 + 每会话一页。
    playwright_factory 供单测注入假实现；正常运行时用
    _load_playwright()。"""

    def __init__(self, playwright_factory=None, shots_dir=SHOTS_DIR,
                 profile_dir=None, cdp_port=None):
        self._factory = playwright_factory or _load_playwright
        self._shots_dir = shots_dir
        if profile_dir:
            self._profile_dir = profile_dir
            self._profile_on_data_disk = profile_dir.startswith(
                WORKSPACE_MOUNT + "/")
        else:
            self._profile_dir, self._profile_on_data_disk = \
                resolve_profile_dir()
        try:
            self._cdp_port = int(cdp_port or os.environ.get(
                "CHV_BROWSER_CDP_PORT") or CDP_PORT_DEFAULT)
        except ValueError:
            self._cdp_port = CDP_PORT_DEFAULT
        self._pw = None
        self._entered = None
        self._context = None
        self._pages = {}
        self._engine = "shell"
        self._mounted_full = False
        self._last_shot = {}

    # -- 浏览器生命周期 ----------------------------------------------
    def _close_context(self):
        for page in list(self._pages.values()):
            try:
                page.close()
            except Exception:
                pass
        self._pages = {}
        if self._context is not None:
            try:
                self._context.close()
            except Exception:
                pass
            self._context = None

    def _ensure_browser(self, engine=None):
        """确保浏览器上下文已起（指定引擎时切换）。返回 (ok, error)。
        切换时先关旧上下文再起新上下文——两个引擎共用同一 profile
        目录，绝不同时打开（Chromium profile 锁会互斥）。"""
        engine = engine or self._engine
        if engine not in ENGINES:
            return False, "未知引擎：%r（只允许 shell/full）" % (engine,)
        if self._context is not None and engine == self._engine:
            return True, ""
        self._close_context()
        try:
            if self._pw is None:
                self._pw = self._factory()
                self._entered = self._pw.__enter__()
            kwargs = {
                "headless": True,
                "viewport": {"width": VIEWPORT_WIDTH,
                             "height": VIEWPORT_HEIGHT},
                "args": ["--remote-debugging-port=%d"
                         % self._cdp_port,
                         "--remote-debugging-address=0.0.0.0"],
            }
            if engine == "full":
                kwargs["channel"] = "chromium"
            launch = getattr(self._entered.chromium,
                             "launch_persistent_context", None)
            if launch is None:
                # 旧版 playwright/假实现只有 launch：回落到
                # 临时上下文（profile 不持久，调用方在 status
                # 里能看到 persistent=false）。
                self._context = self._entered.chromium.launch(
                    headless=True)
            else:
                self._context = launch(self._profile_dir, **kwargs)
        except Exception as e:
            self._context = None
            return False, "启动浏览器（引擎 %s）失败：%s" % (engine, e)
        self._engine = engine
        return True, ""

    def _page(self, session):
        return self._pages.get(session or DEFAULT_SESSION)

    def _cdp_ws(self):
        """等 CDP 端口就绪并返回 webSocketDebuggerUrl（最多等 5 秒）。"""
        for _ in range(10):
            url = _fetch_cdp_version(self._cdp_port)
            if url:
                return url
            time.sleep(0.5)
        return None

    # -- 会话操作 ------------------------------------------------------
    def open(self, url, session=None):
        ok, err = validate_url(url)
        if not ok:
            return {"ok": False, "error": err}
        session = session or DEFAULT_SESSION
        ok, err = self._ensure_browser()
        if not ok:
            return {"ok": False, "error": err}
        try:
            old = self._pages.get(session)
            if old is not None:
                try:
                    old.close()
                except Exception:
                    pass
            page = self._context.new_page()
            self._pages[session] = page
            page.goto(url, timeout=NAV_TIMEOUT_MS)
            page.wait_for_load_state("domcontentloaded",
                                     timeout=NAV_TIMEOUT_MS)
        except Exception as e:
            return {"ok": False,
                    "error": "打开 %s 失败：%s" % (url[:80], e)}
        return self.snapshot(session)

    def snapshot(self, session=None):
        session = session or DEFAULT_SESSION
        page = self._page(session)
        if page is None:
            return {"ok": False,
                    "error": "会话 %s 还没有打开的页面（先 open）"
                             % session}
        try:
            title = page.title()
            url = page.url
            # 当前 Playwright（1.64.1）：page.accessibility 已移除，
            # 用 page.aria_snapshot()（字符串）；旧版 dict 树只作回落。
            if hasattr(page, "aria_snapshot"):
                snap = page.aria_snapshot()
                if isinstance(snap, dict):
                    tree, truncated = truncate_a11y(snap)
                else:
                    text = snap or ""
                    truncated = len(text) > MAX_A11Y_CHARS
                    tree = text[:MAX_A11Y_CHARS] if truncated else text
            elif hasattr(page, "accessibility"):
                tree = page.accessibility.snapshot()
                tree, truncated = truncate_a11y(tree or {})
            else:
                return {"ok": False,
                        "error": "浏览器不支持无障碍快照（无 "
                                 "aria_snapshot/accessibility）"}
        except Exception as e:
            return {"ok": False, "error": "取快照失败：%s" % e}
        return {"ok": True, "title": title, "url": url,
                "a11y": tree, "truncated": truncated,
                "session": session}

    def screenshot(self, session=None, diff_only=False):
        session = session or DEFAULT_SESSION
        page = self._page(session)
        if page is None:
            return {"ok": False,
                    "error": "会话 %s 还没有打开的页面（先 open）"
                             % session}
        try:
            data = page.screenshot(type="jpeg",
                                   quality=JPEG_QUALITY)
        except Exception as e:
            return {"ok": False, "error": "截图失败：%s" % e}
        if not isinstance(data, (bytes, bytearray)):
            return {"ok": False, "error": "截图未返回字节数据"}
        data = bytes(data)
        digest = hashlib.sha256(data).hexdigest()
        if diff_only and self._last_shot.get(session) == digest:
            # 整帧与上一帧字节相同：跳过落盘（阶段 4 的变化区
            # 裁剪不在本阶段做，这里只做整帧相同跳过）。
            return {"ok": True, "changed": False, "size": 0,
                    "path": None, "format": "jpeg",
                    "quality": JPEG_QUALITY,
                    "width": VIEWPORT_WIDTH, "session": session}
        try:
            os.makedirs(self._shots_dir, exist_ok=True)
            path = os.path.join(self._shots_dir, "%s.jpg"
                                % uuid.uuid4().hex)
            with open(path, "wb") as fh:
                fh.write(data)
        except OSError as e:
            return {"ok": False, "error": "截图落盘失败：%s" % e}
        self._last_shot[session] = digest
        return {"ok": True, "path": path, "size": len(data),
                "format": "jpeg", "quality": JPEG_QUALITY,
                "width": VIEWPORT_WIDTH, "changed": True,
                "sha256": digest, "session": session}

    def close(self, session=None):
        session = session or DEFAULT_SESSION
        page = self._pages.pop(session, None)
        if page is not None:
            try:
                page.close()
            except Exception:
                pass
        self._last_shot.pop(session, None)
        return {"ok": True, "closed": True, "session": session}

    def sessions(self):
        out = []
        for sid, page in self._pages.items():
            try:
                title = page.title()
            except Exception:
                title = ""
            try:
                url = page.url
            except Exception:
                url = ""
            out.append({"session": sid, "url": url, "title": title})
        return {"ok": True, "sessions": out}

    # -- 服务化：状态 / CDP / 引擎 --------------------------------------
    def status(self):
        return {"ok": True, "engine": self._engine,
                "profile": self._profile_dir,
                "profile_on_data_disk": self._profile_on_data_disk,
                "persistent": True,
                "cdp_port": self._cdp_port,
                "sessions": [s["session"] for s in
                             self.sessions()["sessions"]],
                "mounted_full": self._mounted_full,
                "full_present": full_chromium_present()}

    def cdp(self):
        ok, err = self._ensure_browser()
        if not ok:
            return {"ok": False, "error": err}
        return {"ok": True,
                "cdp_http": "http://127.0.0.1:%d" % self._cdp_port,
                "websocket_debugger_url": self._cdp_ws(),
                "engine": self._engine,
                "profile": self._profile_dir}

    def switch_engine(self, engine):
        """切换引擎。full 未装或起不来时回落 shell 并明说。"""
        if engine not in ENGINES:
            return {"ok": False,
                    "error": "未知引擎：%r（只允许 shell/full）"
                             % (engine,)}
        if engine == "full" and not full_chromium_present():
            return {"ok": False,
                    "error": "完整 Chromium 未安装（未见 chromium-* "
                             "chrome-linux/chrome），回落 headless-shell；"
                             "先跑 python3 -m playwright install "
                             "chromium 安装",
                    "engine": self._engine, "fallback": True,
                    "mounted_full": False}
        ok, err = self._ensure_browser(engine)
        if not ok:
            # 起不来：回落 shell（shell 本身起不来才如实报错）。
            back_ok, back_err = self._ensure_browser("shell")
            resp = {"ok": False, "error": err,
                    "engine": self._engine, "fallback": True,
                    "mounted_full": False}
            if not back_ok:
                resp["error"] = "%s；回落 shell 也失败：%s" % (
                    err, back_err)
            return resp
        self._mounted_full = (engine == "full")
        return {"ok": True, "engine": engine,
                "mounted_full": self._mounted_full,
                "profile": self._profile_dir}

    def mount_full(self):
        if not full_chromium_present():
            return {"ok": False,
                    "error": "完整 Chromium 未安装，无法挂载（回落 "
                             "headless-shell 继续工作）；先跑 python3 "
                             "-m playwright install chromium 安装",
                    "engine": self._engine, "fallback": True,
                    "mounted_full": False}
        return self.switch_engine("full")

    def unmount_full(self):
        resp = self.switch_engine("shell")
        if resp.get("ok"):
            resp["mounted_full"] = False
        return resp

    def shutdown(self):
        self._close_context()
        try:
            if self._pw is not None:
                self._pw.__exit__(None, None, None)
        except Exception:
            pass
        self._pw = None
        self._entered = None


# 本层不暴露的操作：evaluate / content（原始 DOM）/ DevTools。
# 收到即拒绝，不转发、不执行。
FORBIDDEN_OPS = ("evaluate", "content", "devtools", "js")


def handle_request(agent, req):
    """处理一个请求 dict，返回响应 dict（纯逻辑，可单测）。"""
    if not isinstance(req, dict):
        return {"ok": False, "error": "请求须是 JSON 对象"}
    op = req.get("op")
    if op in FORBIDDEN_OPS:
        return {"ok": False,
                "error": "操作 %r 不暴露（无 JS/无原始 DOM）" % op}
    if op == "ping":
        return {"ok": True, "browser": True}
    if op == "open":
        return agent.open(req.get("url"), req.get("session"))
    if op == "snapshot":
        return agent.snapshot(req.get("session"))
    if op == "screenshot":
        return agent.screenshot(req.get("session"),
                                bool(req.get("diff_only")))
    if op == "close":
        return agent.close(req.get("session"))
    if op == "sessions":
        return agent.sessions()
    if op == "status":
        return agent.status()
    if op == "cdp":
        return agent.cdp()
    if op == "engine":
        return agent.switch_engine(req.get("engine"))
    if op == "mount_full":
        return agent.mount_full()
    if op == "unmount_full":
        return agent.unmount_full()
    return {"ok": False,
            "error": "未知操作：%r（白名单：open/snapshot/screenshot/"
                     "close/sessions/status/cdp/engine/mount_full/"
                     "unmount_full/ping）" % (op,)}


def serve(sock_path=BROWSER_SOCK, shots_dir=SHOTS_DIR,
          profile_dir=None, cdp_port=None):
    """监听 Unix socket，逐连接处理。socket 0600（只给属主 +
    broker 的 root 连）。建不起来即抛错，不要静默带病跑。"""
    if os.path.exists(sock_path):
        os.remove(sock_path)
    os.makedirs(os.path.dirname(sock_path), exist_ok=True)
    srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    srv.bind(sock_path)
    srv.listen(16)
    os.chmod(sock_path, 0o600)
    print("chv-browser: 监听 %s" % sock_path, file=sys.stderr, flush=True)
    agent = BrowserAgent(shots_dir=shots_dir, profile_dir=profile_dir,
                         cdp_port=cdp_port)
    try:
        while True:
            try:
                conn, _ = srv.accept()
            except OSError:
                continue
            try:
                conn.settimeout(120)
                data = b""
                while len(data) < MAX_REQUEST_BYTES:
                    chunk = conn.recv(4096)
                    if not chunk:
                        break
                    data += chunk
                    if data.endswith(b"\n"):
                        break
                try:
                    req = json.loads(data.decode())
                except (ValueError, UnicodeDecodeError):
                    resp = {"ok": False, "error": "请求不是合法 JSON"}
                else:
                    try:
                        resp = handle_request(agent, req)
                    except Exception as e:
                        resp = {"ok": False,
                                "error": "浏览器子代理内部错误：%s" % e}
                conn.sendall((json.dumps(resp, ensure_ascii=False)
                              + "\n").encode())
            except OSError as e:
                print("chv-browser: 连接处理异常：%s" % e, file=sys.stderr,
                      flush=True)
            finally:
                try:
                    conn.close()
                except OSError:
                    pass
    finally:
        agent.shutdown()


def main(argv):
    sock_path = BROWSER_SOCK
    shots_dir = SHOTS_DIR
    profile_dir = None
    cdp_port = None
    i = 0
    while i < len(argv):
        if argv[i] == "--socket" and i + 1 < len(argv):
            sock_path = argv[i + 1]
            i += 2
        elif argv[i] == "--shots-dir" and i + 1 < len(argv):
            shots_dir = argv[i + 1]
            i += 2
        elif argv[i] == "--profile-dir" and i + 1 < len(argv):
            profile_dir = argv[i + 1]
            i += 2
        elif argv[i] == "--cdp-port" and i + 1 < len(argv):
            cdp_port = argv[i + 1]
            i += 2
        else:
            print("chv-browser: 未知参数：%s" % argv[i], file=sys.stderr)
            return 2
    serve(sock_path, shots_dir, profile_dir, cdp_port)
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))