"""Relational core for spec §2.5 — bills ↔ committees ↔ members. The rest of the pipeline persists a FLAT table (``daily_file_records`` written by ``scraper.io.write_sqlite``). QA and the downstream UI read that flat contract, so this lane NEVER mutates it. Instead it ADDS a normalised relational schema and a flat-preserving VIEW on top of the same SQLite database, and derives the relations (bill↔committee, committee↔member) from the flat rows + an optional committee-roster source. FROZEN CONTRACTS honoured (see task t_2736f40d / parent t_21cd58a9): * ``daily_file_records`` stays fully intact (CREATE IF NOT EXISTS only). * New tables: ``bills``, ``committees``, ``members``, ``committee_members`` (committee↔member join), ``bill_committees`` (bill↔committee join), plus a ``schema_migrations`` ledger and a ``v_daily_file_records`` view. * FOREIGN KEYS are enabled on every connection; UNIQUE constraints on the natural keys (bill_id+session+chamber; committee name+session+chamber; committee+member). * Migration and correlation are IDEMPOTENT: CREATE IF NOT EXISTS + INSERT OR IGNORE on deterministic keys. Re-running adds no rows / no errors. * NEVER FABRICATE MEMBERS. ``committee_members`` is populated ONLY from a supplied roster source. When a committee seen in the daily file has no roster entry, it is still created (with its ``session``) but is left with an EMPTY member set and a WARN is logged listing which committees are missing rosters — we record the gap, we do not invent members. This module uses only the stdlib (sqlite3 + a tiny JSON roster reader). No new third-party dependency is introduced. """ from __future__ import annotations import json import logging import os import re import sqlite3 from typing import Any, Dict, Iterable, List, Optional log = logging.getLogger("scraper.relate") # --------------------------------------------------------------------------- # Key constants # --------------------------------------------------------------------------- #: Default database path (spec: db/daily_file.sqlite), resolved relative to the #: repository root so it is stable regardless of cwd. _REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DEFAULT_DB = os.path.join(_REPO_ROOT, "db", "daily_file.sqlite") FLAT_TABLE = "daily_file_records" #: Schema version stamped into ``schema_migrations`` by :func:`migrate`. # v1: bills/committees/members/committee_members/bill_committees + flat view. # v2: adds scheduled_events (the chamber "upcoming events" forward source — the # only genuine future-dated signal; see design/API_CONTRACT.md §5 + the # "what's next" work). Kept OUT of the frozen flat daily_file_records view. SCHEMA_VERSION = 2 #: The flat columns QA + downstream rely on. Imported so the flat-preserving #: view is byte-for-byte aligned with the contract the rest of the pipeline #: writes via ``scraper.io.write_sqlite``. try: # pragma: no cover - import guard (io is a sibling in the same package) from .io import _COLUMNS as FLAT_COLUMNS except Exception: # pragma: no cover # Fallback if io ever changes; keeps the module importable for unit tests. FLAT_COLUMNS = [ ("id", "TEXT"), ("date", "TEXT"), ("house", "TEXT"), ("source", "TEXT"), ("record_type", "TEXT"), ("bill_number", "TEXT"), ("bill_id", "TEXT"), ("session", "TEXT"), ("chamber_origin", "TEXT"), ("author", "TEXT"), ("short_title", "TEXT"), ("description", "TEXT"), ("committee", "TEXT"), ("hearing_time", "TEXT"), ("location", "TEXT"), ("item_number", "INTEGER"), ("supplement", "INTEGER"), ("ics_url", "TEXT"), ("agenda_id", "TEXT"), ("raw_selector", "TEXT"), ] _FLAT_COL_ORDER = [c for c, _ in FLAT_COLUMNS] _FLAT_COL_DEFS = ", ".join(f'"{c}" {t}' for c, t in FLAT_COLUMNS) _SCHEMA_TABLES = ( "bills", "committees", "members", "committee_members", "bill_committees", "schema_migrations", ) # --------------------------------------------------------------------------- # Normalisation helpers # --------------------------------------------------------------------------- _SLUG_RE = re.compile(r"[^a-z0-9]+") def normalise_name(raw: Optional[str]) -> str: """Title-case + trim a committee/author/member display name.""" if not raw: return "" s = re.sub(r"\s+", " ", raw.strip()) return s def slug(raw: Optional[str]) -> str: """Deterministic match key: lowercased, non-alphanumerics -> single '-'. Mirrors spec §6.1 ('Normalise by title-casing and stripping non-alphanumerics') so a roster's committee name resolves to the same committee the daily file references. """ if not raw: return "" s = _SLUG_RE.sub("-", raw.strip().lower()).strip("-") return s or "unknown" def session_key(raw: Optional[str]) -> str: """Coalesce a NULL/empty session to a stable sentinel ('') so the natural-key UNIQUE constraints stay effective and re-runs idempotent.""" return (raw or "").strip() def chamber_key(raw: Optional[str]) -> str: """Normalise chamber origin/house to a stable lowercase key.""" return (raw or "").strip().lower() or "unknown" # --------------------------------------------------------------------------- # Connection + schema DDL # --------------------------------------------------------------------------- def connect(path: str, foreign_keys: bool = True) -> sqlite3.Connection: """Open the DB, enforcing FK integrity (per-connection pragma).""" os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True) conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row if foreign_keys: # Must be set on the connection *before* any DDL/DML. conn.execute("PRAGMA foreign_keys = ON") return conn def _ensure_flat_table(conn: sqlite3.Connection) -> None: """Create the flat contract table ONLY if missing. Never alters it.""" conn.execute( f'CREATE TABLE IF NOT EXISTS {FLAT_TABLE} (' f"{_FLAT_COL_DEFS})" ) def _create_schema(conn: sqlite3.Connection) -> None: """Create all normalised tables + the flat-preserving view idempotently. Every object uses CREATE ... IF NOT EXISTS, so this is safe to run many times. Foreign keys link the relation tables to their parents. """ conn.executescript( """ CREATE TABLE IF NOT EXISTS bills ( id INTEGER PRIMARY KEY, bill_id TEXT NOT NULL, bill_number TEXT, session TEXT NOT NULL DEFAULT '', chamber_origin TEXT NOT NULL DEFAULT 'unknown', house TEXT, author TEXT, short_title TEXT, first_seen TEXT, source TEXT, UNIQUE (bill_id, session, chamber_origin) ); CREATE TABLE IF NOT EXISTS committees ( id INTEGER PRIMARY KEY, name_key TEXT NOT NULL, name_raw TEXT NOT NULL, session TEXT NOT NULL DEFAULT '', chamber TEXT NOT NULL DEFAULT 'unknown', hearing_time TEXT, location TEXT, source TEXT, UNIQUE (name_key, session, chamber) ); CREATE TABLE IF NOT EXISTS members ( id INTEGER PRIMARY KEY, name_norm TEXT NOT NULL, full_name TEXT, chamber TEXT NOT NULL DEFAULT 'unknown', kind TEXT NOT NULL DEFAULT 'roster', source TEXT, UNIQUE (name_norm, chamber, kind) ); CREATE TABLE IF NOT EXISTS committee_members ( id INTEGER PRIMARY KEY, committee TEXT NOT NULL, member TEXT NOT NULL, session TEXT NOT NULL DEFAULT '', chamber TEXT NOT NULL DEFAULT 'unknown', role TEXT, source TEXT, UNIQUE (committee, member, session, chamber) ); CREATE TABLE IF NOT EXISTS bill_committees ( id INTEGER PRIMARY KEY, bill_id TEXT NOT NULL, committee TEXT NOT NULL, session TEXT NOT NULL DEFAULT '', chamber TEXT NOT NULL DEFAULT 'unknown', source TEXT, UNIQUE (bill_id, committee, session, chamber) ); CREATE TABLE IF NOT EXISTS schema_migrations ( version INTEGER NOT NULL, applied_at TEXT, note TEXT, PRIMARY KEY (version) ); -- v2: the chamber "upcoming events" forward source. This is the ONLY -- genuine *future-dated* signal (the daily file is a backward snapshot; -- leginfo bill status/history is backward-only). It holds the next -- scheduled floor/committee events parsed from the chamber events -- calendar (assembly.ca.gov/events, senate.ca.gov/calendar). It is -- deliberately OUT of the frozen flat daily_file_records view. CREATE TABLE IF NOT EXISTS scheduled_events ( id INTEGER PRIMARY KEY, date TEXT NOT NULL, -- ISO YYYY-MM-DD (event day) chamber TEXT NOT NULL, -- ASSEMBLY | SENATE kind TEXT, -- FLOOR | COMMITTEE | OTHER title TEXT NOT NULL, -- human event title source_url TEXT, -- where the event came from fetched_at TEXT, -- when this page was read source TEXT, -- assembly_events | senate_events UNIQUE (date, chamber, kind, title, source) ); """ ) # Flat-preserving view: a drop-in mirror of the table QA/UI read. Column # order and row set are identical to daily_file_records, ordered # deterministically so the "flat contract unchanged" test is stable. # (Rebuild every migrate so the view tracks the current flat schema.) conn.execute(f"DROP VIEW IF EXISTS v_{FLAT_TABLE}") cols = ", ".join(f'"{c}"' for c in _FLAT_COL_ORDER) conn.execute( f"CREATE VIEW v_{FLAT_TABLE} AS SELECT {cols} FROM {FLAT_TABLE} " "ORDER BY date, house, bill_id, committee, id" ) conn.commit() def _stamp_migration(conn: sqlite3.Connection, version: int = SCHEMA_VERSION) -> None: conn.execute( "INSERT OR IGNORE INTO schema_migrations (version, applied_at, note) " "VALUES (?, ?, ?)", (version, __import__("datetime").date.today().isoformat(), "relational schema: bills/committees/members/committee_members/" "bill_committees + flat view + scheduled_events (v2)"), ) conn.commit() # --------------------------------------------------------------------------- # migrate / correlate # --------------------------------------------------------------------------- def migrate(path: str = DEFAULT_DB, flat_table: str = FLAT_TABLE, drop_flat: bool = False) -> Dict[str, Any]: """Create the relational schema idempotently, preserving the flat table. Safe on a fresh DB (creates everything) and on an existing DB (no-op for objects that already exist). Returns a small summary dict. """ conn = connect(path) try: _ensure_flat_table(conn) _create_schema(conn) _stamp_migration(conn) n_flat = conn.execute(f"SELECT COUNT(*) FROM {FLAT_TABLE}").fetchone()[0] return {"db": path, "migrated": True, "flat_rows": n_flat, "version": SCHEMA_VERSION, "objects": _list_objects(conn)} finally: conn.close() def _list_objects(conn: sqlite3.Connection) -> List[str]: rows = conn.execute( "SELECT name FROM sqlite_master WHERE type IN ('table','view') " "ORDER BY type DESC, name").fetchall() return [r["name"] for r in rows] def _load_flat_rows(conn: sqlite3.Connection) -> Iterable[sqlite3.Row]: return conn.execute(f"SELECT * FROM {FLAT_TABLE}").fetchall() def load_rosters(path) -> Dict: """Load committee rosters from a JSON file (the "create new info" path). Format (a list of committee blocks):: [ {"committee": "Budget", "session": "2025-2026", "chamber": "assembly", "members": [ {"name": "Roxas", "role": "Chair"}, {"name": "Liu", "role": "Vice Chair"} ]} ] Returns a dict keyed by ``(slug, session, chamber)`` -> block, so a committee that appears in distinct (session, chamber) slots keeps separate memberships instead of clobbering. Missing/empty file -> empty dict (caller then logs a WARN for the gap). A malformed file raises ValueError so the CLI can report it cleanly rather than silently producing no members. """ out: Dict[tuple, Dict[str, Any]] = {} if not path: return out with open(path, encoding="utf-8") as fh: data = json.load(fh) blocks = data if isinstance(data, list) else data.get("committees", []) for blk in blocks: name = blk.get("committee") or blk.get("name") if not name: continue key = (slug(name), session_key(blk.get("session")), chamber_key(blk.get("chamber"))) out[key] = { "name": normalise_name(name), "session": session_key(blk.get("session")), "chamber": chamber_key(blk.get("chamber")), "members": [ {"name": normalise_name(m.get("name")), "role": m.get("role")} for m in (blk.get("members") or []) if m.get("name") ], } return out def _ingest_relations(conn: sqlite3.Connection, rosters: Dict[str, Any]) -> Dict[str, Any]: """Derive + backfill normalised relations from the flat table. * ``bills`` <- flat rows with a bill_id * ``committees`` <- flat rows with a committee (hearing / committee tables) AND committee-like authors ('Committee on ...' strings name a committee too) * ``bill_committees`` <- flat rows that carry BOTH a bill_id and a committee (the "reviewed by" relation) * ``members`` <- roster members (kind='roster') plus bill authors (kind='author-seed') — real names, not guesses * ``committee_members`` <- roster membership ONLY (never fabricated) Returns a counts summary + the list of committees missing a roster. """ counts = {"bills": 0, "committees": 0, "bill_committees": 0, "members": 0, "committee_members": 0, "committees_missing_roster": 0} # Helper: run an INSERT OR IGNORE and report whether it changed a row # (rowcount == 0 means the natural key already existed -> idempotent dedup). def ins_added(cur) -> int: return 1 if cur and cur.rowcount else 0 seen_committees = set() # slugs we saw in the daily file for row in _load_flat_rows(conn): bill_id = row["bill_id"] if "bill_id" in row.keys() else None committee = row["committee"] if "committee" in row.keys() else None author = row["author"] if "author" in row.keys() else None sess = session_key(row["session"] if "session" in row.keys() else None) house = row["house"] if "house" in row.keys() else None origin = row["chamber_origin"] if "chamber_origin" in row.keys() else None # chamber_origin is the origin house of a bill; hearing rows have no # bill so chamber_origin is None -> fall back to the publishing `house` # (e.g. 'assembly') so the committee slot joins to its roster cleanly. cham = chamber_key(origin) if origin else chamber_key(house) seen = "daily_file" # --- bills ------------------------------------------------------- if bill_id: if ins_added(conn.execute( "INSERT OR IGNORE INTO bills " "(bill_id, bill_number, session, chamber_origin, house, " " author, short_title, first_seen, source) " "VALUES (?,?,?,?,?,?,?,?,?)", (bill_id, row["bill_number"], sess, cham, house, author, row["short_title"], row["first_seen"] if "first_seen" in row.keys() else None, seen))): counts["bills"] += 1 # --- committee-like author names a committee --------------------- author_committee = None if author and author.strip().lower().startswith("committee"): # "Committee on Budget and Fiscal Review" -> "Budget and Fiscal Review" author_committee = re.sub(r"^committees?\s+on\s+", "", author.strip(), flags=re.I) or author.strip() cham = chamber_key(row["chamber_origin"] if "chamber_origin" in row.keys() else None) for cname in {c for c in (committee, author_committee) if c}: ck = slug(cname) seen_committees.add((ck, sess, cham)) if ins_added(conn.execute( "INSERT OR IGNORE INTO committees " "(name_key, name_raw, session, chamber, hearing_time, " " location, source) VALUES (?,?,?,?,?,?,?)", (ck, normalise_name(cname), sess, cham, row["hearing_time"] if "hearing_time" in row.keys() and committee else None, row["location"] if "location" in row.keys() and committee else None, seen))): counts["committees"] += 1 # --- bill <-> committee ("reviewed by") -------------------------- if bill_id and committee: if ins_added(conn.execute( "INSERT OR IGNORE INTO bill_committees " "(bill_id, committee, session, chamber, source) " "VALUES (?,?,?,?,?)", (bill_id, slug(committee), sess, cham, seen))): counts["bill_committees"] += 1 # --- author -> member seed (a real legislator, not a guess) ----- if author and not author.strip().lower().startswith("committee"): stem = re.sub(r"\s+et al\.$", "", author.strip()) if re.fullmatch(r"[\w'\-]+", stem): if ins_added(conn.execute( "INSERT OR IGNORE INTO members " "(name_norm, full_name, chamber, kind, source) " "VALUES (?,?,?,?,?)", (slug(stem), stem, cham, "author-seed", seen))): counts["members"] += 1 # --- roster-driven committee_members (NEVER fabricated) ------------- # `rosters` is keyed by (slug, session, chamber); populate members for each # roster's own slot. A roster slot may use a (session, chamber) chosen by the # roster source; it is not required to mirror every seen slot exactly. roster_slugs = {k[0] for k in rosters} for blk in rosters.values(): for m in blk["members"]: if not m["name"]: continue if ins_added(conn.execute( "INSERT OR IGNORE INTO members " "(name_norm, full_name, chamber, kind, source) " "VALUES (?,?,?,?,?)", (slug(m["name"]), m["name"], blk["chamber"], "roster", "roster"))): counts["members"] += 1 if ins_added(conn.execute( "INSERT OR IGNORE INTO committee_members " "(committee, member, session, chamber, role, source) " "VALUES (?,?,?,?,?,?)", (slug(blk["name"]), slug(m["name"]), blk["session"], blk["chamber"], m.get("role"), "roster"))): counts["committee_members"] += 1 # --- gap reporting: committees seen but with NO roster (slug level) -- # A seen committee slot is "missing a roster" when no roster block covers # that slug at all -- we still created the committee row, but its member # set is left empty. Per the "create new info / never fabricate" rule we # record the gap (+WARN) instead of inventing members. missing = set() for (ck, sess, cham) in seen_committees: if ck not in roster_slugs: missing.add((ck, sess, cham)) counts["committees_missing_roster"] = len(missing) for (ck, sess, cham) in sorted(missing): log.warning( "committee %r (session=%s chamber=%s) has no roster source; " "membership left empty (not fabricated). Provide --roster to " "backfill.", ck, sess or "-", cham) conn.commit() return counts def correlate(path: str = DEFAULT_DB, rosters: Optional[str] = None, rebuild: bool = False, quiet: bool = False) -> Dict[str, Any]: """Derive relations from the flat table into the normalised tables. Idempotent by default (INSERT OR IGNORE on deterministic natural keys). ``rebuild=True`` clears the derived join/parent tables first so a run reflects the current flat snapshot exactly. """ conn = connect(path) try: _ensure_flat_table(conn) _create_schema(conn) _stamp_migration(conn) roster_data = load_rosters(rosters) if rosters else {} if rebuild: for t in ("bill_committees", "committee_members", "members", "committees", "bills"): conn.execute(f"DELETE FROM {t}") conn.commit() counts = _ingest_relations(conn, roster_data) # Final absolute counts (post-dedup) for reporting. for k in ("bills", "committees", "bill_committees", "members", "committee_members"): counts[k] = conn.execute(f"SELECT COUNT(*) FROM {k}").fetchone()[0] if not quiet: log.info("correlate: %s", json.dumps(counts, sort_keys=True)) return counts finally: conn.close() # --------------------------------------------------------------------------- # Demonstration / verification helpers # --------------------------------------------------------------------------- def sample_selects(path: str = DEFAULT_DB, limit: int = 5) -> Dict[str, Any]: """Pull a few human-readable sample SELECTs for the completion report.""" conn = connect(path) try: def q(sql: str, args: tuple = ()): return [dict(r) for r in conn.execute(sql, args).fetchall()] return { "counts": {t: conn.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] for t in ("bills", "committees", "members", "committee_members", "bill_committees")}, "sample_bill_committees": q( "SELECT bc.bill_id, c.name_raw AS committee, bc.chamber, " "bc.session FROM bill_committees bc " "JOIN committees c " " ON c.name_key = bc.committee " " AND c.chamber = bc.chamber " " AND c.session = bc.session LIMIT ?", (limit,)), "sample_committee_members": q( "SELECT DISTINCT c.name_raw AS committee, m.full_name AS " "member, cm.role, cm.chamber FROM committee_members cm " "JOIN committees c ON c.name_key = cm.committee " "JOIN members m ON m.name_norm = cm.member " "ORDER BY c.name_raw, m.full_name LIMIT ?", (limit,)), "committees_without_roster": q( "SELECT name_raw, chamber, session FROM committees c " "WHERE NOT EXISTS (SELECT 1 FROM committee_members cm " " WHERE cm.committee = c.name_key " " AND cm.chamber = c.chamber " " AND cm.session = c.session)"), } finally: conn.close() # --------------------------------------------------------------------------- # scheduled_events: the chamber "upcoming events" forward source (v2) # --------------------------------------------------------------------------- def _scheduled_events_table(conn: sqlite3.Connection, name: str = "scheduled_events") -> bool: cur = conn.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,)) return cur.fetchone() is not None def store_scheduled_events(events: Iterable[Dict[str, Any]], path: str = DEFAULT_DB, fetched_at: str = "") -> Dict[str, Any]: """Insert parsed chamber events into ``scheduled_events`` idempotently. Each ``event`` is a dict with keys ``date`` (ISO), ``chamber`` (upper), ``kind`` (``FLOOR``/``COMMITTEE``/``OTHER``), ``title``, ``source_url``, ``source``. The natural key ``(date, chamber, kind, title, source)`` makes re-runs a no-op (``INSERT OR IGNORE``), so the existing-cron behaviour ("re-parse doesn't re-hit") holds: a second parse of the same calendar day adds no rows. Ensures the table exists (via :func:`_create_schema`). Returns a small summary ``{stored, cleared_stale, total, fetched_at}``. """ conn = connect(path) try: _create_schema(conn) # CREATE IF NOT EXISTS; cheap + idempotent # Drop a stale day's rows first so a re-published calendar (a moved or # renamed hearing) reflects the latest read rather than a frozen # snapshot. Only that day is cleared; other days are untouched. days = {str(e.get("date")) for e in events if e.get("date")} deleted = 0 for d in days: cur = conn.execute( "DELETE FROM scheduled_events WHERE date=? AND fetched_at<>?", (d, fetched_at)) deleted += cur.rowcount rows = [] for e in events: if not e.get("date") or not e.get("title"): continue rows.append(( str(e["date"]).strip(), (e.get("chamber") or "unknown").strip().upper(), (e.get("kind") or "OTHER").strip().upper() or "OTHER", str(e["title"]).strip(), e.get("source_url") or None, fetched_at or None, (e.get("source") or "assembly_events").strip() or "assembly_events", )) if rows: cur = conn.executemany( "INSERT OR IGNORE INTO scheduled_events " "(date, chamber, kind, title, source_url, fetched_at, " "source) VALUES (?,?,?,?,?,?,?)", rows) stored = cur.rowcount else: stored = 0 conn.commit() total = conn.execute( "SELECT COUNT(*) FROM scheduled_events").fetchone()[0] return {"stored": stored, "cleared_stale": deleted, "total": total, "fetched_at": fetched_at or None} finally: conn.close() def load_scheduled_events(path: str = DEFAULT_DB, as_of: Optional[str] = None, chamber: Optional[str] = None, limit: int = 200) -> List[Dict[str, Any]]: """Return scheduled (chamber) events, optionally future-dated from ``as_of``. A ``None``/absent ``as_of`` keeps every event; a date string keeps ``date >= as_of`` (a T-1 event is excluded, matching API_CONTRACT §5). Degrades to ``[]`` when the table does not exist (pre-v2 DBs / fresh schema). """ conn = connect(path, foreign_keys=False) try: if not _scheduled_events_table(conn): return [] sql = ("SELECT date, chamber, kind, title, source_url, fetched_at, " "source FROM scheduled_events WHERE 1=1") clauses: list = [] args: list = [] if as_of: clauses.append("date >= ?") args.append(str(as_of)) if chamber: clauses.append("UPPER(chamber) = ?") args.append(chamber.strip().upper()) if clauses: sql += " AND " + " AND ".join(clauses) sql += " ORDER BY date, chamber, kind, title LIMIT ?" args.append(int(limit)) rows = conn.execute(sql, args).fetchall() finally: conn.close() return [dict(r) for r in rows] def next_legislative_day(path: str = DEFAULT_DB, as_of: Optional[str] = None, chamber: Optional[str] = None) -> Dict[str, Any]: """The earliest scheduled event on/after ``as_of`` (DESIGN §6.1 "next legislative day"), or a ``{date: None, recess: True}`` marker when nothing is scheduled (a genuine recess). This is the dashboard's "what's next" answer when the daily-file bill queue has no future rows: it surfaces the chamber events calendar. ``recess`` is True when there is no event after ``as_of``. """ events = load_scheduled_events(path, as_of=as_of, chamber=chamber, limit=1) if not events: return {"date": None, "recess": True} return {**events[0], "recess": False}