# Lead Pipeline — Spécification Complète > Document de référence pour reconstruire le pipeline de génération de leads en Python local. > Couvre : architecture actuelle (OpenClaw), logique métier, email finder, CRM, outreach. --- ## 1. Vue d'ensemble ### Flow global ``` ┌─────────────────────────────────────────────────────────────────────┐ │ LEAD PIPELINE │ │ │ │ ┌──────────┐ ┌───────────┐ ┌──────────┐ ┌─────────────┐ │ │ │ Discovery │───▶│ Enrichment│───▶│ Email │───▶│ Outreach │ │ │ │ (search) │ │ (scrape) │ │ (finder) │ │ (sequence) │ │ │ └──────────┘ └───────────┘ └──────────┘ └─────────────┘ │ │ │ │ │ │ │ │ Firecrawl Firecrawl SMTP/DNS Himalaya │ │ Search API Scrape API Raw UDP SMTP │ │ │ │ │ │ │ │ ▼ ▼ ▼ ▼ │ │ ┌──────────────────────────────────────────────────────────────┐ │ │ │ EspoCRM (source de vérité) │ │ │ │ New → Researching → Contacted → Follow-up → Replied → ... │ │ │ └──────────────────────────────────────────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────┐ │ │ │ SQLite │ ← anti-doublon local + historique + logs │ │ │ (local) │ │ │ └──────────┘ │ └─────────────────────────────────────────────────────────────────────┘ ``` ### Statuts pipeline (EspoCRM) | Statut | Signification | Transition | |--------|---------------|------------| | **New** | Lead identifié, pas encore qualifié | → Researching (pas d'email) ou Contacted (email valide) | | **Researching** | Email personnel manquant | → Contacted (email trouvé par finder) | | **Contacted** | Premier email envoyé (J+0) | → Follow-up (après J+3) | | **Follow-up** | Relances en cours (J+3/7/14) | → Replied ou Recycled | | **Replied** | Le prospect a répondu | → Converted ou Recycled | | **Converted** | Client signé | FIN | | **Recycled** | Pas intéressé, recontacter dans 90j | → New (après cooldown) | | **Dead** | Refus définitif ou entreprise fermée | FIN | --- ## 2. Architecture actuelle (OpenClaw + Docker) ### Composants serveur (sbproduction) ``` /data/compose/openclaw/build/ ├── docker-compose.yml ├── docker-compose.override.yml # Env vars, volumes, entrypoint ├── .env # Secrets (ANTHROPIC_OAUTH_TOKEN, GEMINI_API_KEY, etc.) └── sync-auth.sh # Injecte ANTHROPIC_OAUTH_TOKEN → auth-profiles.json /data/openclaw-data/ ├── config/ │ ├── openclaw.json # Config agent (sandbox, models, fallbacks) │ ├── cron/jobs.json # 7 cron jobs (Lead Pipeline, Morning Briefing, etc.) │ ├── agents/main/agent/ │ │ └── auth-profiles.json # Tokens Anthropic/Google │ └── .config/himalaya/ │ └── config.toml # IMAP/SMTP o2switch ├── workspace/ │ ├── TOOLS.md # Instructions pour l'agent │ ├── lead-hunter/ │ │ ├── icp.yaml # Profil client idéal │ │ ├── README.md │ │ └── outreach/ │ │ └── sequence-config.yaml # 4 touches cold email │ └── memory/ # Mémoire agent ├── skills-internal/ │ └── espocrm/ │ ├── SKILL.md # Doc skill lue par l'agent │ └── scripts/ │ ├── espocrm.py # CLI CRM (search, create, update, delete, list, link) │ └── email_finder.py # Permutation + SMTP (ping, verify, find, lead, batch) └── tools/ └── himalaya # Binaire email CLI (x86_64) ``` ### Crons actifs | Cron | Schedule | sessionTarget | Action | |------|----------|---------------|--------| | **Lead Pipeline Night** | `*/30 22-23,0-7 * * *` (Europe/Paris) | `last` | 10 leads/run via Firecrawl → EspoCRM | | **Morning Briefing** | `0 9 * * *` | `last` | Recap leads nocturnes + Plane + emails | | **Cold Outreach** | `0 16 * * 1-5` | `last` | Envoi séquence emails (lun-ven) | | **Weekly Report** | `0 10 * * 1` | `last` | Recap hebdo pipeline + Plane | ### Services externes | Service | URL | Auth | Usage | |---------|-----|------|-------| | **EspoCRM** | `https://espo.elevaseo.app` | API Key (`X-Api-Key`) | CRM source de vérité | | **Firecrawl** | `https://crawl.elevaseo.app` | API Key (`Authorization: Bearer`) | Scrape + search web | | **Himalaya** | o2switch `arome.o2switch.net` | IMAP:993 / SMTP:465 TLS | Envoi cold emails | ### Problèmes connus de l'implémentation actuelle | Problème | Impact | |----------|--------| | Bot LLM comme orchestrateur | Lent (~5-10 min/lead), non déterministe, coûteux en tokens | | Pas de DB locale anti-doublon | Repose sur EspoCRM search (lent, rate-limited) | | Pas de logs structurés | Debug difficile, pas de métriques | | Synchrone (1 lead à la fois) | 10 leads/30 min max | | Tokens Anthropic révoqués | Pipeline bloqué la nuit | | Firecrawl dans sandbox | Accès indirect via curl, pas de retry | | Niche hardcodée dans le prompt | Pas de rotation niches/régions | --- ## 3. Email Finder — Logique complète ### 3.1 Normalisation des noms ```python import unicodedata def normalize_name(name: str) -> str: """Normalise un nom pour la génération de permutations email. - NFD decomposition : é → e, è → e, ç → c, ü → u - Espaces → tirets : "Munoz Ruiz" → "munoz-ruiz" - Tout en minuscules """ nfkd = unicodedata.normalize("NFKD", name) ascii_name = "".join(c for c in nfkd if not unicodedata.combining(c)) ascii_name = ascii_name.replace(" ", "-") return ascii_name.lower().strip() ``` ### 3.2 Permutations email (24-29 patterns) Pour `first="Jean-Pierre"`, `last="Dupont"`, `domain="example.com"` : ``` # Patterns standard (24) jean-pierre@example.com # prenom@ dupont@example.com # nom@ jean-pierre.dupont@example.com # prenom.nom@ dupont.jean-pierre@example.com # nom.prenom@ jean-pierredupont@example.com # prenomnom@ dupontjean-pierre@example.com # nomprenom@ jdupont@example.com # initial+nom@ j.dupont@example.com # initial.nom@ j-dupont@example.com # initial-nom@ j_dupont@example.com # initial_nom@ jean-pierred@example.com # prenom+initial@ jean-pierre.d@example.com # prenom.initial@ dupontj@example.com # nom+initial@ dupont.j@example.com # nom.initial@ djean-pierre@example.com # initial_nom+prenom@ d.jean-pierre@example.com # initial_nom.prenom@ jean-pierre-dupont@example.com # prenom-nom@ dupont-jean-pierre@example.com # nom-prenom@ jean-pierre_dupont@example.com # prenom_nom@ dupont_jean-pierre@example.com # nom_prenom@ jd@example.com # initiales@ j.d@example.com # initiales.@ jddupont@example.com # initiales+nom@ j.d.dupont@example.com # initiales.nom@ # Patterns composés (5 supplémentaires si prénom contient "-") jeanpierre@example.com # prenom sans tiret@ jeanpierredupont@example.com # prenom_sans_tiret+nom@ jeanpierre.dupont@example.com # prenom_sans_tiret.nom@ jpdupont@example.com # initiales_composé+nom@ jp.dupont@example.com # initiales_composé.nom@ ``` ### 3.3 Résolution MX (DNS UDP brut, 0 dépendance) ```python def get_mx_records(domain: str) -> list[str]: """Résout les MX records via DNS UDP. 1. Construit un paquet DNS query (type MX = 15) 2. Envoie en UDP sur port 53 (8.8.8.8 puis 1.1.1.1) 3. Parse la réponse avec support compression DNS 4. Trie par priorité (préférence basse = prioritaire) 5. Fallback : domaine lui-même si aucun MX """ ``` **Format paquet DNS :** - Header : 12 bytes (ID, flags, counts) - Question : nom encodé (longueur-prefixed labels) + type MX (15) + class IN (1) - Answer : name + type + class + TTL + rdlength + preference (uint16) + mx_name ### 3.4 Vérification SMTP (RCPT TO) ```python def verify_email_smtp(email: str) -> str: """Vérifie un email via SMTP RCPT TO. 1. Résout MX records du domaine 2. Se connecte au MX primaire (port 25, timeout 10s) 3. HELO elevaseo.com 4. MAIL FROM: verify@elevaseo.com 5. RCPT TO: → code 250 = accepté 6. RCPT TO: → test catch-all 7. QUIT Returns: "valid" — 250 + random rejeté (email existe vraiment) "invalid" — 550/551/552/553 (rejeté par le serveur) "catch_all" — 250 + random aussi accepté (domaine catch-all) "error" — timeout, connexion refusée, etc. """ ``` **Codes SMTP importants :** - `250` : OK, mailbox exists - `550` : Mailbox not found - `551` : User not local - `552` : Exceeded storage - `553` : Mailbox name not allowed ### 3.5 Détection emails génériques Préfixes blacklistés (non utilisables pour cold outreach) : ```python GENERIC_PREFIXES = [ "contact", "info", "hello", "bonjour", "hey", "sav", "support", "admin", "webmaster", "commercial", "direction", "accueil", "noreply", "no-reply", "postmaster", ] ``` --- ## 4. CRM — API EspoCRM ### Endpoints utilisés | Méthode | Endpoint | Usage | |---------|----------|-------| | `GET` | `/api/v1/Lead?searchParams=...` | Recherche/liste leads | | `GET` | `/api/v1/Lead/{id}` | Lecture lead | | `POST` | `/api/v1/Lead` | Création lead | | `PUT` | `/api/v1/Lead/{id}` | Mise à jour lead | | `DELETE` | `/api/v1/Lead/{id}` | Suppression lead | ### Headers ```http X-Api-Key: {api_key} Content-Type: application/json User-Agent: OpenClaw/1.0 ``` ### Champs Lead | Champ | Type | Description | |-------|------|-------------| | `id` | string | UUID auto-généré | | `firstName` | string | Prénom du contact | | `lastName` | string | Nom du contact | | `accountName` | string | Nom de l'entreprise | | `emailAddress` | string | Email principal | | `website` | string | URL du site web | | `status` | enum | New, Researching, Contacted, Follow-up, Replied, Converted, Recycled, Dead | | `description` | text | Notes libres | | `cEmailInvalide` | bool | Custom : email générique ou bounced | | `cEmailNotes` | varchar | Custom : notes validation email | | `cNiche` | enum | Custom : 15 niches possibles | | `createdAt` | datetime | Date de création | | `assignedUserId` | string | User assigné | ### searchParams (filtrage) ```json { "where": [ {"type": "equals", "attribute": "status", "value": "Researching"}, {"type": "contains", "attribute": "accountName", "value": "Magiline"} ], "maxSize": 50, "offset": 0, "orderBy": "createdAt", "order": "desc", "select": ["id", "firstName", "lastName", "emailAddress", "status"] } ``` --- ## 5. Discovery — Firecrawl API ### 5.1 Search (trouver des prospects) ```bash POST https://crawl.elevaseo.app/v1/search Authorization: Bearer {api_key} Content-Type: application/json { "query": "constructeur piscine france site officiel", "limit": 10 } ``` **Réponse :** ```json { "data": [ { "url": "https://www.magiline.com", "title": "Magiline — Constructeur de piscines", "description": "..." } ] } ``` ### 5.2 Scrape (enrichissement) ```bash POST https://crawl.elevaseo.app/v1/scrape Authorization: Bearer {api_key} Content-Type: application/json { "url": "https://www.magiline.com/equipe", "formats": ["markdown"] } ``` **Usage :** Extraire le nom du dirigeant, les pages contact, l'équipe, la techno du site. ### 5.3 Stratégies de recherche par niche ```yaml queries: piscines: - "constructeur piscine france site officiel" - "pisciniste {region} contact dirigeant" - "fabricant piscine coque polyester france" salles_de_sport: - "salle de sport fitness france site web" - "club fitness {region} dirigeant fondateur" - "salle crossfit {region}" e_commerce_cosmetiques: - "marque cosmétique bio france e-commerce" - "boutique en ligne beauté naturelle france" regions: - Île-de-France - Auvergne-Rhône-Alpes - Provence-Alpes-Côte d'Azur - Nouvelle-Aquitaine - Occitanie - Bretagne - Hauts-de-France - Grand Est - Pays de la Loire - Normandie - Centre-Val de Loire - Bourgogne-Franche-Comté - Corse ``` --- ## 6. ICP — Scoring ### Critères (score /100, seuil ≥ 60) | Critère | Points | Comment évaluer | |---------|--------|----------------| | Site web actif avec formulaire | 20 | Firecrawl scrape → chercher `
`, page contact | | Pas de SEO technique | 20 | Lighthouse score < 70, pas de schema markup | | Dirigeant identifiable | 20 | Scrape page équipe/à-propos + search LinkedIn | | CA estimé 1M-20M€ | 15 | Societe.com, nombre employés, taille du site | | Blog absent ou inactif | 10 | Pas de /blog ou dernier article > 6 mois | | Problèmes techniques visibles | 15 | Core Web Vitals rouges, LCP > 4s, pas de HTTPS | ### Niches EspoCRM (champ `cNiche`) ``` e-commerce, retail, services-b2b, saas, industrie, piscines, fitness, cosmetiques, immobilier, restauration, tourisme, formation, sante, juridique, artisanat ``` --- ## 7. Outreach — Séquence Cold Email ### Config sender ```yaml sender: name: "Sebastien Dacosta" email: "sebastien@sbinnovate.net" title: "Co-fondateur, SBInnovate" smtp: arome.o2switch.net:465 (TLS) imap: arome.o2switch.net:993 (TLS) ``` ### Séquence 4 touches | Étape | Délai | Sujet type | Contenu | |-------|-------|------------|---------| | **J+0** | Immédiat | "Question rapide sur {website}" | Audit technique personnalisé (3 problèmes SEO trouvés) | | **J+3** | 3 jours | "Les scores de {company} sur Google" | Mini-rapport Lighthouse (LCP, CLS, ressources bloquantes) | | **J+7** | 7 jours | "+{X}% de trafic pour {case_study}" | Cas concret chiffré d'un client similaire | | **J+14** | 14 jours | "Mes notes sur {company}" | Dernier message, pas d'insistance | ### Règles de rédaction (tone_rules) - Vouvoiement naturel (pas guindé) - Phrases courtes, comme à l'oral - Jamais de tirets longs (—), pas de listes à puces - Maximum 8-10 lignes de corps - Sujet court et mystérieux (pas de résumé) - P.S. technique et utile (pas un argument de vente) - **Pas de mots AI** : significativement, véritablement, fondamentalement ### Tracking ```yaml pause_si: bounce_rate > 5% reply_rate < 5% review_apres: 20 emails envoyés cooldown: 90 jours avant recontact ``` --- ## 8. Nouvelle Architecture Proposée (Python Local) ### Stack technique ``` Python 3.12+ ├── asyncio + aiohttp # Async HTTP (Firecrawl, EspoCRM) ├── aiosmtplib # Async SMTP verification ├── aiodns # Async DNS MX lookup ├── sqlite3 (aiosqlite) # DB locale anti-doublon + historique ├── structlog # Logs structurés JSON ├── rich # CLI output (tables, progress bars) ├── pydantic # Validation config + data models ├── typer # CLI framework └── uv # Package manager ``` ### Structure projet ``` lead-pipeline/ ├── pyproject.toml ├── config/ │ ├── niches.yaml # Définition niches + queries de recherche │ ├── regions.yaml # Régions France │ ├── icp.yaml # Scoring ICP (poids, seuil) │ ├── outreach.yaml # Séquence cold email (templates, rules) │ └── settings.yaml # API keys refs, sender config ├── src/ │ ├── __init__.py │ ├── cli.py # Point d'entrée CLI (typer) │ ├── config.py # Chargement config (pydantic) │ ├── db/ │ │ ├── __init__.py │ │ ├── models.py # Modèles SQLAlchemy/dataclass │ │ ├── repository.py # CRUD leads, anti-doublon │ │ └── migrations.py # Schema init │ ├── discovery/ │ │ ├── __init__.py │ │ ├── firecrawl.py # Client Firecrawl async │ │ ├── enrichment.py # Scrape page équipe, contact, tech stack │ │ └── scorer.py # ICP scoring │ ├── email/ │ │ ├── __init__.py │ │ ├── dns.py # MX lookup (aiodns ou raw UDP) │ │ ├── permutations.py # Génération 24-29 patterns │ │ ├── smtp_verifier.py # RCPT TO async │ │ └── finder.py # Orchestrateur (permutations → verify → résultat) │ ├── crm/ │ │ ├── __init__.py │ │ └── espocrm.py # Client REST async │ ├── outreach/ │ │ ├── __init__.py │ │ ├── templating.py # Rendu templates (Jinja2) │ │ ├── sender.py # Envoi SMTP (himalaya ou aiosmtplib) │ │ └── sequencer.py # Logique J+0/3/7/14 │ ├── pipeline/ │ │ ├── __init__.py │ │ ├── runner.py # Orchestrateur principal async │ │ ├── steps.py # Chaque étape du pipeline │ │ └── scheduler.py # Cron-like scheduler (APScheduler) │ └── utils/ │ ├── __init__.py │ └── logging.py # Config structlog ├── tests/ │ ├── test_permutations.py │ ├── test_smtp.py │ ├── test_scorer.py │ └── test_pipeline.py └── data/ └── leads.db # SQLite (auto-créée) ``` ### Schéma SQLite ```sql CREATE TABLE leads ( id TEXT PRIMARY KEY, -- UUID local espocrm_id TEXT UNIQUE, -- ID EspoCRM (sync) first_name TEXT NOT NULL, last_name TEXT NOT NULL, account_name TEXT NOT NULL, domain TEXT NOT NULL, -- Domaine extrait du website website TEXT, email TEXT, email_status TEXT DEFAULT 'unknown', -- valid, invalid, catch_all, generic, unknown email_notes TEXT, status TEXT DEFAULT 'new', -- new, researching, contacted, follow_up, replied, converted, recycled, dead niche TEXT, region TEXT, icp_score INTEGER DEFAULT 0, source TEXT, -- firecrawl_search, manual, import description TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, contacted_at TIMESTAMP, -- Date du premier email last_outreach_step INTEGER DEFAULT 0, -- 0=pas contacté, 1-4=étape séquence cooldown_until TIMESTAMP, -- Date avant laquelle ne pas recontacter UNIQUE(domain, first_name, last_name) -- Anti-doublon ); CREATE TABLE email_verifications ( id INTEGER PRIMARY KEY AUTOINCREMENT, lead_id TEXT REFERENCES leads(id), email TEXT NOT NULL, result TEXT NOT NULL, -- valid, invalid, catch_all, error mx_host TEXT, smtp_code INTEGER, response_time_ms INTEGER, verified_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE outreach_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, lead_id TEXT REFERENCES leads(id), step INTEGER NOT NULL, -- 1-4 subject TEXT, sent_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, bounced BOOLEAN DEFAULT FALSE, opened BOOLEAN DEFAULT FALSE, replied BOOLEAN DEFAULT FALSE ); CREATE TABLE pipeline_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, niche TEXT, region TEXT, started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, finished_at TIMESTAMP, leads_found INTEGER DEFAULT 0, leads_created INTEGER DEFAULT 0, emails_found INTEGER DEFAULT 0, emails_verified INTEGER DEFAULT 0, errors TEXT, -- JSON array d'erreurs duration_ms INTEGER ); -- Index pour les requêtes fréquentes CREATE INDEX idx_leads_status ON leads(status); CREATE INDEX idx_leads_domain ON leads(domain); CREATE INDEX idx_leads_niche ON leads(niche); CREATE INDEX idx_leads_espocrm ON leads(espocrm_id); CREATE INDEX idx_verifications_lead ON email_verifications(lead_id); CREATE INDEX idx_outreach_lead ON outreach_logs(lead_id); ``` ### Pipeline Runner (pseudo-code async) ```python async def run_pipeline(niche: str, region: str, batch_size: int = 15): """Pipeline principal — une exécution complète.""" run = PipelineRun(niche=niche, region=region) log.info("pipeline.start", niche=niche, region=region, batch_size=batch_size) # 1. DISCOVERY — recherche prospects via Firecrawl queries = build_queries(niche, region) # depuis niches.yaml prospects = [] for query in queries: results = await firecrawl.search(query, limit=batch_size * 2) prospects.extend(results) if len(prospects) >= batch_size * 3: break # 2. DEDUP — filtrer les doublons (DB locale + EspoCRM) new_prospects = [] for p in prospects: domain = extract_domain(p.url) if not domain: continue if await db.exists(domain=domain): log.debug("pipeline.skip_duplicate", domain=domain, source="local_db") continue if await espocrm.search_lead(domain): log.debug("pipeline.skip_duplicate", domain=domain, source="espocrm") await db.insert_known_domain(domain) # cache local continue new_prospects.append(p) # 3. ENRICHMENT — scrape + identification dirigeant (parallel) enriched = await asyncio.gather(*[ enrich_prospect(p) for p in new_prospects[:batch_size] ], return_exceptions=True) # 4. SCORING — ICP qualified = [] for lead in enriched: if isinstance(lead, Exception): log.error("pipeline.enrichment_error", error=str(lead)) continue lead.icp_score = score_icp(lead) if lead.icp_score >= 60: qualified.append(lead) else: log.info("pipeline.below_threshold", lead=lead.account_name, score=lead.icp_score) # 5. CREATE — insérer dans DB locale + EspoCRM for lead in qualified: lead.id = await db.create_lead(lead) espo_result = await espocrm.create_lead(lead.to_espo_dict()) lead.espocrm_id = espo_result["id"] await db.update_espocrm_id(lead.id, lead.espocrm_id) run.leads_created += 1 # 6. EMAIL — ping générique + finder (parallel) email_tasks = [] for lead in qualified: email_tasks.append(process_lead_email(lead)) await asyncio.gather(*email_tasks, return_exceptions=True) # 7. SYNC — mettre à jour EspoCRM avec les résultats for lead in qualified: await espocrm.update_lead(lead.espocrm_id, lead.to_espo_update()) run.finish() log.info("pipeline.done", **run.to_dict()) return run async def enrich_prospect(prospect) -> Lead: """Enrichit un prospect avec scrape Firecrawl.""" # Scrape page principale page = await firecrawl.scrape(prospect.url) # Chercher page équipe/à-propos about_urls = [ f"{prospect.url}/equipe", f"{prospect.url}/a-propos", f"{prospect.url}/about", f"{prospect.url}/team", f"{prospect.url}/qui-sommes-nous", ] contact_name = None for url in about_urls: try: about_page = await firecrawl.scrape(url) contact_name = extract_contact_from_page(about_page) if contact_name: break except Exception: continue # Fallback : search LinkedIn/Google if not contact_name: search_result = await firecrawl.search( f"PDG fondateur dirigeant {prospect.title} site:linkedin.com" ) contact_name = extract_contact_from_search(search_result) if not contact_name: raise ValueError(f"Dirigeant non identifiable pour {prospect.url}") return Lead( first_name=contact_name.first, last_name=contact_name.last, account_name=prospect.title, domain=extract_domain(prospect.url), website=prospect.url, ) async def process_lead_email(lead: Lead): """Vérifie email générique puis lance le finder si nécessaire.""" # 1. Si email générique trouvé sur le site → ping if lead.email and not is_generic_email(lead.email): result = await smtp_verify(lead.email) if result == "valid": lead.status = "contacted" lead.email_status = "valid" return # 2. Email finder (permutations) result = await find_email(lead.first_name, lead.last_name, lead.domain) if result.status == "valid": lead.email = result.email lead.status = "contacted" lead.email_status = "valid" elif result.status == "catch_all": lead.email = result.email lead.status = "researching" lead.email_status = "catch_all" lead.email_notes = f"Catch-all domain. Likely: {result.email}" else: lead.status = "researching" lead.email_status = "not_found" ``` ### CLI (typer) ```bash # Pipeline complet lead-pipeline run --niche piscines --region "Île-de-France" --batch 15 # Pipeline toutes niches, toutes régions (night mode) lead-pipeline run-all --batch-per-niche 5 --parallel 3 # Email finder standalone lead-pipeline find-email --first Jean --last Dupont --domain example.com lead-pipeline ping user@example.com lead-pipeline verify --status contacted --update # Batch email finder sur leads Researching lead-pipeline find-emails --status researching --limit 50 # Outreach lead-pipeline outreach --dry-run # Prévisualise sans envoyer lead-pipeline outreach --send # Envoie les emails dus # Stats lead-pipeline stats # Dashboard leads par niche/statut/région lead-pipeline stats --niche piscines # Stats filtrées # Sync EspoCRM lead-pipeline sync --push # Local → EspoCRM lead-pipeline sync --pull # EspoCRM → Local # Scheduler (mode daemon) lead-pipeline daemon # Lance le scheduler cron-like ``` ### Logging (structlog) ```python import structlog log = structlog.get_logger() # Exemples de logs structurés log.info("smtp.verify", email="user@example.com", result="valid", mx="mail.example.com", duration_ms=1200) log.warning("smtp.timeout", email="user@example.com", mx="mail.example.com", timeout_s=10) log.error("firecrawl.scrape_failed", url="https://magiline.com", status=403, retry=2) log.info("pipeline.lead_created", account="Magiline", score=72, niche="piscines", region="PACA") # Format JSON en fichier, rich en console # Fichier : data/logs/pipeline-2026-02-21.jsonl ``` ### Scheduler (mode daemon) ```python # APScheduler ou custom asyncio scheduler schedule: lead_pipeline_night: cron: "*/30 22-23,0-7 * * *" timezone: "Europe/Paris" action: run_pipeline params: rotate_niches: true # Alterne entre les niches à chaque run rotate_regions: true # Alterne entre les régions batch_size: 15 email_finder_batch: cron: "0 */4 * * *" # Toutes les 4h action: find_emails_batch params: status: "researching" limit: 30 verify_contacted: cron: "0 8 * * *" # Chaque matin 8h action: verify_emails params: status: "contacted" update: true outreach: cron: "0 16 * * 1-5" # Lun-ven 16h action: send_outreach params: dry_run: false morning_report: cron: "0 9 * * *" action: generate_report params: type: "daily" ``` --- ## 9. Données de référence ### Config actuelle espocrm.py - **Auth** : `X-Api-Key` header, env var `ESPOCRM_URL` + `ESPOCRM_API_KEY` - **Note sandbox** : env vars avec suffixes `_API_KEY`, `_TOKEN`, `_PASSWORD` sont bloquées par le sanitizer OpenClaw → renommer en `_CREDENTIAL` ou `_CRED` - **Commandes** : search, list, get, create, update, delete, link, related - **Timeout** : 30s par requête ### Config actuelle email_finder.py - **SMTP timeout** : 10s - **SMTP FROM** : `verify@elevaseo.com` - **DNS servers** : 8.8.8.8, 1.1.1.1 - **Délai entre permutations** : 0.5s - **Délai entre leads (batch)** : 2s - **24-29 permutations** selon si prénom composé ou non ### Himalaya (envoi emails) ```toml # config.toml [accounts.sbinnovate] email = "sebastien@sbinnovate.net" display-name = "Sebastien Dacosta" [accounts.sbinnovate.imap] host = "arome.o2switch.net" port = 993 encryption = "tls" [accounts.sbinnovate.smtp] host = "arome.o2switch.net" port = 465 encryption = "tls" [accounts.sbinnovate.message.send] save-copy = false # o2switch gère côté serveur ``` --- ## 10. Migration : ce qui change | Aspect | Avant (OpenClaw) | Après (Python local) | |--------|-------------------|----------------------| | Orchestration | Bot LLM (non déterministe) | Script Python async | | Anti-doublon | EspoCRM search uniquement | SQLite local + EspoCRM sync | | Logs | stdout brut, docker logs | structlog JSON + fichiers rotatifs | | Email finder | Synchrone, 1 lead à la fois | Async, parallel MX + SMTP | | Niches | Hardcodée dans le prompt cron | Config YAML rotative | | Régions | Pas de segmentation | Par région avec rotation | | Scheduling | OpenClaw cron (gateway) | APScheduler ou cron system | | Coût | ~0.10€/lead (tokens LLM) | ~0€/lead (scripts directs) | | Fiabilité | Dépend de tokens Anthropic | Indépendant, retry automatique | | Vitesse | ~5 min/lead | ~10-30s/lead (async) |