From abdadf04a45e15e2c516b61a4803da1d8ebfd80e Mon Sep 17 00:00:00 2001 From: anusbutt Date: Mon, 27 Jul 2026 14:44:17 +0500 Subject: [PATCH] feat(source): return only new companies, in a stable order Two changes to `prospector source`, both about repeat runs. **Only new companies pass the gate.** Sourcing had no memory: running it twice returned the same businesses both times. Companies already in the vault are now dropped immediately after dedupe and BEFORE the fetch loop, so a repeat sweep costs Places queries and nothing else. Verified against the live vault: replaying the 108 previously sourced companies drops all 108, with no false passes. Matching is by company slug, with the website domain as a second key. Slug is the only key available before the homepage fetch, which is what lets the gate run early; domain catches a business whose Places listing name has drifted since it was first sourced. Email was rejected as a key for two reasons: it is only known after fetching, and 17 of the 62 addresses in the live ledger are on free providers, where the email domain says nothing about the company. The gate is the vault, not the send ledger. The ask was "don't show me companies I already emailed", but on the real data only 51 of the 108 known companies had been emailed -- gating on the ledger alone would have let 57 already-researched, already-drafted companies back through. The vault is a superset, so the stated requirement still holds. Dropped companies are counted and reported, never silently discarded, and a sweep that finds nothing new says so instead of printing an empty table. `--include-known` bypasses the gate; `--vault` points at another vault; a run with no vault yet suppresses nothing. **Output is deterministic.** Places can return the same businesses in a different order on different days, which used to change both the row order and -- via first-seen-wins dedupe -- which of two duplicates survived. Rows are now sorted by company, and dedupe ranks candidates by metro position then place_id before collapsing, so metro precedence is unchanged but the winner no longer depends on API response order. An unchanged result set now produces an unchanged file. Tests shuffle the Places response and assert a byte-identical CSV rather than just running twice -- a run-twice test passed before this change and proved nothing. --- README.md | 17 ++ prospector/cli.py | 21 +- prospector/source.py | 110 +++++++++- .../test_source_gate_and_determinism.py | 206 ++++++++++++++++++ 4 files changed, 346 insertions(+), 8 deletions(-) create mode 100644 tests/integration/test_source_gate_and_determinism.py diff --git a/README.md b/README.md index 5de3e8a..f1f8bbe 100644 --- a/README.md +++ b/README.md @@ -314,6 +314,23 @@ prospector source --out candidates.csv --max-queries 30 Without `--keyword`, the profile's first keyword is used. +**Only new companies come back.** Anything already in your vault is dropped +before its website is fetched, so a repeat sweep spends Places queries and +nothing else, and the CSV contains only companies you have not seen. Matching is +by company slug, with the website domain as a second key to catch a business +whose listing name has changed. Dropped companies are counted in the summary +rather than silently discarded: + +```text + already in vault (skipped): 40 new companies: 7 +``` + +Use `--include-known` to keep them, and `--vault` to point at a different vault. +A first run with no vault yet suppresses nothing. + +Output rows are sorted by company name, so an unchanged result set produces an +unchanged file — two candidate CSVs can be diffed meaningfully. + `source` uses Google Places Text Search, deduplicates results, fetches each candidate's own website, and checks retrieved markup for Meta Pixel signals without contacting Facebook. By default it writes pixel-positive candidates; diff --git a/prospector/cli.py b/prospector/cli.py index 7a08d07..793cf49 100644 --- a/prospector/cli.py +++ b/prospector/cli.py @@ -114,9 +114,15 @@ def source( keep_all: bool = typer.Option(False, "--all", help="Keep every discovered candidate (default: only ad_signal: pixel)"), max_queries: int = typer.Option(60, "--max-queries", help="Places request budget for this run"), limit: int = typer.Option(None, "--limit", help="Stop after N metros (testing)"), + vault: Path = typer.Option(None, "--vault", help="Vault to check for already-known companies (default: Vault/Outreach)"), + include_known: bool = typer.Option(False, "--include-known", help="Do not drop companies already in the vault"), verbose: bool = typer.Option(False, "--verbose", help="Per-step logging to stderr"), ): - """Discover companies for a keyword across US metros and write a candidate CSV.""" + """Discover companies for a keyword across US metros and write a candidate CSV. + + Companies already in the vault are dropped before any homepage is fetched, + so a repeat sweep returns only what is new. Use --include-known to keep them. + """ from prospector.source import load_metros, run_sourcing # deferred: keeps --help fast settings = load_settings() @@ -138,6 +144,8 @@ def source( max_queries=max_queries, limit=limit, verbose=verbose, + vault_dir=vault or settings.vault_dir, + include_known=include_known, ) except ConfigError as exc: typer.echo(f"error: {exc}", err=True) @@ -284,10 +292,19 @@ def _print_sourcing_summary(summary) -> None: typer.echo( f" discovered: {summary.discovered} duplicates collapsed: {summary.duplicates_collapsed}" ) + # Never a silent drop: the gate is the reason a repeat sweep looks empty. + typer.echo( + f" already in vault (skipped): {summary.already_known} new companies: {summary.kept_with_all}" + ) typer.echo( f" pixel-positive: {summary.pixel_positive} emails found: {summary.emails_found} rows written: {summary.written}" ) - if summary.written == 0 and summary.kept_with_all > 0: + if summary.kept_with_all == 0 and summary.already_known: + typer.echo( + "\n Nothing new: every company found is already in your vault." + "\n Try another keyword or more metros, or --include-known to list them anyway." + ) + elif summary.written == 0 and summary.kept_with_all > 0: typer.echo(f" note: 0 rows written; --all would have kept {summary.kept_with_all}") if summary.failures: typer.echo("") diff --git a/prospector/source.py b/prospector/source.py index be3b04b..8db5788 100644 --- a/prospector/source.py +++ b/prospector/source.py @@ -18,6 +18,7 @@ import httpx +from prospector import vault from prospector.config import ConfigError from prospector.extract import EMAIL_RE, _plausible_email, extract_public_email from prospector.fetch import BlockedHostError, Fetcher, FetchError, is_blocked_host @@ -57,6 +58,7 @@ class SourcingSummary: query_budget: int = 0 discovered: int = 0 duplicates_collapsed: int = 0 + already_known: int = 0 # dropped at the gate: already in the vault kept_with_all: int = 0 # unique candidates (what --all would write) pixel_positive: int = 0 emails_found: int = 0 @@ -156,13 +158,26 @@ def candidate_from_place(place: dict, metro: str) -> Candidate | None: ) -def dedupe(candidates: list[Candidate], summary: SourcingSummary) -> list[Candidate]: +def dedupe( + candidates: list[Candidate], summary: SourcingSummary, metros: list[str] | None = None +) -> list[Candidate]: """Collapse duplicates by place_id, then website domain. First seen wins - (metro-list order, so bigger metros win ties — research.md R6).""" + (metro-list order, so bigger metros win ties — research.md R6). + + "First seen" is made explicit rather than inherited from the order Places + happened to answer in: candidates are ranked by metro position, then by + place_id, before the pass. Metro precedence is unchanged; what changes is + that two runs over the same results now collapse to the same winner even if + the API returned them in a different order.""" + order = {metro: i for i, metro in enumerate(metros or [])} + ranked = sorted( + candidates, + key=lambda c: (order.get(c.metro, len(order)), c.place_id, c.company), + ) unique: list[Candidate] = [] seen_ids: set[str] = set() seen_domains: set[str] = set() - for candidate in candidates: + for candidate in ranked: if candidate.place_id and candidate.place_id in seen_ids: summary.duplicates_collapsed += 1 continue @@ -177,6 +192,65 @@ def dedupe(candidates: list[Candidate], summary: SourcingSummary) -> list[Candid return unique +def known_companies(vault_dir: str | Path | None) -> tuple[set[str], set[str]]: + """Slugs and website domains of every company already in the vault. + + Sourcing is for finding companies you do NOT have yet. Anything already in + the vault has been researched, drafted, and possibly emailed, so re-finding + it costs Places quota and a homepage fetch to produce a row you would throw + away. A missing vault is not an error — it just means nothing is known yet. + + Matching is by company slug, with the website domain as a second key. Slug + is the only one of the two available before the homepage fetch, which is why + the gate can run early; domain catches a business whose Places display name + has drifted since it was first sourced.""" + slugs: set[str] = set() + domains: set[str] = set() + if vault_dir is None: + return slugs, domains + vault_dir = Path(vault_dir) + if not vault_dir.is_dir(): + return slugs, domains + for path in sorted(vault_dir.glob("*.md")): + if path.name.startswith("_"): # _Dashboard.md and friends + continue + slugs.add(path.stem) + try: + frontmatter, _ = vault.parse_note(path.read_text(encoding="utf-8")) + except OSError: + continue + website = (frontmatter.get("website") or "").strip().lower() + if not website: + continue + # Notes store a display form ("acme.com/about"), not a URL. + domain = _domain(website if "//" in website else f"https://{website}") + if domain: + domains.add(domain) + return slugs, domains + + +def drop_known( + candidates: list[Candidate], + known_slugs: set[str], + known_domains: set[str], + summary: SourcingSummary, +) -> list[Candidate]: + """Keep only companies the vault has never seen (the sourcing gate). + + Runs before any homepage is fetched, so a repeat sweep costs Places queries + and nothing else. Dropped companies are counted, never silently discarded.""" + fresh: list[Candidate] = [] + for candidate in candidates: + if vault.slugify(candidate.company) in known_slugs: + summary.already_known += 1 + continue + if candidate.domain and candidate.domain in known_domains: + summary.already_known += 1 + continue + fresh.append(candidate) + return fresh + + def _city_state(address: str) -> str | None: # "123 Main St, Boston, MA 02101, USA" -> "Boston, MA" parts = [p.strip() for p in address.split(",")] @@ -311,12 +385,23 @@ def capture_email(candidate: Candidate, html: str, fetcher: Fetcher) -> None: CSV_HEADER = ["company", "email", "website", "city", "ad_signal"] +def _row_sort_key(candidate: Candidate) -> tuple[str, str, str]: + """Canonical row order: by company, then domain, then place_id. + + Places can return the same businesses in a different order on different + days. Sorting on the candidate's own values means the output file depends on + WHAT was found, not on the order it arrived in — so an unchanged result set + produces an unchanged file. Company first because the CSV is read by a human + before it is fed to `run`.""" + return (candidate.company.casefold(), candidate.domain or "", candidate.place_id) + + def write_candidates_csv(candidates: list[Candidate], out: Path) -> int: """Write the candidate CSV (header always; zero rows -> header-only file).""" with out.open("w", newline="", encoding="utf-8") as fh: writer = csv.writer(fh) writer.writerow(CSV_HEADER) - for c in candidates: + for c in sorted(candidates, key=_row_sort_key): writer.writerow([c.company, c.email or "", c.domain or "", c.city, c.ad_signal]) return len(candidates) @@ -333,8 +418,10 @@ def run_sourcing( verbose: bool = False, searcher: PlacesSearcher | None = None, fetcher: Fetcher | None = None, + vault_dir: str | Path | None = None, + include_known: bool = False, ) -> SourcingSummary: - """Full sourcing pipeline: search -> parse -> dedupe -> classify -> write CSV.""" + """Full sourcing pipeline: search -> dedupe -> drop known -> classify -> CSV.""" import sys log = (lambda msg: print(msg, file=sys.stderr)) if verbose else None @@ -345,7 +432,18 @@ def run_sourcing( candidates = discover( searcher, keyword, metros, max_queries=max_queries, limit=limit, summary=summary, log=log ) - unique = dedupe(candidates, summary) + unique = dedupe(candidates, summary, metros) + + # The gate sits BEFORE the fetch loop: a company already in the vault costs + # a Places result and nothing more. Fetching its homepage again to classify + # a row that would be discarded is the expensive mistake this avoids. + if not include_known: + known_slugs, known_domains = known_companies(vault_dir) + before = len(unique) + unique = drop_known(unique, known_slugs, known_domains, summary) + if log and before != len(unique): + log(f"gate: dropped {before - len(unique)} already in the vault") + summary.kept_with_all = len(unique) for candidate in unique: diff --git a/tests/integration/test_source_gate_and_determinism.py b/tests/integration/test_source_gate_and_determinism.py new file mode 100644 index 0000000..728f511 --- /dev/null +++ b/tests/integration/test_source_gate_and_determinism.py @@ -0,0 +1,206 @@ +"""Sourcing finds only NEW companies, and finds them in a stable order. + +Two properties, both about repeat runs: + +- **The gate.** A company already in the vault has been researched, drafted and + possibly emailed. Re-finding it spends Places quota and a homepage fetch to + produce a row that gets thrown away, so it is dropped before the fetch loop. +- **Determinism.** Places can return the same businesses in a different order on + different days. The output file must depend on *what* was found, not on the + order it arrived in — otherwise a diff of two candidate CSVs is noise. + +The determinism test shuffles the API response rather than simply running twice: +a run-twice test passes today and proves nothing. +""" + +from pathlib import Path + +import httpx +import pytest +import respx + +from prospector.config import Settings +from prospector.fetch import Fetcher +from prospector.source import PLACES_URL, run_sourcing + +PIXEL_HTML = ( + "" + "Co mail" +) + + +def settings(tmp_path): + return Settings( + openrouter_key=None, openrouter_model="test/model", places_key="places-x", + hunter_key=None, vault_dir=tmp_path / "Vault" / "Outreach", + ) + + +def quick_fetcher(): + return Fetcher(client=httpx.Client(follow_redirects=True), host_interval=0.0, sleep=lambda s: None) + + +def place(pid, name, host): + return { + "id": pid, + "displayName": {"text": name}, + "websiteUri": f"https://{host}", + "formattedAddress": "1 Main St, Denver, CO 80202, USA", + } + + +PLACES = [ + place("p1", "Acme Duct Cleaning", "acme.com"), + place("p2", "Beta Vents", "beta.com"), + place("p3", "Gamma Air Care", "gamma.com"), +] + + +def stub(places): + respx.post(PLACES_URL).mock(return_value=httpx.Response(200, json={"places": places})) + for p in places: + host = p["websiteUri"].removeprefix("https://") + respx.get(f"https://{host}/").mock( + return_value=httpx.Response(200, text=PIXEL_HTML.format(host=host)) + ) + + +def note(vault_dir: Path, slug: str, company: str, website: str = "") -> None: + """A vault note as `run` would have written it.""" + vault_dir.mkdir(parents=True, exist_ok=True) + (vault_dir / f"{slug}.md").write_text( + f"---\ncompany: {company}\nemail: x@y.com\nchannel: email\nstatus: to-send\n" + f"website: {website}\ntags: [outreach]\n---\n\n## Draft\n**Subject:** S\n\nB\n\n## Log\n-\n", + encoding="utf-8", + ) + + +def source(tmp_path, out_name="c.csv", **kwargs): + out = tmp_path / out_name + summary = run_sourcing( + settings(tmp_path), keyword="duct cleaning", metros=["Denver, CO"], out=out, + keep_all=True, fetcher=quick_fetcher(), **kwargs + ) + return summary, out.read_text(encoding="utf-8") + + +class TestGateDropsAlreadyKnownCompanies: + @respx.mock + def test_company_in_the_vault_is_dropped(self, tmp_path): + stub(PLACES) + vault = tmp_path / "Vault" / "Outreach" + note(vault, "acme-duct-cleaning", "Acme Duct Cleaning", "acme.com") + + summary, csv_text = source(tmp_path, vault_dir=vault) + + assert summary.already_known == 1 + assert "Acme Duct Cleaning" not in csv_text + assert "Beta Vents" in csv_text and "Gamma Air Care" in csv_text + + @respx.mock + def test_known_company_is_never_fetched(self, tmp_path): + """The whole point of gating before the fetch loop.""" + stub(PLACES) + acme = respx.get("https://acme.com/") + vault = tmp_path / "Vault" / "Outreach" + note(vault, "acme-duct-cleaning", "Acme Duct Cleaning", "acme.com") + + source(tmp_path, vault_dir=vault) + + assert acme.call_count == 0 + + @respx.mock + def test_renamed_company_still_caught_by_domain(self, tmp_path): + """Slug misses a renamed business; the website domain catches it.""" + stub(PLACES) + vault = tmp_path / "Vault" / "Outreach" + note(vault, "acme-duct-cleaning-llc", "Acme Duct Cleaning LLC", "acme.com") + + summary, csv_text = source(tmp_path, vault_dir=vault) + + assert summary.already_known == 1 + assert "Acme Duct Cleaning" not in csv_text + + @respx.mock + def test_second_sweep_returns_nothing_new(self, tmp_path): + """Every company known -> empty result, and it is reported, not silent.""" + stub(PLACES) + vault = tmp_path / "Vault" / "Outreach" + for pid, name, host in [("p1", "Acme Duct Cleaning", "acme.com"), + ("p2", "Beta Vents", "beta.com"), + ("p3", "Gamma Air Care", "gamma.com")]: + note(vault, name.lower().replace(" ", "-"), name, host) + + summary, csv_text = source(tmp_path, vault_dir=vault) + + assert summary.already_known == 3 + assert summary.kept_with_all == 0 + assert summary.written == 0 + assert csv_text.strip() == "company,email,website,city,ad_signal" + + @respx.mock + def test_include_known_bypasses_the_gate(self, tmp_path): + stub(PLACES) + vault = tmp_path / "Vault" / "Outreach" + note(vault, "acme-duct-cleaning", "Acme Duct Cleaning", "acme.com") + + summary, csv_text = source(tmp_path, vault_dir=vault, include_known=True) + + assert summary.already_known == 0 + assert "Acme Duct Cleaning" in csv_text + + @respx.mock + def test_missing_vault_suppresses_nothing(self, tmp_path): + """A first run has no vault yet; that is not an error.""" + stub(PLACES) + summary, csv_text = source(tmp_path, vault_dir=tmp_path / "nope") + + assert summary.already_known == 0 + assert summary.kept_with_all == 3 + + @respx.mock + def test_dashboard_note_is_not_treated_as_a_company(self, tmp_path): + stub(PLACES) + vault = tmp_path / "Vault" / "Outreach" + note(vault, "acme-duct-cleaning", "Acme Duct Cleaning", "acme.com") + (vault / "_Dashboard.md").write_text("# Outreach Dashboard\n", encoding="utf-8") + + summary, _ = source(tmp_path, vault_dir=vault) + + assert summary.already_known == 1 # the dashboard is not a 4th company + + +class TestOutputIsDeterministic: + @respx.mock + def test_shuffled_places_order_yields_an_identical_file(self, tmp_path): + """The file depends on WHAT was found, not the order it arrived in.""" + stub(PLACES) + _, first = source(tmp_path, "a.csv", vault_dir=None) + respx.reset() + stub(list(reversed(PLACES))) + _, second = source(tmp_path, "b.csv", vault_dir=None) + + assert first == second + + @respx.mock + def test_rows_are_sorted_by_company(self, tmp_path): + stub(list(reversed(PLACES))) + _, csv_text = source(tmp_path, vault_dir=None) + + companies = [line.split(",")[0] for line in csv_text.strip().splitlines()[1:]] + assert companies == sorted(companies, key=str.casefold) + + @respx.mock + def test_duplicate_domain_collapses_to_the_same_winner_either_way(self, tmp_path): + """Two listings sharing a domain must not swap winners on reordering.""" + dupes = [ + place("p9", "Zed Ducts", "shared.com"), + place("p1", "Acme Ducts", "shared.com"), + ] + stub(dupes) + _, first = source(tmp_path, "a.csv", vault_dir=None) + respx.reset() + stub(list(reversed(dupes))) + _, second = source(tmp_path, "b.csv", vault_dir=None) + + assert first == second