#!/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/登录态仍在。 * 多会话：默认会话（"default"）走单一持久上下文（profile 在 数据盘，登录态跨浏览器重启保留，阶段 3 口径不变）；具名 会话各建一个独立 BrowserContext（同浏览器、不同存储分区， cookie/localStorage 互不串），每会话的 storage-state 落 数据盘 .chv-browser-sessions/<会话>/storage-state.json， open 后与 close 时各存一次；重开同一会话 id 时从文件恢复， 不同会话 id 永远看不到对方的态。登录态跨会话保留的含义 是“同一会话 id 重开能读回自己的态”，不是“新会话能读别 人的态”（那正是隔离要防的）。localStorage 的隔离与 cookie 同源：同一浏览器的不同 context 自带独立存储分区。 * 阶段 5（feat-chv-opt-landing）：具名会话独立 context + storage-state 落盘恢复（见上）；会话 id 只允许 [A-Za-z0-9_-]（首字符须为字母数字，最长 64），防路径穿越 进 storage-state 目录；close/kill 只动该会话自己的 context 与任务，不动其它会话。 * 双浏览器：默认 chromium-headless-shell（随 base 常驻）；完整 Chromium 按需挂载（op `mount_full`：切到 full 引擎）、用完 卸载（op `unmount_full`：切回 shell）。完整版不 bake 进 base； 未安装时 mount_full 明确报错并回落 shell，不硬凑通过。 完整版装在数据盘 /home/chv/workspace/.chv-browsers（根盘只剩 约 600MB，装根盘会 ENOSPC；full 引擎启动时直接传 executable_path，不走 channel/env）。 * 截图：JPEG quality 70，viewport 宽封顶 1280（上下文创建时 定死 1280x800）；diff_only 时与上一帧比对，字节相同即不落盘、 返回 changed=false（整帧相同跳过）。 * 阶段 4（feat-chv-opt-landing）：截图三级管线。 一级 a11y+DOM 主通道：snapshot 在 aria_snapshot 字符串基础 上纯字符串解析出角色计数（button/link/textbox/heading 等） 与 interactive 总数、a11y 字符数与 a11y_hash——不跑 JS、不取 原始 DOM、不新增 evaluate/content 通路，禁 op 保持拒绝。 二级差分截图（op `diff_shot`）：先截 PNG 全帧，用 Pillow 解码与上一帧逐像素比对——像素判变阈值 DIFF_PIXEL_THRESHOLD =10（单通道差值 >10 才算变，滤掉抗锯齿 1 位抖动）；无变化 返回 changed:false size 0；有变化算出变化区 bbox（外扩 margin 8px、钳制到视口内、宽高下限 32px），再用 page.screenshot(type=jpeg, quality=70, clip=bbox) 只截变 化区。Pillow 缺失时 diff_shot 如实报错并指引回落全帧 screenshot，绝不静默假装裁剪。full_size 是本帧 PNG 经 Pillow 以 JPEG q70 编码得到的全帧字节数（与浏览器直出 JPEG q70 同参数、字节数可能有数百字节差异，仅作对照）。 三级全屏校验（op `verify_shot`）：全帧 JPEG q70，专用于 少数校验场景。事件驱动口径：日常只在动作后调 diff_shot （变化才出图），verify_shot 只在检查点调用。计数器 shots_full/shots_diff/shots_verify/shots_skipped 由 status 返回，verify_share = shots_verify / (shots_full+shots_diff+shots_verify)，分母为 0 时记 0.0。 计数口径：shots_full = screenshot 落盘数；shots_diff = diff_shot 返回 changed:true 的数；shots_verify = verify_shot 成功数；shots_skipped = screenshot diff_only 跳过数 + diff_shot changed:false 数。 对齐 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, a11y_roles, a11y_interactive, a11y_chars, a11y_hash} screenshot {session?, diff_only?} -> JPEG 落盘 /run/chv/shots/.jpg， 返回 {path, size, format, quality, width, changed, sha256} diff_shot {session?} -> PNG 全帧逐像素比对；无变化 {changed:false, size:0}；有变化只截变 化区 JPEG，返回 {changed:true, bbox, size, full_size, path, sha256} verify_shot {session?} -> 全帧 JPEG q70（只用于检查点）， 返回 {path, size, format, quality, width, 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" # 具名会话的 storage-state 目录：profile 在数据盘上时跟 profile # 走数据盘（与 profile 同级）；回落本机盘时跟回落 profile 走 # /var/lib/chv-browser/sessions（chv-browser 家目录，服务可写）。 SESSIONS_FALLBACK_DIR = "/var/lib/chv-browser/sessions" # 会话 id 白名单（阶段 5）：只允许字母数字/_/-，首字符须为字母 # 数字，最长 64——storage-state 按会话 id 落子目录，非法 id # 绝不能拼出 ../ 穿越。 SESSION_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$") # 完整 Chromium 不 bake 进 base、不装根盘：根盘只剩 ~600MB， # playwright install chromium 会 ENOSPC（2026-11-08 构建 #8 驱动 # 实证：109 秒后写满失败）。按需装到数据盘（9.3G 空闲），用完可删。 FULL_BROWSERS_DATA_DIR = "/home/chv/workspace/.chv-browsers" FULL_BROWSERS_FALLBACK_DIR = "/var/lib/chv-browser/browsers" 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 # 阶段 4 差分截图口径（写死在代码里，驱动与报告按此口径实测）： # 像素判变阈值：单通道差值 >10 才算变（滤抗锯齿 1 位抖动）； # bbox 外扩 margin 8px；变化区宽高下限 32px（不足则绕中心扩到 # 32px 再钳制到视口内）。 DIFF_PIXEL_THRESHOLD = 10 DIFF_MARGIN_PX = 8 DIFF_MIN_SIZE_PX = 32 # a11y 角色里算 interactive 的集合（aria_snapshot 字符串解析用； # searchbox 是 textbox 的搜索变体，同样算可交互）。 INTERACTIVE_ROLES = frozenset([ "button", "link", "textbox", "searchbox", "checkbox", "radio", "combobox", "listbox", "menuitem", "tab", "switch", "slider", "spinbutton", ]) # 拒绝的 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_executable(browsers_path=None): """完整 Chromium 可执行文件路径（数据盘安装目录下），未装返回 None。 headless-shell 目录不算。布局有两种：经典 chrome-linux/chrome， 与 CFT（Playwright 1.64.1 下的 chrome-linux64.zip）chrome-linux64/ chrome——2026-11-08 构建 #8 驱动实证：装好的是后者，旧检查只认 前者而漏判，mount_full 误报未安装。""" 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 for sub in ("chrome-linux", "chrome-linux64"): exe = os.path.join(cand, sub, "chrome") if os.path.isfile(exe): return exe return None def full_chromium_present(browsers_path=None): """完整 Chromium 是否已装：ms-playwright 下有 chromium-* 目录且含 chrome-linux/chrome 或 chrome-linux64/chrome （headless-shell 目录不算）。""" return full_chromium_executable(browsers_path) is not None def resolve_full_browsers_dir(profile_on_data_disk, env=None): """完整 Chromium 的安装目录：profile 在数据盘上时跟 profile 走 数据盘（/home/chv/workspace/.chv-browsers），否则回落本机盘。 env 显式指定（CHV_BROWSER_FULL_BROWSERS_DIR）时优先。""" if env is None: env = os.environ explicit = (env.get("CHV_BROWSER_FULL_BROWSERS_DIR") or "").strip() if explicit: return explicit if profile_on_data_disk: return FULL_BROWSERS_DATA_DIR return FULL_BROWSERS_FALLBACK_DIR def resolve_sessions_dir(profile_dir, profile_on_data_disk, env=None): """具名会话 storage-state 的根目录（阶段 5）。 env 显式指定（CHV_BROWSER_SESSIONS_DIR）时优先；否则 profile 在数据盘上时取 profile 的同级目录 /home/chv/workspace/.chv-browser-sessions（数据盘，随 快照恢复保留），回落时取 /var/lib/chv-browser/sessions。""" if env is None: env = os.environ explicit = (env.get("CHV_BROWSER_SESSIONS_DIR") or "").strip() if explicit: return explicit if profile_on_data_disk: return os.path.join(os.path.dirname(profile_dir), ".chv-browser-sessions") return SESSIONS_FALLBACK_DIR def validate_session_id(session): """会话 id 校验（阶段 5）：空/None 视为 default；非法返回 (False, error)，合法返回 (True, "")。storage-state 按 id 落子目录，校验是防穿越的唯一关口。""" if session is None or session == "": return True, "" if not isinstance(session, str) or not SESSION_ID_RE.match(session): return False, ("非法会话 id：%r（只允许字母数字/_/-，" "首字符须为字母数字，最长 64）" % (session,)) return True, "" 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 parse_a11y(a11y): """解析无障碍快照出角色计数（纯函数，可单测）。 输入是 aria_snapshot 字符串（主路）或旧版 dict 树（回落）。 字符串按行解析：行形如 `- link "More information..." [ref=e3]`， 嵌套行带前导空格；角色取 `- ` 后的第一个 token。dict 树则 递归数每个节点的 role 字段。返回 (roles, interactive, chars, a11y_hash)：roles 是 {角色: 个数}，interactive 是 INTERACTIVE_ROLES 内的角色总数，chars 是快照字符数， a11y_hash 是快照规范文本的 sha256（字符串用原文，dict 用 sort_keys 的 JSON）。不跑 JS、不取原始 DOM。""" roles = {} if isinstance(a11y, str): text = a11y for line in text.splitlines(): m = re.match(r"^\s*-\s+([A-Za-z][\w-]*)", line) if m: role = m.group(1).lower() roles[role] = roles.get(role, 0) + 1 elif isinstance(a11y, dict): text = json.dumps(a11y, ensure_ascii=False, sort_keys=True) def walk(n): if not isinstance(n, dict): return role = n.get("role") if isinstance(role, str) and role: key = role.lower() roles[key] = roles.get(key, 0) + 1 for value in n.values(): if isinstance(value, list): for child in value: walk(child) elif isinstance(value, dict): walk(value) walk(a11y) else: text = "" interactive = sum(count for role, count in roles.items() if role in INTERACTIVE_ROLES) digest = hashlib.sha256(text.encode()).hexdigest() return roles, interactive, len(text), digest def expand_bbox(x0, y0, x1, y1, width, height, margin=DIFF_MARGIN_PX, min_size=DIFF_MIN_SIZE_PX): """变化区 bbox 外扩 + 钳制 + 最小尺寸（纯函数，可单测）。 输入是像素差分的包围盒（左闭右开 [x0,y0,x1,y1)）。先外扩 margin，再钳制到 [0,0,width,height] 内；宽/高不足 min_size 时绕中心扩到 min_size 再钳制（图本身比 min_size 还小时取 整幅宽/高）。返回 Playwright clip 形 {"x","y","width", "height"}（均为 int）。""" x0 = max(0, x0 - margin) y0 = max(0, y0 - margin) x1 = min(width, x1 + margin) y1 = min(height, y1 + margin) # 宽/高不足 min_size 时绕中心扩；一侧贴边扩不动时向另一侧 # 补足（只有图本身比 min_size 还窄/矮时才取整幅宽/高）。 for axis in ("x", "y"): lo, hi, size = (x0, x1, width) if axis == "x" \ else (y0, y1, height) if hi - lo < min_size: need = min_size - (hi - lo) lo = max(0, lo - need // 2) hi = min(size, hi + (need - need // 2)) if hi - lo < min_size: if lo == 0: hi = min(size, lo + min_size) elif hi == size: lo = max(0, hi - min_size) if hi - lo < min_size: lo, hi = 0, size if axis == "x": x0, x1 = lo, hi else: y0, y1 = lo, hi return {"x": int(x0), "y": int(y0), "width": int(x1 - x0), "height": int(y1 - y0)} def _load_pillow(): """懒加载 Pillow（diff_shot 专用）；缺失时报明错并指引回落 全帧 screenshot，绝不静默假装裁剪。返回 (Image, ImageChops)。""" try: from PIL import Image, ImageChops except ImportError as e: raise RuntimeError( "diff_shot 需要 Pillow（%s）：镜像里 pip 未装 pillow，" "无法逐像素比对与裁剪；回落：用 screenshot 全帧 JPEG " "q70（op screenshot，不带 diff_only）" % e) return Image, ImageChops def compute_diff_bbox(prev_png, cur_png, margin=DIFF_MARGIN_PX, min_size=DIFF_MIN_SIZE_PX, pillow=None): """两帧 PNG 的逐像素比对（纯函数，可单测，需 Pillow）。 返回 (changed, bbox)：无变化时 (False, None)；有变化时 (True, clip dict)。像素判变：逐通道差值 > DIFF_PIXEL_THRESHOLD 才算变（见模块 docstring）。两帧尺寸不同时视为整帧变化， bbox 即当前帧全幅。pillow 供单测注入 (Image, ImageChops)， 正常运行时用 _load_pillow()。""" Image, ImageChops = pillow or _load_pillow() import io prev = Image.open(io.BytesIO(prev_png)).convert("RGB") cur = Image.open(io.BytesIO(cur_png)).convert("RGB") width, height = cur.size if prev.size != cur.size: return True, {"x": 0, "y": 0, "width": width, "height": height} diff = ImageChops.difference(prev, cur) # 逐通道阈值掩码：任一通道差值 >10 即变（convert("L") 会按 # 亮度加权、漏掉单通道小变化，故三通道分别取阈值再取并）。 masks = [] for channel in diff.split(): masks.append(channel.point( lambda p: 255 if p > DIFF_PIXEL_THRESHOLD else 0)) mask = masks[0] for extra in masks[1:]: mask = ImageChops.lighter(mask, extra) box = mask.getbbox() if box is None: return False, None return True, expand_bbox(box[0], box[1], box[2], box[3], width, height, margin, min_size) 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, full_browsers_dir=None, sessions_dir=None): self._factory = playwright_factory or _load_playwright self._shots_dir = shots_dir self._profile_explicit = bool(profile_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() self._sessions_dir = ( sessions_dir or resolve_sessions_dir( self._profile_dir, self._profile_on_data_disk)) self._sessions_explicit = bool(sessions_dir) self._full_browsers_explicit = bool(full_browsers_dir) self._full_browsers_dir = ( full_browsers_dir or resolve_full_browsers_dir( self._profile_on_data_disk)) 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 = {} # 阶段 5：具名会话的独立 context（session → context）与 # 每页所属 context 种类（"shared" 默认持久上下文 / # "isolated" 独立 context）；_storage_restored 记本次 # context 创建时是否从 storage-state 文件恢复过。 self._contexts = {} self._page_context = {} self._storage_restored = {} self._engine = "shell" self._mounted_full = False self._last_shot = {} # 阶段 4：上一帧 PNG（diff_shot 的比对基线）与截图计数器。 self._last_png = {} self._counters = {"shots_full": 0, "shots_diff": 0, "shots_verify": 0, "shots_skipped": 0} # -- 浏览器生命周期 ---------------------------------------------- def _close_context(self): # 阶段 5：先把各具名会话的 storage-state 落盘，再关它们 # 的独立 context——引擎切换/服务退出时态不丢，下次重开 # 同一会话 id 能恢复；关不掉的只跳过，不挡持久上下文关闭。 for session in list(self._contexts): self._save_storage_state(session) ctx = self._contexts.pop(session) try: ctx.close() except Exception: pass self._page_context = {} self._storage_restored = {} 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 _norm_session(self, session): """会话 id 归一化 + 校验：返回 (session, error)。 error 非空时调用方直接返回 {"ok": False, "error": error}。""" session = session or DEFAULT_SESSION ok, err = validate_session_id(session) if not ok: return None, err return session, "" def _session_storage_path(self, session): """具名会话的 storage-state 文件路径（阶段 5）。""" return os.path.join(self._sessions_dir, session, "storage-state.json") def _save_storage_state(self, session): """把具名会话的 context 状态落盘（阶段 5）。 成功返回 True；无独立 context 或写失败返回 False（调用 方只记录，不因此谎报 open/close 失败——存储失败由 storage_saved 字段如实报告）。""" ctx = self._contexts.get(session) if ctx is None: return False try: path = self._session_storage_path(session) os.makedirs(os.path.dirname(path), exist_ok=True) ctx.storage_state(path=path) except (OSError, ValueError, TypeError, AttributeError): return False except Exception: return False return True def _ensure_isolated_context(self, session): """具名会话的独立 context（阶段 5）：已建则复用；未建则 从持久上下文的 browser 上 new_context（同浏览器、独立 存储分区），storage-state 文件在位时恢复。返回 (context, restored, error)。""" ctx = self._contexts.get(session) if ctx is not None: return ctx, self._storage_restored.get(session, False), "" if self._context is None: return None, False, "浏览器尚未启动" browser = getattr(self._context, "browser", None) if callable(browser): browser = browser() if browser is None or not hasattr(browser, "new_context"): return None, False, ("当前浏览器不支持独立 context" "（无 browser.new_context）") kwargs = {"viewport": {"width": VIEWPORT_WIDTH, "height": VIEWPORT_HEIGHT}} state_path = self._session_storage_path(session) restored = os.path.isfile(state_path) if restored: kwargs["storage_state"] = state_path try: ctx = browser.new_context(**kwargs) except Exception as e: return None, False, "创建独立 context 失败：%s" % e self._contexts[session] = ctx self._storage_restored[session] = restored return ctx, restored, "" def _context_for(self, session): """会话对应的 context：default 走共享持久上下文，具名走 独立 context。返回 (context, kind, error)。""" if session == DEFAULT_SESSION: return self._context, "shared", "" ctx, restored, err = self._ensure_isolated_context(session) if err: return None, "isolated", err return ctx, "isolated", "" 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() if not self._profile_explicit: # 迟挂载自愈：首次启动（或旧上下文已关）时按当前挂载状态 # 重解析 profile——进程启动时数据盘若还没挂上，__init__ # 记的是回落路径；等挂盘完成后再 open 时应拿到数据盘 # profile，而不是把回落态锁死到服务重启。已跑起来的 # 上下文不受影响（上面已提前返回）。 self._profile_dir, self._profile_on_data_disk = \ resolve_profile_dir() if not self._sessions_explicit: # 阶段 5：sessions 目录跟 profile 走——迟挂载自愈 # 到数据盘时，storage-state 也落数据盘而非回落盘。 self._sessions_dir = resolve_sessions_dir( self._profile_dir, self._profile_on_data_disk) if not self._full_browsers_explicit: self._full_browsers_dir = resolve_full_browsers_dir( self._profile_on_data_disk) try: # profile 目录自己建好并收紧到 0700：不依赖 # chv-data-mount 的时序（它在 cloud-final 才跑， # 可能晚于浏览器首次启动），登录态目录绝不留 0755。 os.makedirs(self._profile_dir, exist_ok=True) os.chmod(self._profile_dir, 0o700) except OSError: pass # 只读/无权限时由 Playwright 启动报错，status 明说 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": # 完整浏览器在数据盘安装目录里：直接给可执行文件路径， # 不用 channel（channel 与 executable_path 互斥，且 # channel 要走 PLAYWRIGHT_BROWSERS_PATH——驱动进程 # 启动时才读 env，运行中改 env 对已起的驱动无效）。 exe = full_chromium_executable( self._full_browsers_dir) if exe is None: self._context = None return False, ("完整 Chromium 未安装（%s 下未见 " "chromium-*/chrome-linux/chrome）" % self._full_browsers_dir) kwargs["executable_path"] = exe 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, err = self._norm_session(session) if err: return {"ok": False, "error": err} ok, err = self._ensure_browser() if not ok: return {"ok": False, "error": err} ctx, kind, err = self._context_for(session) if err: return {"ok": False, "error": err} try: old = self._pages.get(session) if old is not None: try: old.close() except Exception: pass page = ctx.new_page() self._pages[session] = page self._page_context[session] = kind # 注意：open 不清 diff 基线——diff_shot 永远与“上一帧” # 逐像素比对，不管上一帧是哪一页；导航后调 diff_shot # 得到的正是新旧两页渲染的真实差异区（跨页局部变化 # 如搜索词 Cat->Dog 就靠这个）。只有 close（会话结束） # 才清基线，下次 open 后的首次 diff_shot 为全幅首帧。 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)} # 阶段 5：具名会话导航成功后立即把 storage-state 落盘—— # 导航里刚置的 cookie 当场进文件，不等 close；浏览器 # crash 后重开同一会话 id 仍能恢复（Chromium 刷盘竞态 # 绕过去了：storage_state() 读的是 context 的实时态）。 saved = False if kind == "isolated": saved = self._save_storage_state(session) resp = self.snapshot(session) if resp.get("ok"): resp["context"] = kind if kind == "isolated": resp["storage_saved"] = saved resp["storage_restored"] = \ self._storage_restored.get(session, False) return resp def snapshot(self, session=None): session, err = self._norm_session(session) if err: return {"ok": False, "error": err} 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} # 阶段 4 一级通道：纯字符串解析 enrich（不跑 JS、不取 DOM）。 roles, interactive, chars, digest = parse_a11y(tree) return {"ok": True, "title": title, "url": url, "a11y": tree, "truncated": truncated, "a11y_roles": roles, "a11y_interactive": interactive, "a11y_chars": chars, "a11y_hash": digest, "session": session, "context": self._page_context.get(session, "shared")} def screenshot(self, session=None, diff_only=False): session, err = self._norm_session(session) if err: return {"ok": False, "error": err} 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 的变化区 # 裁剪走 diff_shot op，这里只做整帧相同跳过）。 self._counters["shots_skipped"] += 1 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 self._counters["shots_full"] += 1 return {"ok": True, "path": path, "size": len(data), "format": "jpeg", "quality": JPEG_QUALITY, "width": VIEWPORT_WIDTH, "changed": True, "sha256": digest, "session": session} def diff_shot(self, session=None): """二级差分截图（阶段 4）：PNG 全帧逐像素比对，只截变化区。 无变化：{changed:false, size:0}（计 shots_skipped）。 有变化：clip=bbox 的 JPEG q70 落盘，返回 {changed:true, bbox, size, full_size, path, sha256}（计 shots_diff）。 首帧（本会话无上一帧）视为整帧变化，bbox 为全视口。 Pillow 缺失时 ok:false 并指引回落 screenshot 全帧， 不计入计数器、不落盘、不假装裁剪。""" session, err = self._norm_session(session) if err: return {"ok": False, "error": err} page = self._page(session) if page is None: return {"ok": False, "error": "会话 %s 还没有打开的页面（先 open）" % session} try: png = page.screenshot(type="png") except Exception as e: return {"ok": False, "error": "截图失败：%s" % e} if not isinstance(png, (bytes, bytearray)): return {"ok": False, "error": "截图未返回字节数据"} png = bytes(png) try: Image, _ImageChops = _load_pillow() except RuntimeError as e: return {"ok": False, "error": str(e), "fallback": "screenshot"} prev = self._last_png.get(session) if prev is None: changed, bbox = True, {"x": 0, "y": 0, "width": VIEWPORT_WIDTH, "height": VIEWPORT_HEIGHT} else: try: changed, bbox = compute_diff_bbox(prev, png) except Exception as e: return {"ok": False, "error": "差分比对失败：%s" % e} # full_size：本帧 PNG 经 Pillow 以 JPEG q70 编码的全帧 # 字节数（与浏览器直出同参数，仅作对照，不直接等同）。 import io try: buf = io.BytesIO() Image.open(io.BytesIO(png)).convert("RGB").save( buf, "JPEG", quality=JPEG_QUALITY) full_size = len(buf.getvalue()) except Exception: full_size = len(png) self._last_png[session] = png if not changed: self._counters["shots_skipped"] += 1 return {"ok": True, "changed": False, "size": 0, "path": None, "bbox": None, "full_size": full_size, "format": "jpeg", "quality": JPEG_QUALITY, "session": session} try: clip = page.screenshot(type="jpeg", quality=JPEG_QUALITY, clip=bbox) except Exception as e: return {"ok": False, "error": "变化区截图失败：%s" % e} if not isinstance(clip, (bytes, bytearray)): return {"ok": False, "error": "变化区截图未返回字节数据"} clip = bytes(clip) 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(clip) except OSError as e: return {"ok": False, "error": "截图落盘失败：%s" % e} self._counters["shots_diff"] += 1 return {"ok": True, "changed": True, "bbox": bbox, "path": path, "size": len(clip), "full_size": full_size, "format": "jpeg", "quality": JPEG_QUALITY, "sha256": hashlib.sha256(clip).hexdigest(), "session": session} def verify_shot(self, session=None): """三级全屏校验（阶段 4）：全帧 JPEG q70，只用于检查点。 计 shots_verify。日常动作后用 diff_shot，检查点才调它。""" session, err = self._norm_session(session) if err: return {"ok": False, "error": err} 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) 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] = hashlib.sha256(data).hexdigest() self._counters["shots_verify"] += 1 return {"ok": True, "path": path, "size": len(data), "format": "jpeg", "quality": JPEG_QUALITY, "width": VIEWPORT_WIDTH, "changed": True, "sha256": self._last_shot[session], "session": session} def close(self, session=None): session, err = self._norm_session(session) if err: return {"ok": False, "error": err} page = self._pages.pop(session, None) if page is not None: try: page.close() except Exception: pass kind = self._page_context.pop(session, "shared") saved = False if kind == "isolated": # 阶段 5：关会话先落盘 storage-state，再关独立 # context——态进文件，下次重开同一会话 id 恢复； # 只动本会话的 context，其它会话的页与 context # 原样不动。 saved = self._save_storage_state(session) ctx = self._contexts.pop(session, None) if ctx is not None: try: ctx.close() except Exception: pass self._storage_restored.pop(session, None) self._last_shot.pop(session, None) self._last_png.pop(session, None) resp = {"ok": True, "closed": True, "session": session, "context": kind} if kind == "isolated": resp["storage_saved"] = saved return resp 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, "context": self._page_context.get( sid, "shared")}) return {"ok": True, "sessions": out} # -- 服务化：状态 / CDP / 引擎 -------------------------------------- def status(self): total = (self._counters["shots_full"] + self._counters["shots_diff"] + self._counters["shots_verify"]) verify_share = (self._counters["shots_verify"] / total if total else 0.0) 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"]], "isolated_sessions": sorted(self._contexts), "sessions_dir": self._sessions_dir, "mounted_full": self._mounted_full, "full_present": full_chromium_present( self._full_browsers_dir), "full_browsers_dir": self._full_browsers_dir, "counters": dict(self._counters), "verify_share": verify_share} 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( self._full_browsers_dir): return {"ok": False, "error": "完整 Chromium 未安装（%s 下未见 " "chromium-*/chrome-linux/chrome），回落 " "headless-shell；先跑 PLAYWRIGHT_BROWSERS_" "PATH=%s python3 -m playwright install " "chromium 安装（装数据盘，根盘空间不够）" % (self._full_browsers_dir, self._full_browsers_dir), "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(self._full_browsers_dir): return {"ok": False, "error": "完整 Chromium 未安装，无法挂载（回落 " "headless-shell 继续工作）；先跑 " "PLAYWRIGHT_BROWSERS_PATH=%s python3 -m " "playwright install chromium 安装（装数据" "盘，根盘空间不够）" % self._full_browsers_dir, "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 == "diff_shot": return agent.diff_shot(req.get("session")) if op == "verify_shot": return agent.verify_shot(req.get("session")) 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/" "diff_shot/verify_shot/" "close/sessions/status/cdp/engine/mount_full/" "unmount_full/ping）" % (op,)} def _remove_stale_socket(sock_path): """删掉上次残留的 socket。删不掉时不静默吞掉：目录权限不对 （/run/chv 须是 0775 root:chv-browser）就会无限 crash-loop， 报错必须说清修法（2026-11-08 构建 #8 live 实证：cell 曾把组 改回 root，服务 62 次重启才被发现）。""" if not os.path.exists(sock_path): return try: os.remove(sock_path) except PermissionError: raise PermissionError( "删不掉旧 socket %s：/run/chv 须是 0775 root:chv-browser " "（tmpfiles.d/chv.conf），且跑 cell 不可把它 chown 回 " "root:root" % sock_path) def serve(sock_path=BROWSER_SOCK, shots_dir=SHOTS_DIR, profile_dir=None, cdp_port=None): """监听 Unix socket，逐连接处理。socket 0600（只给属主 + broker 的 root 连）。建不起来即抛错，不要静默带病跑。""" _remove_stale_socket(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:]))