"""Pure ranking core for the legislative dashboard (spec §5, DESIGN_DASHBOARD §4). This module is deliberately **I/O-free**: it takes plain ``dict``-shaped bill rows (``scraper.schema.BillAction`` fields, optionally with a ``tags`` list) and returns them in float-to-top order. Keeping the ranking pure is what lets the rank logic be unit-tested without a database, and is the heart of §5. The rule (DESIGN_DASHBOARD §4, verbatim intent):: A bill FLOATS TO THE TOP iff author ∩ watched_legislators ≠ ∅ (author path) OR bill.tags ∩ opted_in_tags ≠ ∅ (tag path) OR (committee-of-watched-legistrator path) # committee -> members join Scoring (design default, tunable — Open Decision D3):: score = 3·(author is watched) + 2·(author sits-in a watched legislator's committee) + 2·( |tags ∩ opted_in_tags| ) + 1·(chamber == user's home chamber hint) Band membership = ``score > 0``. Intra-band order = descending score, then ``hearing_sort`` ascending (a future, earlier hearing floats ahead of a later one with the same score). The general band keeps source order. Nothing here is "hidden": floated bills are *promoted* (a ``Relevant to you`` marker / a 3px meaning bar), never removed — so a bill appears in both the personal band and the full queue. """ from __future__ import annotations import re from typing import Dict, List, Optional, Sequence, Set from . import PERSONAL_BAND_N # --------------------------------------------------------------------------- # hearing_sort (API_CONTRACT §1) # --------------------------------------------------------------------------- # "Upon Call of the Chair" / "Call of the Chair" sort LAST; a missing/unknown # time also sinks to the end of its day (design: time TBD note). A large # sentinel keeps them last without inventing a wall-clock value. _SORT_LAST = 24 * 60 * 60 + 60000 # > any HH:MM, so call-of-chair always sinks _CALL_OF_CHAIR_RE = re.compile(r"call of the chair", re.I) _AMPM_RE = re.compile(r"^(\d{1,2})(?::(\d{2}))?\s*(am|pm)$", re.I) # A bare clock like "9:30" (no am/pm) — treat as 24h HH:MM. _HHMM_RE = re.compile(r"^(\d{1,2}):([0-5]?\d)$") def hearing_sort(hearing_time: Optional[str]) -> int: """Derive a sortable integer for ``hearing_time``. API_CONTRACT §1: * ``"9:30am"`` / ``"10 minutes prior to Session"``-style wall times -> seconds-from-midnight. * ``"Upon Call of the Chair"`` / ``"Call of the Chair"`` -> **max** (last). * ``None`` / unparseable -> **max** (last), with the row still carrying ``hearing_time: None`` so the UI can render a "time TBD" note. Returns an int suitable for ascending sort within a day. """ if not hearing_time: return _SORT_LAST s = str(hearing_time).strip() if not s: return _SORT_LAST if _CALL_OF_CHAIR_RE.search(s): return _SORT_LAST m = _AMPM_RE.match(s) if m: hour = int(m.group(1)) minute = int(m.group(2) or 0) period = m.group(3).lower() if period == "pm" and hour != 12: hour += 12 if period == "am" and hour == 12: # 12am == midnight hour = 0 return hour * 3600 + minute * 60 m = _HHMM_RE.match(s) if m: return int(m.group(1)) * 3600 + int(m.group(2)) * 60 # Unparseable (e.g. "10 minutes prior to Session") -> sinks last; that is # the designed "time TBD" behaviour, not a crash. return _SORT_LAST # --------------------------------------------------------------------------- # Relevance scoring (DESIGN_DASHBOARD §4, weights per Open Decision D3) # --------------------------------------------------------------------------- class Weights: """Scoring weights. Defaults match DESIGN_DASHBOARD §4. These are a *design default*, not frozen — Open Decision D3 flags them as tunable. The :class:`Weights` object is kept explicit (rather than module globals) so tests and the API can pass alternative weights without global mutation. """ def __init__(self, author_watched: int = 3, committee_watched: int = 2, tag_watched: int = 2, home_chamber: int = 1): self.author_watched = author_watched self.committee_watched = committee_watched self.tag_watched = tag_watched self.home_chamber = home_chamber DEFAULT_WEIGHTS = Weights() def normalise_author(author: Optional[str]) -> str: """Reduce a bill ``author`` string ("Schiavo et al.", "Macedo") to the primary legislator's slug, matching the member slugs the resolve/prefs layer keys on. Mirrors ``relational.slug`` / author-stem stripping.""" if not author: return "" s = re.sub(r"\s+et al\.$", "", author.strip()) s = re.sub(r"\s+et al\s+", " ", s, flags=re.I) s = re.sub(r"\s+", " ", s).strip() return s.lower() def _slug(name: Optional[str]) -> str: if not name: return "" return re.sub(r"[^a-z0-9]+", "-", name.strip().lower()).strip("-") or "unknown" def _match_tags(bill_tags: Sequence[str], opted: Set[str]) -> List[str]: """Opted-in tags that actually appear on this bill (intersection, stable order = opted-in order), so the "Why is this here?" list is deterministic.""" if not opted: return [] present = {str(t).strip().lower() for t in (bill_tags or [])} return [t for t in opted if str(t).strip().lower() in present] def score_bill(bill: Dict, watched_authors: Set[str], watched_committees: Set[str], opted_in_tags: Set[str], home_chamber: Optional[str] = None, weights: Optional[Weights] = None) -> Dict: """Compute a relevance ``{score, matched[]}`` for one bill against a user's preferences. Parameters ---------- bill: a dict-like with at least ``bill_id``, ``author``, ``committee``, ``chamber_origin`` and optionally ``tags`` (list of tag strings). watched_authors: lower-cased primary-author slugs the user watches (``author ∩ watched_legislators`` path). watched_committees: slugs of committees the user watches (committee-of- watched-legislator path; ``∅`` when the correlation lane found no roster — never fabricated). opted_in_tags: tag strings the user opted into (``tags ∩ opted_in_tags``). home_chamber: optional chamber hint ("assembly"/"senate") for the +1 term. weights: scoring weights (defaults to DESIGN_DASHBOARD §4). Returns ------- A ``{"score": int, "matched": [{"type", "value"}, ...]}`` dict, matching the ``Bill.relevance`` shape in API_CONTRACT §1. ``matched`` powers the "Why is this here?" transparency toggle. A single rule suffices to float (OR); stronger matches score higher. """ w = weights or DEFAULT_WEIGHTS matched: List[Dict[str, str]] = [] score = 0 # --- author path ------------------------------------------------------ author = bill.get("author") norm_author = normalise_author(author) if norm_author and norm_author in {a.lower() for a in watched_authors}: score += w.author_watched matched.append({ "type": "watched-legislator", "value": "Author is a watched legislator" + ("" if not author else " — " + str(author)), }) # --- committee-of-watched-legislator path ---------------------------- # The bill sits in a committee the user's watched legislators chair/serve. # `bill.committee` is the daily-file committee name; watched_committees is # the slug set of the watched set's committees. (Degrades to a no-op when # the correlation lane found no roster — honest, never fabricated.) committee = bill.get("committee") if committee and _slug(committee) in {c.lower() for c in watched_committees}: score += w.committee_watched matched.append({ "type": "committee-of-watched-legislator", "value": "Sits in a committee your watched legislator chair/serves " + ("— " + str(committee) if committee else ""), }) # --- tag path --------------------------------------------------------- tag_hits = _match_tags(bill.get("tags") or [], {t.lower() for t in opted_in_tags}) if tag_hits: score += w.tag_watched * len(tag_hits) for t in tag_hits: matched.append({"type": "tag", "value": "Topic: " + str(t)}) # --- home-chamber hint (optional, +1) -------------------------------- if home_chamber and bill.get("chamber_origin"): if str(bill["chamber_origin"]).lower() == str(home_chamber).lower(): score += w.home_chamber matched.append({ "type": "home-chamber", "value": "Matches your home chamber (" + str(bill["chamber_origin"]) + ")", }) return {"score": int(score), "matched": matched} # --------------------------------------------------------------------------- # Bill normalisation # --------------------------------------------------------------------------- # Fields the UI consumes from a BillAction row, in API_CONTRACT §1 display order. _BILL_FIELDS = ( "bill_id", "bill_number", "chamber_origin", "author", "short_title", "description", "committee", "hearing_time", "location", "date", "session", "item_number", "supplement", "source", "tags", ) def normalise_bill(row: Dict, tags: Optional[Sequence[str]] = None) -> Dict: """Project a raw DB row onto the API_CONTRACT §1 ``Bill`` shape. * Keeps the exact field names the contract freezes. * Adds the derived ``hearing_sort`` (backend-owned, §1). * Carries ``tags`` from the LLM lane (may be ``[]``/``None`` -> no chips, ranking degrades to legislator-match only, per §6.6). * Nulls are preserved (the UI renders TBD, never crashes). """ out: Dict = {} for f in _BILL_FIELDS: if f == "tags": out["tags"] = None if tags is None else list(tags) else: val = row.get(f) out[f] = None if val == "" else val out["hearing_sort"] = hearing_sort(row.get("hearing_time")) return out # --------------------------------------------------------------------------- # Float-to-top ordering (DESIGN_DASHBOARD §4) # --------------------------------------------------------------------------- def is_floated(bill: Dict) -> bool: """True if the bill carries a re: positive score (i.e. it floats). ``relevance`` is only populated for logged-in requests (§1); anon bills have no ``relevance`` key. """ rel = bill.get("relevance") return bool(rel and int(rel.get("score", 0)) > 0) def float_to_top(bills: Sequence[Dict], personal_band_n: int = PERSONAL_BAND_N) -> List[Dict]: """Stably re-sort ``bills`` so floated bills (``relevance.score > 0``) come first, ordered by ``(score desc, hearing_sort asc)``; non-floated bills keep their existing relative order. This is the "promote, never hide" step of DESIGN_DASHBOARD §4: every floated bill also remains visible in the full queue (marked), only its *position* changes. Idempotent: running it twice is a no-op. """ def key(bill: Dict) -> tuple: rel = bill.get("relevance") or {} score = int(rel.get("score", 0)) # -score so higher scores sort earlier; hearing earlier-in-day sorts # earlier; tie-break on bill_id for total determinism. return (-score, int(bill.get("hearing_sort") or _SORT_LAST), str(bill.get("bill_id") or "")) # Stable sort (Python's sort is stable; the bill_id tie-break makes it # total anyway). personal_band_n is accepted for signature parity with the # band helpers but does NOT truncate the full queue. _ = personal_band_n ordered = sorted(bills, key=key) return ordered def personal_band(bills: Sequence[Dict], n: int = PERSONAL_BAND_N) -> List[Dict]: """The first ``n`` floated bills (the "Relevant to you" band, top N=10). Derived by "relevance.score > 0, then hearing_sort", per API_CONTRACT §1 — i.e. the head of a :func:`float_to_top` ordering. Returns ``[]`` when nothing is floated (the UI shows the setup / empty state, not a dead band). """ band = [b for b in float_to_top(bills) if is_floated(b)] return band[:n] def full_queue(bills: Sequence[Dict]) -> List[Dict]: """The complete queue, sorted the same way as :func:`float_to_top` but untruncated — every floated bill is marked and every other bill follows.""" return float_to_top(bills)