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