"""What do the top 1,000 websites tell AI crawlers? For each domain in the 1,000-site
test (Tranco list), fetch robots.txt and check whether each major AI user agent may
fetch the homepage, using Protego (Scrapy's robots.txt parser). Also record
Content-Signal lines, llms.txt, TDMRep (/.well-known/tdmrep.json) and ai.txt.
Run on October 12, 2026, from Mumbai, with httpx 0.28 and protego 0.7.0.
Run with: uv run --with httpx --with protego python scan.py"""
import asyncio
import csv
import json
import re
from pathlib import Path

import httpx
from protego import Protego

UA = "DoccraftResearchBot/1.0 (https://doccraft.dev; hello@doccraft.dev) httpx/0.28"
AGENTS = [
    "GPTBot", "OAI-SearchBot", "ChatGPT-User",
    "ClaudeBot", "Claude-SearchBot", "Claude-User",
    "Google-Extended", "Googlebot",
    "PerplexityBot", "Perplexity-User",
    "CCBot", "Applebot-Extended", "Bytespider", "Amazonbot", "meta-externalagent",
]
RAW = Path("robots")  # the raw robots.txt files, for checking results by hand
RAW.mkdir(parents=True, exist_ok=True)
rows = list(csv.DictReader(open("../browser-test-2026/results.csv")))


def plain_text(r):
    """A real text file, not an HTML page served at that path."""
    ctype = r.headers.get("content-type", "")
    return r.status_code == 200 and "html" not in ctype and not r.text.lstrip().startswith("<")


async def check(client, sem, row):
    domain = row["domain"]
    out = {"rank": int(row["rank"]), "domain": domain, "group": row["group"]}
    async with sem:
        base = f"https://{domain}"
        try:
            r = await client.get(f"{base}/robots.txt")
            out["robots_status"] = r.status_code
            out["robots_final"] = str(r.url)
            if plain_text(r):
                text = r.text
                (RAW / f"{domain}.txt").write_text(text)
                rp = Protego.parse(text)
                home = str(r.url.join("/"))
                out["allowed"] = {a: rp.can_fetch(home, a) for a in AGENTS}
                named = {m.strip().lower() for m in re.findall(r"(?im)^\s*user-agent\s*:\s*(\S+)", text)}
                out["named"] = [a for a in AGENTS if a.lower() in named]
                out["content_signal"] = re.findall(r"(?im)^\s*content-signal\s*:\s*(.+?)\s*$", text)
        except Exception as e:
            out["robots_error"] = type(e).__name__
        for key, path in [("llms_txt", "/llms.txt"), ("ai_txt", "/ai.txt"), ("tdmrep", "/.well-known/tdmrep.json")]:
            try:
                r = await client.get(base + path)
                ok = plain_text(r)
                if ok and key == "llms_txt":
                    # The llms.txt format is Markdown: a heading, or links to pages.
                    body = r.text.lstrip()
                    ok = body.startswith("#") or bool(re.search(r"\]\(https?://", body))
                if ok and key == "tdmrep":
                    try:
                        json.loads(r.text)
                    except ValueError:
                        ok = False
                out[key] = ok
            except Exception:
                out[key] = None
    return out


async def main():
    sem = asyncio.Semaphore(20)
    async with httpx.AsyncClient(headers={"User-Agent": UA}, follow_redirects=True, timeout=20) as client:
        results = await asyncio.gather(*(check(client, sem, r) for r in rows))
    json.dump(results, open("results.json", "w"), indent=1)
    print(len(results), "domains,", sum("allowed" in r for r in results), "with a robots.txt")


asyncio.run(main())