#!/usr/bin/env python3
"""
One-off crawl behind /study — "How many websites say who built them", 14 September 2026.

Deliberately NOT application code: no queue job, no artisan command, no model, no scheduler entry. If this has to
run again in a year, running it by hand is cheaper than the abstraction. Python + stdlib only, so it runs on the
host (the app itself has no PHP outside Docker).

    python3 docs/research/who-built-the-web-crawl.py top-1m.csv out.json out.csv [review.json]

Inputs:  the Tranco top-1m list (https://tranco-list.eu/top-1m.csv.zip).
Outputs: a JSON with the aggregates and every positive hit with its snippet, and the full corpus as CSV.
         Both are committed under docs/research/ and are what the page renders; the page never crawls.

Politeness, the same rules CreditedBot follows: robots.txt honoured (our own token first, then "*"), one home page
per host, 12 s timeout, a declared user agent with a URL that explains the crawl.

The positives this writes are CANDIDATES, not findings. Every one of them was read by hand afterwards and the
verdict column in the JSON is that review, not this script's opinion. On the pilot run 4 of 4 regex hits on the
global sample were false positives (TYPO3 boilerplate, a drug "developed by Gedeon Richter", "Built by clinicians"),
which is the entire reason the review step exists.
"""

import csv
import gzip
import json
import pathlib
import random
import re
import socket
import ssl
import sys
import threading
import urllib.error
import urllib.request
import zlib
from concurrent.futures import ThreadPoolExecutor
from html import unescape

USER_AGENT = "CreditedResearchBot/1.0 (+https://credited.dev/study; one-off research crawl)"
ROBOTS_TOKENS = ("creditedresearchbot", "creditedbot", "*")
TIMEOUT = 12
MAX_BYTES = 3_000_000  # the footer is at the END of the document, so the 256 KB head-only ceiling will not do
SAMPLE_SEED = 20260914

# European ccTLDs: the pilot's SMB-skewed sample. The head and this tail behave nothing alike, so they are
# reported as separate strata and never averaged together.
SMB_TLDS = (".es", ".it", ".nl", ".de", ".fr", ".pt", ".be", ".at", ".ch", ".dk", ".se", ".no", ".fi", ".pl", ".cz")

# (name, rank_low, rank_high, size, smb_only, label, description). The SMB tail is drawn first so it gets its
# pick of the hosts; the rank bands then draw from what is left, and no host lands in two strata.
STRATA = [
    ("smb-tail", 250_000, 1_000_000, 1000, True, "The SMB tail",
     "European ccTLDs (.es .it .nl .de .fr .pt and 9 more) ranked below 250,000"),
    ("top-10k", 1, 10_000, 1000, False, "Tranco top 10k", "The corporate head of the web"),
    ("10k-100k", 10_000, 100_000, 1000, False, "Tranco 10k–100k", "Large sites, mostly in-house or agency-retained"),
    ("100k-500k", 100_000, 500_000, 1000, False, "Tranco 100k–500k", "The middle of the list, global"),
    ("500k-1m", 500_000, 1_000_000, 1000, False, "Tranco 500k–1m", "The global tail, every TLD"),
]
# Head first on the page; the SMB tail is drawn first but read last.
DISPLAY_ORDER = ["top-10k", "10k-100k", "100k-500k", "500k-1m", "smb-tail"]

# Multilingual footer credits. Each one names a builder; the review step decides whether the thing named is a
# person or agency (real) or a platform, a theme, a medicine or a group of clinicians (noise).
CREDIT_PATTERNS = [
    ("en", r"(?:built|designed|developed|made|created|crafted|handcrafted|coded|produced)\s+by"),
    ("en", r"(?:web\s*)?(?:site|website|web\s*design|design(?:\s+and\s+development)?|development)\s+by"),
    ("es", r"(?:dise[ñn]ado|desarrollado|creado|realizado|hecho|programado)\s+por"),
    ("es", r"dise[ñn]o(?:\s+y\s+desarrollo)?\s+(?:web\s+)?por"),
    ("fr", r"(?:con[çc]u|r[ée]alis[ée]|d[ée]velopp[ée]|cr[ée][ée])\s+par"),
    ("fr", r"(?:conception|cr[ée]ation|r[ée]alisation|d[ée]veloppement)\s*(?:du\s+site\s*)?(?::|\s+par)"),
    ("de", r"(?:erstellt|umgesetzt|realisiert|entwickelt|gestaltet|programmiert)\s+von"),
    ("de", r"(?:umsetzung|programmierung|webdesign|gestaltung|konzept(?:ion)?)\s*:"),
    ("it", r"(?:realizzato|sviluppato|creato|progettato|curato)\s+da"),
    ("it", r"a\s+cura\s+di"),
    ("nl", r"(?:gemaakt|ontwikkeld|ontworpen|gerealiseerd|verzorgd)\s+door"),
    ("nl", r"(?:realisatie|ontwerp|website)\s*(?::|\s+door)"),
    ("pt", r"(?:desenvolvido|criado|feito|concebido)\s+por"),
    ("pl", r"(?:wykonanie|realizacja|projekt\s+i\s+wykonanie)\s*:"),
]
CREDIT_RE = [(lang, re.compile(p, re.I)) for lang, p in CREDIT_PATTERNS]

# A bare credit: "by" (in any of the six languages) immediately followed by an outbound link. The loose pattern
# from the pilot; it is noisier than the phrase list, so it is recorded separately and reviewed separately.
BY_LINK_RE = re.compile(
    r"\b(?:by|por|par|von|da|di|door)\s*(?:&nbsp;|\s){0,3}<a\b[^>]*href=[\"']?(?P<href>[^\"'\s>]+)",
    re.I,
)

# Known boilerplate that matches the patterns above and never means "this site was built by". Flagged, not
# dropped: the row stays in the corpus with boilerplate=1 so the filter itself can be audited.
BOILERPLATE_RE = re.compile(
    r"kasper\s+skaarh[oø]j"          # TYPO3 ships this in every install
    r"|powered\s+by\s+"
    r"|(?:designed|developed|made|built)\s+by\s+(?:wordpress|shopify|wix|squarespace|webflow|joomla|drupal|elementor|woocommerce)"
    r"|made\s+with\s+(?:love|❤)"
    r"|(?:created|made|designed)\s+by\s+you\b",
    re.I,
)

CMS_FINGERPRINTS = [
    ("WordPress", re.compile(r"/wp-(?:content|includes)/|wp-json", re.I)),
    ("Shopify", re.compile(r"cdn\.shopify\.com|shopify\.theme", re.I)),
    ("Wix", re.compile(r"static\.wixstatic\.com|wix\.com/website", re.I)),
    ("Squarespace", re.compile(r"squarespace\.com|static1\.squarespace", re.I)),
    ("Webflow", re.compile(r"webflow\.(?:com|io)|data-wf-page", re.I)),
    ("Joomla", re.compile(r"/media/jui/|joomla", re.I)),
    ("Drupal", re.compile(r"drupal(?:settings|\.js)|/sites/default/files/", re.I)),
    ("TYPO3", re.compile(r"typo3(?:conf|temp)", re.I)),
    ("PrestaShop", re.compile(r"prestashop", re.I)),
    ("Magento", re.compile(r"/static/version\d+/frontend/|\bmage/|magento", re.I)),
    ("Ghost", re.compile(r"ghost\.io|content/themes/casper", re.I)),
    ("HubSpot", re.compile(r"hs-scripts\.com|hubspot", re.I)),
    ("Tilda", re.compile(r"tildacdn", re.I)),
    ("Bitrix", re.compile(r"bitrix", re.I)),
    ("Duda", re.compile(r"dudaone|d\.dudamobile", re.I)),
    ("Jimdo", re.compile(r"jimdo", re.I)),
    ("Weebly", re.compile(r"weebly", re.I)),
]

GENERATOR_RE = re.compile(r"<meta[^>]+name=[\"']?generator[\"']?[^>]*>", re.I)
CONTENT_RE = re.compile(r"content=[\"']([^\"']*)[\"']", re.I)
SCRIPT_RE = re.compile(r"<(script|style|noscript)\b.*?</\1>|<!--.*?-->", re.I | re.S)
TAG_RE = re.compile(r"<[^>]+>")
WS_RE = re.compile(r"\s+")

# ponytail: one unverified SSL context for the whole run. A study of who a site credits does not care whether its
# certificate chain validates, and refusing on a bad chain would silently bias the tail (expired certs cluster
# there). Upgrade to per-host verification only if a finding ever depends on TLS.
CTX = ssl.create_default_context()
CTX.check_hostname = False
CTX.verify_mode = ssl.CERT_NONE

print_lock = threading.Lock()


def fetch(url, timeout=TIMEOUT):
    """GET url once. Returns (status, final_url, body_text) or raises."""
    request = urllib.request.Request(url, headers={
        "User-Agent": USER_AGENT,
        "Accept": "text/html,application/xhtml+xml,*/*;q=0.8",
        "Accept-Encoding": "gzip, deflate, identity",
        "Accept-Language": "en;q=0.9,*;q=0.5",
    })
    with urllib.request.urlopen(request, timeout=timeout, context=CTX) as response:
        raw = response.read(MAX_BYTES)
        encoding = (response.headers.get("Content-Encoding") or "").lower()
        if "gzip" in encoding:
            raw = _gunzip(raw)
        elif "deflate" in encoding:
            raw = zlib.decompressobj().decompress(raw)
        charset = response.headers.get_content_charset()
        if not charset:
            match = re.search(rb"charset=[\"']?([\w-]+)", raw[:4096], re.I)
            charset = match.group(1).decode("ascii", "replace") if match else "utf-8"
        try:
            text = raw.decode(charset, "replace")
        except LookupError:
            text = raw.decode("utf-8", "replace")
        return response.status, response.geturl(), text


def _gunzip(raw):
    """A stream truncated at MAX_BYTES is not a complete gzip member, so fall back to a raw inflater that
    returns everything up to the cut instead of raising."""
    try:
        return gzip.decompress(raw)
    except (EOFError, zlib.error, OSError):
        return zlib.decompressobj(16 + zlib.MAX_WBITS).decompress(raw)


def blocked_by_robots(origin):
    """True when robots.txt disallows "/" for us. Errors, absence, or an unparseable file never block."""
    try:
        _, _, body = fetch(f"{origin}/robots.txt", timeout=TIMEOUT)
    except Exception:
        return False

    groups, open_group = [], False
    for line in body.splitlines():
        line = re.sub(r"#.*", "", line).strip()
        agent = re.match(r"user-agent:\s*(.+)$", line, re.I)
        disallow = re.match(r"disallow:\s*(.*)$", line, re.I)
        if agent:
            if not open_group:
                groups.append({"agents": [], "disallow": []})
                open_group = True
            groups[-1]["agents"].append(agent.group(1).strip().lower())
        elif disallow and groups:
            groups[-1]["disallow"].append(disallow.group(1).strip())
            open_group = False
        elif line == "":
            open_group = False

    for token in ROBOTS_TOKENS:
        for group in groups:
            if token in group["agents"]:
                return "/" in group["disallow"]
    return False


def visible_text(html):
    return WS_RE.sub(" ", unescape(TAG_RE.sub(" ", SCRIPT_RE.sub(" ", html)))).strip()


def snippet(haystack, start, end, width=90):
    return WS_RE.sub(" ", haystack[max(0, start - width):end + width]).strip()


def inspect(html):
    """Everything we look for in one page. Returns a dict of findings."""
    text = visible_text(html)
    # The credit lives in the footer. Scanning the last slice of the rendered text (and of the raw HTML for the
    # link pattern) is what keeps "developed by Gedeon Richter" in a product page's body out of the count — and
    # it still let three of those through on the pilot, hence the hand review.
    tail_text = text[-3000:]
    tail_html = html[-25000:]

    found = {}
    for lang, pattern in CREDIT_RE:
        match = pattern.search(tail_text)
        if match:
            found = {
                "credit_lang": lang,
                "credit_match": match.group(0),
                "credit_snippet": snippet(tail_text, match.start(), match.end()),
                "credit_boilerplate": int(bool(BOILERPLATE_RE.search(snippet(tail_text, match.start(), match.end())))),
            }
            break

    by_link = {}
    for match in BY_LINK_RE.finditer(tail_html):
        href = match.group("href")
        if href.startswith("#") or href.startswith("/") or "javascript:" in href.lower():
            continue
        by_link = {
            "by_link_href": href[:200],
            "by_link_snippet": visible_text(tail_html[max(0, match.start() - 200):match.end() + 300])[:300],
        }
        break

    generator = ""
    tag = GENERATOR_RE.search(html)
    if tag:
        content = CONTENT_RE.search(tag.group(0))
        generator = unescape(content.group(1))[:120] if content else ""

    cms = next((name for name, pattern in CMS_FINGERPRINTS if pattern.search(html[:200_000])), "")

    return {"generator": generator, "cms": cms, **found, **by_link}


def probe(row):
    rank, host, stratum = row
    result = {
        "rank": rank, "host": host, "stratum": stratum, "reachable": 0, "status": "", "error": "",
        "final_url": "", "robots_blocked": 0, "generator": "", "cms": "", "humans_txt": 0,
        "credit_lang": "", "credit_match": "", "credit_snippet": "", "credit_boilerplate": 0,
        "by_link_href": "", "by_link_snippet": "",
    }

    origin = f"https://{host}"
    if blocked_by_robots(origin):
        result["robots_blocked"] = 1
        result["error"] = "robots"
        return result

    html = None
    for candidate in (origin, f"http://{host}"):
        try:
            status, final_url, html = fetch(f"{candidate}/")
            result["status"], result["final_url"] = status, final_url[:300]
            break
        except urllib.error.HTTPError as e:
            # The host answered, it just refused us. Retrying the same refusal over http:// buys nothing and
            # costs the site a second request, so stop here and record the status.
            result["status"], result["error"] = e.code, "http"
            break
        except (urllib.error.URLError, socket.timeout, ssl.SSLError, OSError, ValueError) as e:
            result["error"] = type(e).__name__
        except Exception as e:  # a malformed response should cost one host, not the run
            result["error"] = type(e).__name__

    if html is None:
        return result

    result["reachable"] = 1
    result.update(inspect(html))

    try:
        status, _, body = fetch(f"{origin}/humans.txt", timeout=TIMEOUT)
        result["humans_txt"] = int(status == 200 and "<html" not in body[:2000].lower() and len(body.strip()) > 0)
    except Exception:
        pass

    return result


def sample(list_path):
    """Stratified draw, deterministic under SAMPLE_SEED. A host is drawn into exactly one stratum."""
    ranked = []
    with open(list_path, newline="", encoding="utf-8", errors="replace") as handle:
        for line in csv.reader(handle):
            if len(line) >= 2:
                ranked.append((int(line[0]), line[1].strip().lower()))

    rng = random.Random(SAMPLE_SEED)
    taken, drawn = set(), []
    for name, low, high, size, smb_only, *_ in STRATA:
        pool = [
            (rank, host) for rank, host in ranked
            if low <= rank <= high and host not in taken
            and (not smb_only or host.endswith(SMB_TLDS))
        ]
        for rank, host in rng.sample(pool, min(size, len(pool))):
            taken.add(host)
            drawn.append((rank, host, name))
    return drawn


def main():
    list_path, json_path, csv_path = sys.argv[1], sys.argv[2], sys.argv[3]
    # The hand review, as {host: [verdict, reason, quote]}. Without it every candidate reads "unreviewed" and
    # StudyTest fails, which is the point: the parser's opinion is not publishable on its own.
    review = json.loads(pathlib.Path(sys.argv[4]).read_text(encoding="utf-8")) if len(sys.argv) > 4 else {}
    rows = sample(list_path)
    print(f"sampled {len(rows)} hosts across {len(STRATA)} strata", flush=True)

    results, done = [], 0
    with ThreadPoolExecutor(max_workers=48) as pool:
        for result in pool.map(probe, rows):
            results.append(result)
            done += 1
            if done % 250 == 0:
                with print_lock:
                    print(f"{done}/{len(rows)}", flush=True)

    results.sort(key=lambda r: (DISPLAY_ORDER.index(r["stratum"]), r["rank"]))
    for row in results:
        if row["credit_match"] or row["by_link_href"]:
            verdict, reason, _ = review.get(row["host"], ("unreviewed", "", ""))
            row["verdict"], row["review_reason"] = verdict, reason
        else:
            row["verdict"], row["review_reason"] = "", ""
    fields = list(results[0].keys())
    with open(csv_path, "w", newline="", encoding="utf-8") as handle:
        # LF, not the csv module's default CRLF: the file is committed and read back by the app.
        writer = csv.DictWriter(handle, fieldnames=fields, lineterminator="\n")
        writer.writeheader()
        writer.writerows(results)

    with open(json_path, "w", encoding="utf-8") as handle:
        json.dump(summarise(results, review), handle, indent=2, ensure_ascii=False)
    print(f"wrote {csv_path} and {json_path}", flush=True)


def summarise(results, review):
    """The aggregates the page renders, plus every candidate with its snippet and its hand-review verdict."""
    def rate(hits, base):
        return round(hits / base * 100, 1) if base else 0.0

    labels = {name: (label, description) for name, _, _, _, _, label, description in STRATA}
    credited = {host for host, (verdict, *_) in review.items() if verdict == "credit"}

    strata = []
    for name in DISPLAY_ORDER:
        rows = [r for r in results if r["stratum"] == name]
        live = [r for r in rows if r["reachable"]]
        platform = sum(1 for r in live if r["cms"] or r["generator"])
        confirmed = sum(1 for r in live if r["host"] in credited)
        strata.append({
            "stratum": name,
            "label": labels[name][0],
            "description": labels[name][1],
            "sampled": len(rows),
            "reachable": len(live),
            "reachable_pct": rate(len(live), len(rows)),
            "credit_candidates": sum(1 for r in live if r["credit_match"]),
            "by_link_candidates": sum(1 for r in live if r["by_link_href"]),
            "credits_confirmed": confirmed,
            "credit_pct": rate(confirmed, len(live)),
            "platform": platform,
            "platform_pct": rate(platform, len(live)),
            "cms_identified": sum(1 for r in live if r["cms"]),
            "cms_pct": rate(sum(1 for r in live if r["cms"]), len(live)),
            "generator": sum(1 for r in live if r["generator"]),
            "generator_pct": rate(sum(1 for r in live if r["generator"]), len(live)),
            "humans_txt": sum(1 for r in live if r["humans_txt"]),
            "humans_txt_pct": rate(sum(1 for r in live if r["humans_txt"]), len(live)),
        })

    live = [r for r in results if r["reachable"]]
    platform = sum(1 for r in live if r["cms"] or r["generator"])
    confirmed = sum(1 for r in live if r["host"] in credited)
    candidates = []
    for r in live:
        if not (r["credit_match"] or r["by_link_href"]):
            continue
        verdict, reason, quote = review.get(r["host"], ("unreviewed", "", ""))
        candidates.append({
            **{k: r[k] for k in ("rank", "host", "stratum", "credit_lang", "credit_match", "credit_snippet",
                                 "credit_boilerplate", "by_link_href", "by_link_snippet", "cms", "generator")},
            "stratum_label": labels[r["stratum"]][0],
            "verdict": verdict,
            "reason": reason,
            "quote": quote or r["credit_snippet"] or r["by_link_snippet"],
        })

    return {
        "crawled_on": "2026-09-14",
        "source": "Tranco top-1m",
        "user_agent": USER_AGENT,
        "cms_fingerprints": len(CMS_FINGERPRINTS),
        "sampled": len(results),
        "reachable": len(live),
        "reachable_pct": rate(len(live), len(results)),
        "robots_blocked": sum(1 for r in results if r["robots_blocked"]),
        "credits_confirmed": confirmed,
        "credit_pct": rate(confirmed, len(live)),
        "platform": platform,
        "platform_pct": rate(platform, len(live)),
        "candidates_raised": len(candidates),
        "strata": strata,
        "candidates": sorted(candidates, key=lambda c: (c["verdict"] != "credit", DISPLAY_ORDER.index(c["stratum"]), c["host"])),
    }


def self_check():
    """The parser is the only thing here with branches worth a check. One run, asserts, no framework."""
    page = "<html><body><p>Ibuprofen, developed by Gedeon Richter.</p>" + ("filler " * 900) + \
           "<footer>Diseno web por <a href='https://ideahands.com'>Ideahands</a></footer></body></html>"
    found = inspect(page)
    assert found["credit_lang"] == "es", found
    assert "ideahands" in found["by_link_href"], found
    # The drug is in the body, 6 000 characters above the footer: the tail window is what keeps it out.
    assert "Gedeon" not in found["credit_snippet"], found

    typo3 = "<html><body><footer>This website is powered by TYPO3, created by Kasper Skaarhoj.</footer></body></html>"
    assert inspect(typo3)["credit_boilerplate"] == 1

    meta = inspect('<html><head><meta name="generator" content="WordPress 6.5"></head>'
                   '<body><img src="/wp-content/a.png"></body></html>')
    assert meta["generator"] == "WordPress 6.5" and meta["cms"] == "WordPress", meta

    # An <img src="/images/..."> must not read as Magento.
    assert inspect('<html><body><img src="/images/a.png"></body></html>')["cms"] == ""

    # robots.txt parsing: our token wins over "*", and a rule closes the agent list.
    assert visible_text("<script>var a = 1;</script><p>Hello &amp; bye</p>") == "Hello & bye"
    print("self-check ok")


if __name__ == "__main__":
    if len(sys.argv) == 2 and sys.argv[1] == "--self-check":
        self_check()
    else:
        main()