Spaces:
Running
Running
| """ | |
| check_lead_quality.py — guards the anti-fabrication rules. | |
| Every check here exists because the code once did the opposite and shipped | |
| invented contact data to the CRM. They are pure-logic assertions: no network, no | |
| database, no API keys, so this runs anywhere and is safe to run repeatedly. | |
| uv run python scripts/check_lead_quality.py | |
| Exit code 0 means every guard holds. Non-zero prints which one broke. | |
| The functions under test are extracted with `ast` and exec'd in isolation rather | |
| than imported, so this works without google-genai / requests / ddgs installed and | |
| covers BOTH copies — the root Streamlit module and the backend port, which must | |
| not drift apart. | |
| """ | |
| from __future__ import annotations | |
| import ast | |
| import io | |
| import re | |
| import sys | |
| from pathlib import Path | |
| REPO_ROOT = Path(__file__).resolve().parent.parent.parent | |
| AGENT_CORE = [ | |
| REPO_ROOT / "agent_core.py", | |
| REPO_ROOT / "outreach-backend" / "app" / "services" / "agent_core.py", | |
| ] | |
| LEAD_ENGINE = [ | |
| REPO_ROOT / "lead_engine.py", | |
| REPO_ROOT / "outreach-backend" / "app" / "services" / "lead_engine.py", | |
| ] | |
| WANT_FN = {"_distinctive_tokens", "_is_non_personal", "site_matches_lead"} | |
| WANT_CONST = {"_NON_PERSONAL", "_COMPANY_STOPWORDS", "INDUSTRY_KEYWORDS"} | |
| def _extract(path: Path) -> dict: | |
| """Pull the pure helpers out of a module without importing its dependencies.""" | |
| tree = ast.parse(io.open(path, encoding="utf-8").read()) | |
| keep: list = [] | |
| for node in tree.body: | |
| if isinstance(node, ast.FunctionDef) and node.name in WANT_FN: | |
| keep.append(node) | |
| elif isinstance(node, ast.Assign): | |
| if any(isinstance(t, ast.Name) and t.id in WANT_CONST for t in node.targets): | |
| keep.append(node) | |
| ns: dict = {"re": re} | |
| mod = ast.Module(body=keep, type_ignores=[]) | |
| exec(compile(ast.fix_missing_locations(mod), str(path), "exec"), ns) # noqa: S102 | |
| return ns | |
| class Checker: | |
| def __init__(self) -> None: | |
| self.failures = 0 | |
| def check(self, label: str, got, expect) -> None: | |
| ok = got == expect | |
| if not ok: | |
| self.failures += 1 | |
| print(f" [{'ok ' if ok else 'FAIL'}] {label}: {got!r}" + ("" if ok else f" (expected {expect!r})")) | |
| def absent(self, label: str, path: Path, needle: str) -> None: | |
| # Comments are allowed to name the old bug; code is not. | |
| hits = [ | |
| f"{path.name}:{i}" | |
| for i, line in enumerate(io.open(path, encoding="utf-8"), 1) | |
| if needle in line and not line.lstrip().startswith("#") | |
| ] | |
| ok = not hits | |
| if not ok: | |
| self.failures += 1 | |
| print(f" [{'ok ' if ok else 'FAIL'}] {label}" + ("" if ok else f" -> found at {', '.join(hits)}")) | |
| def main() -> int: | |
| c = Checker() | |
| for path in AGENT_CORE: | |
| print(f"\n===== {path.relative_to(REPO_ROOT)} =====") | |
| ns = _extract(path) | |
| nonpersonal = ns["_is_non_personal"] | |
| verdict = ns["site_matches_lead"] | |
| # --- placeholder contacts must not reach a metered API ------------------ | |
| # The old test was `name.lower() in _NON_PERSONAL`, an equality check, so | |
| # only a bare "Team" was caught. _NON_PERSONAL also carries a trailing "" | |
| # entry, which a naive substring rewrite turns into "matches everything". | |
| for name in ("Team", "Custom Sales Team", "Sales Desk", "Decision Maker", "unknown", ""): | |
| c.check(f"_is_non_personal({name!r})", nonpersonal(name), True) | |
| for name in ("Hal Garrett", "Jonathan Ward", "Maria Lopez"): | |
| c.check(f"_is_non_personal({name!r})", nonpersonal(name), False) | |
| # --- every standard category needs INDUSTRY_KEYWORDS ------------------- | |
| # A category with no entry got [] from .get() and fell straight through | |
| # to "mismatch" whenever the company's name tokens missed. | |
| missing = [k for k in ns["INDUSTRY_KEYWORDS"] if not ns["INDUSTRY_KEYWORDS"][k]] | |
| c.check("no empty INDUSTRY_KEYWORDS entry", missing, []) | |
| c.check( | |
| "tech category can match", | |
| verdict( | |
| "Nordic Widget Labs", | |
| "⚡ Tech & Commercial Products", | |
| "We are a custom product manufacturer building bespoke industrial equipment.", | |
| )[0], | |
| "match", | |
| ) | |
| # --- the two verdict invariants ---------------------------------------- | |
| c.check( | |
| "blocked beats mismatch", | |
| verdict("Acme", "⛳ Custom Golf Carts", "totally unrelated text here ok", blocked=True)[0], | |
| "blocked", | |
| ) | |
| c.check( | |
| "real mismatch still caught", | |
| verdict( | |
| "Papilio Bespoke Kitchens", | |
| "\U0001f6cb️ Luxury Furniture & Interiors", | |
| "Read the latest novels and browse our paperback catalogue today.", | |
| )[0], | |
| "mismatch", | |
| ) | |
| # --- no invented email, no manufactured URL ---------------------------- | |
| c.absent("no info@{domain} fabrication", path, 'f"info@{domain}"') | |
| c.absent("no manufactured www..com fallback", path, 'f"https://www.{extract_domain') | |
| c.absent("no hardcoded verified_direct on an empty phone", path, '"phone_status": "verified_direct"') | |
| c.absent("no hardcoded verified_direct kwarg", path, 'phone_status="verified_direct",') | |
| for path in LEAD_ENGINE: | |
| print(f"\n===== {path.relative_to(REPO_ROOT)} =====") | |
| c.absent("no Apollo contact@ fabrication", path, "contact@{org.get") | |
| print() | |
| if c.failures: | |
| print(f"FAILED — {c.failures} guard(s) broken") | |
| return 1 | |
| print("All lead-quality guards hold.") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |