#!/usr/bin/env python3
"""Egress test: does this machine's address get the answers the collector gets at home?

Stream E, task E1. Runs anywhere with Python 3 and nothing else (no app import, no
third-party package), so the same file runs on a fresh droplet and on the dev server.
It sends exactly the headers `app.services.collectors.fetch.fetch()` sends, so the only
variable between two runs is the source address. Two requests per host: robots.txt and
one product page the collector has already read (URLs taken from the dev database on
2026-09-10). Verdicts mirror `fetch._looks_blocked` and `fetch.render_refusal`: a
refusal is decided on content and headers, never on the status code alone.

    python3 deploy/egress-test.py --label mosiah --out egress-mosiah.json
    python3 deploy/egress-test.py --label droplet --out egress-droplet.json
    python3 deploy/egress-test.py --compare egress-mosiah.json egress-droplet.json > egress.md

Polite by construction: one robots read and one page per host, a pause between the
two, hosts in series. Dubai refuses the declared reader by name at its edge already
(running list `issue-dubai-refuses-by-name`), so a refusal there in BOTH runs is the
expected control, not a finding.
"""

import argparse
import gzip
import json
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone

# Mirrors fetch.py; BOT_NAME is "DutyFreeProfessorBot" there (tests pin the pair equal).
USER_AGENT = "Mozilla/5.0 (compatible; DutyFreeProfessorBot/0.1; +https://bot.dutyfreeprofessor.com)"
TIMEOUT = 40
PAUSE_SAME_HOST = 5.0
PAUSE_BETWEEN_HOSTS = 2.0

# (host, one product URL the collector has read). Rendered sources (Changi, Shilla)
# answer a text fetch with a shell; what matters for them is status and headers.
TARGETS = [
    ("athens.shopdutyfree.com", "https://athens.shopdutyfree.com/en/137/chivas-regal-18-year-old-blended-scotch-whisky-scotland-1l-1000089p1134280"),
    ("barcelona.shopdutyfree.com", "https://barcelona.shopdutyfree.com/en/13/estrella-galicia-especial-33cl"),
    ("buenosaires.shopdutyfree.com", "https://buenosaires.shopdutyfree.com/en/54/dewars-white-label-70cl"),
    ("hongkong.shopdutyfree.com", "https://hongkong.shopdutyfree.com/en/120/xi-jiu-jiaocang-1988-50cl"),
    ("jfk.shopdutyfree.com", "https://jfk.shopdutyfree.com/en/91/basil-haydens-kentucky-straight-bourbon-whiskey-1l"),
    ("london-heathrow.worlddutyfree.com", "https://london-heathrow.worlddutyfree.com/en/64/gabbiano-venezie-pinot-grigio-75cl"),
    ("madrid.shopdutyfree.com", "https://madrid.shopdutyfree.com/en/15/estrella-galicia-especial-33cl"),
    ("mexicocity.shopdutyfree.com", "https://mexicocity.shopdutyfree.com/en/47/esperanto-reposado-tequila-75cl-2585472p1136762"),
    ("toronto.shopdutyfree.com", "https://toronto.shopdutyfree.com/en/20/bottega-gold-15l"),
    ("zurich.shopdutyfree.com", "https://zurich.shopdutyfree.com/en/53/penfolds-koonunga-hill-retro-76-syrahcabernet-75cl"),
    ("co.attenza.net", "https://co.attenza.net/products/jack-daniel-s-old-no-7-tennessee-whiskey"),
    ("es.attenza.net", "https://es.attenza.net/products/buchanans-deluxe-aged-12-years-blended-scotch-whisky"),
    ("pa.attenza.net", "https://pa.attenza.net/products/hydra-beauty-essence-mist"),
    ("www.montrealdutyfree.ca", "https://www.montrealdutyfree.ca/products/ciroc-blue-stone-1l"),
    ("www.bordershop.com", "https://www.bordershop.com/grimbergen/grimbergen-double-ambree-6-5vol-12x0-33l-can/p/000000000001772124/"),
    ("www.heinemann-shop.com", "https://www.heinemann-shop.com/bruichladdich/bruichladdich-black-art-propheta-islay-single-malt-scotch-whisky-48-4vol-0-7l-gift-pack/p/000000000001795511/"),
    ("www.heinemanndutyfree.com.au", "https://www.heinemanndutyfree.com.au/jose-cuervo/jose-cuervo-250th-aniversario-extra-aged-tequila-40vol-0-75l-wooden-gift-box/p/000000000001590840/"),
    ("www.islanddutyfree.is", "https://www.islanddutyfree.is/stella-artois/stella-artois-6x0-33l-can/p/000000000000178850/"),
    ("www.dubaidutyfree.com", "https://www.dubaidutyfree.com/penderyn-faraday-welsh-single-malt-700ml/product/113708219"),
    ("www.dublinandcorkdutyfree.ie", "https://www.dublinandcorkdutyfree.ie/alcohol/whiskey/irish/sommelier-selection-rioja-cask-70cl/402251.html?lang=en_IE"),
    ("www.extime.com", "https://www.extime.com/en/paris/product/wild-vetiver-111650944"),
    ("www.ishopchangi.com", "https://www.ishopchangi.com/en/product/elizabeth-arden-eight-hour-hydraplay-cleanser-mp00685707"),
    ("www.shilladfs.com", "https://www.shilladfs.com/estore/kr/en/p/4802935"),
]

CHALLENGE_MARKERS = (
    b"just a moment", b"attention required", b"access denied", b"are you a robot",
    b"verify you are human", b"checking your browser", b"enable javascript and cookies to continue",
)
WATCH_HEADERS = ("server", "cf-mitigated", "cf-ray", "x-akamai-request-id", "content-type", "content-encoding")


def probe(url: str, accept: str) -> dict:
    request = urllib.request.Request(
        url,
        headers={
            "User-Agent": USER_AGENT,
            "Accept": accept,
            "Accept-Language": "en",
            "Accept-Encoding": "gzip",
        },
    )
    started = time.monotonic()
    status, body, headers, error = None, b"", {}, None
    try:
        with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
            status, body = response.status, response.read()
            headers = {k.lower(): v for k, v in response.headers.items()}
    except urllib.error.HTTPError as exc:
        status = exc.code
        body = exc.read() if exc.fp else b""
        headers = {k.lower(): v for k, v in exc.headers.items()} if exc.headers else {}
    except Exception as exc:  # URLError, timeout, TLS: the kind is the finding
        error = f"{type(exc).__name__}: {exc}"
    elapsed_ms = int((time.monotonic() - started) * 1000)
    if headers.get("content-encoding", "").lower() == "gzip" and body:
        try:
            body = gzip.decompress(body)
        except OSError:
            pass
    lowered = body[:4096].lower()
    markers = [m.decode() for m in CHALLENGE_MARKERS if m in lowered]
    if error:
        verdict = "error"
    elif headers.get("cf-mitigated") == "challenge" or markers:
        verdict = "challenge"
    elif status in (401, 403, 406, 429) or (status == 202 and len(body) < 512) or not body.strip():
        verdict = "refused"
    elif status and status >= 400:
        verdict = f"http-{status}"
    else:
        verdict = "page"
    return {
        "url": url,
        "status": status,
        "bytes": len(body),
        "elapsed_ms": elapsed_ms,
        "verdict": verdict,
        "markers": markers,
        "headers": {k: headers[k] for k in WATCH_HEADERS if k in headers},
        "error": error,
    }


def run(label: str) -> dict:
    ip = None
    try:
        with urllib.request.urlopen("https://api.ipify.org", timeout=10) as r:
            ip = r.read().decode().strip()
    except Exception:
        pass
    results = []
    for i, (host, product_url) in enumerate(TARGETS):
        if i:
            time.sleep(PAUSE_BETWEEN_HOSTS)
        robots = probe(f"https://{host}/robots.txt", "text/plain,*/*")
        time.sleep(PAUSE_SAME_HOST)
        page = probe(product_url, "text/html,application/xhtml+xml,*/*")
        results.append({"host": host, "robots": robots, "page": page})
        print(f"{host:38} robots {robots['verdict']:9} {str(robots['status']):4} "
              f"page {page['verdict']:9} {str(page['status']):4} {page['bytes']:>8} B {page['elapsed_ms']:>6} ms",
              file=sys.stderr, flush=True)
    return {
        "label": label,
        "egress_ip": ip,
        "user_agent": USER_AGENT,
        "started_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "results": results,
    }


def compare(control: dict, candidate: dict) -> str:
    by_host = {r["host"]: r for r in candidate["results"]}
    lines = [
        f"# Egress test: {candidate['label']} ({candidate['egress_ip']}) against {control['label']} ({control['egress_ip']})",
        "",
        f"Control run {control['started_utc']}, candidate run {candidate['started_utc']}. Same user agent, same headers, same URLs; the address is the only variable.",
        "",
        "| Host | robots | page (control) | page (candidate) | bytes c/c | edge | finding |",
        "|---|---|---|---|---|---|---|",
    ]
    worse, same, better = [], [], []
    for c in control["results"]:
        d = by_host.get(c["host"])
        if not d:
            continue
        cp, dp = c["page"], d["page"]
        edge = dp["headers"].get("server", "") + (" cf-mitigated=" + dp["headers"]["cf-mitigated"] if "cf-mitigated" in dp["headers"] else "")
        if cp["verdict"] == dp["verdict"]:
            finding = "same"
            same.append(c["host"])
        elif dp["verdict"] == "page":
            finding = "better on candidate"
            better.append(c["host"])
        else:
            finding = "WORSE on candidate"
            worse.append(c["host"])
        lines.append(
            f"| {c['host']} | {c['robots']['verdict']}/{d['robots']['verdict']} | {cp['verdict']} {cp['status']} | {dp['verdict']} {dp['status']} | {cp['bytes']}/{dp['bytes']} | {edge} | {finding} |"
        )
    lines += [
        "",
        f"**Summary:** {len(same)} same, {len(better)} better on the candidate, {len(worse)} worse on the candidate.",
    ]
    if worse:
        lines.append(f"Worse: {', '.join(worse)}.")
    return "\n".join(lines) + "\n"


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--label", default="unnamed")
    parser.add_argument("--out", help="write the JSON record here")
    parser.add_argument("--compare", nargs=2, metavar=("CONTROL", "CANDIDATE"), help="print a markdown comparison")
    args = parser.parse_args()
    if args.compare:
        with open(args.compare[0]) as f:
            control = json.load(f)
        with open(args.compare[1]) as f:
            candidate = json.load(f)
        sys.stdout.write(compare(control, candidate))
        return 0
    record = run(args.label)
    if args.out:
        with open(args.out, "w") as f:
            json.dump(record, f, indent=1)
    else:
        json.dump(record, sys.stdout, indent=1)
    return 0


if __name__ == "__main__":
    sys.exit(main())