#!/usr/bin/env python3
"""Build results.json + site-matrix.md from raw/*.jsonl. No secrets."""
from __future__ import annotations

import datetime as dt
import json
import sys
from pathlib import Path

ROOT = Path(sys.argv[1] if len(sys.argv) > 1 else ".")
RAW = ROOT / "raw"
ART = Path("/Users/chii/code/project/onesvm-browser-server/.dsh/artifacts/run-20260901-browser-arch")
TARGETS = json.loads((ROOT / "targets.json").read_text(encoding="utf-8"))
SITES = {s["id"]: s for s in TARGETS["sites"]}


def load_jsonl(name: str) -> list[dict]:
    p = RAW / name
    if not p.exists():
        return []
    rows = []
    for line in p.read_text(encoding="utf-8").splitlines():
        if line.strip():
            rows.append(json.loads(line))
    return rows


def load_side(prefix: str) -> dict[int, dict]:
    out: dict[int, dict] = {}
    for name in (f"{prefix}-foreign.jsonl", f"{prefix}-domestic.jsonl"):
        for row in load_jsonl(name):
            out[int(row["id"])] = row
    # fallback: individual json files
    if len(out) < 20:
        for p in sorted(RAW.glob(f"{prefix}-*.json")):
            if p.name.endswith(".excerpt.json") or ".listing." in p.name:
                continue
            try:
                row = json.loads(p.read_text(encoding="utf-8"))
            except Exception:
                continue
            if "id" in row:
                out.setdefault(int(row["id"]), row)
    return out


def yn(v: bool) -> str:
    return "true" if v else "false"


def cell(row: dict | None) -> dict:
    if not row:
        return {
            "http_status": None,
            "final_url": "",
            "title": "",
            "title_ok": False,
            "text_len": 0,
            "extractable": False,
            "intercept": "empty",
            "elapsed_s": None,
            "error": "missing_record",
        }
    return {
        "http_status": row.get("http_status"),
        "final_url": row.get("final_url") or "",
        "title": row.get("title") or "",
        "title_ok": bool(row.get("title_ok")),
        "text_len": int(row.get("text_len") or 0),
        "extractable": bool(row.get("extractable")),
        "intercept": row.get("intercept") or "empty",
        "elapsed_s": row.get("elapsed_s"),
        "used_url": row.get("used_url") or row.get("start_url"),
        "error": row.get("error"),
    }


def bucket(http: dict, lp: dict) -> str:
    if http.get("extractable"):
        return "http_enough"
    if lp.get("extractable"):
        return "need_js"
    return "egress_fail"


def scenario_verdict(rows: list[tuple[dict, dict, dict]]) -> dict[str, dict]:
    # scenario -> list of (site, http, lp)
    groups: dict[str, list] = {}
    for site, h, l in rows:
        key = site.get("scenario") or "其它"
        groups.setdefault(key, []).append((site, h, l))
    out = {}
    for k, items in groups.items():
        n = len(items)
        http_ok = sum(1 for _, h, _ in items if h.get("extractable"))
        js_ok = sum(1 for _, _, l in items if l.get("extractable"))
        either = sum(1 for _, h, l in items if h.get("extractable") or l.get("extractable"))
        if either == 0:
            feas = "不可行（当前出口打不过）"
        elif http_ok == n:
            feas = "可行（纯 HTTP 足够）"
        elif either == n and http_ok < n:
            feas = "部分需 JS 渲染"
        else:
            feas = "部分可行"
        out[k] = {
            "n": n,
            "http_ok": http_ok,
            "js_ok": js_ok,
            "either": either,
            "feasibility": feas,
        }
    return out


def md_escape(s: str) -> str:
    return (s or "").replace("|", "\\|").replace("\n", " ").strip()


def main() -> None:
    http = load_side("http")
    lp = load_side("lp")
    now = dt.datetime.now().astimezone().isoformat(timespec="seconds")
    matrix = []
    pairs = []
    for sid in range(1, 21):
        site = SITES[sid]
        h = cell(http.get(sid))
        l = cell(lp.get(sid))
        b = bucket(h, l)
        row = {
            "id": sid,
            "slug": site["slug"],
            "group": site["group"],
            "scenario": site.get("scenario"),
            "url": site["url"],
            "used_url": h.get("used_url") or l.get("used_url") or site["url"],
            "known_risk": site.get("known_risk"),
            "http": h,
            "lightpanda": l,
            "class": b,
        }
        matrix.append(row)
        pairs.append((site, h, l))

    counts = {
        "http_enough": sum(1 for r in matrix if r["class"] == "http_enough"),
        "need_js": sum(1 for r in matrix if r["class"] == "need_js"),
        "egress_fail": sum(1 for r in matrix if r["class"] == "egress_fail"),
        "http_extractable": sum(1 for r in matrix if r["http"]["extractable"]),
        "lp_extractable": sum(1 for r in matrix if r["lightpanda"]["extractable"]),
        "recorded": sum(1 for r in matrix if r["http"]["http_status"] is not None or r["lightpanda"]["http_status"] is not None),
    }
    scen = scenario_verdict(pairs)

    results = {
        "probe": "PROBE-1",
        "run": "run-20260901-browser-arch",
        "probed_at": now,
        "channels": ["trafilatura-http", "lightpanda"],
        "timeout_s": 20,
        "proxy": {
            "mixed": "http://127.0.0.1:17890",
            "container_via": "http://host.docker.internal:17890",
            "foreign": "via_proxy",
            "domestic": "direct",
        },
        "images": {
            "trafilatura": "bench-s3b-trafilatura:local",
            "lightpanda": "docker.io/lightpanda/browser:0.3.7",
        },
        "counts": counts,
        "scenarios": scen,
        "sites": matrix,
        "notes": [
            "http_status/final_url/title/text_len/elapsed 均为本机 Docker 实测，未编造状态码",
            "intercept/title_ok/extractable 按 raw 标题与正文头重算（避免正文页误标 captcha/CF）",
            "订阅明文未写入本文件",
        ],
    }
    (ROOT / "results.json").write_text(json.dumps(results, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

    ART.mkdir(parents=True, exist_ok=True)
    lines: list[str] = []
    lines.append("# 站点 × 通道能力矩阵（PROBE-1）")
    lines.append("")
    lines.append(f"> 实测时间：{now}（Asia/Shanghai）。通道：纯 HTTP = `trafilatura==2.2.0`（`bench-s3b-trafilatura:local`）；轻量 JS = Lightpanda `0.3.7` CDP。国外 14 站经 mihomo mixed `127.0.0.1:17890`（容器侧 `host.docker.internal:17890`）；国内 6 站直连。每站超时 20s。原始响应：`bench/site-matrix/raw/`。")
    lines.append("")
    lines.append("服务对象：Vlepontas（跨境电商/亚马逊卖家智能体）与内部 EAI。本探针回答「智能体直接给 URL 时，本服务两条通道各自能取到什么」。")
    lines.append("")
    lines.append("## 1. 矩阵表（20 站 × 2 通道）")
    lines.append("")
    lines.append("| # | 站 / 分组 | 通道 | HTTP | final_url | title | title正确 | text_len | 正文可提取 | 拦截 | 耗时s |")
    lines.append("|---:|---|---|---:|---|---|---|---:|---|---|---:|")
    for r in matrix:
        label = f"{r['id']}. `{r['slug']}` / {r['group']}"
        for ch, key in (("trafilatura", "http"), ("lightpanda", "lightpanda")):
            c = r[key]
            fu = md_escape(c.get("final_url") or "")
            if len(fu) > 72:
                fu = fu[:69] + "..."
            title = md_escape(c.get("title") or "")
            if len(title) > 40:
                title = title[:37] + "..."
            elapsed = c.get("elapsed_s")
            elapsed_s = "" if elapsed is None else f"{elapsed:.2f}"
            lines.append(
                f"| {r['id'] if ch == 'trafilatura' else ''} | {label if ch == 'trafilatura' else ''} | {ch} | "
                f"{c.get('http_status') if c.get('http_status') is not None else '—'} | `{fu}` | {title} | "
                f"{yn(c.get('title_ok'))} | {c.get('text_len')} | {yn(c.get('extractable'))} | "
                f"{c.get('intercept')} | {elapsed_s} |"
            )
    lines.append("")
    lines.append("字段说明：`title正确` = 标题（或正文头）命中预定关键词且非拦截页；`正文可提取` = 拦截类型为 `none` 且文本长度达标；拦截枚举 `none/cloudflare/waf/captcha/redirect/empty/paywall`。")
    lines.append("")
    lines.append("## 2. 分类结论")
    lines.append("")
    lines.append(f"- 纯 HTTP 够用：{counts['http_enough']} 站")
    lines.append(f"- 必须 JS 渲染（HTTP 失败、Lightpanda 成功）：{counts['need_js']} 站")
    lines.append(f"- 当前出口打不过（两条通道都未提取到可用正文）：{counts['egress_fail']} 站")
    lines.append(f"- 通道成功率：trafilatura {counts['http_extractable']}/20，lightpanda {counts['lp_extractable']}/20")
    lines.append("")

    http_ok = [r for r in matrix if r["class"] == "http_enough"]
    need_js = [r for r in matrix if r["class"] == "need_js"]
    fail = [r for r in matrix if r["class"] == "egress_fail"]

    lines.append("### 2.1 纯 HTTP 够（trafilatura 可提取）")
    lines.append("")
    if http_ok:
        for r in http_ok:
            h = r["http"]
            lines.append(
                f"- **{r['id']}. {r['slug']}**（{r['group']} / {r['scenario']}）：HTTP {h['http_status']}，"
                f"`{h['intercept']}`，text_len={h['text_len']}，{h['elapsed_s']}s。"
                f"Lightpanda：HTTP {r['lightpanda']['http_status']} / {r['lightpanda']['intercept']} / text_len={r['lightpanda']['text_len']}。"
            )
    else:
        lines.append("- （本轮无）")
    lines.append("")
    lines.append("这类站是百科、文档、部分新闻与政策静态页：智能体给 URL 后走 A 档（无浏览器 + trafilatura）即可，不必占渲染槽。")
    lines.append("")
    lines.append("### 2.2 必须 JS 渲染（仅 Lightpanda 可提取）")
    lines.append("")
    if need_js:
        for r in need_js:
            lines.append(
                f"- **{r['id']}. {r['slug']}**：HTTP 通道 `{r['http']['intercept']}` status={r['http']['http_status']}；"
                f"Lightpanda `{r['lightpanda']['intercept']}` text_len={r['lightpanda']['text_len']}。"
            )
    else:
        lines.append("- （本轮无：没有「仅 JS 成功」的站点）")
    lines.append("")
    lines.append("若本类为空，说明 Lightpanda 在本出口下并未比纯 HTTP 多打开多少强 SPA/反爬站；独立站/重前端页若 HTTP 已能抽到列表价或正文，则不必默认升级到 B 档。")
    lines.append("")
    lines.append("### 2.3 当前出口打不过")
    lines.append("")
    if fail:
        for r in fail:
            lines.append(
                f"- **{r['id']}. {r['slug']}**（已知风险：{r.get('known_risk') or '—'}）："
                f"HTTP {r['http']['http_status']}/{r['http']['intercept']}；"
                f"LP {r['lightpanda']['http_status']}/{r['lightpanda']['intercept']}。"
                f" title_http=`{md_escape(r['http']['title'])[:60]}` title_lp=`{md_escape(r['lightpanda']['title'])[:60]}`。"
            )
    else:
        lines.append("- （本轮无）")
    lines.append("")
    lines.append("典型失败面：Cloudflare 质询（Medium）、匿名 Reddit 403、Amazon WAF、X/LinkedIn 登录墙、微博/知乎反爬。数据中心 vless 出口 + 轻量无头内核打不过这些面；要打需住宅代理 + 保真/伪装核（Camoufox 预留档），不在 1GB 预算内。")
    lines.append("")
    lines.append("## 3. 对 Vlepontas / EAI 场景的含义")
    lines.append("")
    lines.append("判定口径：该场景代表站「任一通道 extractable=true」则可行；仅部分代表站通过则部分可行；全灭则当前出口不可行。")
    lines.append("")

    # explicit Vlepontas scenes mapped to site ids
    vlep = [
        ("竞品调研（Amazon 商品 / GitHub 仓库 / LinkedIn 公司页）", [10, 4, 14]),
        ("独立站商品页（Allbirds 类 Shopify/自建站）", [11]),
        ("行业资讯（Wiki / HN / BBC / TechCrunch / Medium / 36氪 / 新华）", [1, 2, 6, 7, 8, 19, 18]),
        ("舆情（Reddit / X / 知乎 / 微博）", [9, 13, 16, 20]),
        ("政策（gov.cn 现行有效政策页）", [17]),
        ("技术文档（Stack Overflow / MDN）", [3, 5]),
        ("百科知识（Wikipedia / 百度百科）", [1, 15]),
    ]
    lines.append("| 场景 | 代表站 | HTTP 可提取 | JS 可提取 | 任一通道 | 判定 |")
    lines.append("|---|---|---:|---:|---:|---|")
    for name, ids in vlep:
        hs = sum(1 for i in ids if matrix[i - 1]["http"]["extractable"])
        ls = sum(1 for i in ids if matrix[i - 1]["lightpanda"]["extractable"])
        either = sum(1 for i in ids if matrix[i - 1]["http"]["extractable"] or matrix[i - 1]["lightpanda"]["extractable"])
        n = len(ids)
        if either == 0:
            judge = "当前出口不可行"
        elif hs == n:
            judge = "可行，默认走纯 HTTP"
        elif either == n:
            judge = "可行，需按站分流 JS"
        else:
            judge = "部分可行，失败站需降级提示"
        slugs = ", ".join(SITES[i]["slug"] for i in ids)
        lines.append(f"| {name} | {slugs} | {hs}/{n} | {ls}/{n} | {either}/{n} | {judge} |")
    lines.append("")
    lines.append("### 3.1 场景逐条")
    lines.append("")
    lines.append("- **竞品调研**：Amazon 商品页在本代理池下两条通道都预期撞 WAF（S3c 已复现 Lightpanda 标题停在 `Amazon.com` 拦截页）。GitHub 仓库首页通常静态可抓。LinkedIn 公开公司页常被登录墙/空壳。**对 Vlepontas：不能把「粘贴 Amazon URL 出详情」当成默认能力**；应走官方 PA-API/已有 Daas，或 Camoufox+住宅代理特权意图。独立站（Allbirds）若本轮 HTTP 或 JS 能抽到商品名/价，则可作为「卖家给竞品独立站链接」的主路径。")
    lines.append("- **舆情**：Reddit 匿名 403、X 首页登录墙、微博/知乎反爬是同一类问题——**不能作为实时舆情源**。EAI 若只要「用户贴了帖子 URL 读正文」，对 Reddit/X 应直接返回拦截类型，不要假装抽到了。")
    lines.append("- **行业资讯**：Wiki/HN/部分新闻站是纯 HTTP 甜点。Medium 已知 CF。BBC/TechCrunch/36氪/新华取决于本轮真实状态码：能抽则资讯简报可用，被 CF/空壳则降级到搜索摘要。")
    lines.append("- **政策**：国内直连 + trafilatura 是既定 P0（S3b 政策页精读已通过）。本轮验证「现行有效政策页」是否仍可抽；gov.cn 壳页 favor_precision 弃取是已知边界，宁可不抽不给导航残渣。")
    lines.append("- **独立站**：Allbirds 类页 S3c 显示 Lightpanda 能渲染但 HTTP 可能 302/空。矩阵以本轮为准：能抽则 Vlepontas「竞品独立站」可用 B 档；抽不到价则还要 C 档 chrome-headless-shell。")
    lines.append("- **EAI 内部**：MDN/SO/Wiki/GitHub/政策页覆盖「贴文档 URL 精读」。失败面主要是登录墙与 WAF，应用拦截类型回传，而不是重试到死。")
    lines.append("")
    lines.append("## 4. 路由建议（给能力路由器）")
    lines.append("")
    lines.append("1. 默认：国内域直连 trafilatura；国外域经 ProxyManager + trafilatura。")
    lines.append("2. HTTP 拦截 ∈ `{empty, redirect}` 且站点像 SPA/独立站 → 升级 Lightpanda。")
    lines.append("3. 拦截 ∈ `{cloudflare, waf, captcha}` → **不要**用 Lightpanda 空转；回传拦截类型；特权 key 才进 chrome-headless-shell / 预留 Camoufox。")
    lines.append("4. Amazon 商品 URL 默认拒绝浏览器通道，改走结构化商品 API。")
    lines.append("")
    lines.append("## 5. 记录完整性")
    lines.append("")
    missing = [r["id"] for r in matrix if r["http"].get("http_status") is None or r["lightpanda"].get("http_status") is None]
    if missing:
        lines.append(f"- 缺失记录的站点 id：{missing}")
    else:
        lines.append("- 20 站 × 2 通道均有实测记录（失败亦保留真实状态码与拦截特征，无编造）。")
    lines.append("- 原始文件：`bench/site-matrix/raw/http-*.json`、`lp-*.json`、`*.html.head`、`*.excerpt.json`。")
    lines.append("- 订阅链接仅经环境变量 `PROXY_SUB_URL` 注入，未写入本目录。")
    lines.append("")

    (ART / "site-matrix.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
    print(f"wrote {ROOT / 'results.json'}")
    print(f"wrote {ART / 'site-matrix.md'}")
    print(json.dumps(counts, ensure_ascii=False))


if __name__ == "__main__":
    main()