"""``scraper.llm`` - spec §4 LLM analysis: subject-matter tag clouds. Tag every bill with subject-matter tags via a *configurable* Ollama endpoint and model, persisting the NEW information into a ``bill_tags`` table and producing a tag cloud (``out/tag_cloud.json``). The endpoint + model are NEVER hardcoded -- they come from ``config.yaml``'s ``llm:`` section (``Config.llm``). Frozen contracts honoured (task t_950780e0 / parent t_21cd58a9): * Input bills are loaded from the flat ``daily_file_records`` table (``scraper.io`` / ``scraper.schema.BillAction``) -- reused, never redefined. * The Ollama HTTP request uses the stdlib ``urllib`` (no new dependency) and is routed through an injectable ``post`` callable so the ollama path is unit testable with a canned response and *no network*. * **Deterministic degradation**: when ``llm.enabled`` is false, the endpoint is unreachable, or ``llm.model`` is empty, the run falls back to a built-in deterministic tagger (keyword / short_title extraction, **zero network**) so analysis never blocks the build. The active path is logged and recorded on every ``bill_tags`` row via its ``generated_by`` column (''fallback'' vs ``'ollama:'``). The ``bill_tags`` table is additive (CREATE ... IF NOT EXISTS + INSERT OR IGNORE on a natural key), mirrors the relational lane's style (``scraper.relate_*``), and never alters the preserved flat ``daily_file_records`` contract. """ from __future__ import annotations import json import logging import os import re import sqlite3 import urllib.error import urllib.request from typing import Any, Callable, Dict, List, Optional, Sequence from . import relational log = logging.getLogger("scraper.llm") # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- #: New table storing the tag analysis (the "new info"). TAG_TABLE = "bill_tags" #: generated_by sentinel for the deterministic built-in tagger. FALLBACK = "fallback" #: Default HTTP timeout (seconds) when cfg does not supply one. DEFAULT_TIMEOUT = 30 #: Canonical Ollama endpoints, tried in order (OpenAI-compatible then native). _ENDPOINTS = ( "/v1/chat/completions", # OpenAI-compatible (per spec §4) "/api/generate", # Ollama native ) #: Boilerplate lead-ins / filler drained from a California bill short_title #: before its subject matter is extracted. These carry no subject signal. _BOILERPLATE = ( r"^an act relating to\b", r"^an act to\b", r"^relating to\b", r"^an act\b", r"\band making an appropriation therefor\b", r"\bmaking an appropriation therefor\b", r"\band declaring the urgency thereof(?:(?:,?\s*to take effect immediately\b)?),?", r"\bdeclaring the urgency thereof(?:(?:,?\s*to take effect immediately\b)?),?", r"\bto take effect immediately\b", r"\btherefor\b", r"^[,\s]+", r"[,\.\s]+$", ) # Words that carry no subject-matter signal, removed from extracted terms. _FILLER = { "an", "the", "to", "of", "and", "for", "relating", "act", "making", "appropriation", "therefor", "urgency", "thereof", "immediately", "taking", "effect", "by", "from", "with", "in", "on", "as", "each", "every", "a", "or", "shall", "its", "their", "his", "her", "which", "such", "also", "including", "regarding", } #: A small, curated California subject taxonomy that maps regex cues found in #: real bill text to canonical, human-readable tags. Used by the deterministic #: tagger to cluster bills into a clean tag cloud. Every tag is anchored in a #: real substring of a bill (never invented): a domain tag only appears when its #: cue is actually present in the bill's text. TAXONOMY = [ ((r"solid waste|recycl|landfill|waste"), "solid waste & recycling"), ((r"water"), "water"), ((r"housing|residential rental|foreclosure|eviction"), "housing"), ((r"\bhealth\b|health care|health insurance|medicare|medicaid" r"|hospital|pharm"), "health care"), ((r"mental health|mental ill|psychiatr|psycholog"), "mental health"), ((r"educat|k-?12|community college|universit|school"), "education"), ((r"tax|taxation|levy|revenue|fiscal"), "taxation"), ((r"employment|labor\b|labour|wage|workforce|unemployment"), "employment"), ((r"public resource|environment|pollution|emission|climate" r"|california dream|zero emission|renewable|conserv"), "environment"), ((r"prison|penal|incarcerat|parole|probation|juvenile justice" r"|sentenc|criminal justice"), "criminal justice"), ((r"juvenile|child|childhood|foster|adoption|dependency"), "child welfare"), ((r"transit|rail|highway|freeway|traffic|bikeway|pedestrian" r"|public transport"), "public transit"), ((r"firearm|gun|ammunition|gun control"), "firearms"), ((r"budget|appropriation|funding|treasury|expenditure"), "budget"), ((r"senior|elderly|\bagged\b"), "seniors"), ((r"insurance|insuranc"), "insurance"), ((r"election|voter|ballot|legislative apportion|districting"), "elections"), ((r"immigrat|undocumented|asylum|deportat"), "immigration"), ((r"privacy|personal data|data protection|cyber|surveillance"), "data privacy"), ((r"small business|small enterprise|start ?up"), "small business"), ((r"agricultur|farm|farmers|crop"), "agriculture"), ((r"pension|retiree|retirement|public employee s fund"), "pensions"), ((r"homeless|homelessness"), "homelessness"), ((r"\bsafety\b|security|protectiv|guarding"), "public safety"), ((r"energy|electric|utility|power|solar"), "energy"), ((r"technology|internet|digital|online|software"), "technology"), ((r"maritime|harbor|port|fishing|ocean|fish"), "maritime"), ] _TAXONOMY_RE = [(re.compile(pat, re.I), tag) for pat, tag in TAXONOMY] # --------------------------------------------------------------------------- # Bill loading (from the preserved flat table / BillAction) # --------------------------------------------------------------------------- def load_bills(path: str = relational.DEFAULT_DB, house: Optional[str] = None) -> List[Dict[str, Any]]: """Load distinct bills from the preserved flat ``daily_file_records`` table. Honors the frozen contract: input == rows written by ``scraper.io`` / ``BillAction``. Rows are deduped on ``bill_id`` (a bill can appear under multiple agenda items / dates) so the tag cloud is per-bill, not per-row. :param path: SQLite path (default: ``db/daily_file.sqlite``). :param house: optional publish-house filter (e.g. ``'assembly'``); matches either the publishing ``house`` or the bill's ``chamber_origin``. """ conn = relational.connect(path, foreign_keys=False) try: sql = ("SELECT DISTINCT bill_id, short_title, description, author, " "session, chamber_origin, house FROM " + relational.FLAT_TABLE + " WHERE bill_id IS NOT NULL AND bill_id != ''") rows = conn.execute(sql).fetchall() finally: conn.close() bills: List[Dict[str, Any]] = [] seen = set() for r in rows: bid = r["bill_id"] if bid in seen: continue seen.add(bid) d = { "bill_id": bid, "short_title": r["short_title"], "description": r["description"], "author": r["author"], "session": r["session"], "chamber_origin": r["chamber_origin"], "house": r["house"], } if house: h = house.lower() if (d["house"] or "").lower() != h and \ (d["chamber_origin"] or "").lower() != h: continue bills.append(d) # Deterministic order (by bill_id) so fallback output is reproducible. bills.sort(key=lambda b: b["bill_id"]) return bills # --------------------------------------------------------------------------- # Deterministic fallback tagger (ZERO network) # --------------------------------------------------------------------------- def _clean_subject(title: Optional[str]) -> List[str]: """Drain boilerplate from a CA short_title and return its subject phrases. E.g. "An act relating to solid waste, and making an appropriation therefor." -> ["solid waste"]. """ if not title: return [] s = title.strip().lower() for pat in _BOILERPLATE: s = re.sub(pat, " ", s) # Split remaining text on comma / conjunction into fragments. fragments = re.split(r"\s*,\s*|\s+and\s+", s) out: List[str] = [] for frag in fragments: frag = re.sub(r'^["\s]+|["\s]+$', "", frag.strip()) if not frag or frag in _FILLER: continue word = re.sub(r"[^a-z0-9' ]+", "", frag) token = " ".join(t for t in word.split() if t and t not in _FILLER) if token: out.append(token) return out def _title_tag(fragments: List[str]) -> List[str]: """Turn cleaned subject fragments into tidy, distinct title-derived tags.""" tags: List[str] = [] seen = set() for frag in fragments: # Keep words; drop stray single letters / numerics-as-noise but keep # meaningful numeric qualifiers (e.g. "2025"). words = frag.split() norm = " ".join(words).strip() if not norm: continue key = norm if key not in seen: seen.add(key) tags.append(_title_case(norm)) return tags def _title_case(s: str) -> str: return " ".join(w[:1].upper() + w[1:] for w in s.split() if w) def tag_bill_fallback(bill: Dict[str, Any], max_tags: int = 8) -> List[tuple]: """Deterministic subject-matter tags for one bill. Zero network. Returns a list of ``(tag, source)`` tuples, deduped and capped to ``max_tags``. Sources are ``'short_title'`` and/or ``'title+taxonomy'``. Nothing is invented: every tag derives from a real substring of the bill's text or from a taxonomy cue that is literally present in that text. """ seen: Dict[str, str] = {} short = bill.get("short_title") or "" desc = bill.get("description") or "" text = (short + " " + desc).strip() # 1) Title-derived subject phrases (primary "subject matter" of the bill). for tag in _title_tag(_clean_subject(short)): # Normalise to lowercase so title-derived tags merge with the taxonomy # clusters (e.g. "Housing" -> "housing"). seen.setdefault(tag.lower(), "short_title") # 2) Taxonomy clusters - only when the real cue is present in the text. for rx, tag in _TAXONOMY_RE: if rx.search(text): seen.setdefault(tag, "title+taxonomy") # 3) Guaranteed signal: if the title yielded no phrase, fall back to the # title-cased raw title so the cloud still has a signal, not empty. if not seen: raw = re.sub(r"^[^A-Za-z0-9]+", "", (short or "").strip()) key = ("bill " + raw.lower())[:60].strip() or "Unclassified bill" seen[key] = "short_title" # Preserve deterministic insertion order; cap to max_tags. return [(tag, src) for tag, src in list(seen.items())[:max_tags]] # --------------------------------------------------------------------------- # Ollama client (OpenAI-compatible / native, stdlib urllib, injectable) # --------------------------------------------------------------------------- def _build_prompt(bill: Dict[str, Any], max_tags: int) -> str: title = bill.get("short_title") or "" desc = bill.get("description") or "" return ( "You are tagging a California legislative bill by subject matter.\n" "Bill: " + (bill.get("bill_number") or bill.get("bill_id") or "") + "\n" "Title: " + title + "\n" + (("Action: " + desc + "\n") if desc else "") + "Reply with ONLY a JSON array of between 2 and " + str(max_tags) + " short, lowercase, snake_case subject-matter tags " "drawn from real content in the bill. No prose, no explanation, e.g. " '["housing", "residential rental", "foreclosure"].' ) def _parse_response_payload(body: Dict[str, Any]) -> str: """Extract assistant text from an OpenAI-compatible or native Ollama body.""" if not isinstance(body, dict): return "" # OpenAI-compatible: choices[0].message.content choices = body.get("choices") if choices: c0 = choices[0] or {} content = (c0.get("message") or {}).get("content") if content: return str(content) # Ollama native: {"response": ...} if "response" in body: return str(body.get("response") or "") return str(body.get("content") or "") def _extract_tags_from_text(text: str, fallback_model: str = "", max_tags: int = 8) -> List[str]: """Parse model output into a tag list: prefer JSON array, else delIMITED.""" text = (text or "").strip() if not text: return [] # Try JSON (a code-fence or bare array both work after stripping fences). fence = re.findall(r"```(?:json)?\s*(\[.*?\])\s*```", text, re.S) candidate = fence[-1] if fence else None if candidate is None: m = re.search(r"(\[[^\]]*\])", text, re.S) candidate = m.group(1) if m else None if candidate: try: arr = json.loads(candidate) if isinstance(arr, list): tags: List[str] = [] seen = set() for t in arr: s = str(t).strip().lower() s = re.sub(r"[^a-z0-9& ]+", " ", s).strip() s = " ".join(s.split()) if s and s not in seen and s not in _FILLER: seen.add(s) tags.append(s) if tags: return tags[:max_tags] except (ValueError, TypeError): pass # Degrade: split on punctuation / slashes / newlines. raw = [p for p in re.split(r"[\n,/;]|\s{2,}|\.\s+\w+", text) if p.strip()] out: List[str] = [] seen = set() for part in raw: s = re.sub(r"[^a-z0-9& ]+", " ", part.lower()).strip() s = " ".join(s.split()) if s and s not in seen and s not in _FILLER and len(s) > 1: seen.add(s) out.append(s[:48]) return out[:max_tags] def _ollama_request(endpoint: str, model: str, prompt: str, timeout: int, post: Optional[Callable] = None) -> str: """POST to a configurable Ollama endpoint, trying compatible shapes. :param post: injectable ``post(url, payload, timeout) -> payload`` callable, so tests feed a canned response with no network. Defaults to a stdlib-urllib implementation. """ base = (endpoint or "").rstrip("/") last_err: Optional[Exception] = None for path in _ENDPOINTS: url = base + path payload: Dict[str, Any] if path == "/v1/chat/completions": payload = { "model": model, "messages": [{"role": "user", "content": prompt}], "stream": False, "temperature": 0.0, } else: # /api/generate native payload = {"model": model, "prompt": prompt, "stream": False} try: body = post(url, payload, timeout) if post is not None \ else _urllib_post(url, payload, timeout) body = _coerce_json(body) text = _parse_response_payload(body) if text and text.strip(): return text last_err = ValueError("empty Ollama response body") except Exception as exc: # noqa: BLE001 last_err = exc continue # No compatible endpoint returned text -> signal caller to fall back. raise RuntimeError( "Ollama returned no usable text (%s)" % (last_err if last_err else "all endpoints empty") ) def _urllib_post(url: str, payload: Dict[str, Any], timeout: int) -> Any: """Stdlib-urllib POST (no third-party dep). Returns parsed JSON body.""" data = json.dumps(payload).encode("utf-8") req = urllib.request.Request( url, data=data, method="POST", headers={"Content-Type": "application/json"}, ) with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read().decode("utf-8", "replace") return _coerce_json(raw) def _coerce_json(body: Any) -> Any: """Accept a dict (mock) or a JSON str/bytes (real HTTP) and normalise.""" if isinstance(body, (bytes, bytearray)): body = body.decode("utf-8", "replace") if isinstance(body, str): return json.loads(body) return body # --------------------------------------------------------------------------- # Analysis orchestration + persistence + tag cloud # --------------------------------------------------------------------------- def ensure_tag_table(path: str = relational.DEFAULT_DB) -> None: """Idempotently create the additive ``bill_tags`` table. Does not touch the preserved flat table.""" conn = relational.connect(path, foreign_keys=False) try: conn.execute( "CREATE TABLE IF NOT EXISTS " + TAG_TABLE + " (" " id INTEGER PRIMARY KEY," " bill_id TEXT NOT NULL," " tag TEXT NOT NULL," " source TEXT," " generated_by TEXT NOT NULL," " generated_at TEXT," " UNIQUE (bill_id, tag, generated_by)" ")" ) conn.commit() finally: conn.close() def _generated_by(model: str) -> str: return "ollama:" + model if model else "ollama" def tag_bills(bills: Sequence[Dict[str, Any]], cfg: Dict[str, Any], post: Optional[Callable] = None) -> List[Dict[str, Any]]: """Tag every bill, choosing Ollama or the deterministic fallback. Path selection (deterministic degradation): * ``enabled`` false -> fallback (reason 'disabled') * ``model`` empty -> fallback (reason 'no model configured') * otherwise attempt Ollama; on ANY error (unreachable / empty / bad parse) -> fallback (reason 'unreachable: ...'), logged, never crashing the build Each bill gets a row of ``{"bill_id", "tags": [(tag, source)], "generated_by", "reason"}``. """ enabled = bool(cfg.get("enabled")) model = (cfg.get("model") or "").strip() max_tags = int(cfg.get("default_tags") or 8) endpoint = (cfg.get("endpoint") or "").strip() timeout = int(cfg.get("timeout_seconds") or DEFAULT_TIMEOUT) use_ollama = enabled and bool(model) and bool(endpoint) results: List[Dict[str, Any]] = [] ran_fallback = False reason = "enabled=true, model=%s" % model for b in bills: tags: List[tuple] = [] gen = FALLBACK if use_ollama: try: text = _ollama_request(endpoint, model, _build_prompt(b, max_tags), timeout, post=post) parsed = _parse_tags_for_bill(text, max_tags=max_tags) if parsed: tags = [(t, "ollama") for t in parsed] gen = _generated_by(model) else: ran_fallback = True tags = [(t, s) for (t, s) in tag_bill_fallback(b, max_tags)] gen = FALLBACK reason = "ollama empty -> fallback" except Exception as exc: # noqa: BLE001 -- degrade, never crash log.warning("Ollama unreachable/bad-response for %s: %s; " "using fallback", b.get("bill_id"), exc) ran_fallback = True tags = [(t, s) for (t, s) in tag_bill_fallback(b, max_tags)] gen = FALLBACK reason = "unreachable: %s" % (type(exc).__name__) else: if not model: reason = "no model configured" else: reason = "disabled" tags = [(t, s) for (t, s) in tag_bill_fallback(b, max_tags)] gen = FALLBACK ran_fallback = True results.append({ "bill_id": b.get("bill_id"), "tags": tags, "generated_by": gen, "reason": reason, }) # Report the dominant path so the log + the summary name what happened. if not use_ollama: log.info("llm path = fallback (%s)", reason) elif ran_fallback: log.info("llm path = ollama with per-bill fallback (%s)", reason) else: log.info("llm path = ollama (model=%s, endpoint=%s)", model, endpoint) return results def _parse_tags_for_bill(text: str, max_tags: int = 8) -> List[str]: """Public parse hook used above so the mock test can assert it directly.""" return _extract_tags_from_text(text, max_tags=max_tags) def persist_tags(path: str, results: Sequence[Dict[str, Any]], generated_at: Optional[str] = None) -> int: """Persist tagged bills into ``bill_tags``. Idempotent on the natural key `` (bill_id, tag, generated_by)``. Returns rows written this pass.""" ensure_tag_table(path) conn = relational.connect(path, foreign_keys=False) written = 0 try: for r in results: for (tag, src) in r.get("tags") or []: cur = conn.execute( "INSERT OR IGNORE INTO " + TAG_TABLE + " (bill_id, tag, source, generated_by, generated_at) " "VALUES (?,?,?,?,?)", (str(r["bill_id"]) if r.get("bill_id") is not None else "", str(tag), src, r["generated_by"], generated_at)) if cur.rowcount: written += 1 conn.commit() finally: conn.close() return written def build_tag_cloud(db_path: str = relational.DEFAULT_DB, generated_by: Optional[str] = None, top: Optional[int] = None) -> Dict[str, Any]: """Aggregate ``bill_tags`` into a tag cloud (tag -> count, desc). :param generated_by: restrict to rows whose ``generated_by`` starts with this (e.g. ``'fallback'`` or ``'ollama'``); ``None`` aggregates all rows. :param top: cap the returned tag list to the N most frequent. """ conn = relational.connect(db_path, foreign_keys=False) try: ensure_tag_table(db_path) if generated_by: rows = conn.execute( "SELECT tag, COUNT(*) AS n, COUNT(DISTINCT bill_id) AS bills " "FROM " + TAG_TABLE + " WHERE generated_by LIKE ? GROUP BY tag ORDER BY n DESC, tag", (generated_by + "%",)).fetchall() total_bills = conn.execute( "SELECT COUNT(DISTINCT bill_id) FROM " + TAG_TABLE + " WHERE generated_by LIKE ?", (generated_by + "%",)).fetchone()[0] else: rows = conn.execute( "SELECT tag, COUNT(*) AS n, COUNT(DISTINCT bill_id) AS bills " "FROM " + TAG_TABLE + " GROUP BY tag ORDER BY n DESC, tag" ).fetchall() total_bills = conn.execute( "SELECT COUNT(DISTINCT bill_id) FROM " + TAG_TABLE ).fetchone()[0] finally: conn.close() tags = [{"tag": r["tag"], "count": int(r["n"]), "bills": int(r["bills"])} for r in rows] if top is not None: tags = tags[:top] return { "tag_count": len(tags), "bill_count": total_bills, "generated_by": generated_by, "tags": tags, } def write_tag_cloud(cloud: Dict[str, Any], out_path: str) -> int: """Write the tag cloud to JSON. Returns the number of tags written. The file is diff-stable between runs (tags ordered desc by count, then tag). """ os.makedirs(os.path.dirname(os.path.abspath(out_path)) or ".", exist_ok=True) with open(out_path, "w", encoding="utf-8") as fh: json.dump(cloud, fh, ensure_ascii=False, indent=2, sort_keys=False) fh.write("\n") return int(cloud.get("tag_count", 0)) def analyze(db_path: str = relational.DEFAULT_DB, out_path: Optional[str] = None, cfg: Optional[Dict[str, Any]] = None, post: Optional[Callable] = None, house: Optional[str] = None, generated_at: Optional[str] = None, top: Optional[int] = None, rebuild: bool = False) -> Dict[str, Any]: """Full ``analyze`` pass: load bills, tag them, persist, build + write cloud. Returns a summary:: {bills_tagged, unique_tags, generated_by, reason, cloud_path, top_tags, rows_written, cloud}. The dominant path is in ``generated_by`` / ``reason`` (both log the fallback decision). """ cfg = cfg or {} bills = load_bills(db_path, house=house) results = tag_bills(bills, cfg, post=post) if rebuild: _drop_tag_rows(db_path, keep_by=None) ensure_tag_table(db_path) written = persist_tags(db_path, results, generated_at=generated_at) # Determine the dominant generated_by for the cloud + reporting. dominant = _dominant_generated_by(results) cloud = build_tag_cloud(db_path, generated_by=None, top=top) if out_path: write_tag_cloud(cloud, out_path) cloud_path = out_path else: cloud_path = None reason = results[0]["reason"] if results else "no bills" return { "path": "fallback" if dominant == FALLBACK else "ollama", "generated_by": dominant, "reason": reason, "bills_tagged": len(results), "unique_tags": cloud["tag_count"], "rows_written": written, "cloud": cloud, "cloud_path": cloud_path, "out_path": out_path, "top_tags": cloud["tags"][: (top or 10)], } # --------------------------------------------------------------------------- # Internals: rebuild + dominance # --------------------------------------------------------------------------- def _drop_tag_rows(path: str, keep_by: Optional[str] = None) -> None: """Clear ``bill_tags`` rows (optionally keep rows for one generated_by).""" ensure_tag_table(path) conn = relational.connect(path, foreign_keys=False) try: if keep_by: conn.execute("DELETE FROM " + TAG_TABLE + " WHERE generated_by NOT LIKE ?", (keep_by + "%",)) else: conn.execute( "DELETE FROM " + TAG_TABLE + " WHERE 1=1" ) conn.commit() finally: conn.close() def _dominant_generated_by(results: Sequence[Dict[str, Any]]) -> str: """Which path won: 'fallback' if any bill fell back, else the ollama model.""" counts: Dict[str, int] = {} for r in results: gb = r["generated_by"] counts[gb] = counts.get(gb, 0) + 1 if not counts: return FALLBACK if counts.get(FALLBACK, 0) > 0: return FALLBACK first = next(iter(counts)) return first