onesvm-browser-server/bench/site-matrix/reclassify.py

#!/usr/bin/env python3
"""Recompute intercept/title_ok/extractable from recorded fields. No re-fetch, no secrets."""
from __future__ import annotations

import json
from pathlib import Path

from classify import BLOCK_TITLE
from urllib.parse import urlparse

ROOT = Path(__file__).resolve().parent
RAW = ROOT / "raw"
TARGETS = json.loads((ROOT / "targets.json").read_text(encoding="utf-8"))
SITES = {s["slug"]: s for s in TARGETS["sites"]}


def classify(row: dict) -> str:
    title = (row.get("title") or "").strip()
    text = row.get("text_head") or ""
    text_len = int(row.get("text_len") or 0)
    status = int(row.get("http_status") or 0)
    final = row.get("final_url") or ""
    start = row.get("used_url") or row.get("start_url") or ""
    hops = len(row.get("redirect_chain") or [])
    tl = title.lower()
    text_l = text.lower()

    if tl in {"just a moment...", "attention required! | cloudflare"} or BLOCK_TITLE.search(title):
        return "cloudflare"
    if title == "百度安全验证" or "请向右滑动完成拼图" in text or "请完成下方验证" in text:
        return "captcha"
    if "you've been blocked by network security" in text_l:
        return "waf"
    if "click the button below to continue shopping" in text_l:
        return "waf"
    if "正在进行安全检测" in text:
        return "captcha"
    if "sina visitor system" in tl or "passport.weibo.com" in final:
        return "redirect"
    if status == 403 and text_len < 80 and ("zhihu.com" in start or "zhihu.com" in final):
        return "waf"
    if status == 403 and text_len < 80:
        return "waf"
    if status == 404:
        return "empty"
    if text_len < 500 and any(
        x in text_l for x in ("continue with google", "continue with apple", "already have an account?")
    ):
        return "redirect"

    # substantial real page
    if status == 200 and text_len >= 400 and title and not BLOCK_TITLE.search(title):
        sp = urlparse(start).path or ""
        fp = urlparse(final).path or ""
        if "/products/" in sp and fp and sp.rstrip("/") != fp.rstrip("/"):
            return "redirect"
        return "none"

    if hops and "passport" in final:
        return "redirect"
    if hops and "/products/" in (urlparse(start).path or "") and urlparse(start).path != urlparse(final).path:
        return "redirect"
    if status == 200 and text_len < 80:
        return "empty"
    if status == 0:
        return "empty"
    if status >= 400:
        return "waf" if status in (401, 403, 406, 429, 451, 503) else "empty"
    return "none"


def title_ok(row: dict, intercept: str) -> bool:
    if intercept in {"cloudflare", "waf", "captcha"}:
        return False
    title = (row.get("title") or "").strip()
    if not title or BLOCK_TITLE.search(title):
        return False
    site = SITES.get(row.get("slug") or "")
    needles = (site or {}).get("title_needles") or []
    blob = (title + "\n" + (row.get("text_head") or "")[:1500]).lower()
    if any(n.lower() in blob for n in needles):
        return True
    if title.lower().strip() in {"x", "x.com"}:
        return True
    if any(x in title.lower() for x in ("visitor system", "passport", "安全验证", "just a moment")):
        return False
    # real headline on a non-block page
    if intercept == "none" and len(title) >= 8:
        return True
    return False


def extractable(row: dict, intercept: str, text_len: int, tok: bool) -> bool:
    if intercept in {"cloudflare", "waf", "captcha", "empty"}:
        return False
    text_l = (row.get("text_head") or "").lower()
    if any(x in text_l for x in ("continue with google", "already have an account?", "sina visitor")):
        return False
    if intercept == "redirect" and text_len < 400:
        return False
    if text_len >= 200:
        return True
    return False


def apply_one(path: Path) -> dict:
    row = json.loads(path.read_text(encoding="utf-8"))
    intercept = classify(row)
    tok = title_ok(row, intercept)
    row["intercept"] = intercept
    row["title_ok"] = bool(tok)
    row["extractable"] = extractable(row, intercept, int(row.get("text_len") or 0), tok)
    path.write_text(json.dumps(row, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return row


def write_jsonl(prefix: str, group: str, rows: list[dict]) -> None:
    rows = sorted(rows, key=lambda r: int(r["id"]))
    p = RAW / f"{prefix}-{group}.jsonl"
    p.write_text("".join(json.dumps(r, ensure_ascii=False) + "\n" for r in rows), encoding="utf-8")


def main() -> None:
    http_f, http_d, lp_f, lp_d = [], [], [], []
    for p in sorted(RAW.glob("http-*.json")):
        if ".listing." in p.name or p.name.endswith(".container.log"):
            continue
        row = apply_one(p)
        (http_f if row.get("group") == "foreign" else http_d).append(row)
    for p in sorted(RAW.glob("lp-*.json")):
        if ".listing." in p.name or p.name.endswith(".excerpt.json"):
            continue
        row = apply_one(p)
        (lp_f if row.get("group") == "foreign" else lp_d).append(row)
    write_jsonl("http", "foreign", http_f)
    write_jsonl("http", "domestic", http_d)
    write_jsonl("lp", "foreign", lp_f)
    write_jsonl("lp", "domestic", lp_d)
    print(f"reclassified http_f={len(http_f)} http_d={len(http_d)} lp_f={len(lp_f)} lp_d={len(lp_d)}")
    for r in http_f + http_d + lp_f + lp_d:
        print(
            f"  {r['channel'][:4]} {r['id']:02d} {r['slug']:12} "
            f"st={r.get('http_status')} {r['intercept']:10} ext={r['extractable']} "
            f"tok={r['title_ok']} n={r.get('text_len')}"
        )


if __name__ == "__main__":
    main()