cali_scraper/scraper/parsers.py

"""HTML parsers: transform cached raw Daily File HTML into BillAction records.

Selector contracts follow DAILY_FILE_PARSE_SPEC.md section 3:

   - Committee hearings : table.committee-hearing-table / tr.committee-hearing-details
   - Bills on the floor : table.floor-bills-table       / tr.floor-bill-details
   - Bills in committee : table.committee-bills-table   / tr[class~=committee-bills-details]

Robustness (task acceptance criteria):
   * malformed rows (missing bill link, no number, decode failures) are skipped
     with a logged warning and never abort the batch;
   * encoding issues are handled by a charset decode with fallback + logging;
   * multi-page / multi-file input is handled by iterating every *.html the
     input resolves to.
"""

from __future__ import annotations

import glob
import logging
import os
import re
from datetime import datetime
from typing import List, Optional

from .bill_id import parse_bill_id, parse_display, parse_supplement, origin_house
from .schema import BillAction, ASSEMBLY_DAILY_FILE

log = logging.getLogger("scraper.parsers")

# ---- section selectors ---------------------------------------------------
HEARING_SELECTOR = "table.committee-hearing-table tr.committee-hearing-details"
FLOOR_SELECTOR = "table.floor-bills-table tr.floor-bill-details"
COMMITTEE_SELECTOR = "table.committee-bills-table tr[class~=committee-bills-details]"

# Looser fallbacks in case a table class is missing but rows are still present.
HEARING_ROW = "tr[class*=committee-hearing-details]"
FLOOR_ROW = "tr[class*=floor-bill-details]"
COMMITTEE_ROW = "tr[class*=committee-bills-details]"

_SUPP = re.compile(r"Supplemental\s+(\d+)", re.IGNORECASE)
_DATE_SPLIT = re.compile(r"-\d+$")   # strip display suffix like '-001'


# --- cell helpers ---------------------------------------------------------
def _attr_or_text(td, prefer: str = "data-sort") -> Optional[str]:
    """Return td[prefer] when present else trimmed textContent."""
    if prefer and td is not None and td.get(prefer):
        return td.get(prefer).strip()
    return td.get_text(" ", strip=True) if td is not None else None


def normalise_date(raw: Optional[str]) -> Optional[str]:
    """'08/31/26-001' / '08/31/26' -> '2026-08-31'. Non-date -> None."""
    if not raw:
        return None
    s = _DATE_SPLIT.sub("", raw.strip()).strip()
    if not s:
        return None
    for fmt in ("%m/%d/%y", "%m/%d/%Y", "%Y-%m-%d"):
        try:
            return datetime.strptime(s, fmt).strftime("%Y-%m-%d")
        except ValueError:
            continue
    return None


# --- section parsers ------------------------------------------------------
def _parse_hearings(soup, house, source, file_date=None):
    rows = soup.select(HEARING_SELECTOR) or soup.select(HEARING_ROW)
    records = []
    for i, tr in enumerate(rows):
        try:
            date_td = tr.select_one("td.committee_hearing-date")
            time_td = tr.select_one("td.committee_hearing-time")
            comm_td = tr.select_one("td.committee_hearing-name")
            loc_td = tr.select_one("td.committee_hearing-location")
            menu = tr.select_one("td.committee_hearing-menu")

            date = normalise_date(_attr_or_text(date_td))
            committee = _attr_or_text(comm_td, "data-sort")
            ics = None
            agenda = None
            if menu is not None:
                ics_a = menu.select_one("a[href^='/api/ics/generate']")
                if ics_a:
                    ics = ics_a.get("href")
                ag_a = menu.select_one("a[href^='/api/dailyfile/agenda/']")
                if ag_a:
                    agenda = ag_a.get("href", "").rsplit("/", 1)[-1]

            if not committee and date is None:
                raise ValueError("hearing row has neither committee nor date")

            records.append(BillAction(
                date=date,
                house=house,
                source=source,
                record_type="hearing",
                committee=committee,
                hearing_time=_attr_or_text(time_td, "text") or None,
                location=_attr_or_text(loc_td, "data-sort"),
                ics_url=ics,
                agenda_id=agenda,
                raw_selector="committee-hearing-table",
            ))
        except Exception as e:  # noqa: BLE001
            log.warning("skip malformed hearing row #%d: %s", i, e)
    # backfill a missing per-row date from the file date
    for h in records:
        if not h.date and file_date:
            h.date = file_date
    return records


def _extract_bill_measure(measure_td):
    """Return (bill_number, bill_id, supplement) from a measure / number cell."""
    supplement = None
    a = measure_td.select_one("a")
    bill_id = None
    bill_number = None
    if a is not None and a.get("href"):
        m = re.search(r"bill_id=([A-Za-z0-9]+)", a["href"])
        if m:
            bill_id = m.group(1)
            dec = parse_bill_id(bill_id)
            if dec:
                bill_number = dec["display"]
    if not bill_number:
        # Fallback: the display text before any <br> (e.g. "AB 28"). Only accept
        # a clean "PREFIX N" form; anything with extra text is a malformed row
        # and is left None so the caller can skip it.
        display = a.get_text(" ", strip=True) if a is not None else measure_td.get_text(" ", strip=True)
        m2 = re.match(r"^(AB|SB|ACR|SCR|HR|SJR|ABR)\s*(\d+)$", (display or "").strip())
        if m2:
            bill_number = f"{m2.group(1)} {int(m2.group(2))}"
            bill_id = None    # display only; leave bill_id None
    parsed_supp = parse_supplement(measure_td.get_text(" ", strip=True))
    if parsed_supp:
        supplement = parsed_supp
    return bill_number, bill_id, supplement


def _parse_floor_bills(soup, house, source, file_date):
    rows = soup.select(FLOOR_SELECTOR) or soup.select(FLOOR_ROW)
    records = []
    for i, tr in enumerate(rows):
        try:
            measure = tr.select_one("td.floor_bill-measure")
            if measure is None:
                raise ValueError("missing floor_bill-measure cell")
            bill_number, bill_id, supplement = _extract_bill_measure(measure)
            if not (bill_number or bill_id):
                raise ValueError("no bill number/id in floor row")

            dec = parse_bill_id(bill_id) if bill_id else None
            chamber_origin = dec["house"].upper() if dec else origin_house((bill_number or "").split(" ")[0])
            session = dec["session"] if dec else None

            view = tr.select_one("td.floor_bill-view-details")
            description = None
            if view is not None:
                description = view.get_text(" ", strip=True) or None

            item_td = tr.select_one("td.floor_bill-item")
            item_text = _attr_or_text(item_td, "data-sort") or _attr_or_text(item_td, "text")
            item_number = int(item_text.strip()) if item_text and item_text.strip().isdigit() else None

            short_title = _attr_or_text(tr.select_one("td.floor_bill-subject"), "text")
            records.append(BillAction(
                date=file_date,
                house=house,
                source=source,
                record_type="bill",
                bill_number=bill_number,
                bill_id=bill_id,
                session=session,
                chamber_origin=chamber_origin,
                author=_attr_or_text(tr.select_one("td.floor_bill-author"), "data-sort"),
                short_title=short_title,
                description=description or short_title,
                location=_attr_or_text(tr.select_one("td.floor_bill-location"), "data-sort"),
                item_number=item_number,
                supplement=supplement,
                raw_selector="floor-bills-table",
            ))
        except Exception as e:  # noqa: BLE001
            log.warning("skip malformed floor row #%d: %s", i, e)
    return records


def _parse_committee_bills(soup, house, source, file_date):
    rows = soup.select(COMMITTEE_SELECTOR) or soup.select(COMMITTEE_ROW)
    records = []
    for i, tr in enumerate(rows):
        try:
            num_td = tr.select_one("td.committee_bill-number")
            if num_td is None:
                raise ValueError("missing committee_bill-number cell")
            bill_number, bill_id, _ = _extract_bill_measure(num_td)
            if not (bill_number or bill_id):
                raise ValueError("no bill number/id in committee-bill row")

            dec = parse_bill_id(bill_id) if bill_id else None
            chamber_origin = dec["house"].upper() if dec else origin_house((bill_number or "").split(" ")[0])
            short_title = _attr_or_text(tr.select_one("td.committee_bill-subject"), "text")
            records.append(BillAction(
                date=file_date,
                house=house,
                source=source,
                record_type="bill",
                bill_number=bill_number,
                bill_id=bill_id,
                session=dec["session"] if dec else None,
                chamber_origin=chamber_origin,
                author=_attr_or_text(tr.select_one("td.committee_bill-author"), "data-sort"),
                short_title=short_title,
                description=short_title,
                committee=_attr_or_text(tr.select_one("td.committee_bill-committee"), "data-sort"),
                raw_selector="committee-bills-table",
            ))
        except Exception as e:  # noqa: BLE001
            log.warning("skip malformed committee-bill row #%d: %s", i, e)
    return records


def parse_html(html: str, *, house: str = "assembly",
               source: str = ASSEMBLY_DAILY_FILE, parser: str = "lxml") -> List[BillAction]:
    """Parse a single HTML document into a list of BillAction records.

    The date is derived from the committee-hearing table (the daily file's date)
    and applied to the bill rows, per spec 3.2 (bills carry no per-row date).
    """
    from bs4 import BeautifulSoup

    engine = parser if parser in ("lxml", "html.parser") else "html.parser"
    try:
        soup = BeautifulSoup(html, engine)
    except Exception as e:  # noqa: BLE001
        log.error("failed to parse document (%s): %s", type(e).__name__, e)
        return []

    hearings = _parse_hearings(soup, house, source, file_date=None)
    file_date = next((h.date for h in hearings if h.date), None)
    for h in hearings:
        if not h.date:
            h.date = file_date

     # Bills first (floor, then committee), then hearings — so a parsed batch
     # leads with bill records, each carrying bill_number/session/etc.
    records = []
    records.extend(_parse_floor_bills(soup, house, source, file_date))
    records.extend(_parse_committee_bills(soup, house, source, file_date))
    records.extend(hearings)
    return records


def _read_text(path: str) -> str:
    """Read a file with charset detection + an encoding fallback."""
    with open(path, "rb") as fh:
        raw = fh.read()
    head = raw[:2048].decode("utf-8", "ignore").lower()
    charset = "utf-8"
    meta = re.search(r"content-type=\"[^\"]*?charset=([a-z0-9_-]+)", head)
    if meta:
        charset = meta.group(1)
    else:
        decl = re.search(r'charset\s*=\s*["\']?\s*([a-z0-9_-]+)', head)
        if decl:
            charset = decl.group(1)
    try:
        text = raw.decode(charset)
    except (UnicodeDecodeError, LookupError):
        log.warning("bad charset %r in %s, falling back to UTF-8/replace", charset, path)
        text = raw.decode("utf-8", "replace")
    if "\ufffd" in text:
        log.warning("replacement chars detected while decoding %s", path)
    return text


def _resolve_inputs(input_path: str) -> List[str]:
    """Expand an input into HTML files: a file, a directory (all *.html), or a glob."""
    if os.path.isdir(input_path):
        return sorted(glob.glob(os.path.join(input_path, "*.html")))
    if any(ch in input_path for ch in "*?[]"):
        return sorted(glob.glob(input_path))
    return [input_path]


def parse(input_path: str, *, house: str = "assembly",
          source: str = ASSEMBLY_DAILY_FILE, parser: str = "lxml",
          include_hearings: bool = True) -> List[BillAction]:
    """Parse one or more cached HTML files into a flat list of records.

    Robust to malformed rows (logged + skipped), bad encodings (logged +
    fallback), and multi-file/multi-page input. Never aborts the batch on a
    single bad file or row.
    """
    files = _resolve_inputs(input_path)
    if not files:
        log.warning("no input files matched %r", input_path)
        return []
    records: List[BillAction] = []
    for path in files:
        if not os.path.exists(path):
            log.warning("input file not found: %s", path)
            continue
        try:
            html = _read_text(path)
        except OSError as e:
            log.warning("cannot read %s: %s", path, e)
            continue
        doc_records = parse_html(html, house=house, source=source, parser=parser)
        if not include_hearings:
            doc_records = [r for r in doc_records if r.record_type != "hearing"]
        records.extend(doc_records)
    return records

# -------------------------------------------------------------------
# Chamber "upcoming events" calendar (scheduled_events source, v2)
#
# The chamber calendar (assembly.ca.gov/events/upcoming, senate.ca.gov/
# calendar) is a Drupal "views" table (<table class="cols-3"> with Date /
# Title / Video columns). Each row is a scheduled hearing / floor session --
# the only *forward-looking* signal (the daily file is a backward snapshot).
# Rows are stored in scheduled_events (relational.store_scheduled_events).
# An empty page (a recess) is a legitimate empty result, not an error.
# -------------------------------------------------------------------

# Title keywords that classify an event's kind (best-effort, never crash).
_EV_KIND_PATTERNS = (
    (re.compile(r"\bfloor\b", re.I), "FLOOR"),
    (re.compile(r"\bcommittee\b", re.I), "COMMITTEE"),
)


def classify_event_kind(title):
    """Map an event title to a kind: FLOOR / COMMITTEE / OTHER."""
    t = title or ""
    for pat, kind in _EV_KIND_PATTERNS:
        if pat.search(t):
            return kind
    return "OTHER"


def parse_events_page(html, chamber="assembly", source="assembly_events",
                      base_url="", parser="lxml"):
    """Parse a chamber events calendar page into scheduled-event dicts.

    Returns a list of dicts with keys: date (ISO YYYY-MM-DD), chamber
    (upper), kind (FLOOR|COMMITTEE|OTHER), title, source_url. An empty /
    recess page yields []. Rows without a resolvable date are skipped
    (logged). Header rows (<th>) are ignored.
    """
    from bs4 import BeautifulSoup

    engine = parser if parser in ("lxml", "html.parser") else "html.parser"
    try:
        soup = BeautifulSoup(html, engine)
    except Exception as e:   # noqa: BLE001
        log.error("events: failed to parse document (%s): %s",
                  type(e).__name__, e)
        return []

    out = []
     # Fall back to any <tr> carrying a <time> cell when table.cols-3 is
     # absent, so a layout change still yields rows rather than a silent
     # empty page.
    base = (base_url or "").rstrip("/")
    host = base.rsplit("/", 2)[0] if base else "https://www.assembly.ca.gov"
    tables = soup.select("table.cols-3")
    rows = [tr for tbl in tables for tr in tbl.find_all("tr")]
    if not rows:
        rows = [tr for tr in soup.find_all("tr") if tr.find("time")]
    for i, tr in enumerate(rows):
         # Skip header rows (column labels render as <th>).
        if tr.find("th"):
            continue
        try:
            time_el = tr.find("time")
            if time_el is None:
                raise ValueError("event row has no date cell")
             # Prefer the datetime attr: "2026-09-01T00:00:00-07:00".
            dt = time_el.get("datetime") or ""
            date = dt[:10] if len(dt) >= 10 and dt[4:5] == "-" else None
            if not date:
                date = normalise_date(time_el.get_text(strip=True))

            title_td = tr.select_one("td.views-field-title")
            if title_td is not None:
                title = title_td.get_text(" ", strip=True)
            else:
                t = tr.find("time")
                if t is not None:
                    t.decompose()
                title = tr.get_text(" ", strip=True)
            if not date or not title:
                raise ValueError("event row has no date or title")

              # source_url: the row's own link (event / video detail) when
              # present, else the calendar page (base_url) as provenance.
            a = tr.find("a", href=True)
            source_url = None
            if a is not None:
                href = (a.get("href") or "").strip()
                if href.startswith("http"):
                    source_url = href
                elif href.startswith("/"):
                    source_url = host + href
            if not source_url:
                source_url = base or None

            out.append({
                "date": date,
                "chamber": chamber.strip().upper(),
                "kind": classify_event_kind(title),
                "title": title,
                "source_url": source_url,
                "source": source,
             })
        except Exception as e:   # noqa: BLE001
            log.warning("skip malformed event row #%d: %s", i, e)
    return out