From 1ced6c5d966428822a7bb4838642a2417f654b8f Mon Sep 17 00:00:00 2001 From: Nathan Date: Sun, 23 Aug 2026 18:31:11 -0400 Subject: [PATCH] fix(lobbying): weekly-scraper cursor exceeds Firestore's 1MB document limit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The live weekly cursor (scrapers/lobbying) stored the entire processed-URL history and per-registrant summary cache as two fields on a single Firestore document. Firestore documents have a hard 1MB size limit; once run against the full corpus, this document grew past it partway through, and every write after that point failed silently — meaning every registrant processed for the rest of that run (and any future run building on that cursor) was silently skipped rather than captured. Moved to subcollections — scrapers/lobbying/processedUrls/{hash} and scrapers/lobbying/summaryCache/{hash}, one small doc per URL — mirroring the pattern the backfill cursor (scrapers/lobbyingBackfill/processedUrls) already used elsewhere in this same file. Point lookups replace the in-memory set/dict that used to get loaded and saved as one blob on every write. Tested with a small in-memory Firestore fake (tests/test_scrape.py) real enough to simulate write-then-read-back across calls, including a regression test asserting the parent cursor doc never grows beyond a handful of small fields. Confirmed this test fails against the pre-fix code by checking it out and rerunning the suite, then passes again with the fix restored. Also validated live against a project with existing data. Co-Authored-By: Claude Sonnet 4.6 --- lobbying-scraper/scrape.py | 96 +++++++++---- lobbying-scraper/tests/test_scrape.py | 198 ++++++++++++++++++++++++++ lobbying-scraper/writer.py | 2 + 3 files changed, 266 insertions(+), 30 deletions(-) create mode 100644 lobbying-scraper/tests/test_scrape.py diff --git a/lobbying-scraper/scrape.py b/lobbying-scraper/scrape.py index 657c59603..470732c66 100644 --- a/lobbying-scraper/scrape.py +++ b/lobbying-scraper/scrape.py @@ -38,43 +38,79 @@ from writer import ( BACKFILL_DOC, BACKFILL_URLS_COLLECTION, + PROCESSED_URLS_COLLECTION, SCRAPER_DOC, + SUMMARY_CACHE_COLLECTION, write_filings, write_registrant, ) # ── 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()} ) @@ -117,7 +153,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 @@ -136,33 +172,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/tests/test_scrape.py b/lobbying-scraper/tests/test_scrape.py new file mode 100644 index 000000000..e2de50768 --- /dev/null +++ b/lobbying-scraper/tests/test_scrape.py @@ -0,0 +1,198 @@ +"""Unit tests for the weekly-mode 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_weekly: subcollection cursor 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 + + +def test_weekly_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, 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() + ): + n = scrape.run_weekly(None, years=[2024], dry_run=True) + + assert n == 1 + assert db.store == {} diff --git a/lobbying-scraper/writer.py b/lobbying-scraper/writer.py index fefd81aaf..c6d4958f8 100644 --- a/lobbying-scraper/writer.py +++ b/lobbying-scraper/writer.py @@ -23,6 +23,8 @@ 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"