Use append-only event log checkpoints and keep 'none' in CSV

Author: JCOM127

README.md

@@ -63,8 +63,7 @@ To persist intermediate pipeline data for verification/debugging, set:
 
 - `LEADGEN_CHECKPOINT_DIR` (example: `checkpoints`)
 
-This will write JSON checkpoint files such as `queries.validated.json`, `search_results.<query_id>.json`,
-`leads.raw.json`, `leads.deduped.json`, `leads.filtered.json`, and `leads.final.json`.
+This will append newline-delimited JSON events to `checkpoints/events.jsonl` (append-only event log).
 
 ## Inputs and outputs
 
@@ -78,4 +77,4 @@ Output CSV columns:
 Notes:
 
 - Leads with neither a valid email nor a valid phone are removed.
-- Missing `email`/`phone` values are written as empty cells in the CSV.
+- Missing `email`/`phone` values are written as the literal `none`.

leadgen/models/schemas.py

@@ -176,14 +176,12 @@ def merged_with(self, other: ExtractionResult) -> ExtractionResult:
 
 
 def lead_row(lead: Lead) -> dict[str, Any]:
-    email = "" if is_missing(lead.email) else lead.email
-    phone = "" if is_missing(lead.phone) else lead.phone
     return {
         "title": lead.title or "",
         "url": lead.url,
         "domain": lead.domain,
-        "email": email,
-        "phone": phone,
+        "email": lead.email,
+        "phone": lead.phone,
         "city": lead.city,
         "state": lead.state,
         "industry": lead.industry,

leadgen/pipeline/checkpoints.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+
+def _ensure_dir(path: Path) -> None:
+    path.mkdir(parents=True, exist_ok=True)
+
+
+@dataclass
+class EventLogWriter:
+    directory: Path
+    filename: str = "events.jsonl"
+    _seq: int = 0
+
+    def append(self, event: str, data: Any) -> None:
+        _ensure_dir(self.directory)
+        self._seq += 1
+        record = {"seq": self._seq, "event": event, "data": data}
+        line = json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n"
+        path = self.directory / self.filename
+
+        with path.open("a", encoding="utf-8", newline="\n") as f:
+            f.write(line)
+            f.flush()
+            os.fsync(f.fileno())

leadgen/pipeline/event_log.py

@@ -23,8 +23,8 @@
     lead_row,
     normalize_domain_from_url,
 )
-from leadgen.pipeline.checkpoints import CheckpointWriter
 from leadgen.pipeline.dedupe import dedupe_leads
+from leadgen.pipeline.event_log import EventLogWriter
 from leadgen.scraping.discovery import discover_contact_like_urls, homepage_url
 from leadgen.scraping.extractor import extract_contacts
 from leadgen.scraping.fetcher import Fetcher
@@ -109,27 +109,27 @@ def __init__(
         self._scrape_delay_seconds = scrape_delay_seconds
         self._max_output_leads = max_output_leads
         self._sleep = sleeper
-        self._checkpoints = CheckpointWriter(checkpoint_dir) if checkpoint_dir else None
+        self._event_log = EventLogWriter(checkpoint_dir) if checkpoint_dir else None
 
     def run(self, *, queries_path: Path, output_csv_path: Path) -> PipelineOutput:
         queries_file = load_queries(queries_path)
 
         logger.info("Loaded %s queries from %s", len(queries_file.queries), queries_path)
-        if self._checkpoints:
-            self._checkpoints.write_json(
-                "queries.validated",
-                queries_file.model_dump(),
-            )
+        if self._event_log:
+            self._event_log.append("queries.validated", queries_file.model_dump())
         domain_contexts: dict[str, _DomainContext] = {}
         leads: list[Lead] = []
         for query in queries_file.queries:
             logger.info("Searching query_id=%s", query.id)
             results = self._search.search(query)
             logger.info("Search results query_id=%s count=%s", query.id, len(results))
-            if self._checkpoints:
-                self._checkpoints.write_json(
-                    f"search_results.{query.id}",
-                    [r.model_dump() | {"domain": r.domain} for r in results],
+            if self._event_log:
+                self._event_log.append(
+                    "search.results",
+                    {
+                        "query_id": query.id,
+                        "results": [r.model_dump() | {"domain": r.domain} for r in results],
+                    },
                 )
             leads.extend(
                 self._scrape_results(
@@ -139,19 +139,24 @@ def run(self, *, queries_path: Path, output_csv_path: Path) -> PipelineOutput:
                     domain_contexts=domain_contexts,
                 )
             )
-            if self._checkpoints:
-                self._checkpoints.write_json(
-                    f"leads.scraped.{query.id}",
-                    [lead.model_dump() for lead in leads if lead.source_query_id == query.id],
+            if self._event_log:
+                self._event_log.append(
+                    "leads.scraped",
+                    {
+                        "query_id": query.id,
+                        "leads": [
+                            lead.model_dump() for lead in leads if lead.source_query_id == query.id
+                        ],
+                    },
                 )
 
         logger.info("Collected %s raw leads (pre-dedupe)", len(leads))
-        if self._checkpoints:
-            self._checkpoints.write_json("leads.raw", [lead.model_dump() for lead in leads])
+        if self._event_log:
+            self._event_log.append("leads.raw", [lead.model_dump() for lead in leads])
         deduped = dedupe_leads(leads)
         logger.info("Deduped leads by domain: %s -> %s", len(leads), len(deduped))
-        if self._checkpoints:
-            self._checkpoints.write_json("leads.deduped", [lead.model_dump() for lead in deduped])
+        if self._event_log:
+            self._event_log.append("leads.deduped", [lead.model_dump() for lead in deduped])
 
         filtered = [
             lead for lead in deduped if (is_valid_email(lead.email) or is_valid_phone(lead.phone))
@@ -166,9 +171,9 @@ def run(self, *, queries_path: Path, output_csv_path: Path) -> PipelineOutput:
             )
         )
         limited = filtered[: self._max_output_leads]
-        if self._checkpoints:
-            self._checkpoints.write_json("leads.filtered", [lead.model_dump() for lead in filtered])
-            self._checkpoints.write_json("leads.final", [lead.model_dump() for lead in limited])
+        if self._event_log:
+            self._event_log.append("leads.filtered", [lead.model_dump() for lead in filtered])
+            self._event_log.append("leads.final", [lead.model_dump() for lead in limited])
 
         written = _write_csv(output_csv_path, limited)
         logger.info("Wrote output CSV: %s", output_csv_path)

leadgen/pipeline/runner.py

@@ -1,2 +1,2 @@
 title,url,domain,email,phone,city,state,industry,source_query_id
-Example Domain,https://example.com,example.com,,,Austin,TX,Software,q1
+Example Domain,https://example.com,example.com,none,none,Austin,TX,Software,q1

leads.example.csv

@@ -63,7 +63,7 @@ def test_pipeline_end_to_end_dedupes_and_writes_csv(tmp_path: Path) -> None:
     assert out.written_rows == 2
     csv_text = output.read_text(encoding="utf-8")
     assert "example.com,[email protected],+14155552671" in csv_text
-    assert "other.com,[email protected]," in csv_text
+    assert "other.com,[email protected],none" in csv_text
 
 
 def test_pipeline_crawls_contact_link_from_homepage(tmp_path: Path) -> None: