kfj001
03eeb6bb62
The daily-file bill queue is a backward snapshot, so "upcoming" could only resurface the latest past day (the 08-31 snapshot read as today's queue) and there was no "what is scheduled next." leginfo bill status/history are backward-only, so the genuine future-dated signal is the chambers' "upcoming events" calendars. - api.py: floor the "upcoming" view at *today* (not the DB max date) when no as_of is pinned; surface next_legislative_day + scheduled_events (next 7 days) on /bills. Add _today_iso / _next_legislative_day / _scheduled_events, all degrading to a recess marker / [] (never 5xx). - relational.py: schema v2 adds the scheduled_events table + store_scheduled_events (idempotent per-day refresh), load_scheduled_events, next_legislative_day. - parsers.py: parse_events_page + classify_event_kind for the chamber events calendar; skip empty-title rows (decompose the <time> before the fallback). - ingest.py + cli.py: --store-events lane (cache-first, recess-safe) that parses the calendar HTML into scheduled_events. - sources.py + config.yaml: register assembly_events / senate_events (kind: events, empty_is_ok: true). - run_ingest.sh: run the events lane after the daily-file lane as non-fatal (skippable via CALI_SCRAPER_SKIP_EVENTS=1); daily-file exit still gates cron. - frontend: "Next legislative day" line + "What's next · next 7 days" band with honest recess empty states. - DASHBOARD.md: document the forward signal + known limitations (leginfo backward-only gap, recess empty state, 08-31 snapshot is analysis-only). - tests: 5 new API tests + 18 parse-events tests; 170 pass, 92% coverage.
391 lines
14 KiB
Python
391 lines
14 KiB
Python
# Orchestrator: ingest one or more chamber daily-file pages for a date.
|
|
# Ties together the cache, HTTP client, source planner, and alerting hooks.
|
|
#
|
|
# ing = Ingestor("config.yaml")
|
|
# result = ing.ingest("2026-09-02") # both chambers
|
|
# result = ing.ingest("2026-09-02", source="assembly") # one chamber
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
from .alerts import AlertContext, fire
|
|
from .cache import RawCache
|
|
from .config import load_config, Config
|
|
from .http_client import (
|
|
AntibotError,
|
|
FetchResult,
|
|
HttpClient,
|
|
NotFoundError,
|
|
TransientError,
|
|
)
|
|
from .request_log import RequestLog
|
|
from .sources import SourcePlan, build_source_plan, parse_senate_pdf_filename
|
|
|
|
|
|
@dataclass
|
|
class ChamberResult:
|
|
# Per-chamber outcome for one date.
|
|
source: str
|
|
date: str
|
|
status: str # ok | cached | not_found | antibot | transient
|
|
path: Optional[Path] = None
|
|
bytes: int = 0
|
|
attempts: int = 0
|
|
cache_hit: bool = False
|
|
url: Optional[str] = None
|
|
as_of: Optional[str] = None
|
|
note: str = ""
|
|
|
|
|
|
@dataclass
|
|
class IngestResult:
|
|
date: str
|
|
chambers: List[ChamberResult] = field(default_factory=list)
|
|
|
|
@property
|
|
def all_ok(self) -> bool:
|
|
return all(c.status in ("ok", "cached") for c in self.chambers)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"date": self.date,
|
|
"all_ok": self.all_ok,
|
|
"chambers": [
|
|
{
|
|
"source": c.source,
|
|
"status": c.status,
|
|
"path": str(c.path) if c.path else None,
|
|
"bytes": c.bytes,
|
|
"attempts": c.attempts,
|
|
"cache_hit": c.cache_hit,
|
|
"url": c.url,
|
|
"as_of": c.as_of,
|
|
"note": c.note,
|
|
}
|
|
for c in self.chambers
|
|
],
|
|
}
|
|
|
|
|
|
class Ingestor:
|
|
# One instance per process; reuses the HTTP connection pool.
|
|
def __init__(self, config_path=None, force=False):
|
|
self.cfg = load_config(config_path)
|
|
self.cache = RawCache(self.cfg.raw_dir)
|
|
self.log = RequestLog(self.cfg.request_log)
|
|
self.force = force
|
|
self.http = HttpClient(
|
|
user_agent=self.cfg.http.get("user_agent", "research-scanner/1.0"),
|
|
timeout_seconds=self.cfg.http.get("timeout_seconds", 20),
|
|
max_retries=self.cfg.http.get("max_retries", 3),
|
|
backoff_base_seconds=self.cfg.http.get("backoff_base_seconds", 1.0),
|
|
backoff_cap_seconds=self.cfg.http.get("backoff_cap_seconds", 30.0),
|
|
retry_status=self.cfg.http.get("retry_status", [500, 502, 503, 504]),
|
|
antibot_status=self.cfg.http.get("antibot_status", [403]),
|
|
log_request=self._log_event,
|
|
)
|
|
|
|
# -- log --
|
|
def _log_event(self, event):
|
|
event.setdefault(
|
|
"log_ts",
|
|
datetime.datetime.now(datetime.timezone.utc).isoformat(
|
|
timespec="seconds"))
|
|
self.log.record(event)
|
|
|
|
# -- main API --
|
|
def ingest(self, date_iso, source=None, sources=None):
|
|
names = sources or ([source] if source else self.cfg.source_names())
|
|
result = IngestResult(date=date_iso)
|
|
for name in names:
|
|
if name not in self.cfg.sources:
|
|
raise ValueError(
|
|
"Unknown source '%s'. Known: %s"
|
|
% (name, self.cfg.source_names()))
|
|
cr = self._ingest_chamber(name, date_iso)
|
|
result.chambers.append(cr)
|
|
return result
|
|
|
|
# -- chamber events forward source (scheduled_events, schema v2) --
|
|
def ingest_events(self, date_iso, source=None, sources=None, db_path=None):
|
|
"""Fetch + cache + parse + store the chamber "upcoming events" source.
|
|
|
|
Like :meth:`ingest`, this is cache-first (a date already on disk is served
|
|
with no network) and reuses the same source planner (the events source is
|
|
``kind: events``). It parses the calendar HTML into scheduled-event rows
|
|
and stores them in the ``scheduled_events`` table.
|
|
|
|
``db_path`` defaults to the relational lane DEFAULT_DB so the dashboard
|
|
can read what this writes. A recess (an empty calendar) is a legitimate
|
|
no-op, not an error: nothing is cleared and nothing is stored.
|
|
"""
|
|
from . import relational
|
|
from . import parsers
|
|
|
|
db = db_path or relational.DEFAULT_DB
|
|
relational.migrate(db) # ensure the scheduled_events table exists
|
|
names = sources or ([source] if source else self.cfg.source_names())
|
|
result = IngestResult(date=date_iso)
|
|
for name in names:
|
|
cfg = self.cfg.source(name)
|
|
if cfg.get("kind") != "events":
|
|
raise ValueError(
|
|
"ingest_events needs an events source; %r is kind %r"
|
|
% (name, cfg.get("kind")))
|
|
cr = self._ingest_chamber(name, date_iso)
|
|
|
|
events = []
|
|
if cr.status == "ok" and cr.path is not None:
|
|
try:
|
|
html = cr.path.read_text(encoding="utf-8",
|
|
errors="replace")
|
|
except OSError as e:
|
|
html = None
|
|
cr.note += " | cannot read cached page: %s" % e
|
|
if html:
|
|
base = getattr(cr, "url", "") or ""
|
|
events = parsers.parse_events_page(
|
|
html, chamber=cfg.get("chamber") or cr.source, source=
|
|
name, base_url=base,)
|
|
|
|
# Store when a page was read (fresh or served from cache). A fresh
|
|
# timestamp lets a re-published calendar refresh its day on the next
|
|
# run while an unchanged read stays a no-op. An empty parse (recess)
|
|
# clears no days, so existing scheduled rows are preserved.
|
|
when = datetime.datetime.now(datetime.timezone.utc).strftime(
|
|
"%Y-%m-%dT%H:%M:%SZ")
|
|
summary = relational.store_scheduled_events(events, path=db,
|
|
fetched_at=when)
|
|
cr.note += (" | events: %d stored, %d stale cleared, %d total"
|
|
% (summary["stored"], summary["cleared_stale"],
|
|
summary["total"]))
|
|
result.chambers.append(cr)
|
|
return result
|
|
|
|
# -- per-chamber --
|
|
def _ingest_chamber(self, name, date_iso):
|
|
cfg = self.cfg.source(name)
|
|
plan = build_source_plan(name, date_iso, cfg)
|
|
|
|
# 1. Cache first -- if raw bytes are present and not forced, no network.
|
|
if not self.force:
|
|
cached = self.cache.get(plan.source, date_iso, plan.ext)
|
|
if cached is not None:
|
|
self._alert(cfg, AlertContext(
|
|
kind="ok",
|
|
level="info",
|
|
message="cache HIT: serving from local cache (no network)",
|
|
source=name,
|
|
date=date_iso,
|
|
url=plan.primary_url,
|
|
extra={"as_of": plan.as_of,
|
|
"cache_entry": str(self.cache.meta_path(plan.source, date_iso))},
|
|
))
|
|
return ChamberResult(
|
|
source=name,
|
|
date=date_iso,
|
|
status="cached",
|
|
path=self.cache.raw_dir / plan.source / f"{date_iso}.{plan.ext}",
|
|
bytes=len(cached),
|
|
attempts=0,
|
|
cache_hit=True,
|
|
url=plan.primary_url,
|
|
as_of=plan.as_of,
|
|
note="served from local cache",
|
|
)
|
|
|
|
# 2. Network fetch with retry + alerting.
|
|
try:
|
|
fr = self._fetch(plan, cfg)
|
|
except NotFoundError as exc:
|
|
return self._handle_404(plan, name, cfg, exc)
|
|
except AntibotError as exc:
|
|
return self._handle_403(plan, name, cfg, exc)
|
|
except TransientError as exc:
|
|
return self._handle_transient(plan, name, cfg, exc)
|
|
|
|
# 3. Write to cache + alert success.
|
|
fetched_at = datetime.datetime.now(datetime.timezone.utc).isoformat(
|
|
timespec="seconds")
|
|
out_path = self.cache.put(
|
|
plan.source, date_iso, plan.ext, fr.content,
|
|
url=fr.url,
|
|
status=fr.status,
|
|
fetched_at=fetched_at,
|
|
)
|
|
self._alert(cfg, AlertContext(
|
|
kind="ok",
|
|
level="info",
|
|
message="Fetched + cached %s (%d bytes, %d attempt(s))"
|
|
% (fr.url, len(fr.content), fr.attempts),
|
|
source=name,
|
|
date=date_iso,
|
|
url=fr.url,
|
|
extra={"status_code": fr.status, "as_of": plan.as_of},
|
|
))
|
|
return ChamberResult(
|
|
source=name,
|
|
date=date_iso,
|
|
status="ok",
|
|
path=out_path,
|
|
bytes=len(fr.content),
|
|
attempts=fr.attempts,
|
|
cache_hit=False,
|
|
url=fr.url,
|
|
as_of=plan.as_of,
|
|
)
|
|
|
|
# -- fetch dispatch --
|
|
def _fetch(self, plan, cfg):
|
|
if plan.kind == "pdf":
|
|
return self._fetch_pdf(plan, cfg)
|
|
return self.http.get(plan.primary_url, source=plan.source)
|
|
|
|
def _fetch_pdf(self, plan, cfg):
|
|
# Step 1: fetch the live index HTML (index.html carries no antibot).
|
|
index_fr = self.http.get(plan.primary_url, source=plan.source)
|
|
index_text = index_fr.content.decode("utf-8", errors="replace")
|
|
|
|
# Step 2: find the PDF filename that matches the requested date.
|
|
target_date = plan.date
|
|
found = None
|
|
for pdf_url in _iter_senate_pdf_urls(index_text, plan.pdf_url):
|
|
ref = parse_senate_pdf_filename(pdf_url)
|
|
if ref and ref["date_iso"] == target_date:
|
|
found = pdf_url
|
|
break
|
|
if found is None:
|
|
found = plan.pdf_url # fall back to the predicted URL
|
|
|
|
# Step 3: fetch the PDF itself (may hit antibot -> AntibotError).
|
|
return self.http.get(found, source=plan.source)
|
|
|
|
# -- error handlers --
|
|
def _handle_404(self, plan, name, cfg, exc):
|
|
if cfg.get("missing_is_expected_404", False):
|
|
# 404 on the Senate PDF is normal on non-legislative days.
|
|
self._alert(cfg, AlertContext(
|
|
kind="not_found",
|
|
level="info",
|
|
message="No publication for %s (404 expected on non-legislative days)"
|
|
% plan.date,
|
|
source=name,
|
|
date=plan.date,
|
|
url=plan.pdf_url,
|
|
))
|
|
return ChamberResult(
|
|
source=name,
|
|
date=plan.date,
|
|
status="not_found",
|
|
url=plan.pdf_url,
|
|
as_of=plan.as_of,
|
|
attempts=getattr(self.http, "last_attempts", 1),
|
|
note="404 no publication (expected)",
|
|
)
|
|
# Unexpected 404 -> alert as a warning.
|
|
self._alert(cfg, AlertContext(
|
|
kind="not_found",
|
|
level="warning",
|
|
message="Unexpected 404 for %s: %s" % (name, exc.url),
|
|
source=name,
|
|
date=plan.date,
|
|
url=exc.url,
|
|
extra={"status": exc.status},
|
|
))
|
|
return ChamberResult(
|
|
source=name,
|
|
date=plan.date,
|
|
status="not_found",
|
|
url=exc.url,
|
|
as_of=plan.as_of,
|
|
attempts=getattr(self.http, "last_attempts", 1),
|
|
note="unexpected 404",
|
|
)
|
|
|
|
def _handle_403(self, plan, name, cfg, exc):
|
|
self._alert(cfg, AlertContext(
|
|
kind="antibot",
|
|
level="error",
|
|
message="Antibot 403 on %s: %s\n"
|
|
"Mitigation: blocked-page-recovery skill or a headless browser "
|
|
"(antibot session pre-warm)." % (name, exc.url),
|
|
source=name,
|
|
date=plan.date,
|
|
url=exc.url,
|
|
extra={
|
|
"status": exc.status,
|
|
"body_len": len(exc.body),
|
|
"body_preview": exc.body[:200].decode("utf-8", errors="replace"),
|
|
},
|
|
))
|
|
return ChamberResult(
|
|
source=name,
|
|
date=plan.date,
|
|
status="antibot",
|
|
url=exc.url,
|
|
as_of=plan.as_of,
|
|
attempts=getattr(self.http, "last_attempts", 1),
|
|
note="antibot 403 (mitigate per blocked-page-recovery skill)",
|
|
)
|
|
|
|
def _handle_transient(self, plan, name, cfg, exc):
|
|
self._alert(cfg, AlertContext(
|
|
kind="transient",
|
|
level="error",
|
|
message="Transient after %d attempt(s) for %s: %s"
|
|
% (exc.attempts, name, exc),
|
|
source=name,
|
|
date=plan.date,
|
|
url=exc.url,
|
|
extra={"status": exc.status, "attempts": exc.attempts},
|
|
))
|
|
return ChamberResult(
|
|
source=name,
|
|
date=plan.date,
|
|
status="transient",
|
|
url=exc.url,
|
|
as_of=plan.as_of,
|
|
attempts=exc.attempts,
|
|
note="transient after %d retries" % exc.attempts,
|
|
)
|
|
|
|
# -- alert convenience --
|
|
def _alert(self, cfg, ctx):
|
|
fire(self.cfg.alerts, ctx)
|
|
|
|
# -- lifecycle --
|
|
def close(self):
|
|
self.http.close()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
self.close()
|
|
|
|
|
|
# -- helpers --
|
|
|
|
def _iter_senate_pdf_urls(index_text, fallback):
|
|
# Extract PDF download URLs referenced by the Senate index page HTML.
|
|
import re
|
|
urls = re.findall(
|
|
r'/dailyfile/publications/download/[^"<>\s]+\.PDF',
|
|
index_text,
|
|
re.IGNORECASE,
|
|
)
|
|
base = "https://www.senate.ca.gov"
|
|
seen = set()
|
|
out = []
|
|
for u in urls:
|
|
full = base + u if u.startswith("/") else u
|
|
if full not in seen:
|
|
out.append(full)
|
|
seen.add(full)
|
|
out.append(fallback)
|
|
return out
|