onesvm-browser-server/bench/searxng-cn/eval_t1.py

#!/usr/bin/env python3
"""Assert one SearXNG T1 JSON body. stdout: JSON verdict."""
from __future__ import annotations

import json
import re
import sys
from urllib.parse import urlparse

POLICY = re.compile(
    r"(gov\.cn|customs\.gov\.cn|chinatax\.gov\.cn|mofcom\.gov\.cn|"
    r"yidaiyilu\.gov\.cn|news\.cn|people\.com\.cn|xinhuanet\.com)$",
    re.I,
)
T2_PREF = re.compile(r"(gov\.cn|news\.cn|people\.com\.cn)$", re.I)


def first_cjk4(text: str) -> str:
    chars = re.findall(r"[\u4e00-\u9fff]", text or "")
    return "".join(chars[:4])


def eval_body(raw: str) -> dict:
    out = {
        "assert_ok": False,
        "fail_class": "",
        "note": "",
        "n_results": 0,
        "n_rich": 0,
        "engines": [],
        "engine_counts": {},
        "unresponsive_engines": [],
        "policy_hosts": [],
        "hosts": [],
        "t2_url": "",
        "t2_title": "",
        "t2_title_cjk4": "",
        "sample_titles": [],
    }
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        out["fail_class"] = "json_403"
        out["note"] = "body_not_json"
        return out
    if isinstance(data, dict) and data.get("error"):
        out["fail_class"] = "json_403"
        out["note"] = str(data.get("error"))[:200]
        return out
    results = data.get("results") if isinstance(data, dict) else None
    if not isinstance(results, list):
        out["fail_class"] = "json_403"
        out["note"] = "no_results_array"
        return out
    out["n_results"] = len(results)
    engines = []
    counts = {}
    hosts = []
    policy = []
    rich = 0
    t2_url = ""
    t2_title = ""
    for i, item in enumerate(results):
        if not isinstance(item, dict):
            continue
        title = (item.get("title") or "").strip()
        url = (item.get("url") or "").strip()
        content = (item.get("content") or "").strip()
        eng = item.get("engine") or ""
        if isinstance(item.get("engines"), list) and not eng:
            eng = item["engines"][0] if item["engines"] else ""
        if eng:
            engines.append(str(eng))
            counts[str(eng)] = counts.get(str(eng), 0) + 1
        host = urlparse(url).hostname or ""
        if host:
            hosts.append(host.lower())
        if POLICY.search(host.lower()):
            policy.append(host.lower())
        if title and url.startswith("http") and len(content) >= 8:
            rich += 1
        if i < 3:
            out["sample_titles"].append({"title": title[:120], "url": url[:200], "engine": eng, "host": host})
        if not t2_url and url.startswith("http") and T2_PREF.search(host.lower()):
            t2_url = url
            t2_title = title
    out["n_rich"] = rich
    out["engines"] = sorted(set(engines))
    out["engine_counts"] = counts
    out["hosts"] = hosts[:20]
    out["policy_hosts"] = sorted(set(policy))
    out["t2_url"] = t2_url
    out["t2_title"] = t2_title
    out["t2_title_cjk4"] = first_cjk4(t2_title)
    unresp = data.get("unresponsive_engines") or []
    out["unresponsive_engines"] = unresp
    notes = []
    if unresp:
        notes.append(f"unresponsive={unresp}")

    ok = True
    if out["n_results"] < 5:
        ok = False
        out["fail_class"] = "engine_outage"
        notes.append("results<5")
    if rich < 5:
        # first 10 rich count — we counted all; restrict to first 10
        rich10 = 0
        for item in results[:10]:
            if not isinstance(item, dict):
                continue
            title = (item.get("title") or "").strip()
            url = (item.get("url") or "").strip()
            content = (item.get("content") or "").strip()
            if title and url.startswith("http") and len(content) >= 8:
                rich10 += 1
        out["n_rich"] = rich10
        if rich10 < 5:
            ok = False
            out["fail_class"] = out["fail_class"] or "thin_snippets"
            notes.append("rich<5_in_top10")
    if len(out["engines"]) < 2:
        # not a hard fail per select.md: 单引擎全灭记 engine_outage，不直接判方案失败
        # but plan T1 assert 4 says engines >= 2
        ok = False
        out["fail_class"] = out["fail_class"] or "engine_outage"
        notes.append("engines<2")
    if not out["policy_hosts"]:
        ok = False
        notes.append("no_policy_host")
        # plan: assert_ok=false but keep host list
    out["assert_ok"] = ok
    out["note"] = "; ".join(notes)
    return out


def main() -> None:
    path = sys.argv[1]
    raw = open(path, "r", encoding="utf-8", errors="replace").read()
    print(json.dumps(eval_body(raw), ensure_ascii=False))


if __name__ == "__main__":
    main()