""" Extract election calendar rows, candidacies, and ballot measures from jurisdiction HTML. Best-effort heuristics for city/county sites — not a substitute for official election files, but suitable for bronze + c1 promotion when no external API is used. """ from __future__ import annotations import json import re from datetime import date, datetime from typing import Any from urllib.parse import urljoin, urlparse from bs4 import BeautifulSoup ELECTION_LINK_RE = re.compile( r"\b(?:election|elections|ballot|ballots|candidate|candidates|voting|vote|voter|" r"poll(?:ing)?|sample[- ]ballot|election[- ]results?|runoff|primary|general)\b", re.I, ) ELECTION_HEADING_RE = re.compile( r"\b(?:election|ballot|candidate|voting|vote|primary|general|runoff|special)\b", re.I, ) DATE_PATTERNS: tuple[re.Pattern[str], ...] = ( re.compile( r"\b(\d{4})-(\d{1,2})-(\d{1,2})\b", ), re.compile( r"\b(\d{1,2})[/.-](\d{1,2})[/.-](\d{2,4})\b", ), re.compile( r"\b(January|February|March|April|May|June|July|August|September|October|November|December)" r"\s+(\d{1,2})(?:st|nd|rd|th)?,?\s+(\d{4})\b", re.I, ), ) _MONTH = { "january": 1, "february": 2, "march": 3, "april": 4, "may": 5, "june": 6, "july": 7, "august": 8, "september": 9, "october": 10, "november": 11, "december": 12, } ELECTION_PROBE_PATHS: tuple[str, ...] = ( "/elections", "/elections/", "/election-information", "/election-info", "/vote", "/voting", "/voter-information", "/ballot", "/candidates", "/election-results", "/government/elections", "/departments/city-clerk/elections", "/departments/city-clerk/election-information", "/county-clerk/elections", ) def _same_site(url: str, base: str) -> bool: try: return urlparse(url).netloc.lower() == urlparse(base).netloc.lower() except Exception: return False def discover_election_page_urls(homepage_url: str, html: str, *, max_urls: int = 12) -> list[str]: """Collect same-site links that look election-related.""" if not homepage_url or not html: return [] out: list[str] = [] seen: set[str] = set() soup = BeautifulSoup(html, "html.parser") for a in soup.find_all("a", href=True): href = (a.get("href") or "").strip() if not href or href.startswith(("#", "mailto:", "tel:")): continue text = a.get_text(" ", strip=True) if not ELECTION_LINK_RE.search(href) and not ELECTION_LINK_RE.search(text): continue abs_url = urljoin(homepage_url, href) if not _same_site(abs_url, homepage_url): continue norm = abs_url.split("#", 1)[0].rstrip("/") if norm in seen: continue seen.add(norm) out.append(abs_url) if len(out) >= max_urls: break return out def candidate_election_urls(homepage_url: str) -> list[str]: """Build ordered ``/elections``-style URLs relative to the jurisdiction homepage.""" if not homepage_url: return [] out: list[str] = [] seen: set[str] = set() for path in ELECTION_PROBE_PATHS: url = urljoin(homepage_url, path) if url in seen: continue seen.add(url) out.append(url) return out def probe_election_path_urls(homepage_url: str, session) -> list[str]: """HEAD/GET common ``/elections``-style paths on the site host.""" from scrapers.discovery.mayor_url_discovery import probe_urls return probe_urls(candidate_election_urls(homepage_url), session=session) def _parse_date_from_text(text: str) -> date | None: blob = (text or "").strip() if not blob: return None m = DATE_PATTERNS[0].search(blob) if m: try: return date(int(m.group(1)), int(m.group(2)), int(m.group(3))) except ValueError: pass m = DATE_PATTERNS[1].search(blob) if m: mo, da, yr = int(m.group(1)), int(m.group(2)), int(m.group(3)) if yr < 100: yr += 2000 try: return date(yr, mo, da) except ValueError: pass m = DATE_PATTERNS[2].search(blob) if m: mo = _MONTH.get(m.group(1).lower()) if mo: try: return date(int(m.group(3)), mo, int(m.group(2))) except ValueError: pass return None def _infer_election_type(text: str) -> str: low = (text or "").lower() for label in ("primary", "general", "runoff", "special", "municipal"): if label in low: return label return "unknown" def _json_ld_events(soup: BeautifulSoup, page_url: str) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for script in soup.find_all("script", attrs={"type": re.compile(r"ld\+json", re.I)}): raw = (script.string or script.get_text() or "").strip() if not raw: continue try: data = json.loads(raw) except (json.JSONDecodeError, TypeError, ValueError): continue def walk(node: Any) -> None: if isinstance(node, list): for item in node: walk(item) elif isinstance(node, dict): t = node.get("@type") or "" types = t if isinstance(t, list) else [t] if any(str(x).lower() == "event" for x in types): name = (node.get("name") or "").strip() start = node.get("startDate") or node.get("startdate") or "" if name and ELECTION_HEADING_RE.search(name): rows.append({ "name": name, "election_date": _parse_date_from_text(str(start)) or _parse_date_from_text(name), "election_type": _infer_election_type(name), "source_url": page_url, "extraction_method": "json_ld_event", "raw_snippet": name, }) for v in node.values(): walk(v) walk(data) return rows def _heading_blocks(soup: BeautifulSoup, page_url: str) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for tag in soup.find_all(re.compile(r"^h[1-3]$", re.I)): title = tag.get_text(" ", strip=True) if not title or not ELECTION_HEADING_RE.search(title): continue sibling_text = [] for sib in tag.find_next_siblings(limit=4): if getattr(sib, "name", None) and re.match(r"^h[1-3]$", sib.name, re.I): break sibling_text.append(sib.get_text(" ", strip=True)) block = " ".join([title, *sibling_text]) rows.append({ "name": title[:500], "election_date": _parse_date_from_text(block), "election_type": _infer_election_type(title), "source_url": page_url, "extraction_method": "heading_block", "raw_snippet": block[:2000], }) return rows def _table_candidacies(soup: BeautifulSoup, page_url: str) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for table in soup.find_all("table"): headers = [ th.get_text(" ", strip=True).lower() for th in table.find_all("th") ] if not headers: first = table.find("tr") if first: headers = [c.get_text(" ", strip=True).lower() for c in first.find_all(["td", "th"])] header_blob = " ".join(headers) if not re.search(r"candidate|name|office|position|party|seat", header_blob, re.I): continue name_idx = party_idx = office_idx = -1 for i, h in enumerate(headers): if "name" in h or h == "candidate": name_idx = i if "party" in h: party_idx = i if "office" in h or "position" in h or "seat" in h: office_idx = i for tr in table.find_all("tr"): cells = [td.get_text(" ", strip=True) for td in tr.find_all(["td", "th"])] if len(cells) < 2: continue if name_idx >= 0 and name_idx < len(cells): person = cells[name_idx] else: person = cells[0] if not person or len(person) < 2 or person.lower() in headers: continue if not re.search(r"[A-Za-z]{2,}", person): continue party = cells[party_idx] if 0 <= party_idx < len(cells) else None office = cells[office_idx] if 0 <= office_idx < len(cells) else None rows.append({ "person_name": person[:300], "party": (party or "")[:120] or None, "office": (office or "")[:300] or None, "status": "candidate", "source_url": page_url, "extraction_method": "html_table", }) return rows def _list_ballot_measures(soup: BeautifulSoup, page_url: str) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for li in soup.find_all("li"): text = li.get_text(" ", strip=True) if not text or len(text) < 12: continue if not re.search(r"\b(?:proposition|measure|question|referendum|amendment)\b", text, re.I): continue if not re.search(r"\b\d+\b", text) and "?" not in text: continue rows.append({ "title": text[:500], "summary": text[:2000], "classification": "referendum", "source_url": page_url, "extraction_method": "list_item", }) return rows def extract_election_bundle_from_html(html: str, page_url: str) -> dict[str, Any]: """ Parse one HTML page into elections, candidacies, and ballot-measure candidates. """ soup = BeautifulSoup(html or "", "html.parser") elections: list[dict[str, Any]] = [] elections.extend(_json_ld_events(soup, page_url)) elections.extend(_heading_blocks(soup, page_url)) # Page title as weak election row when election-themed title = (soup.title.string or "").strip() if soup.title else "" if title and ELECTION_HEADING_RE.search(title): elections.append({ "name": title[:500], "election_date": _parse_date_from_text(title), "election_type": _infer_election_type(title), "source_url": page_url, "extraction_method": "document_title", "raw_snippet": title, }) candidacies = _table_candidacies(soup, page_url) measures = _list_ballot_measures(soup, page_url) return { "page_url": page_url, "elections": elections, "candidacies": candidacies, "ballot_measures": measures, } def merge_election_bundles(bundles: list[dict[str, Any]]) -> dict[str, Any]: """Deduplicate merged bundles from multiple pages.""" elections: list[dict[str, Any]] = [] candidacies: list[dict[str, Any]] = [] measures: list[dict[str, Any]] = [] seen_e: set[tuple[str, str]] = set() seen_c: set[tuple[str, str, str]] = set() seen_m: set[str] = set() for bundle in bundles: for e in bundle.get("elections") or []: key = ((e.get("name") or "").lower(), str(e.get("election_date") or "")) if key in seen_e: continue seen_e.add(key) elections.append(e) for c in bundle.get("candidacies") or []: key = ( (c.get("person_name") or "").lower(), (c.get("office") or "").lower(), c.get("source_url") or "", ) if key in seen_c: continue seen_c.add(key) candidacies.append(c) for m in bundle.get("ballot_measures") or []: key = (m.get("title") or "").lower() if key in seen_m: continue seen_m.add(key) measures.append(m) return { "elections": elections, "candidacies": candidacies, "ballot_measures": measures, "pages_scraped": [b.get("page_url") for b in bundles if b.get("page_url")], }