contentm_agent/执行会话/目标-根据1a需求文档生成MCN短视频整合营销-d85687/ui/_tools/cap.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""取证小工具（③段 3b/3d 用）。

  --mode dom  : 取渲染后的 DOM（自检输出、数值取证都从这里读）
  --mode shot : 截图（视口帧）

用法示例：
  python _tools/cap.py --mode dom  --route selftest --out _scratch/selftest.html
  python _tools/cap.py --mode shot --route p/p1     --viewport 1440x900 --out _gate_shots/P1-1440.png

🔴 2026-10-08 改：**连已有浏览器实例，不再自己起 headless Chrome**。
   旧写法是 `chrome.exe --headless=new --dump-dom/--screenshot`，两个后果：
     ① 违反本机明令（见 `skills/browser-harness/SKILL.md` 开头「先探测、能复用就复用」）；
     ② 只能靠 `--window-size` 猜视口 —— 而且**两条通道对同一个 window-size 的解释不同**
        （dom 通道会扣掉 16 / 95，截图通道不扣），还得在调用侧补偏移量；
        更糟的是它有**视口宽度下限 500**（Chrome 窗口最小 516），390px 根本量不了。
   现在改用 CDP 的 `Emulation.setDeviceMetricsOverride`：视口是**声明式的，要多少是多少**，
   上下限与偏移量的坑一并消失（390px 现在能量了）。
   实例地址默认 `http://127.0.0.1:9223`（可用环境变量 `BU_CDP_URL` 覆盖）。
   ⚠️ 实例没起来时**直接报错退出**，⛔ 不自动拉起浏览器 —— 拉起谁、拉几个，那是人的决定。
"""
import argparse, os, socket, subprocess, sys, tempfile
from urllib.parse import quote, urlparse

HERE = os.path.dirname(os.path.abspath(__file__))
PROTO = os.path.abspath(os.path.join(HERE, "..", "mcn-workbench.html"))
HARNESS = os.path.expanduser("~/.local/bin/browser-harness.exe")
BU_URL = os.environ.get("BU_CDP_URL", "http://127.0.0.1:9223")

SCRIPT = r"""
import io, json
BU = {bu!r}
url = {url!r}
out = {out!r}
mode = {mode!r}
W, H = {w}, {h}
t = new_tab(url)
cdp("Emulation.setDeviceMetricsOverride", width=W, height=H, deviceScaleFactor=1, mobile=False)
wait_for_load()
wait(0.8)
if mode == 'dom':
    vw = js("innerWidth"); vh = js("innerHeight")
    assert (vw, vh) == (W, H), "视口不符：要 %dx%d，实到 %dx%d" % (W, H, vw, vh)
    html = js("document.documentElement.outerHTML")
    io.open(out, 'w', encoding='utf-8').write("<!DOCTYPE html>\n" + html)
    print("VIEWPORT %dx%d" % (vw, vh))
else:
    vw = js("innerWidth"); vh = js("innerHeight")
    assert (vw, vh) == (W, H), "视口不符：要 %dx%d，实到 %dx%d" % (W, H, vw, vh)
    capture_screenshot(path=out, full=False)
    print("VIEWPORT %dx%d" % (vw, vh))
cdp("Emulation.clearDeviceMetricsOverride")
close_tab()
"""


def _probe(host="127.0.0.1", port=9223):
    s = socket.socket()
    s.settimeout(1.5)
    try:
        s.connect((host, port))
        return True
    except OSError:
        return False
    finally:
        s.close()


def url(route, proto=None):
    p = (proto or PROTO).replace("\\", "/")
    h = route.strip().lstrip("#").lstrip("/")
    return "file:///" + quote(p, safe="/:") + ("#/" + h if h else "")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--mode", choices=["dom", "shot"], required=True)
    ap.add_argument("--route", default="p/p1/ok", help="例：selftest ／ p/p4/empty（⛔ 不带 #/）")
    ap.add_argument("--hash", default=None, help="兼容旧参数（不带 #/）")
    ap.add_argument("--viewport", default="1440x900", help="目标 CSS 视口，例 1440x900 / 500x844 / 390x844")
    ap.add_argument("--proto", default=None, help="要渲染的文件（默认本目录的 mcn-workbench.html；变异对照用）")
    ap.add_argument("--out", required=True)
    a = ap.parse_args()
    route = a.hash or a.route

    vw, vh = [int(x) for x in a.viewport.lower().replace(",", "x").split("x")]
    out = os.path.abspath(a.out)
    os.makedirs(os.path.dirname(out), exist_ok=True)

    host = urlparse(BU_URL).hostname or "127.0.0.1"
    port = urlparse(BU_URL).port or 9223
    if not _probe(host, port):
        sys.stderr.write("⛔ %s 上没有浏览器实例在跑。\n"
                         "   本工具**只连不起**（本机明令）。先把常驻实例起来，或者\n"
                         "   用 browser-harness 自带的 daemon 流程，再重跑本命令。\n" % BU_URL)
        return 2

    script = SCRIPT.format(bu=BU_URL, url=url(route, a.proto), out=out, mode=a.mode, w=vw, h=vh)
    tf = tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8")
    tf.write(script)
    tf.close()

    env = dict(os.environ)
    for k in ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"):
        env.pop(k, None)
    env["NO_PROXY"] = "127.0.0.1,localhost"
    env["BU_CDP_URL"] = BU_URL

    r = subprocess.run([HARNESS], stdin=None, capture_output=True, env=env,
                       input=open(tf.name, "rb").read())
    os.unlink(tf.name)
    err = r.stderr.decode("utf-8", "replace")
    ok = r.stdout.decode("utf-8", "replace")

    if a.mode == "dom":
        sys.stderr.write("dom → %s ｜ %s ｜ bytes=%d\n"
                         % (out, ok.strip() or "（没说视口）", os.path.getsize(out) if os.path.exists(out) else -1))
    else:
        # 🔴 截图出来必须**当面对尺寸**：这条通道曾经因为偏移量算错，把 1456×995 当成
        #    「1440×900」交付出去（同一份报告里 dump-dom 又按 640 说话 ⇒ 两份证据互相打架）。
        #    现在直接从 PNG 的 IHDR 里读真实像素，跟目标视口比，不等就报红。
        if not os.path.exists(out):
            sys.stderr.write("shot FAILED\n")
        else:
            with open(out, "rb") as f:
                head = f.read(24)
            pw = int.from_bytes(head[16:20], "big")
            ph = int.from_bytes(head[20:24], "big")
            if (pw, ph) == (vw, vh):
                sys.stderr.write("shot ok: %s（视口 %dx%d，PNG 实测 %dx%d ✓）\n" % (out, vw, vh, pw, ph))
            else:
                sys.stderr.write("shot ⛔ 尺寸不符：目标视口 %dx%d，PNG 实际 %dx%d ⇒ 别拿它当 %dx%d 的证据\n"
                                 % (vw, vh, pw, ph, vw, vh))
    if "Traceback (most recent call last)" in err:
        sys.stderr.write(err[-1500:] + "\n")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())