Spaces:
Running
Running
| """ | |
| check_duplicate_discovered_leads.py -- read-only report on duplicate/near-duplicate leads from AI | |
| Discovery and Google Maps sourcing. | |
| User report (2026-10-06): "AI or Google Maps wali leads jo add ki hain, woh repeated toh nahi hain -- | |
| pehle bhi repeat ho rahi thin" (have these been repeating again, same as before). | |
| Background: crud.is_duplicate() -- called by both run_discover() (AI Discovery) and qualify.run_candidates() | |
| (Google Maps + qualified AI discovery) before every save -- only ever compares company_name, case- | |
| insensitive and trimmed, nothing else. That catches an EXACT repeat of the same name string, but AI/Maps | |
| sourcing can name the same real company slightly differently between runs ("Murvi Motorhomes" vs "Murvi | |
| Motorhomes Ltd", a stray "&"/"and", extra whitespace, ...) -- an exact-string check misses all of those, | |
| even though the underlying company (same website) is identical. | |
| This script reports BOTH kinds, read-only, no writes at all: | |
| 1. Exact-name duplicates (case-insensitive) -- if any exist, the existing is_duplicate() check itself is | |
| somehow not catching what it's designed to catch (a real bug in that function or its wiring). | |
| 2. Same-domain duplicates -- different company_name strings that share the same normalized website | |
| domain (agent_core.extract_domain() -- strips scheme/path/"www."). These are the near-duplicates the | |
| current exact-name check structurally cannot catch; they're the likely source of "repeat" leads a | |
| rep would recognize as the same company sourced twice. | |
| uv run python scripts/check_duplicate_discovered_leads.py | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from collections import defaultdict | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) | |
| from sqlalchemy import select # noqa: E402 | |
| from app.database import SessionLocal # noqa: E402 | |
| from app.models import Lead # noqa: E402 | |
| from app.services.agent_core import extract_domain # noqa: E402 | |
| def main() -> int: | |
| db = SessionLocal() | |
| try: | |
| leads = list(db.scalars(select(Lead).order_by(Lead.created_at)).all()) | |
| print(f"Checked {len(leads)} total leads in the CRM.\n") | |
| # 1. Exact-name duplicates -- what crud.is_duplicate() is SUPPOSED to prevent. | |
| by_name: dict[str, list[Lead]] = defaultdict(list) | |
| for lead in leads: | |
| key = (lead.company_name or "").strip().lower() | |
| if key: | |
| by_name[key].append(lead) | |
| name_dupes = {k: v for k, v in by_name.items() if len(v) > 1} | |
| print("=" * 90) | |
| print(f"1. EXACT company_name duplicates (case-insensitive): {len(name_dupes)} name(s) repeated") | |
| print("=" * 90) | |
| if not name_dupes: | |
| print(" none -- the existing is_duplicate() check is doing its one job correctly.") | |
| for key, rows in name_dupes.items(): | |
| print(f" \"{rows[0].company_name}\" -- {len(rows)} rows: " + ", ".join(f"#{r.id} ({r.source_prompt[:30] or 'n/a'}, {r.created_at:%Y-%m-%d})" for r in rows)) | |
| # 2. Same-domain near-duplicates -- the gap is_duplicate() can't see (different name text, same | |
| # real company website). Only leads with a usable website are comparable this way. | |
| by_domain: dict[str, list[Lead]] = defaultdict(list) | |
| for lead in leads: | |
| domain = extract_domain(lead.company_website or "") | |
| if domain: | |
| by_domain[domain].append(lead) | |
| domain_dupes = {d: v for d, v in by_domain.items() if len(v) > 1} | |
| print() | |
| print("=" * 90) | |
| print(f"2. Same-website near-duplicates (different company_name, same domain): {len(domain_dupes)} domain(s)") | |
| print("=" * 90) | |
| if not domain_dupes: | |
| print(" none found.") | |
| for domain, rows in domain_dupes.items(): | |
| names = {r.company_name for r in rows} | |
| if len(names) == 1: | |
| continue # already counted above as an exact-name duplicate, not a near-duplicate | |
| print(f" {domain} -- {len(rows)} rows, {len(names)} different name(s):") | |
| for r in rows: | |
| print(f" #{r.id:<6} \"{r.company_name}\" created_at={r.created_at:%Y-%m-%d %H:%M} source={r.source_prompt[:50] or 'n/a'}") | |
| return 0 | |
| finally: | |
| db.close() | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |