"""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 / 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/.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/.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()