improving lobbying data pipeline reliability

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)
 

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 == {}

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"