""" check_qualify.py — run the qualification funnel end to end and audit what it wrote. This is the real integration test for phases 3-6: grounded discovery, the site scan, contact gathering, validation, scoring, and the "reject, don't persist" rule. It costs provider tokens and does real scans, so keep --max small. uv run python scripts/check_qualify.py uv run python scripts/check_qualify.py --icp golf --max 2 uv run python scripts/check_qualify.py --keep # don't delete the leads it creates By default every lead it creates is DELETED at the end (along with its contacts), so it is safe to run against the live database repeatedly. Pass --keep to inspect them in the UI afterwards. What it asserts: * discovery was grounded (the run reports a provider + model, not an error) * every persisted lead has qualification_status="qualified" * every persisted lead has a resolved website and at least one usable contact * no persisted lead has has_3d=True (those must be rejected at the fit stage) * rejected candidates were NOT written to the leads table * the rejected list explains itself (each entry has a stage and a reason) """ from __future__ import annotations import argparse import asyncio import sys 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, LeadContact, RevealRequest # noqa: E402 from app.services import qualify # noqa: E402 ICP_PRESETS = { "golf": ( "US custom golf cart and luxury utility vehicle builders doing $1M-$10M revenue " "whose websites still show static photo galleries instead of an interactive 3D " "configurator" ), "kitchens": "Luxury bespoke kitchen cabinet manufacturers in the UAE or UK without a 3D configurator", "yachts": "Custom boat and small-yacht builders in the US selling configurable vessels with static photos only", } class Suite: def __init__(self) -> None: self.failed = 0 self.ran = 0 def check(self, label: str, cond: bool, extra: str = "") -> None: self.ran += 1 self.failed += not cond print(f" [{'ok ' if cond else 'FAIL'}] {label}" + (f" — {extra}" if extra else "")) async def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--icp", default="golf", help="preset name or free text") ap.add_argument("--max", type=int, default=3) ap.add_argument("--provider", default="") ap.add_argument("--keep", action="store_true", help="keep the leads this run creates") args = ap.parse_args() icp = ICP_PRESETS.get(args.icp.strip().lower(), args.icp) print(f"ICP: {icp}") print(f"max_leads: {args.max} provider: {args.provider or 'auto'}\n") def progress(cur, total, msg): print(f" [{cur}/{total}] {msg}") outcome = await qualify.run(icp, args.max, progress=progress, provider_name=args.provider) print("\n" + "=" * 64) print(f"provider : {outcome['provider']} ({outcome['model']})") print(f"discovered : {outcome['discovered']}") print(f"qualified : {outcome['qualified']}") print(f"skipped duplicates : {outcome['skipped_duplicates']}") print(f"rejected : {len(outcome['rejected'])}") for r in outcome["rejected"]: print(f" - {r['company']:32} [{r['stage']}] {r['reason']}") if outcome["error"]: print(f"error : {outcome['error']}") s = Suite() db = SessionLocal() created_ids = list(outcome["saved_ids"]) try: print("\nassertions") s.check("discovery produced a provider + model (was grounded)", bool(outcome["provider"]) and not outcome["error"], outcome["error"] or "") s.check("every rejection names a stage and a reason", all(r.get("stage") and r.get("reason") for r in outcome["rejected"])) rejected_names = {r["company"].strip().lower() for r in outcome["rejected"]} leads = db.scalars(select(Lead).where(Lead.id.in_(created_ids))).all() if created_ids else [] s.check("saved_ids all resolve to real rows", len(leads) == len(created_ids), f"{len(leads)}/{len(created_ids)}") for lead in leads: tag = f"lead {lead.id} ({lead.company_name})" s.check(f"{tag}: qualification_status = qualified", lead.qualification_status == "qualified", lead.qualification_status) s.check(f"{tag}: has a resolved website", bool((lead.company_website or "").startswith("http")), lead.company_website) s.check(f"{tag}: has_3d is False (a configurator site must be rejected at 'fit')", lead.has_3d is False) s.check(f"{tag}: site_verdict is not 'mismatch'", lead.site_verdict != "mismatch", lead.site_verdict) contacts = db.scalars(select(LeadContact).where(LeadContact.lead_id == lead.id)).all() has_usable = any((c.email or c.phone) for c in contacts) s.check(f"{tag}: at least one contact with an email or phone", has_usable, f"{len(contacts)} contact rows") s.check(f"{tag}: every contact row carries a source", all((c.source or "").strip() for c in contacts)) s.check(f"{tag}: lead_score reflects the funnel (1..100)", 1 <= int(lead.lead_score or 0) <= 100, str(lead.lead_score)) s.check(f"{tag}: was not also in the rejected list", lead.company_name.strip().lower() not in rejected_names) # The core guarantee: a rejected candidate is not in the leads table. if rejected_names: all_company_names = { (n or "").strip().lower() for n in db.scalars(select(Lead.company_name)).all() } leaked = rejected_names & all_company_names s.check("no rejected candidate was persisted to leads", not leaked, ", ".join(sorted(leaked)) if leaked else "") finally: if created_ids and not args.keep: db.query(RevealRequest).filter(RevealRequest.lead_id.in_(created_ids)).delete( synchronize_session=False ) db.query(LeadContact).filter(LeadContact.lead_id.in_(created_ids)).delete( synchronize_session=False ) db.query(Lead).filter(Lead.id.in_(created_ids)).delete(synchronize_session=False) db.commit() print(f"\ncleaned up {len(created_ids)} lead(s): {created_ids}") elif created_ids: print(f"\nkept {len(created_ids)} lead(s): {created_ids}") db.close() print() if s.failed: print(f"FAILED — {s.failed} of {s.ran} assertions") return 1 print(f"PASS — {s.ran} assertions, {outcome['qualified']} qualified, " f"{len(outcome['rejected'])} rejected, none leaked into the table.") return 0 if __name__ == "__main__": sys.exit(asyncio.run(main()))