Spaces:
Running
Running
| """ | |
| 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())) | |