validate_links.py

#!/usr/bin/env python3
"""
Link validator — HEAD-checks every source link in the GC + SP + jurisprudence
metadata, updates `lastVerifiedAt` for successful checks, and writes a status
report.

Designed to run both:
  - Locally:           python3 validate_links.py
  - GitHub Actions:    .github/workflows/link-check.yml (weekly cron)

Behaviour:
  • Reads crc_gc_info.json + specialprocedures_info.json + jurisprudence_info.json.
  • For each unique URL, sends a HEAD request (falls back to GET on 405).
  • docs.un.org is a client-side viewer and answers 200 for ANY string, so a
    link there is resolved through documents.un.org/api/symbol/access instead.
  • Considers status 2xx as OK; anything else as broken.
  • Records with no checkable source — no link at all, or a link still on a
    retired OHCHR host — are counted under `knownUnavailable`, not `broken`:
    UN Documents has no entry for those symbols, so there is nothing to fix
    and no reason to reopen an issue about them every week.
  • docstore.ohchr.org answers 200 with an HTML error page for a dead
    FilesHandler token, so those links are judged by content type (status -2).
  • For OK links: bumps `lastVerifiedAt` to today's date in the metadata.
  • For broken links: collects (signature, link, status, reason) and writes
    a JSON report.
  • Optional: if --strict is set, exits 1 when any link is broken (so
    GitHub Actions can fail the workflow).

Outputs:
  link_status.json — full report, committed alongside the dataset:
    {
      "checkedAt":   "2026-04-28T08:00:00Z",
      "totalUnique": 250,
      "okCount":     245,
      "brokenCount": 5,
      "broken": [{ "signature": ..., "link": ..., "status": 403, "reason": "..." }]
    }
"""
from __future__ import annotations

import argparse
import json
import os
import ssl
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import date, datetime, timezone
from pathlib import Path

# Default paths — overridden in CLI / GH Actions if needed.
ROOT = Path(__file__).resolve().parent
GC_META = ROOT / 'mysite_pythonanywhere' / 'crc_gc_info.json'
SP_META = ROOT / 'mysite_pythonanywhere' / 'specialprocedures_info.json'
JUR_META = ROOT / 'mysite_pythonanywhere' / 'jurisprudence_info.json'
STATUS_OUT = ROOT / 'link_status.json'

# The jurisprudence metadata predates the GC/SP files and uses lower-case
# field names. Read and write through these rather than assuming one shape.
FIELD_ALIASES = {
    'link':      ('Link', 'link'),
    'signature': ('Signature', 'signature'),
    'name':      ('Name', 'name'),
    'committee': ('Committee', 'committee'),
}


def field(record: dict, logical: str) -> str:
    for key in FIELD_ALIASES[logical]:
        value = record.get(key)
        if value:
            return str(value)
    return ''


# Hosts that were retired when OHCHR took its document servers down. A record
# still pointing at one of them has no UN Documents entry to move to — the
# decision was published only inside a committee's annual report — so it is
# reported separately rather than counted as fresh link rot. See the
# 2026-08 migration in docs/assets/app.js (officialSourceUrl).
# docstore.ohchr.org is deliberately not listed: it came back, and its
# FilesHandler links were re-keyed record by record in 2026-09.
RETIRED_HOSTS = ('tbinternet.ohchr.org', 'juris.ohchr.org')


def is_retired_host(url: str) -> bool:
    host = urllib.parse.urlsplit(url).hostname or ''
    return host.lower() in RETIRED_HOSTS

# OHCHR's TLS chain is occasionally flaky from Python's stdlib; mirror what
# browsers tolerate. We do NOT skip this for arbitrary hosts — only OHCHR/UN.
SSL_CTX = ssl.create_default_context()
SSL_CTX.check_hostname = False
SSL_CTX.verify_mode = ssl.CERT_NONE

# UN servers reject Python's default User-Agent + sometimes have TLS chains
# that stdlib doesn't trust. Use a browser-like UA and a relaxed TLS context
# for known UN domains.
USER_AGENT = (
    'Mozilla/5.0 (compatible; GenevaReporter-LinkValidator/1.1; '
    '+https://github.com/lszoszk/generalcomments)'
)

UN_HOSTS = (
    'tbinternet.ohchr.org',
    'docstore.ohchr.org',
    'www.ohchr.org',
    'ohchr.org',
    'undocs.org',
    'docs.un.org',
    'documents.un.org',
    'www.un.org',
    'refworld.org',
    'www.refworld.org',
)


def _is_un(url: str) -> bool:
    return any(h in url for h in UN_HOSTS)


UN_DOC_LANGS = frozenset({'ar', 'zh', 'en', 'fr', 'ru', 'es'})


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def check_un_docs_symbol(url: str, timeout: float = 15.0) -> tuple[int, str]:
    """Resolve a docs.un.org/<lang>/<symbol> link through the UN access API.

    docs.un.org is a client-side viewer: it answers 200 for every symbol,
    including ones the UN document system has never heard of. Fetching the page
    therefore proves nothing. The viewer's own iframe calls
    documents.un.org/api/symbol/access, which redirects to the PDF for a symbol
    it knows and to /error for one it doesn't — that is the real check.
    """
    path = urllib.parse.urlsplit(url).path.strip('/')
    if not path:
        return (-1, 'no symbol in docs.un.org path')
    # The language segment is optional — undocs.org/A/50/440 and
    # docs.un.org/en/A/50/440 name the same document. Only strip a leading
    # segment that really is a language, or "A" gets eaten off the symbol.
    head, _, rest = path.partition('/')
    symbol = urllib.parse.unquote(
        rest if (rest and head.lower() in UN_DOC_LANGS) else path)
    api = ('https://documents.un.org/api/symbol/access?s='
           + urllib.parse.quote(symbol, safe='') + '&l=en&t=pdf')

    opener = urllib.request.build_opener(
        _NoRedirect, urllib.request.HTTPSHandler(context=SSL_CTX))
    try:
        req = urllib.request.Request(api, headers={'User-Agent': USER_AGENT},
                                     method='GET')
        try:
            with opener.open(req, timeout=timeout) as resp:
                # The access API always redirects; a bare 200 is a
                # maintenance or error page, not a verified document.
                return (0, f'no redirect from the access API (HTTP {resp.status})')
        except urllib.error.HTTPError as e:
            if e.code not in (301, 302, 303, 307, 308):
                return (e.code, e.reason or 'HTTPError')
            target = (e.headers.get('Location') or '') if e.headers else ''
            if '/error' in target:
                return (404, f'UN Documents has no record of {symbol}')
            if not target.upper().endswith('.PDF'):
                # A bare /doc/ target means the symbol is indexed but no file
                # sits behind it. E/1991/23 behaves this way; the (SUPP) form
                # of the same report resolves properly.
                return (404, f'{symbol} resolves to no file ({target or "empty"})')
            return (200, 'ok')
    except urllib.error.URLError as e:
        return (0, f'URLError: {e.reason}')
    except Exception as e:
        return (0, f'{type(e).__name__}: {str(e)[:80]}')


def check_url(url: str, timeout: float = 15.0) -> tuple[int, str]:
    """Returns (status_code, reason). status_code 0 means a network error."""
    if not url or not url.startswith('http'):
        return (-1, 'invalid url')

    if urllib.parse.urlsplit(url).hostname in ('docs.un.org', 'undocs.org'):
        return check_un_docs_symbol(url, timeout)

    headers = {'User-Agent': USER_AGENT, 'Accept': '*/*'}
    ctx = SSL_CTX if _is_un(url) else None
    # docstore's FilesHandler answers 200 for any token; a dead one just
    # carries an HTML error page instead of the PDF/DOC, so look at the type.
    docstore = urllib.parse.urlsplit(url).hostname == 'docstore.ohchr.org'

    for method in ('GET',) if docstore else ('HEAD', 'GET'):
        try:
            req = urllib.request.Request(url, headers=headers, method=method)
            with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
                ctype = (resp.headers.get('Content-Type') or '').lower()
                if docstore and (ctype.startswith('text/html')
                                 or is_retired_host(resp.geturl())):
                    return (-2, 'docstore: HTML error page (dead FilesHandler token)')
                return (resp.status, 'ok')
        except urllib.error.HTTPError as e:
            # Some servers don't support HEAD; retry with GET.
            if method == 'HEAD' and e.code in (403, 405, 501):
                continue
            return (e.code, e.reason or 'HTTPError')
        except urllib.error.URLError as e:
            return (0, f'URLError: {e.reason}')
        except Exception as e:
            return (0, f'{type(e).__name__}: {str(e)[:80]}')
    return (0, 'unreachable')


def collect_records() -> list[dict]:
    """Load every metadata file and tag each record with its source for write-back."""
    out = []
    for src_file in (GC_META, SP_META, JUR_META):
        if src_file.exists():
            try:
                data = json.loads(src_file.read_text())
                for r in data:
                    r['_src_file'] = str(src_file)
                    out.append(r)
            except Exception as e:
                print(f'WARNING: failed to parse {src_file}: {e}', file=sys.stderr)
    return out


def write_back(records: list[dict]) -> None:
    """Persist the per-record updates back to their source files."""
    by_file: dict[str, list[dict]] = {}
    for r in records:
        f = r.pop('_src_file', None)
        if f:
            by_file.setdefault(f, []).append(r)
    for f, rs in by_file.items():
        Path(f).write_text(json.dumps(rs, ensure_ascii=False, indent=2))


def run(args) -> int:
    today = date.today().isoformat()
    started = datetime.now(timezone.utc).isoformat()

    records = collect_records()
    if not records:
        print('No metadata records found.', file=sys.stderr)
        return 1

    # Build URL → list of records sharing that URL. Records without a link, and
    # links still on the retired OHCHR hosts, are held aside: they cannot be
    # checked into a healthy state, and re-testing them every week would keep
    # the link-rot issue permanently open over something already known and
    # unfixable.
    by_url: dict[str, list[dict]] = {}
    retired: list[dict] = []
    for r in records:
        link = field(r, 'link').strip()
        if not link or is_retired_host(link):
            retired.append(r)
        else:
            by_url.setdefault(link, []).append(r)

    total = len(by_url)
    print(f'Validating {total} unique URLs across {len(records)} records '
          f'(workers={args.workers}, timeout={args.timeout}s)...')

    results: dict[str, tuple[int, str]] = {}
    t0 = time.time()
    with ThreadPoolExecutor(max_workers=args.workers) as ex:
        futures = {ex.submit(check_url, url, args.timeout): url for url in by_url}
        for i, fut in enumerate(as_completed(futures), 1):
            url = futures[fut]
            try:
                results[url] = fut.result()
            except Exception as e:
                results[url] = (0, f'unexpected: {e}')
            if i % 25 == 0 or i == total:
                elapsed = time.time() - t0
                ok = sum(1 for s, _ in results.values() if 200 <= s < 300)
                print(f'  [{i}/{total}] checked · ok={ok} · {elapsed:.0f}s elapsed')

    # Serial re-check pass for anything that came back non-2xx. The
    # parallel pass hits some UN hosts (notably undocs.org) hard enough
    # that they throttle the runner IP — the requests come back as
    # status 0 (connection reset / timeout), NOT genuine link rot. A
    # single-threaded re-check with a longer timeout and a polite delay
    # recovers those false positives; anything that fails the slow pass
    # too is reported broken. Without this, the weekly CI run flags
    # ~30 working SP-report links every time it runs from GitHub
    # Actions (local runs from a residential IP see 360/360).
    # Only transient outcomes deserve the slow serial pass: network errors
    # (0), throttling and server errors. A 404 from the access API, an
    # invalid URL (-1) or a dead docstore token (-2) will not change.
    recheck = [u for u, (s, _) in results.items()
               if s == 0 or s in (403, 408, 425, 429) or 500 <= s < 600]
    if recheck:
        slow_timeout = max(args.timeout * 2, 40.0)
        print(f'\n  Re-checking {len(recheck)} non-2xx URL(s) serially '
              f'(timeout={slow_timeout:.0f}s)...')
        recovered = 0
        for j, url in enumerate(recheck, 1):
            status, reason = check_url(url, slow_timeout)
            results[url] = (status, reason)
            if 200 <= status < 300:
                recovered += 1
            time.sleep(1.0)   # be polite — throttle was the problem
            if j % 10 == 0 or j == len(recheck):
                print(f'    [{j}/{len(recheck)}] re-checked · recovered={recovered}')
        print(f'  Re-check recovered {recovered}/{len(recheck)} '
              f'(transient throttle/timeout, not link rot).')

    # Build the report
    broken: list[dict] = []
    n_ok = 0
    for url, (status, reason) in results.items():
        if 200 <= status < 300:
            n_ok += 1
            # Bump lastVerifiedAt for every record sharing this URL
            for r in by_url[url]:
                r['lastVerifiedAt'] = today
        else:
            for r in by_url[url]:
                broken.append({
                    'signature': field(r, 'signature'),
                    'docName':   field(r, 'name')[:120],
                    'committee': field(r, 'committee'),
                    'link':      url,
                    'status':    status,
                    'reason':    reason,
                })

    known_unavailable = sorted(
        {field(r, 'signature') for r in retired if field(r, 'signature')})

    report = {
        'schemaVersion': 2,
        'checkedAt':   started,
        'finishedAt':  datetime.now(timezone.utc).isoformat(),
        'totalUnique': total,
        'totalRecords': len(records),
        'okCount':     n_ok,
        'brokenCount': len(broken),
        'broken':      broken,
        # Records whose symbol UN Documents has no entry for — pre-2000
        # communications and early-2000s CAT decisions, published only inside
        # the committees' annual reports. Not actionable, so kept out of
        # brokenCount, but counted so the gap stays visible.
        'knownUnavailableCount': len(retired),
        'knownUnavailable': known_unavailable,
    }
    Path(args.out).write_text(json.dumps(report, ensure_ascii=False, indent=2))

    # Persist lastVerifiedAt updates
    if not args.dry_run:
        write_back(records)

    print()
    print(f'Result: {n_ok}/{total} OK · {len(broken)} broken')
    if retired:
        print(f'        {len(retired)} record(s) skipped — no UN Documents entry '
              f'for the symbol (no link, or still on a retired OHCHR host)')
    print(f'Status report: {args.out}')
    if broken[:5]:
        print('First broken links:')
        for b in broken[:5]:
            print(f'  [{b["status"]}] {b["signature"]:30s} → {b["link"][:80]}')

    if args.strict and len(broken) > 0:
        return 1
    return 0


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument('--out', default=str(STATUS_OUT),
                    help='Status report output path (default: link_status.json)')
    ap.add_argument('--workers', type=int, default=8,
                    help='Concurrent HTTP workers (default: 8)')
    ap.add_argument('--timeout', type=float, default=15.0,
                    help='Per-request timeout in seconds (default: 15)')
    ap.add_argument('--dry-run', action='store_true',
                    help="Don't persist lastVerifiedAt updates back to metadata files")
    ap.add_argument('--strict', action='store_true',
                    help='Exit 1 if any link is broken (for CI use)')
    args = ap.parse_args()

    return run(args)


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