#!/usr/bin/env python3
"""Analysis for "Blocking GPTBot does not stop ChatGPT reading the page".

Input:  raw_1_20000.jsonl.gz  (one JSON object per domain, both user-agents)
Output: robots_20000.csv      (one row per domain, per-crawler state)
        numbers.json          (every figure quoted in the report)
        lookalike_sites.csv   (every site with a non-ASCII user-agent token)

Run: python3 analyse.py
"""
import csv, gzip, json, re, collections, unicodedata, sys, os
import robots_rfc9309 as R

OUT = sys.argv[1] if len(sys.argv) > 1 else "out"
RAWS = sys.argv[2:] or ["raw_part1.jsonl.gz", "raw_part2.jsonl.gz"]

# ---------------------------------------------------------------- vendor sets
# Token lists come from each vendor's own current documentation (see
# vendors.json for the doc URL behind every list). "bulk" is the crawler that
# collects pages in advance for training or indexing; "live" are the same
# vendor's crawlers that fetch a page while answering, of which "user" are the
# ones documented as triggered by a user's question.
VENDORS = json.load(open("vendors.json"))
ALL_TOKENS = sorted({t["token"].lower() for d in VENDORS.values() for t in d["tokens"]})
GAP = {v: d["gap"] for v, d in VENDORS.items() if d.get("gap")}
for g in GAP.values():
    for k in ("bulk", "live", "user"):
        g[k] = [t.lower() for t in g[k]]
UNDOC = {v: [t["token"].lower() for t in d["tokens"] if t["kind"] == "undocumented"]
         for v, d in VENDORS.items()}

# Characters that look like ASCII hyphen-minus but are not, plus any other
# non-ASCII character inside a product token (RFC 9309 restricts product tokens
# to a-z, A-Z, 0-9, "_" and "-").
HYPHEN_LOOKALIKES = {
    0x2010: "HYPHEN", 0x2011: "NON-BREAKING HYPHEN", 0x2012: "FIGURE DASH",
    0x2013: "EN DASH", 0x2014: "EM DASH", 0x2015: "HORIZONTAL BAR",
    0x2212: "MINUS SIGN", 0xFE58: "SMALL EM DASH", 0xFE63: "SMALL HYPHEN-MINUS",
    0xFF0D: "FULLWIDTH HYPHEN-MINUS", 0x00AD: "SOFT HYPHEN", 0x2043: "HYPHEN BULLET",
}


def band(rank):
    for hi, name in ((1000, "1-1,000"), (5000, "1,001-5,000"), (10000, "5,001-10,000"), (20000, "10,001-20,000")):
        if rank <= hi:
            return name
    return "other"


def is_html(body, ctype):
    """Is the body served at /robots.txt an HTML page rather than a robots.txt?

    A single-page-application catch-all or a soft 404 returns HTTP 200 with a
    web page, so the operator believes there is a robots.txt where there is
    none. Conservative: an HTML marker at the top of the body, or an HTML
    content-type on a body that contains no user-agent line anywhere.
    """
    head = body[:600].lower()
    if "<html" in head or "<!doctype html" in head or "<head>" in head or "<head " in head:
        return True
    if "text/html" in (ctype or "").lower():
        return not re.search(r"(?im)^\s*user-agent\s*:", body)
    return False


CHALLENGE = (
    "recaptcha", "captcha", "just a moment", "attention required", "cf-browser-verification",
    "cf_chl", "checking your browser", "perimeterx", "datadome", "px-captcha",
    "access denied", "request unsuccessful", "incapsula", "imperva", "are you a robot",
    "enable javascript and cookies", "ddos-guard", "verifying you are human",
)


def is_challenge(body):
    """An anti-bot interstitial served at /robots.txt. It is not a soft 404 and
    must not be counted as one: the site has a robots.txt, we were not shown it."""
    head = body[:4000].lower()
    return any(m in head for m in CHALLENGE)


rows, lookalikes, agent_counter = [], [], collections.Counter()
counts = collections.Counter()
n = 0


def records(paths):
    """Yield one record per domain; the first file to carry a domain wins."""
    seen = set()
    for p in paths:
        for line in gzip.open(p, "rt"):
            d = json.loads(line)
            if d["domain"] in seen:
                continue
            seen.add(d["domain"])
            yield d


for d in records(RAWS):
    n += 1
    rank, dom = d["rank"], d["domain"]
    rs, bs = d["r_status"], d["b_status"]
    r_body = d.get("r_body") or ""
    b_body = d.get("b_body") or (r_body if d["r_sha"] == d["b_sha"] else "")

    # Which response do we treat as the site's robots.txt? The research agent's,
    # unless it was refused and the browser agent was served, in which case the
    # site is discriminating by user-agent and the browser response is the file.
    ua_blocked = False
    if rs == 200:
        body, served, ctype = r_body, "research", d.get("r_ctype", "")
    elif bs == 200:
        body, served, ctype = b_body, "browser", d.get("b_ctype", "")
        ua_blocked = True
    else:
        body, served, ctype = "", "none", ""

    row = {"rank": rank, "domain": dom, "band": band(rank),
           "status_research": rs, "status_browser": bs, "served_to": served,
           "fetched_via": d.get("fetched_via", "direct"),
           "ua_discriminated": int(ua_blocked)}
    if row["fetched_via"] == "proxy":
        counts["via_proxy"] += 1
        # Only a domain the direct run could not get a 200 from, either agent,
        # counts as recovered by the proxy; the rest were re-fetched because one
        # agent saw a refusal and the proxy result simply replaced a good one.
        if (rs == 200 or bs == 200) and not (d.get("direct_r_status") == 200
                                             or d.get("direct_b_status") == 200):
            counts["proxy_rescued"] += 1
    html_at_path = bool(body.strip()) and is_html(body, ctype)
    challenge = html_at_path and is_challenge(body)
    row["bot_challenge"] = int(challenge)
    row["html_at_robots"] = int(html_at_path and not challenge)
    usable = bool(body.strip()) and not html_at_path
    row["usable_robots"] = int(usable)

    counts["fetched"] += 1
    if ua_blocked:
        counts["ua_discriminated"] += 1
    if rs != 200 and bs != 200:
        counts["no_robots_either_agent"] += 1
    if challenge:
        counts["bot_challenge"] += 1
    elif html_at_path:
        counts["html_at_robots"] += 1
    if not usable:
        rows.append(row)
        continue
    counts["usable"] += 1

    groups, sitemaps, _other = R.parse(body)
    tokens = {t for ts, _ in groups for t in ts}
    for t in tokens:
        agent_counter[t] += 1
    row["n_groups"] = len(groups)
    row["n_tokens"] = len(tokens)
    row["star"] = R.star_state(groups)
    for tok in ALL_TOKENS:
        row["ua:" + tok] = R.state(groups, tok)

    bad = []
    for t in tokens:
        if t == "*":
            continue
        odd = [c for c in t if ord(c) > 127]
        if not odd:
            continue
        bad.append({
            "token": t,
            "codepoints": ["U+%04X" % ord(c) for c in odd],
            "names": [HYPHEN_LOOKALIKES.get(ord(c)) or unicodedata.name(c, "?") for c in odd],
            "hyphen_lookalike": any(ord(c) in HYPHEN_LOOKALIKES for c in odd),
        })
    row["nonascii_tokens"] = len(bad)
    if bad:
        counts["nonascii_ua_sites"] += 1
        if any(x["hyphen_lookalike"] for x in bad):
            counts["hyphen_lookalike_sites"] += 1
        for x in bad:
            lookalikes.append({"rank": rank, "domain": dom, "token": x["token"],
                               "codepoints": " ".join(x["codepoints"]),
                               "characters": "; ".join(x["names"]),
                               "hyphen_lookalike": x["hyphen_lookalike"]})
    rows.append(row)

USABLE = [r for r in rows if r.get("usable_robots")]


def st(r, tok):
    """State of the group that names this crawler, or 'absent' if unnamed."""
    return r.get("ua:" + tok, "absent")


def names(r, tok):
    return st(r, tok) != "absent"


def blocks(r, tok):
    """The file names this crawler and disallows it something. This is the
    deliberate act we are counting: somebody wrote a rule about this bot."""
    return st(r, tok) in ("block", "partial")


def blocks_root(r, tok):
    return st(r, tok) == "block"


def restricted(r, tok):
    """Whether this crawler is in fact kept out of anything. A crawler the file
    never names falls back to the '*' group (RFC 9309 section 2.2.1), so a site
    with a blanket Disallow: / for '*' does restrict it even without naming it."""
    s = st(r, tok)
    if s != "absent":
        return s in ("block", "partial")
    return r.get("star") == "block"


def restricted_root(r, tok):
    s = st(r, tok)
    if s != "absent":
        return s == "block"
    return r.get("star") == "block"


def open_to(r, tok):
    return not restricted(r, tok)


def any_of(r, toks, fn):
    return any(fn(r, t) for t in toks)


def all_of(r, toks, fn):
    return bool(toks) and all(fn(r, t) for t in toks)


BANDS = ("1-1,000", "1,001-5,000", "5,001-10,000", "10,001-20,000")


def pct(a, b):
    return round(100 * a / b, 1) if b else None


N = {}
N["meta"] = {
    "tranco_list_id": "K9QPW",
    "tranco_list_date": "2026-09-01",
    "tranco_md5": "ed970e1b159546a9ade66e72ae76eb3b",
    "measured_on": "2026-09-02",
    "domains_requested": n,
}
N["fetch"] = {
    "domains": n,
    "usable_robots": counts["usable"],
    "usable_pct": round(100 * counts["usable"] / n, 1),
    "no_robots_either_agent": counts["no_robots_either_agent"],
    "html_at_robots": counts["html_at_robots"],
    "html_at_robots_pct": round(100 * counts["html_at_robots"] / n, 1),
    "bot_challenge_at_robots": counts["bot_challenge"],
    "bot_challenge_at_robots_pct": round(100 * counts["bot_challenge"] / n, 1),
    "ua_discriminated": counts["ua_discriminated"],
    "ua_discriminated_pct": round(100 * counts["ua_discriminated"] / n, 2),
    "proxy_result_used": counts["via_proxy"],
    "recovered_by_proxy_only": counts["proxy_rescued"],
    "recovered_by_proxy_only_pct": round(100 * counts["proxy_rescued"] / n, 1),
}

# ------------------------------------------------------------------- vendors
vend = {}
for v, g in GAP.items():
    bulk, live, user = g["bulk"], g["live"], g["user"]
    search = [t for t in live if t not in user]
    blockers = [r for r in USABLE if any_of(r, bulk, blocks)]
    complete = [r for r in blockers if all_of(r, live, restricted)]
    no_live_named = [r for r in blockers if not any_of(r, live, names) and r.get("star") != "block"]
    user_open = [r for r in blockers if user and not any_of(r, user, restricted)]
    search_open = [r for r in blockers if search and not any_of(r, search, restricted)]
    reverse = [r for r in USABLE if any_of(r, live, blocks) and not any_of(r, bulk, blocks)]
    und = UNDOC.get(v) or []
    und_any = [r for r in USABLE if und and any_of(r, und, names)]
    und_only = [r for r in USABLE if und and any_of(r, und, names) and not any_of(r, bulk + live, names)]
    e = {
        "docs": VENDORS[v]["docs"],
        "bulk_tokens": bulk, "live_tokens": live, "user_tokens": user,
        "undocumented_tokens": und,
        "names_bulk": sum(1 for r in USABLE if any_of(r, bulk, names)),
        "block_bulk": len(blockers),
        "block_bulk_pct_of_usable": pct(len(blockers), len(USABLE)),
        "complete": len(complete),
        "complete_pct": pct(len(complete), len(blockers)),
        "incomplete": len(blockers) - len(complete),
        "incomplete_pct": pct(len(blockers) - len(complete), len(blockers)),
        "user_agent_open": len(user_open),
        "user_agent_open_pct": pct(len(user_open), len(blockers)),
        "search_agent_open": len(search_open),
        "search_agent_open_pct": pct(len(search_open), len(blockers)),
        "no_live_sibling_named_at_all": len(no_live_named),
        "no_live_sibling_named_pct": pct(len(no_live_named), len(blockers)),
        "reverse_only_live_blocked": len(reverse),
        "undocumented_named_any": len(und_any),
        "undocumented_named_any_pct": pct(len(und_any), len(USABLE)),
        "undocumented_named_only": len(und_only),
        "undocumented_named_only_domains": sorted((r["rank"], r["domain"]) for r in und_only)[:80],
        "bands": {},
    }
    for bn in BANDS:
        bl = [r for r in blockers if r["band"] == bn]
        co = [r for r in bl if all_of(r, live, restricted)]
        e["bands"][bn] = {"blockers": len(bl), "complete": len(co),
                          "incomplete_pct": pct(len(bl) - len(co), len(bl))}
    vend[v] = e
N["vendors"] = vend

# ------------------------------------------------- the headline: any vendor
HL = [v for v in GAP if GAP[v]["headline"]]
allai = [t for t in ALL_TOKENS]
anyblock = [r for r in USABLE if any_of(r, allai, blocks)]
tr_block = [r for r in USABLE if any(any_of(r, GAP[v]["bulk"], blocks) for v in HL)]
gap = [r for r in tr_block
       if any(any_of(r, GAP[v]["bulk"], blocks) and not all_of(r, GAP[v]["live"], restricted) for v in HL)]
gap_user = [r for r in tr_block
            if any(any_of(r, GAP[v]["bulk"], blocks) and not any_of(r, GAP[v]["user"], restricted) for v in HL)]
gap_unnamed = [r for r in tr_block
               if any(any_of(r, GAP[v]["bulk"], blocks) and not any_of(r, GAP[v]["live"], names) for v in HL)
               and r.get("star") != "block"]
N["headline"] = {
    "vendors_considered": HL,
    "sites_blocking_any_ai_token": len(anyblock),
    "sites_blocking_any_ai_token_pct": pct(len(anyblock), len(USABLE)),
    "sites_blocking_a_bulk_crawler": len(tr_block),
    "sites_blocking_a_bulk_crawler_pct": pct(len(tr_block), len(USABLE)),
    "with_a_live_sibling_left_open": len(gap),
    "with_a_live_sibling_left_open_pct": pct(len(gap), len(tr_block)),
    "with_the_user_agent_left_open": len(gap_user),
    "with_the_user_agent_left_open_pct": pct(len(gap_user), len(tr_block)),
    "live_sibling_never_named": len(gap_unnamed),
    "live_sibling_never_named_pct": pct(len(gap_unnamed), len(tr_block)),
    "bands": {},
}
for bn in BANDS:
    tb = [r for r in tr_block if r["band"] == bn]
    gb = [r for r in gap if r["band"] == bn]
    ub = [r for r in USABLE if r["band"] == bn]
    N["headline"]["bands"][bn] = {
        "usable_robots": len(ub), "blocking_a_bulk_crawler": len(tb),
        "blocking_pct_of_usable": pct(len(tb), len(ub)),
        "live_sibling_left_open": len(gb),
        "live_sibling_left_open_pct": pct(len(gb), len(tb)),
    }

# Strict variant: count only whole-site blocks (Disallow: /), ignoring sites that
# merely disallow a directory to the bulk crawler.
def _all(r, toks, fn):
    return bool(toks) and all(fn(r, t) for t in toks)


tr_s = [r for r in USABLE if any(any(blocks_root(r, t) for t in GAP[v]["bulk"]) for v in HL)]
gap_s = [r for r in tr_s
         if any(any(blocks_root(r, t) for t in GAP[v]["bulk"]) and not _all(r, GAP[v]["live"], restricted_root)
                for v in HL)]
N["headline_strict"] = {
    "note": "only sites that disallow the whole site (Disallow: /) to the bulk crawler",
    "sites_blocking_a_bulk_crawler": len(tr_s),
    "with_a_live_sibling_left_open": len(gap_s),
    "with_a_live_sibling_left_open_pct": pct(len(gap_s), len(tr_s)),
}
N["vendors_strict"] = {}
for v, g in GAP.items():
    bl = [r for r in USABLE if any(blocks_root(r, t) for t in g["bulk"])]
    co = [r for r in bl if _all(r, g["live"], restricted_root)]
    uo = [r for r in bl if g["user"] and not any(restricted_root(r, t) for t in g["user"])]
    N["vendors_strict"][v] = {"block_bulk": len(bl), "complete": len(co),
                              "incomplete": len(bl) - len(co), "incomplete_pct": pct(len(bl) - len(co), len(bl)),
                              "user_agent_open": len(uo), "user_agent_open_pct": pct(len(uo), len(bl))}

# ------------------------------------------- context: who names anything at all
AI_TOKENS = [t for t in ALL_TOKENS if t not in ("googlebot", "googlebot-image", "googlebot-video",
                                                "googlebot-news", "facebookexternalhit", "applebot")]
namers = [r for r in USABLE if any_of(r, AI_TOKENS, names)]
gptb = [r for r in USABLE if blocks(r, "gptbot")]
N["context"] = {
    "sites_naming_any_ai_token": len(namers),
    "sites_naming_any_ai_token_pct": pct(len(namers), len(USABLE)),
    "gptbot_blockers": len(gptb),
    "gptbot_blockers_also_naming_ccbot": sum(1 for r in gptb if names(r, "ccbot")),
    "gptbot_blockers_also_naming_anthropic_ai": sum(1 for r in gptb if names(r, "anthropic-ai")),
    # Every live sibling (search or user-triggered) published by the six vendors
    # that run a bulk crawler and a live crawler under separate tokens.
    "gptbot_blockers_naming_no_live_token": sum(
        1 for r in gptb if not any_of(r, ["oai-searchbot", "chatgpt-user", "claude-searchbot",
                                          "claude-user", "meta-webindexer", "meta-externalfetcher",
                                          "amzn-searchbot", "amzn-user", "mistralai-index",
                                          "mistralai-user", "perplexity-user"], names)),
    "gptbot_blockers_naming_neither_openai_live_token": sum(
        1 for r in gptb if not names(r, "oai-searchbot") and not names(r, "chatgpt-user")),
    "gptbot_blockers_naming_neither_openai_live_token_pct": pct(
        sum(1 for r in gptb if not names(r, "oai-searchbot") and not names(r, "chatgpt-user")), len(gptb)),
}

# ----------------------------------------------------------- Google-Extended
ge = [r for r in USABLE if blocks(r, "google-extended")]
N["google"] = {
    "block_google_extended": len(ge),
    "block_google_extended_pct": pct(len(ge), len(USABLE)),
    "also_block_googlebot": sum(1 for r in ge if blocks(r, "googlebot")),
    # "unrestricted" as defined in the report: not named-and-disallowed, and not
    # caught by a blanket Disallow: / in the "*" group.
    "leave_googlebot_open": sum(1 for r in ge if not restricted(r, "googlebot")),
    "leave_googlebot_open_pct": pct(sum(1 for r in ge if not restricted(r, "googlebot")), len(ge)),
}

# ------------------------------------------------------------- broken tokens
N["broken_tokens"] = {
    "sites_with_nonascii_token": counts["nonascii_ua_sites"],
    "sites_with_nonascii_token_pct": round(100 * counts["nonascii_ua_sites"] / len(USABLE), 2),
    "sites_with_hyphen_lookalike": counts["hyphen_lookalike_sites"],
    "distinct_broken_tokens": len({l["token"] for l in lookalikes}),
    "sites": sorted({(l["rank"], l["domain"]) for l in lookalikes}),
    "domains": sorted({l["domain"] for l in lookalikes}),
}
# The single most-copied broken token: "Perplexity-User" written with
# U+2011 NON-BREAKING HYPHEN, which can never match Perplexity's product token.
_nb = sorted({(l["rank"], l["domain"]) for l in lookalikes
              if "U+2011" in l["codepoints"] and l["token"].startswith("perplexity")})
N["broken_tokens"]["nbhyphen_perplexity_user_sites"] = len(_nb)
N["broken_tokens"]["nbhyphen_perplexity_user_domains"] = [d for _, d in _nb]
N["broken_tokens"]["distinct_nonascii_domains"] = len({l["domain"] for l in lookalikes})



# ------------------------------------------------------- candidate examples
def _ex(rs, k=200):
    return [[r["rank"], r["domain"]] for r in sorted(rs, key=lambda x: x["rank"])[:k]]


N["examples"] = {
    "openai_gap": _ex([r for r in USABLE if blocks(r, "gptbot")
                       and not all_of(r, ["oai-searchbot", "chatgpt-user"], restricted)]),
    "openai_gap_user_open": _ex([r for r in USABLE if blocks(r, "gptbot") and open_to(r, "chatgpt-user")]),
    "openai_none_named": _ex([r for r in USABLE if blocks(r, "gptbot")
                              and not names(r, "chatgpt-user") and not names(r, "oai-searchbot")
                              and r.get("star") != "block"]),
    "anthropic_gap_user_open": _ex([r for r in USABLE if blocks(r, "claudebot") and open_to(r, "claude-user")]),
    "anthropic_undocumented_only": _ex([r for r in USABLE
                                        if (names(r, "anthropic-ai") or names(r, "claude-web"))
                                        and not (names(r, "claudebot") or names(r, "claude-user")
                                                 or names(r, "claude-searchbot"))]),
    "perplexity_gap": _ex([r for r in USABLE if blocks(r, "perplexitybot") and open_to(r, "perplexity-user")]),
    "google_extended_googlebot_open": _ex([r for r in USABLE if blocks(r, "google-extended")
                                           and not restricted(r, "googlebot")]),
    "html_at_robots": _ex([r for r in rows if r.get("html_at_robots")]),
    "bot_challenge": _ex([r for r in rows if r.get("bot_challenge")]),
    "ua_discriminated": _ex([r for r in rows if r.get("ua_discriminated")]),
}

N["ai_token_counts"] = {t: agent_counter[t] for t in ALL_TOKENS}
N["top_tokens_overall"] = agent_counter.most_common(40)

os.makedirs(OUT, exist_ok=True)
with open(os.path.join(OUT, "robots_20000.csv"), "w", newline="") as f:
    keys = ["rank", "domain", "band", "status_research", "status_browser", "served_to",
            "fetched_via", "ua_discriminated", "bot_challenge", "html_at_robots", "usable_robots", "n_groups", "n_tokens",
            "star"] + ["ua:" + t for t in ALL_TOKENS] + ["nonascii_tokens"]
    w = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
    w.writeheader()
    for r in rows:
        w.writerow(r)
with open(os.path.join(OUT, "lookalike_sites.csv"), "w", newline="") as f:
    w = csv.DictWriter(f, fieldnames=["rank", "domain", "token", "codepoints", "characters", "hyphen_lookalike"])
    w.writeheader()
    for l in sorted(lookalikes, key=lambda x: x["rank"]):
        w.writerow(l)
json.dump(N, open(os.path.join(OUT, "numbers.json"), "w"), indent=1, default=str)
print(json.dumps({k: v for k, v in N.items() if k in ("fetch", "headline", "google", "broken_tokens")}, indent=1)[:6000])
