311 lines
13 KiB
Python
311 lines
13 KiB
Python
"""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)
|