From 1f5619d16aed93bececb4b6834a98e989b5a34e5 Mon Sep 17 00:00:00 2001 From: Nathan Date: Sun, 23 Aug 2026 09:32:17 -0400 Subject: [PATCH 1/2] fix(lobbying): weekly-scraper cursor & stats reliability at current data scale MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three related Firestore document/field size-limit bugs surfaced when the weekly incremental scraper was run for the first time against the full production-scale dataset (300K+ filings): 1. The live weekly cursor (scrapers/lobbying) stored the entire processed-URL history and summary cache as two fields on one document. That document exceeded Firestore's 1MB limit partway through a run, silently failing (and thus skipping) every registrant processed afterward. Moved to subcollections — one small doc per URL — mirroring the pattern the backfill cursor already used, with point lookups instead of an in-memory set/dict. 2. compute_stats() streamed the full lobbyingFilings/lobbyingRegistrants collections (300K+ docs) in one unbounded query, which timed out server-side; the client library's automatic stream-retry then crashed on an internal AttributeError instead of recovering. Replaced with cursor-paginated batches (50K docs/request) and a manual retry that re-issues a fresh query rather than resuming a broken stream. 3. Once (2) was fixed, compute_stats() reached a third limit: the billSummaries_{court} JSON blob itself exceeded Firestore's 1MB field-size limit for the current session (1,057KB for court 194's ~5,600 bills), with courts 192/193 close behind. Restructured to one small doc per bill in a bills subcollection instead of one JSON blob per court — same fix pattern as (1), applied to writer.py, seedLobbyingStats.ts, and the frontend fetcher in components/db/lobbying.ts. All three fixes validated end-to-end against dev Firestore at current scale (373K filings, 25.6K registrants, 11 courts including the previously-failing 194th). Also includes scripts/firebase-admin/checkLobbyingFreshness.ts, a read-only diagnostic for checking scraper cursor state and data recency. Co-Authored-By: Claude Sonnet 4.6 --- components/db/lobbying.ts | 18 ++-- lobbying-scraper/scrape.py | 96 +++++++++++++------ lobbying-scraper/writer.py | 72 +++++++++++++- .../firebase-admin/checkLobbyingFreshness.ts | 78 +++++++++++++++ scripts/firebase-admin/seedLobbyingStats.ts | 20 +++- 5 files changed, 241 insertions(+), 43 deletions(-) create mode 100644 scripts/firebase-admin/checkLobbyingFreshness.ts diff --git a/components/db/lobbying.ts b/components/db/lobbying.ts index fabae0159..f540c7b5b 100644 --- a/components/db/lobbying.ts +++ b/components/db/lobbying.ts @@ -276,13 +276,19 @@ export type BillRow = { async function fetchLobbyingBillSummaries( court: number ): Promise> { - const snap = await getDoc( - doc(firestore, LOBBYING_STATS_COLLECTION, `billSummaries_${court}`) + const snap = await getDocs( + collection( + firestore, + LOBBYING_STATS_COLLECTION, + `billSummaries_${court}`, + "bills" + ) ) - if (!snap.exists()) return {} - const raw = snap.data() as { data?: string } - if (!raw.data) return {} - return JSON.parse(raw.data) as Record + const result: Record = {} + snap.docs.forEach(d => { + result[d.id] = d.data() as BillSummaryEntry + }) + return result } export function useLobbyingBillSummaries(court: number) { diff --git a/lobbying-scraper/scrape.py b/lobbying-scraper/scrape.py index 55fa793c6..5174406f4 100644 --- a/lobbying-scraper/scrape.py +++ b/lobbying-scraper/scrape.py @@ -38,7 +38,9 @@ from writer import ( BACKFILL_DOC, BACKFILL_URLS_COLLECTION, + PROCESSED_URLS_COLLECTION, SCRAPER_DOC, + SUMMARY_CACHE_COLLECTION, compute_stats, write_filings, write_registrant, @@ -46,36 +48,70 @@ # ── Cursor helpers ──────────────────────────────────────────────────────────── +# +# Weekly-mode cursor state lives in subcollections under SCRAPER_DOC, one small +# doc per URL, mirroring the backfill cursor below. An earlier version stored +# the entire processed-URL history and summary cache as two fields on a single +# document; that doc grew past Firestore's 1MB limit once run against the full +# corpus, silently failing (and thus skipping) every registrant processed +# after the limit was hit. Per-URL docs have no such ceiling. -def _load_live_cursor(db: firestore.Client) -> tuple[set[str], dict[str, list[str]]]: - """Return (processedDiscUrls, summaryDiscCache) from the live scraper doc.""" - doc = db.document(SCRAPER_DOC).get() - data = doc.to_dict() or {} +def _url_hash(url: str) -> str: + return hashlib.sha256(url.encode()).hexdigest()[:40] + + +def _is_processed(db: firestore.Client, disc_url: str) -> bool: + h = _url_hash(disc_url) return ( - set(data.get("processedDiscUrls", [])), - data.get("summaryDiscCache", {}), + db.document(SCRAPER_DOC) + .collection(PROCESSED_URLS_COLLECTION) + .document(h) + .get() + .exists + ) + + +def _mark_processed(db: firestore.Client, disc_url: str) -> None: + h = _url_hash(disc_url) + db.document(SCRAPER_DOC).collection(PROCESSED_URLS_COLLECTION).document(h).set( + {"url": disc_url, "processedAt": datetime.now(tz=timezone.utc).isoformat()} ) -def _save_live_cursor( - db: firestore.Client, - processed: set[str], - cache: dict[str, list[str]], +def _get_cached_disc_urls(db: firestore.Client, summary_url: str) -> list[str] | None: + """Cached disclosure URLs for a registrant's summary page, or None if unseen. + + Only consulted for prior years — the current year is always refetched live + since its disclosures can still change. + """ + h = _url_hash(summary_url) + doc = db.document(SCRAPER_DOC).collection(SUMMARY_CACHE_COLLECTION).document(h).get() + if not doc.exists: + return None + return doc.to_dict().get("discUrls", []) + + +def _cache_disc_urls( + db: firestore.Client, summary_url: str, disc_urls: list[str] ) -> None: - db.document(SCRAPER_DOC).set( - {"processedDiscUrls": list(processed), "summaryDiscCache": cache}, - merge=True, + h = _url_hash(summary_url) + db.document(SCRAPER_DOC).collection(SUMMARY_CACHE_COLLECTION).document(h).set( + { + "summaryUrl": summary_url, + "discUrls": disc_urls, + "cachedAt": datetime.now(tz=timezone.utc).isoformat(), + } ) def _is_backfill_processed(db: firestore.Client, disc_url: str) -> bool: - h = hashlib.sha256(disc_url.encode()).hexdigest()[:40] + h = _url_hash(disc_url) return db.document(BACKFILL_DOC).collection(BACKFILL_URLS_COLLECTION).document(h).get().exists def _mark_backfill_processed(db: firestore.Client, disc_url: str) -> None: - h = hashlib.sha256(disc_url.encode()).hexdigest()[:40] + h = _url_hash(disc_url) db.document(BACKFILL_DOC).collection(BACKFILL_URLS_COLLECTION).document(h).set( {"url": disc_url, "processedAt": datetime.now(tz=timezone.utc).isoformat()} ) @@ -118,7 +154,7 @@ def run_weekly( ) -> int: """Incremental weekly check. Returns number of new disclosures processed.""" current_year = datetime.now(tz=timezone.utc).year - processed, cache = _load_live_cursor(db) if db is not None else (set(), {}) + use_cursor = db is not None and not dry_run session = make_session() new_count = 0 @@ -137,33 +173,33 @@ def run_weekly( print(f" {len(summary_urls)} registrants on portal") for summary_url in summary_urls: - # Use cached disc URLs for prior years; always re-check current year - disc_urls = cache.get(summary_url) - if disc_urls is None or year == current_year: + # Prior years: trust the cache if we have one. Current year: + # always refetch live, since its disclosures can still change. + disc_urls = None + if year != current_year and use_cursor: + disc_urls = _get_cached_disc_urls(db, summary_url) + + if disc_urls is None: try: meta = fetch_disclosure_meta(session, summary_url) disc_urls = meta.disclosure_urls - cache[summary_url] = disc_urls - if not dry_run: - _save_live_cursor(db, processed, cache) + if use_cursor: + _cache_disc_urls(db, summary_url, disc_urls) except Exception as e: print(f" failed to fetch summary {summary_url}: {e}", file=sys.stderr) continue - new_disc_urls = [u for u in disc_urls if u not in processed] - if not new_disc_urls: - continue - - for disc_url in new_disc_urls: + for disc_url in disc_urls: + if use_cursor and _is_processed(db, disc_url): + continue try: comp_n, filing_n = process_disclosure( db, session, summary_url, disc_url, year, dry_run=dry_run ) - processed.add(disc_url) new_count += 1 print(f" processed: {comp_n} clients, {filing_n} filings") - if not dry_run: - _save_live_cursor(db, processed, cache) + if use_cursor: + _mark_processed(db, disc_url) except Exception as e: print(f" failed to process {disc_url}: {e}", file=sys.stderr) diff --git a/lobbying-scraper/writer.py b/lobbying-scraper/writer.py index cfcfbca38..a798f271d 100644 --- a/lobbying-scraper/writer.py +++ b/lobbying-scraper/writer.py @@ -6,9 +6,10 @@ from __future__ import annotations -import json +import time from datetime import datetime, timezone +from google.api_core.exceptions import GoogleAPICallError from google.cloud import firestore from normalize import normalize_entity_name from portal import ( @@ -24,11 +25,56 @@ REGISTRANTS_COLLECTION = "lobbyingRegistrants" FILINGS_COLLECTION = "lobbyingFilings" SCRAPER_DOC = "scrapers/lobbying" +PROCESSED_URLS_COLLECTION = "processedUrls" +SUMMARY_CACHE_COLLECTION = "summaryCache" BACKFILL_DOC = "scrapers/lobbyingBackfill" BACKFILL_URLS_COLLECTION = "processedUrls" STATS_COLLECTION = "lobbyingMeta" STATS_DOC_ID = "stats" +# compute_stats() streams the full filings/registrants collections, which at +# MAPLE's current scale (300K+ docs) can exceed Firestore's server-side query +# timeout. Batching with an explicit cursor keeps each individual RPC small +# and fast; retry=None disables the client library's built-in stream-retry +# (which has a version-skew bug that crashes instead of retrying), and the +# manual retry loop below just re-issues a fresh, small query on failure +# instead of trying to resume a broken stream. +_BATCH_SIZE = 50000 +_MAX_RETRIES = 3 + + +def _iter_collection(db: firestore.Client, collection_name: str): + """Yield every document in a collection via small, cursor-paginated reads.""" + coll_ref = db.collection(collection_name) + last_doc = None + + while True: + query = coll_ref.order_by("__name__").limit(_BATCH_SIZE) + if last_doc is not None: + query = query.start_after(last_doc) + + for attempt in range(_MAX_RETRIES): + try: + batch = list(query.stream(retry=None)) + break + except GoogleAPICallError as e: + if attempt == _MAX_RETRIES - 1: + raise + print( + f" batch read failed ({e}); retrying " + f"({attempt + 1}/{_MAX_RETRIES})…" + ) + time.sleep(2**attempt) + + if not batch: + return + + yield from batch + last_doc = batch[-1] + + if len(batch) < _BATCH_SIZE: + return + def _now() -> datetime: return datetime.now(tz=timezone.utc) @@ -60,7 +106,7 @@ def compute_stats(db: firestore.Client) -> None: bill_entity_sets: dict[int, dict[str, set]] = {} total_filings = 0 - for doc in db.collection(FILINGS_COLLECTION).stream(): + for doc in _iter_collection(db, FILINGS_COLLECTION): d = doc.to_dict() year = str(d.get("year", "")) gc = d.get("generalCourt") @@ -114,7 +160,7 @@ def compute_stats(db: firestore.Client) -> None: spend_by_year: dict[str, float] = {} total_registrants = 0 - for doc in db.collection(REGISTRANTS_COLLECTION).stream(): + for doc in _iter_collection(db, REGISTRANTS_COLLECTION): d = doc.to_dict() year = str(d.get("year", "")) for c in d.get("clients", []): @@ -143,9 +189,25 @@ def compute_stats(db: firestore.Client) -> None: client_filing_counts ) for gc, bills_map in bill_summaries.items(): - db.collection(STATS_COLLECTION).document(f"billSummaries_{gc}").set( - {"data": json.dumps(bills_map)} + # One small doc per bill, not one JSON blob per court: a court's blob + # eventually exceeds Firestore's 1MB field-size limit as its session + # accumulates filings (hit at 1,057KB for court 194 with ~5,600 + # bills). Per-bill docs have no such ceiling. + parent_ref = db.collection(STATS_COLLECTION).document(f"billSummaries_{gc}") + parent_ref.set( + {"billCount": len(bills_map), "updatedAt": _now().isoformat()} ) + bills_coll = parent_ref.collection("bills") + batch = db.batch() + count = 0 + for bill_id, counts in bills_map.items(): + batch.set(bills_coll.document(bill_id), counts) + count += 1 + if count % 400 == 0: + batch.commit() + batch = db.batch() + if count % 400 != 0: + batch.commit() print( f" stats written: {total_filings} filings, " f"{total_registrants} registrants, {len(client_norms)} clients, " diff --git a/scripts/firebase-admin/checkLobbyingFreshness.ts b/scripts/firebase-admin/checkLobbyingFreshness.ts new file mode 100644 index 000000000..16809de03 --- /dev/null +++ b/scripts/firebase-admin/checkLobbyingFreshness.ts @@ -0,0 +1,78 @@ +import { Script } from "./types" + +export const script: Script = async ({ db }) => { + const filingsSnap = await db + .collection("lobbyingFilings") + .orderBy("fetchedAt", "desc") + .limit(5) + .get() + + console.log(`lobbyingFilings: most recently fetched docs`) + filingsSnap.docs.forEach(doc => { + const d = doc.data() + console.log( + ` fetchedAt=${d.fetchedAt?.toDate?.()?.toISOString()} year=${ + d.year + } gc=${d.generalCourt} entity=${d.entityName}` + ) + }) + + const registrantsSnap = await db + .collection("lobbyingRegistrants") + .orderBy("fetchedAt", "desc") + .limit(5) + .get() + + console.log(`\nlobbyingRegistrants: most recently fetched docs`) + registrantsSnap.docs.forEach(doc => { + const d = doc.data() + console.log( + ` fetchedAt=${d.fetchedAt?.toDate?.()?.toISOString()} year=${ + d.year + } entity=${d.entityName}` + ) + }) + + const scraperDoc = await db.doc("scrapers/lobbying").get() + console.log(`\nscrapers/lobbying doc exists: ${scraperDoc.exists}`) + const processedUrlsSnap = await db + .collection("scrapers/lobbying/processedUrls") + .get() + const summaryCacheSnap = await db + .collection("scrapers/lobbying/summaryCache") + .get() + console.log(` processedUrls subcollection: ${processedUrlsSnap.size} URLs`) + console.log( + ` summaryCache subcollection: ${summaryCacheSnap.size} registrant summaries cached` + ) + + const backfillDoc = await db.doc("scrapers/lobbyingBackfill").get() + console.log(`\nscrapers/lobbyingBackfill doc exists: ${backfillDoc.exists}`) + if (backfillDoc.exists) { + console.log(` completedYears: ${backfillDoc.data()?.completedYears}`) + } + + const currentYear = new Date().getFullYear() + const yRegistrants = await db + .collection("lobbyingRegistrants") + .where("year", "==", currentYear) + .get() + const yFilings = await db + .collection("lobbyingFilings") + .where("year", "==", currentYear) + .get() + console.log( + `\n${currentYear}: ${yRegistrants.size} registrants, ${yFilings.size} filings` + ) + + // Grand totals come from the last computed stats doc rather than a live + // full-collection scan — cheap, and the corpus (300K+ docs) makes an + // aggregate scan here wasteful for what's meant to be a quick check. + const statsDoc = await db.doc("lobbyingMeta/stats").get() + const stats = statsDoc.data() + console.log( + `\nlobbyingMeta/stats (as of last compute): ${ + stats?.totalFilings ?? "?" + } filings, ${stats?.totalRegistrants ?? "?"} registrants` + ) +} diff --git a/scripts/firebase-admin/seedLobbyingStats.ts b/scripts/firebase-admin/seedLobbyingStats.ts index 42f959c72..254605bd7 100644 --- a/scripts/firebase-admin/seedLobbyingStats.ts +++ b/scripts/firebase-admin/seedLobbyingStats.ts @@ -151,10 +151,26 @@ export const script: Script = async ({ db }) => { .set(clientFilingCounts) for (const [court, billsMap] of Object.entries(billSummaries)) { - await db + // One small doc per bill, not one JSON blob per court: a court's blob + // eventually exceeds Firestore's 1MB field-size limit as its session + // accumulates filings (hit at 1,057KB for court 194 with ~5,600 bills). + // Per-bill docs have no such ceiling. + const parentRef = db .collection(STATS_COLLECTION) .doc(`billSummaries_${court}`) - .set({ data: JSON.stringify(billsMap) }) + const entries = Object.entries(billsMap) + await parentRef.set({ + billCount: entries.length, + updatedAt: new Date().toISOString() + }) + const billsColl = parentRef.collection("bills") + for (let i = 0; i < entries.length; i += 400) { + const batch = db.batch() + for (const [billId, counts] of entries.slice(i, i + 400)) { + batch.set(billsColl.doc(billId), counts) + } + await batch.commit() + } } console.log(`Written to ${STATS_COLLECTION}/${STATS_DOC_ID}`) From 2d12bb52a7e99b8a2873876f9e07e4eaef89478e Mon Sep 17 00:00:00 2001 From: Nathan Date: Sun, 23 Aug 2026 17:31:34 -0400 Subject: [PATCH 2/2] fix(lobbying): remove permanent completedYears gate from backfill mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit run_backfill() marked a year "complete" after one pass and skipped it forever on every future run. That's wrong for the current (still-accruing) year: a run partway through the year would mark it complete based on whatever existed at that moment, silently missing every disclosure filed afterward — no future backfill run would ever see it again. This is exactly what happened to 2026 in production: marked complete in July with 0 disclosures captured. run_backfill already has a fully correct, granular completeness check — _is_backfill_processed, a per-URL subcollection lookup. The year-level flag only ever bought a coarse fast-path (skip re-listing a year's registrants entirely) and it's what caused the bug. Removed it: every run now always re-lists every requested year (one cheap HTTP request per year) and relies solely on the per-URL cursor for correctness, so no year can ever be skipped wholesale again. Added tests/test_scrape.py with a small in-memory Firestore fake (real enough to simulate write-then-read-back across calls, unlike a plain mock) covering both cursor systems: - Regression test reproducing the exact bug scenario (empty pass, then real data appears for the same year) — fails against the old code with 4/4 backfill tests red, passes with the fix, confirmed by checking out the pre-fix scrape.py and rerunning the suite against it. - Backfill always re-lists every year, per-URL dedup still works, dry-run never touches Firestore, completedYears is never written anywhere. - Weekly-mode cursor sanity checks (prior-year caching, current-year always live, parent doc stays small — all state in subcollections). Also validated live against dev: --mode backfill --year 2005 --limit 2 run twice confirms the year is re-listed both times while already-processed disclosures are correctly not reprocessed (0 new on both runs, as expected since 2005 was already backfilled in July). Co-Authored-By: Claude Sonnet 4.6 --- lobbying-scraper/scrape.py | 41 ++-- lobbying-scraper/tests/test_scrape.py | 322 ++++++++++++++++++++++++++ 2 files changed, 342 insertions(+), 21 deletions(-) create mode 100644 lobbying-scraper/tests/test_scrape.py diff --git a/lobbying-scraper/scrape.py b/lobbying-scraper/scrape.py index 5174406f4..1f4ed23a4 100644 --- a/lobbying-scraper/scrape.py +++ b/lobbying-scraper/scrape.py @@ -207,17 +207,21 @@ def run_weekly( # ── Historical backfill ─────────────────────────────────────────────────────── - - -def _completed_years(db: "firestore.Client") -> set[int]: - data = db.document(BACKFILL_DOC).get().to_dict() or {} - return set(data.get("completedYears", [])) - - -def _mark_year_complete(db: "firestore.Client", year: int) -> None: - db.document(BACKFILL_DOC).set( - {"completedYears": firestore.ArrayUnion([year])}, merge=True - ) +# +# Correctness here relies entirely on the per-URL cursor (_is_backfill_processed +# / _mark_backfill_processed below) — every disclosure URL is checked and +# marked individually, so re-running a backfill is always safe and complete. +# +# An earlier version also tracked a per-year "completedYears" flag as a +# fast-path to skip re-listing a year's registrants at all. That flag was +# permanent once set, which is wrong for the current (still-accruing) year: +# a backfill run partway through the year would mark it complete after +# finding whatever existed at that moment, and every later run would then +# skip it forever — silently missing every disclosure filed afterward. There +# is no reliable way to tell "genuinely finished" apart from "happened to be +# a quiet moment" for a year that's still in progress, so the flag is gone; +# each run always re-lists every requested year's registrants (one cheap +# HTTP request per year) and leans on the per-URL cursor for correctness. def run_backfill( @@ -226,18 +230,15 @@ def run_backfill( limit: int | None = None, dry_run: bool = False, ) -> int: - """Full historical backfill using the subcollection cursor. Resumable.""" + """Full historical backfill using the per-URL subcollection cursor. + + Always resumable and safe to re-run: every disclosure URL is checked + individually against the cursor, so no year is ever skipped wholesale. + """ session = make_session() total_new = 0 - done = _completed_years(db) if db is not None and not dry_run else set() - if done: - print(f"Skipping already-completed years: {sorted(done)}") - for year in years: - if year in done: - continue - print(f"\n── {year} ──") try: summary_urls = fetch_summary_links(session, year) @@ -276,8 +277,6 @@ def run_backfill( print(f" [{i+1}/{len(summary_urls)}] {year_new} new disclosures so far") print(f" {year} complete: {year_new} new disclosures") - if db is not None and not dry_run and not limit: - _mark_year_complete(db, year) return total_new diff --git a/lobbying-scraper/tests/test_scrape.py b/lobbying-scraper/tests/test_scrape.py new file mode 100644 index 000000000..4e8d1557a --- /dev/null +++ b/lobbying-scraper/tests/test_scrape.py @@ -0,0 +1,322 @@ +"""Unit tests for the weekly/backfill cursor logic in scrape.py. + +Uses a tiny in-memory fake standing in for firestore.Client — real enough to +exercise document/subcollection reads and writes statefully across calls +(unlike a plain MagicMock, which can't easily simulate "write now, read back +later"), without needing a live database or emulator. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from portal import DisclosureDetail, DisclosureMeta +import scrape + + +# ── Fake Firestore ──────────────────────────────────────────────────────────── + + +class _FakeSnapshot: + def __init__(self, data): + self._data = data + + @property + def exists(self): + return self._data is not None + + def to_dict(self): + return self._data + + +class _FakeDocRef: + def __init__(self, store, path): + self._store = store + self._path = path + + def get(self): + return _FakeSnapshot(self._store.get(self._path)) + + def set(self, data, merge=False): + if merge and self._path in self._store: + self._store[self._path] = {**self._store[self._path], **data} + else: + self._store[self._path] = dict(data) + + def collection(self, name): + return _FakeCollectionRef(self._store, f"{self._path}/{name}") + + +class _FakeCollectionRef: + def __init__(self, store, path): + self._store = store + self._path = path + + def document(self, doc_id): + return _FakeDocRef(self._store, f"{self._path}/{doc_id}") + + +class FakeFirestore: + """Minimal stand-in for firestore.Client: document()/collection() only.""" + + def __init__(self): + self.store: dict[str, dict] = {} + + def document(self, path): + return _FakeDocRef(self.store, path) + + def collection(self, name): + return _FakeCollectionRef(self.store, name) + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + + +def _meta(summary_url: str, disc_urls: list[str]) -> DisclosureMeta: + return DisclosureMeta( + entity_name=f"Entity for {summary_url}", + year=2024, + reg_type="Employer", + disclosure_urls=disc_urls, + ) + + +# ── run_backfill: no year is ever skipped wholesale ────────────────────────── + + +def test_backfill_relists_every_year_on_every_run(): + """No completedYears-style gate: fetch_summary_links must be called for + every requested year, on every run, regardless of what a prior run did.""" + db = FakeFirestore() + summary_links = {2020: ["https://x/summary/a"], 2021: [], 2022: []} + meta_by_url = {"https://x/summary/a": _meta("https://x/summary/a", [])} + + with patch("scrape.make_session", return_value=None), patch( + "scrape.fetch_summary_links", + side_effect=lambda session, year: summary_links.get(year, []), + ) as fetch_links, patch( + "scrape.fetch_disclosure_meta", + side_effect=lambda session, url: meta_by_url[url], + ): + scrape.run_backfill(db, years=[2020, 2021, 2022]) + scrape.run_backfill(db, years=[2020, 2021, 2022]) + + # 3 years x 2 runs = 6 calls, none skipped by a "completed" flag. + assert fetch_links.call_count == 6 + called_years = sorted(c.args[1] for c in fetch_links.call_args_list) + assert called_years == [2020, 2020, 2021, 2021, 2022, 2022] + + +def test_backfill_does_not_write_completed_years_anywhere(): + """The old completedYears field must never be written by the new code.""" + db = FakeFirestore() + summary_links = {2024: ["https://x/summary/a"]} + meta_by_url = { + "https://x/summary/a": _meta("https://x/summary/a", ["https://x/disc/1"]) + } + + with patch("scrape.make_session", return_value=None), patch( + "scrape.fetch_summary_links", + side_effect=lambda session, year: summary_links.get(year, []), + ), patch( + "scrape.fetch_disclosure_meta", + side_effect=lambda session, url: meta_by_url[url], + ), patch( + "scrape.fetch_disclosure_detail", return_value=DisclosureDetail() + ), patch( + "scrape.write_registrant", return_value=None + ), patch( + "scrape.write_filings", return_value=0 + ): + scrape.run_backfill(db, years=[2024]) + + for path, data in db.store.items(): + assert "completedYears" not in data, f"stale field written at {path}" + + +def test_backfill_regression_partial_year_then_real_data_appears(): + """The exact bug scenario: a backfill run mid-year finds nothing for the + current year (quiet moment), and a later run for the same year finds real + disclosures. The second run must NOT skip the year — it must process the + newly-appeared data. (Previously: the first pass would mark the year + 'complete' with 0 disclosures, and the second run would skip it forever.) + """ + db = FakeFirestore() + summary_url = "https://x/summary/late-filer" + disc_url = "https://x/disc/late-filer-1" + + # First run: this year has no registrants yet. + with patch("scrape.make_session", return_value=None), patch( + "scrape.fetch_summary_links", side_effect=lambda session, year: [] + ) as fetch_links_1: + n1 = scrape.run_backfill(db, years=[2026]) + assert n1 == 0 + assert fetch_links_1.call_count == 1 + + # Second run: a registrant has since filed for the same year. + with patch("scrape.make_session", return_value=None), patch( + "scrape.fetch_summary_links", + side_effect=lambda session, year: [summary_url], + ) as fetch_links_2, patch( + "scrape.fetch_disclosure_meta", + side_effect=lambda session, url: _meta(url, [disc_url]), + ), patch( + "scrape.fetch_disclosure_detail", return_value=DisclosureDetail() + ), patch( + "scrape.write_registrant", return_value=None + ), patch( + "scrape.write_filings", return_value=0 + ): + n2 = scrape.run_backfill(db, years=[2026]) + + # The year was re-listed (not skipped) and the new disclosure was processed. + assert fetch_links_2.call_count == 1 + assert n2 == 1 + + +def test_backfill_skips_already_processed_disclosures_but_not_the_year(): + """Per-URL dedup still works: a disclosure already marked processed is + not reprocessed, even though the year itself is always re-listed.""" + db = FakeFirestore() + summary_url = "https://x/summary/a" + disc_url = "https://x/disc/1" + meta_by_url = {summary_url: _meta(summary_url, [disc_url])} + + def run(): + with patch("scrape.make_session", return_value=None), patch( + "scrape.fetch_summary_links", + side_effect=lambda session, year: [summary_url], + ), patch( + "scrape.fetch_disclosure_meta", + side_effect=lambda session, url: meta_by_url[url], + ), patch( + "scrape.fetch_disclosure_detail", return_value=DisclosureDetail() + ), patch( + "scrape.write_registrant", return_value=None + ), patch( + "scrape.write_filings", return_value=0 + ): + return scrape.run_backfill(db, years=[2024]) + + n1 = run() + n2 = run() + + assert n1 == 1 # first run: one new disclosure + assert n2 == 0 # second run: already processed, correctly skipped + + +def test_backfill_dry_run_never_touches_firestore(): + db = FakeFirestore() + summary_url = "https://x/summary/a" + disc_url = "https://x/disc/1" + meta_by_url = {summary_url: _meta(summary_url, [disc_url])} + + with patch("scrape.make_session", return_value=None), patch( + "scrape.fetch_summary_links", + side_effect=lambda session, year: [summary_url], + ), patch( + "scrape.fetch_disclosure_meta", + side_effect=lambda session, url: meta_by_url[url], + ), patch( + "scrape.fetch_disclosure_detail", return_value=DisclosureDetail() + ): + n = scrape.run_backfill(None, years=[2024], dry_run=True) + + assert n == 1 + assert db.store == {} + + +# ── run_weekly: subcollection cursor (Bug 1 fix) sanity checks ─────────────── + + +def test_weekly_skips_already_processed_disclosure(): + db = FakeFirestore() + summary_url = "https://x/summary/a" + disc_url = "https://x/disc/1" + meta_by_url = {summary_url: _meta(summary_url, [disc_url])} + + def run(year): + with patch("scrape.make_session", return_value=None), patch( + "scrape.fetch_summary_links", + side_effect=lambda session, y: [summary_url], + ), patch( + "scrape.fetch_disclosure_meta", + side_effect=lambda session, url: meta_by_url[url], + ), patch( + "scrape.fetch_disclosure_detail", return_value=DisclosureDetail() + ), patch( + "scrape.write_registrant", return_value=None + ), patch( + "scrape.write_filings", return_value=0 + ): + return scrape.run_weekly(db, years=[year]) + + # Use a prior (non-current) year so caching also gets exercised below. + n1 = run(2020) + n2 = run(2020) + + assert n1 == 1 + assert n2 == 0 + + +def test_weekly_caches_prior_year_but_not_current_year(): + db = FakeFirestore() + current_year = scrape.datetime.now(tz=scrape.timezone.utc).year + prior_year = current_year - 1 + summary_url = "https://x/summary/a" + + with patch("scrape.make_session", return_value=None), patch( + "scrape.fetch_summary_links", + side_effect=lambda session, y: [summary_url], + ), patch( + "scrape.fetch_disclosure_meta", + side_effect=lambda session, url: _meta(url, []), + ) as fetch_meta: + # Prior year twice: second call should hit the cache, not refetch. + scrape.run_weekly(db, years=[prior_year]) + scrape.run_weekly(db, years=[prior_year]) + assert fetch_meta.call_count == 1 + + # Current year twice: must always refetch live. + fetch_meta.reset_mock() + scrape.run_weekly(db, years=[current_year]) + scrape.run_weekly(db, years=[current_year]) + assert fetch_meta.call_count == 2 + + +def test_weekly_cursor_doc_never_exceeds_a_few_small_fields(): + """Regression guard for the original 1MB-doc bug: the parent scraper doc + itself must stay tiny — all real state lives in subcollection docs.""" + db = FakeFirestore() + summary_url = "https://x/summary/a" + disc_url = "https://x/disc/1" + meta_by_url = {summary_url: _meta(summary_url, [disc_url])} + + with patch("scrape.make_session", return_value=None), patch( + "scrape.fetch_summary_links", + side_effect=lambda session, y: [summary_url], + ), patch( + "scrape.fetch_disclosure_meta", + side_effect=lambda session, url: meta_by_url[url], + ), patch( + "scrape.fetch_disclosure_detail", return_value=DisclosureDetail() + ), patch( + "scrape.write_registrant", return_value=None + ), patch( + "scrape.write_filings", return_value=0 + ): + scrape.run_weekly(db, years=[2020]) + + parent = db.store.get(scrape.SCRAPER_DOC) + assert parent is None or "processedDiscUrls" not in parent + assert parent is None or "summaryDiscCache" not in parent + + # The actual state must be in per-URL subcollection docs. + subcollection_paths = [ + p for p in db.store if p.startswith(scrape.SCRAPER_DOC + "/") + ] + assert len(subcollection_paths) >= 2 # one processedUrls doc, one summaryCache doc