"""Re-enrich existing Baserow leads with missing data via Firecrawl + Claude. Fetches leads missing director/email/tech, re-scrapes their sites, extracts structured data with Claude, and PATCHes Baserow. Usage: uv run python scripts/re_enrich_leads.py # all missing uv run python scripts/re_enrich_leads.py --limit 10 # first 10 uv run python scripts/re_enrich_leads.py --dry-run # preview only """ from __future__ import annotations import argparse import asyncio import httpx import structlog from src.config import get_settings from src.discovery import _get_firecrawl_client from src.enrichment import ( _build_rate_limiter, _claude_extract, _email_hunt_firecrawl, _extract_emails_regex, _is_valid_email, _pick_best_email, _scrape_multi_pages, ) log = structlog.get_logger() def _needs_enrichment(lead: dict) -> bool: """True if the lead is missing director OR email.""" has_director = bool(lead.get("Prenom")) and bool(lead.get("Nom")) has_email = bool(lead.get("Email Contact")) return not has_director or not has_email def _build_patch(lead: dict, extracted: dict, regex_email: str) -> dict: """Build PATCH payload with only newly-found non-empty fields.""" patch: dict[str, str] = {} # Director — only fill if currently empty if not lead.get("Prenom") and extracted.get("director_first_name"): patch["Prenom"] = extracted["director_first_name"] if not lead.get("Nom") and extracted.get("director_last_name"): patch["Nom"] = extracted["director_last_name"] # Email — only fill if currently empty, validate format if not lead.get("Email Contact"): email = extracted.get("contact_email", "") or regex_email if email and _is_valid_email(email): patch["Email Contact"] = email # Tech stack — only fill if currently empty if not lead.get("Tech Stack") and extracted.get("technology"): patch["Tech Stack"] = extracted["technology"] # Company name — only fill if currently generic (same as title) if extracted.get("company_name") and lead.get("Entreprise") != extracted["company_name"]: # Only update if it looks more specific (e.g. "SARL X" vs "X: Le site") current = lead.get("Entreprise", "") new = extracted["company_name"] if len(new) > 3 and (not current or len(current) > len(new) * 2): patch["Entreprise"] = new return patch async def _fetch_all_leads() -> list[dict]: """Fetch all leads from Baserow with pagination.""" settings = get_settings() all_leads: list[dict] = [] cursor = None async with httpx.AsyncClient() as client: while True: params: dict[str, str] = {"user_field_names": "true", "size": "200"} if cursor: params["cursor"] = cursor resp = await client.get( f"{settings.baserow_base_url}/api/database/rows/table/{settings.baserow_table_id}/", params=params, headers={"Authorization": f"Token {settings.baserow_token}"}, timeout=30.0, ) resp.raise_for_status() data = resp.json() all_leads.extend(data["results"]) next_url = data.get("next") if not next_url: break from urllib.parse import parse_qs, urlparse qs = parse_qs(urlparse(next_url).query) cursor = qs.get("cursor", [None])[0] if not cursor: break return all_leads async def re_enrich(limit: int | None = None, dry_run: bool = False) -> None: settings = get_settings() fc = _get_firecrawl_client() claude_limiter = _build_rate_limiter("anthropic") # 1. Fetch leads all_leads = await _fetch_all_leads() to_enrich = [l for l in all_leads if _needs_enrichment(l) and l.get("Domaine")] if limit: to_enrich = to_enrich[:limit] total = len(all_leads) target = len(to_enrich) print(f"Total leads: {total}, à re-enrichir: {target}") if dry_run: print("(mode dry-run — aucune mise à jour Baserow)\n") updated = 0 skipped = 0 errors = 0 async with httpx.AsyncClient() as http_client: for i, lead in enumerate(to_enrich, 1): domain = lead["Domaine"] row_id = lead["id"] name = lead.get("Entreprise", domain) url = lead.get("Website", f"https://{domain}") print(f"\n[{i}/{target}] {name} — {domain}") # Scrape try: markdown = await _scrape_multi_pages(fc, url) except Exception as exc: print(f" X Scrape failed: {exc}") errors += 1 continue if not markdown or len(markdown) < 200: print(f" X Contenu trop court ({len(markdown)} chars)") skipped += 1 continue # Claude extraction try: extracted = await _claude_extract(markdown, claude_limiter) except Exception as exc: print(f" X Claude extract failed: {exc}") errors += 1 continue # Email regex fallback regex_emails = _extract_emails_regex(markdown, domain) regex_email = _pick_best_email(regex_emails, domain) if regex_emails else "" # Email hunt fallback (Firecrawl search "@domain") if not extracted.get("contact_email") and not regex_email and not lead.get("Email Contact"): hunt_emails = await _email_hunt_firecrawl(fc, domain) if hunt_emails: regex_email = _pick_best_email(hunt_emails, domain) # Build patch patch = _build_patch(lead, extracted, regex_email) if not patch: print(f" - Rien de nouveau trouvé") skipped += 1 continue # Show what we found for k, v in patch.items(): print(f" + {k}: {v}") if dry_run: updated += 1 continue # PATCH Baserow resp = await http_client.patch( f"{settings.baserow_base_url}/api/database/rows/table/{settings.baserow_table_id}/{row_id}/", params={"user_field_names": "true"}, headers={"Authorization": f"Token {settings.baserow_token}"}, json=patch, timeout=15.0, ) if resp.status_code == 200: print(f" > Baserow OK") updated += 1 else: print(f" ! Baserow error: {resp.status_code}") errors += 1 await asyncio.sleep(0.3) print(f"\n{'='*50}") print(f"Résultat: {updated} mis à jour / {target} traités ({skipped} sans nouveauté, {errors} erreurs)") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Re-enrich leads with missing data") parser.add_argument("--limit", type=int, default=None, help="Max leads to process") parser.add_argument("--dry-run", action="store_true", help="Preview without updating Baserow") args = parser.parse_args() asyncio.run(re_enrich(limit=args.limit, dry_run=args.dry_run))