#!/usr/bin/env python3
"""Validate external HTTP/HTTPS links in every markdown doc.

Requires Python 3.10+ (PEP 604 union types).

Walks every `*.md` file under the repo (excluding `.git/`, `node_modules/`,
`outputs/`), extracts `https?://` URLs from markdown link syntax and bare URLs,
deduplicates, and validates each unique URL by HEAD request (falling back to
GET on 405/501). Results are cached for 7 days at `.link-cache.json` (repo
root, gitignored) so re-runs do not hammer external services.

Stdlib only. No `requests`, no `httpx`.

Usage:
    python3 scripts/link_check.py                       # full check, group by file
    python3 scripts/link_check.py --phase 14            # one phase
    python3 scripts/link_check.py --path README.md      # one file
    python3 scripts/link_check.py --path phases/14-... # one directory
    python3 scripts/link_check.py --strict              # exit 1 on any broken link
    python3 scripts/link_check.py --json                # machine-readable
    python3 scripts/link_check.py --cache 0             # bypass cache for this run
    python3 scripts/link_check.py --timeout 15          # per-request timeout (sec)
    python3 scripts/link_check.py --concurrency 16      # worker threads

Companion to `scripts/audit_lessons.py` (rule L010 validates *internal* links);
this script handles the external HTTP/HTTPS surface.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import socket
import ssl
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable
from urllib import error as urlerror
from urllib import request as urlrequest
from urllib.parse import urlparse

ROOT = Path(__file__).resolve().parent.parent
CACHE_PATH = ROOT / ".link-cache.json"
CACHE_SCHEMA_VERSION = 1
USER_AGENT = (
    "ai-engineering-from-scratch link-check/1.0 "
    "(+https://aiengineeringfromscratch.com)"
)
DEFAULT_TIMEOUT = 10
DEFAULT_CONCURRENCY = 8
DEFAULT_CACHE_DAYS = 7
DEFAULT_SKIP_DOMAINS = (
    "twitter.com",
    "x.com",
    "linkedin.com",
    "instagram.com",
    "medium.com",
)
EXCLUDE_DIRS = {".git", "node_modules", "outputs"}

MD_LINK_RE = re.compile(r"\[[^\]]*\]\((<?)(https?://[^\s)>]+)>?\)")
BARE_URL_RE = re.compile(r"(?<![\w(\[=\"'])(https?://[^\s)\]<>\"'`]+)")
TRAILING_PUNCT = ".,;:!?)\"'>"


@dataclass
class UrlOccurrence:
    file: str
    line: int


@dataclass
class CheckResult:
    url: str
    status: str
    http_status: int | None
    error: str | None
    cached: bool = False

    @property
    def ok(self) -> bool:
        return self.status == "ok"


@dataclass
class Report:
    checked_files: int = 0
    unique_urls: int = 0
    requested: int = 0
    cached_hits: int = 0
    skipped: list[str] = field(default_factory=list)
    failed: list[dict[str, object]] = field(default_factory=list)
    by_file: dict[str, list[dict[str, object]]] = field(default_factory=dict)

    def to_dict(self) -> dict[str, object]:
        return {
            "schema_version": 1,
            "checked_files": self.checked_files,
            "unique_urls": self.unique_urls,
            "requested": self.requested,
            "cached_hits": self.cached_hits,
            "skipped_count": len(self.skipped),
            "failed_count": len(self.failed),
            "skipped": sorted(set(self.skipped)),
            "failed": self.failed,
            "by_file": self.by_file,
        }


def iter_markdown_files(
    root: Path, phase: int | None, path: Path | None
) -> Iterable[Path]:
    if path is not None:
        path = path.resolve()
        if path.is_file():
            if path.suffix == ".md":
                yield path
            return
        roots = [path]
    elif phase is not None:
        phase_prefix = f"{phase:02d}-"
        phases_dir = root / "phases"
        if not phases_dir.is_dir():
            return
        matches = [p for p in phases_dir.iterdir() if p.is_dir() and p.name.startswith(phase_prefix)]
        if not matches:
            return
        roots = matches
    else:
        roots = [root]

    for r in roots:
        if r.is_file():
            if r.suffix == ".md":
                yield r
            continue
        for dirpath, dirnames, filenames in os.walk(r):
            dirnames[:] = [d for d in dirnames if d not in EXCLUDE_DIRS]
            for name in filenames:
                if name.endswith(".md"):
                    yield Path(dirpath) / name


def strip_trailing_punct(url: str) -> str:
    while url and url[-1] in TRAILING_PUNCT:
        url = url[:-1]
    return url


def extract_urls(text: str) -> list[tuple[str, int]]:
    """Return list of (url, line_number) tuples preserving order."""
    out: list[tuple[str, int]] = []
    seen_per_line: set[tuple[int, str]] = set()
    for lineno, line in enumerate(text.splitlines(), start=1):
        for m in MD_LINK_RE.finditer(line):
            url = strip_trailing_punct(m.group(2))
            key = (lineno, url)
            if key in seen_per_line:
                continue
            seen_per_line.add(key)
            out.append((url, lineno))
        masked = MD_LINK_RE.sub(" ", line)
        for m in BARE_URL_RE.finditer(masked):
            url = strip_trailing_punct(m.group(1))
            key = (lineno, url)
            if key in seen_per_line:
                continue
            seen_per_line.add(key)
            out.append((url, lineno))
    return out


def load_cache() -> dict[str, dict[str, object]]:
    if not CACHE_PATH.is_file():
        return {}
    try:
        raw = json.loads(CACHE_PATH.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    if not isinstance(raw, dict):
        return {}
    if raw.get("schema_version") != CACHE_SCHEMA_VERSION:
        return {}
    entries = raw.get("entries")
    return entries if isinstance(entries, dict) else {}


def save_cache(entries: dict[str, dict[str, object]]) -> None:
    payload = {"schema_version": CACHE_SCHEMA_VERSION, "entries": entries}
    try:
        CACHE_PATH.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
    except OSError as exc:
        print(f"warning: could not write {CACHE_PATH.name}: {exc}", file=sys.stderr)


def cache_is_fresh(entry: dict[str, object], cache_days: int) -> bool:
    if cache_days <= 0:
        return False
    checked_at = entry.get("checked_at")
    if not isinstance(checked_at, (int, float)):
        return False
    age = time.time() - float(checked_at)
    return age < cache_days * 86400


def domain_of(url: str) -> str:
    try:
        netloc = urlparse(url).netloc.lower()
    except ValueError:
        return ""
    if netloc.startswith("www."):
        netloc = netloc[4:]
    if ":" in netloc:
        netloc = netloc.split(":", 1)[0]
    return netloc


def should_skip(url: str, skip_domains: set[str]) -> bool:
    domain = domain_of(url)
    if not domain:
        return False
    for sd in skip_domains:
        if domain == sd or domain.endswith("." + sd):
            return True
    return False


def _request(url: str, method: str, timeout: int) -> tuple[int | None, str | None]:
    req = urlrequest.Request(
        url,
        method=method,
        headers={
            "User-Agent": USER_AGENT,
            "Accept": "*/*",
            "Accept-Language": "en-US,en;q=0.9",
        },
    )
    ctx = ssl.create_default_context()
    try:
        with urlrequest.urlopen(req, timeout=timeout, context=ctx) as resp:
            return resp.status, None
    except urlerror.HTTPError as exc:
        return exc.code, f"http {exc.code}"
    except urlerror.URLError as exc:
        reason = getattr(exc, "reason", exc)
        return None, f"url-error: {reason}"
    except socket.timeout:
        return None, "timeout"
    except (TimeoutError, ConnectionError) as exc:
        return None, f"conn-error: {exc}"
    except ssl.SSLError as exc:
        return None, f"ssl-error: {exc}"
    except Exception as exc:
        return None, f"error: {exc.__class__.__name__}: {exc}"


def check_url(url: str, timeout: int) -> CheckResult:
    status_code, err = _request(url, "HEAD", timeout)
    if status_code in (405, 501) or (status_code is None and err and "http" not in err):
        get_status, get_err = _request(url, "GET", timeout)
        if get_status is not None:
            status_code, err = get_status, get_err
        elif status_code is None:
            status_code, err = get_status, get_err

    if status_code is None:
        return CheckResult(url=url, status="error", http_status=None, error=err or "unknown")
    if 200 <= status_code < 400:
        return CheckResult(url=url, status="ok", http_status=status_code, error=None)
    return CheckResult(url=url, status="broken", http_status=status_code, error=err or f"http {status_code}")


def run(args: argparse.Namespace) -> int:
    if args.path is not None:
        path_arg: Path | None = Path(args.path)
        if not path_arg.is_absolute():
            path_arg = (Path.cwd() / path_arg).resolve()
    else:
        path_arg = None

    skip_env = os.environ.get("LINK_CHECK_SKIP", "")
    if skip_env.strip():
        skip_domains = {d.strip().lower() for d in skip_env.split(",") if d.strip()}
    else:
        skip_domains = set(DEFAULT_SKIP_DOMAINS)

    files = sorted(set(iter_markdown_files(ROOT, args.phase, path_arg)))

    occurrences: dict[str, list[UrlOccurrence]] = {}
    for f in files:
        try:
            text = f.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError):
            continue
        try:
            rel = f.relative_to(ROOT).as_posix()
        except ValueError:
            rel = str(f)
        for url, line in extract_urls(text):
            occurrences.setdefault(url, []).append(UrlOccurrence(file=rel, line=line))

    report = Report(checked_files=len(files), unique_urls=len(occurrences))

    cache = load_cache()
    to_check: list[str] = []
    results: dict[str, CheckResult] = {}

    for url in occurrences:
        if should_skip(url, skip_domains):
            report.skipped.append(url)
            continue
        entry = cache.get(url)
        if entry and cache_is_fresh(entry, args.cache):
            status = str(entry.get("status", "error"))
            http_status = entry.get("http_status")
            http_status_int = int(http_status) if isinstance(http_status, (int, float)) else None
            err = entry.get("last_error")
            results[url] = CheckResult(
                url=url,
                status=status,
                http_status=http_status_int,
                error=str(err) if err else None,
                cached=True,
            )
            report.cached_hits += 1
            continue
        to_check.append(url)

    report.requested = len(to_check)

    if to_check:
        with ThreadPoolExecutor(max_workers=max(1, args.concurrency)) as executor:
            futures = {executor.submit(check_url, url, args.timeout): url for url in to_check}
            for fut in as_completed(futures):
                url = futures[fut]
                try:
                    result = fut.result()
                except Exception as exc:
                    result = CheckResult(
                        url=url, status="error", http_status=None, error=f"executor: {exc}"
                    )
                results[url] = result
                cache[url] = {
                    "status": result.status,
                    "http_status": result.http_status,
                    "checked_at": time.time(),
                    "last_error": result.error,
                }
                if not args.json:
                    mark = "OK" if result.ok else "FAIL"
                    code = result.http_status if result.http_status is not None else "-"
                    print(f"  [{mark}] {code} {url}", file=sys.stderr)

    save_cache(cache)

    by_file: dict[str, list[dict[str, object]]] = {}
    for url, occs in occurrences.items():
        result = results.get(url)
        if result is None:
            continue
        if result.ok:
            continue
        for occ in occs:
            entry = {
                "url": url,
                "line": occ.line,
                "status": result.status,
                "http_status": result.http_status,
                "error": result.error,
                "cached": result.cached,
            }
            by_file.setdefault(occ.file, []).append(entry)
            report.failed.append({"file": occ.file, **entry})

    for fname in by_file:
        by_file[fname].sort(key=lambda e: (e["line"], e["url"]))
    report.by_file = dict(sorted(by_file.items()))

    if args.json:
        json.dump(report.to_dict(), sys.stdout, indent=2, sort_keys=True)
        sys.stdout.write("\n")
    else:
        print(f"checked {report.checked_files} markdown files", file=sys.stderr)
        print(f"unique urls: {report.unique_urls}", file=sys.stderr)
        print(f"requested:   {report.requested}", file=sys.stderr)
        print(f"cache hits:  {report.cached_hits}", file=sys.stderr)
        print(f"skipped:     {len(set(report.skipped))} urls in {len(skip_domains)} domains", file=sys.stderr)
        print(f"broken:      {len(report.failed)} occurrences across {len(report.by_file)} files", file=sys.stderr)
        if report.by_file:
            print("", file=sys.stderr)
            for fname, entries in report.by_file.items():
                print(fname, file=sys.stderr)
                for e in entries:
                    code = e["http_status"] if e["http_status"] is not None else e["error"]
                    print(f"  line {e['line']}: [{code}] {e['url']}", file=sys.stderr)

    if args.strict and report.failed:
        return 1
    return 0


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Validate external HTTP/HTTPS links across markdown docs.",
    )
    parser.add_argument("--phase", type=int, default=None, help="restrict to phase NN")
    parser.add_argument("--path", default=None, help="restrict to one file or directory")
    parser.add_argument("--strict", action="store_true", help="exit 1 if any link is broken")
    parser.add_argument("--json", action="store_true", help="emit machine-readable report")
    parser.add_argument(
        "--timeout",
        type=int,
        default=DEFAULT_TIMEOUT,
        help=f"per-request timeout in seconds (default: {DEFAULT_TIMEOUT})",
    )
    parser.add_argument(
        "--concurrency",
        type=int,
        default=DEFAULT_CONCURRENCY,
        help=f"worker threads (default: {DEFAULT_CONCURRENCY})",
    )
    parser.add_argument(
        "--cache",
        type=int,
        default=DEFAULT_CACHE_DAYS,
        help=f"cache TTL in days; 0 disables (default: {DEFAULT_CACHE_DAYS})",
    )
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(sys.argv[1:] if argv is None else argv)
    return run(args)


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