canlex/refresh.py

"""Detect when CanLex's ingested sources have gone stale upstream.

    py -m canlex.refresh --check

For every Justice Laws instrument in config.SOURCES, this fetches just the
opening of the XML (the root <Statute>/<Regulation> element carries the
consolidation date in its lims:current-date attribute, so a few KB is enough)
and compares that upstream date against the current_to stored in the matching
data/processed/<code>.json. Anything where upstream is newer is reported as
drift, and the process exits non-zero -- so this can gate a deploy or run on a
schedule.

It also checks the two delegation instruments against upstream (check_delegation):
the CBSA IRPA delegation -- by diffing the dated 'irpa-lipr-YYYY-MM-DD' instruments
the CBSA legislation index publishes against those delegation.SOURCES ingests, so
a newly-published amendment is flagged -- and the IRCC IL3 instrument, by comparing
the version stamp in the live PDF against the stored one. Finally it prints the
stored currency of the remaining non-XML sources (D-memos, case law, agreements,
directives, benefits, tariff schedule), which expose no comparable upstream
signal, with the command to refresh each; and flags any tracked pending bill
whose coming-into-force date has passed.

This is the staleness detection the corpus previously lacked: legislation was
refreshed by hand, so e.g. Bill C-12 only landed when someone thought to look.

    py -m canlex.refresh --check          # human-readable; exit 1 on drift
    py -m canlex.refresh --check --json    # machine-readable for CI
"""
import datetime
import json
import re
import sys
import tempfile
import time
import urllib.request
from pathlib import Path

from . import delegation
from ._common import BROWSER_UA
from .config import SOURCES, PROCESSED_DIR

_UA = "CanLex-refresh/0.1"
_HEAD_BYTES = 16384      # the root element's attributes sit at the very top
_CURRENT_DATE = re.compile(rb'current-date="(\d{4}-\d{2}-\d{2})"')

# Aggregate (non-Justice-Laws) sources: (label, processed filename, refresh
# command, description). These carry no single upstream consolidation date, so
# the check reports their STORED currency (newest item) and how to refresh them
# rather than diffing against upstream. (Delegation is checked against upstream
# separately, by check_delegation, because its sources DO expose a comparable
# signal -- the CBSA instrument index and the IL3 version string.)
_NON_XML = [
    ("memorandum", "dmemos.json", "py -m canlex.dmemo", "CBSA D-Memoranda"),
    ("caselaw", "caselaw.json", "py -m canlex.caselaw", "Court & tribunal decisions"),
    ("agreement", "agreements.json", "py -m canlex.agreement", "FB collective agreement"),
    ("directive", "directives.json", "py -m canlex.directive", "NJC directives"),
    ("benefits", "benefits.json", "py -m canlex.benefits", "Benefit-plan booklets"),
    ("tariff", "tariff_schedule.json", "py -m canlex.tariff_schedule",
     "Customs Tariff Schedule ch. 98-99"),
]

# CBSA publishes its IRPA delegation as a base instrument plus incremental
# amendments (no consolidation), each a dated 'irpa-lipr-YYYY-MM-DD-eng.html'
# linked from the CBSA legislation index. A date on the index that we have not
# ingested means a new amendment to add to delegation.SOURCES.
_CBSA_INDEX_URL = ("https://www.cbsa-asfc.gc.ca/agency-agence/actreg-loireg/"
                   "legislation-eng.html")
_CBSA_INSTRUMENT_RE = re.compile(r"irpa-lipr-(\d{4}-\d{2}-\d{2})-eng\.html")
# The IRCC IL3 instrument is reissued with a seasonal version stamp on page 1.
_IL3_VERSION_RE = re.compile(r"(?:Spring|Summer|Fall|Winter)\s+\d{4}")

# Bills known to be heading into force that affect the corpus. Surfaced once
# their in-force date has passed, with the codes to re-ingest. (The legislation
# drift check above will independently flag the affected Act once Justice Laws
# consolidates the amendment; this list just names the cause.)
_PENDING_BILLS = [
    {"bill": "C-16", "name": "Protecting Victims Act",
     "in_force": "2026-07-18", "affects": ["C-46"],
     "note": "Criminal Code: coercive control, child-protection reporting "
             "(coercive-control provisions come into force later)."},
]


def _extract_current_date(raw):
    """Pull the lims:current-date (YYYY-MM-DD) from the head of a Justice Laws
    XML document. Pure and offline, for testability. Returns '' if absent."""
    if raw[:3] == b"\xef\xbb\xbf":
        raw = raw[3:]
    m = _CURRENT_DATE.search(raw)
    return m.group(1).decode("ascii") if m else ""


def _local_current_to(code):
    """The current_to stored in data/processed/<code>.json (first chunk), or ''."""
    path = PROCESSED_DIR / f"{code}.json"
    if not path.exists():
        return None
    chunks = json.loads(path.read_text(encoding="utf-8"))
    return chunks[0].get("current_to", "") if chunks else ""


def _remote_current_date(url):
    """Fetch only the head of the XML and extract its consolidation date.
    Returns (date, error): date is '' on a clean fetch that lacked the
    attribute, error is a short string when the fetch itself failed."""
    req = urllib.request.Request(url, headers={"User-Agent": _UA})
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            head = resp.read(_HEAD_BYTES)
        return _extract_current_date(head), ""
    except Exception as exc:                         # network / HTTP / timeout
        return "", f"{type(exc).__name__}: {exc}"


def check_legislation():
    """Compare each Justice Laws source's upstream date to the stored corpus.
    Returns a list of row dicts."""
    rows = []
    for code, src in SOURCES.items():
        local = _local_current_to(code)
        remote, error = _remote_current_date(src["xml_url"])
        if error:
            status = "error"
        elif local is None:
            status = "missing-local"      # configured but never ingested
        elif remote and remote > local:
            status = "stale"              # upstream newer than our corpus
        elif not remote:
            status = "no-date"            # fetched, but no date found (unusual)
        else:
            status = "ok"
        rows.append({"code": code, "short": src["short"], "local": local or "",
                     "remote": remote, "status": status, "error": error})
        time.sleep(0.3)                   # be polite to Justice Laws
    return rows


def non_xml_currency():
    """Stored currency (newest item) of the aggregate sources."""
    rows = []
    for label, fname, cmd, desc in _NON_XML:
        path = PROCESSED_DIR / fname
        if not path.exists():
            rows.append({"label": label, "desc": desc, "cmd": cmd,
                         "newest": "", "chunks": 0, "present": False})
            continue
        chunks = json.loads(path.read_text(encoding="utf-8"))
        newest = max((c.get("current_to", "") for c in chunks), default="")
        rows.append({"label": label, "desc": desc, "cmd": cmd,
                     "newest": newest, "chunks": len(chunks), "present": True})
    return rows


def _cbsa_ingested_dates():
    """The effective dates of the CBSA dated delegation instruments we ingest."""
    return sorted({s["effective"] for s in delegation.SOURCES.values()
                   if s.get("kind") == "html-cbsa" and s.get("effective")})


def _ircc_stored_version():
    """The IL3 version string recorded in the ingested corpus, or ''."""
    path = PROCESSED_DIR / "delegation.json"
    if not path.exists():
        return ""
    chunks = json.loads(path.read_text(encoding="utf-8"))
    versions = [c.get("current_to", "") for c in chunks
                if c.get("act_short") == "IRCC IL3" and c.get("current_to")]
    return max(versions) if versions else ""


def check_delegation():
    """Compare the delegation instruments against upstream. Returns row dicts.

    CBSA: diff the dated instruments published on the legislation index against
    those we ingest -- a new date means an un-ingested amendment. IRCC: compare
    the IL3 version stamp published in the live PDF against the stored one."""
    rows = []

    # --- CBSA: index diff (cheap; one HTML fetch) ---
    have = _cbsa_ingested_dates()
    try:
        req = urllib.request.Request(_CBSA_INDEX_URL,
                                     headers={"User-Agent": BROWSER_UA})
        with urllib.request.urlopen(req, timeout=60) as resp:
            html = resp.read().decode("utf-8", "replace")
        published = sorted(set(_CBSA_INSTRUMENT_RE.findall(html)))
        new = sorted(set(published) - set(have))
        rows.append({"name": "CBSA IRPA delegation (+ amendments)",
                     "status": "stale" if new else "ok",
                     "have": have, "published": published, "new": new,
                     "error": ""})
    except Exception as exc:
        rows.append({"name": "CBSA IRPA delegation (+ amendments)",
                     "status": "error", "have": have, "published": [],
                     "new": [], "error": f"{type(exc).__name__}: {exc}"})

    # --- IRCC IL3: version-stamp compare (re-fetches the PDF via PowerShell) ---
    stored = _ircc_stored_version()
    try:
        with tempfile.TemporaryDirectory() as td:
            data = delegation._fetch(delegation.SOURCES["ircc"]["url"],
                                     Path(td) / "il3.pdf", powershell=True)
        pages = delegation._pdf_pages(data)
        m = _IL3_VERSION_RE.search(pages[0]) if pages else None
        published = m.group(0) if m else ""
        if not published:
            status = "no-version"
        elif published != stored:
            status = "stale"
        else:
            status = "ok"
        rows.append({"name": "IRCC IL3 instrument", "status": status,
                     "have": [stored], "published": [published], "new": [],
                     "error": ""})
    except Exception as exc:
        rows.append({"name": "IRCC IL3 instrument", "status": "error",
                     "have": [stored], "published": [], "new": [],
                     "error": f"{type(exc).__name__}: {exc}"})
    return rows


def pending_in_force(today):
    """Pending bills whose in-force date has passed as of `today` (ISO str)."""
    return [b for b in _PENDING_BILLS if b["in_force"] <= today]


def run(as_json=False):
    today = datetime.date.today().isoformat()
    leg = check_legislation()
    deleg = check_delegation()
    nonxml = non_xml_currency()
    pending = pending_in_force(today)
    stale = [r for r in leg if r["status"] in ("stale", "missing-local")]
    errors = [r for r in leg if r["status"] == "error"]
    deleg_stale = [r for r in deleg if r["status"] == "stale"]

    if as_json:
        print(json.dumps({"checked": today, "legislation": leg,
                          "delegation": deleg, "non_xml": nonxml,
                          "pending_in_force": pending,
                          "stale_count": len(stale) + len(deleg_stale),
                          "error_count": len(errors)}, indent=2))
    else:
        print(f"CanLex corpus staleness check — {today}\n")
        print(f"Justice Laws legislation ({len(leg)} instruments):")
        for r in sorted(leg, key=lambda x: (x["status"] != "stale", x["short"])):
            if r["status"] == "stale":
                print(f"  STALE  {r['short']:34} local {r['local']}  "
                      f"upstream {r['remote']}")
            elif r["status"] == "missing-local":
                print(f"  ABSENT {r['short']:34} configured but not ingested "
                      f"(upstream {r['remote']})")
            elif r["status"] == "error":
                print(f"  ERROR  {r['short']:34} {r['error'][:50]}")
            elif r["status"] == "no-date":
                print(f"  ?      {r['short']:34} no current-date found upstream")
        if not stale and not errors:
            print(f"  all {len(leg)} current (local matches or exceeds upstream)")
        if stale:
            codes = " ".join(r["code"] for r in stale)
            print(f"\n  -> {len(stale)} drifted. Refresh: "
                  f"py -m canlex.ingest --force {codes}\n"
                  f"     then: py -m canlex.embed && py -m canlex.eval")

        print("\nDelegation instruments (upstream check):")
        for r in deleg:
            if r["status"] == "stale" and r["new"]:
                print(f"  STALE  {r['name']}: new upstream {', '.join(r['new'])} "
                      f"not ingested")
            elif r["status"] == "stale":
                print(f"  STALE  {r['name']}: upstream {r['published'][0] or '?'} "
                      f"!= stored {r['have'][0] or '?'}")
            elif r["status"] == "error":
                print(f"  ERROR  {r['name']}: {r['error'][:50]}")
            elif r["status"] == "no-version":
                print(f"  ?      {r['name']}: no version stamp found upstream")
            elif len(r["published"]) > 1:        # CBSA: a set of dated instruments
                print(f"  ok     {r['name']}: current "
                      f"({len(r['published'])} instruments, latest "
                      f"{max(r['published'])})")
            else:                                # IRCC: a single version stamp
                shown = r["published"][0] if r["published"] else "current"
                print(f"  ok     {r['name']}: current ({shown})")
        if deleg_stale:
            print("     -> re-ingest: add any new CBSA dates to delegation.SOURCES, "
                  "then py -m canlex.delegation && py -m canlex.embed")

        print("\nOther sources (stored currency; re-run the ingester to refresh):")
        for r in nonxml:
            mark = " " if r["present"] else "!"
            print(f" {mark}{r['label']:14} {str(r['chunks']):>5} chunks  "
                  f"newest {r['newest'] or 'n/a':12}  {r['cmd']}")

        if pending:
            print("\nPending bills now in force (re-ingest the affected Acts):")
            for b in pending:
                print(f"  Bill {b['bill']} ({b['name']}) in force {b['in_force']} "
                      f"-> {', '.join(b['affects'])}")
                print(f"     {b['note']}")
                print(f"     run: py -m canlex.ingest --force {' '.join(b['affects'])}")

    return 1 if (stale or errors or deleg_stale) else 0


def main():
    args = sys.argv[1:]
    as_json = "--json" in args
    # --check is the only mode; accepted (and default) for forward-compatibility.
    sys.exit(run(as_json=as_json))


if __name__ == "__main__":
    main()