From faecef78b66e8a5812069922c349f387f8a58378 Mon Sep 17 00:00:00 2001 From: Philipp Wenger Lebron Date: Mon, 25 May 2026 19:46:14 +0200 Subject: [PATCH 1/2] docs(i18n): add Spanish translation of README intro MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cherry-picked from #17 onto current main (PR #17's branch base was too old to fast-forward, and the GitHub update-branch API returned a server-side merge conflict). Adds docs/i18n/README.es.md with the first six sections translated (Por qué / Qué es / Estado de mantenimiento / Verlo en acción / Inicio rápido) plus a closing note pointing readers back to the full English README for everything else. Also adds the language switcher to the main README header. Spanish image-alt path corrected (../assets/demo.png → ../assets/demo.png resolves to docs/assets/demo.png from docs/i18n/), and the cross-link back to English fixed from ../README.md to ../../README.md. Closes #10. Credit: original translation by @MD-Mushfiqur123 in #17. Co-Authored-By: Claude Opus 4.7 (1M context) --- README.md | 2 ++ docs/i18n/README.es.md | 40 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+) create mode 100644 docs/i18n/README.es.md diff --git a/README.md b/README.md index 8124368..3f0121b 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,7 @@ # Compabob +[English](README.md) | [Español](docs/i18n/README.es.md) + [![CI](https://github.com/chacosoldier/compabob/actions/workflows/smoke.yml/badge.svg)](https://github.com/chacosoldier/compabob/actions/workflows/smoke.yml) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) [![GitHub stars](https://img.shields.io/github/stars/chacosoldier/compabob?style=social)](https://github.com/chacosoldier/compabob/stargazers) diff --git a/docs/i18n/README.es.md b/docs/i18n/README.es.md new file mode 100644 index 0000000..44b93c6 --- /dev/null +++ b/docs/i18n/README.es.md @@ -0,0 +1,40 @@ +# Compabob + +[English](../../README.md) | [Español](README.es.md) + +## Por qué + +Una sesión nueva de Claude Code es una generalista competente que no sabe nada de ti. Cada vez tienes que volver a explicarle quién eres, cómo quieres las respuestas y en qué estás trabajando. No tiene noción de tus prioridades, ni barreras de seguridad sobre lo que se envía, ni un lugar donde guardar lo que aprende. + +Compabob es la estructura que faltaba. El ciclo fundamental, desde el primer día: + +> Le pides que tome notas de la reunión de planificación de hoy. Las archiva en tu base de conocimiento. La semana siguiente, antes del seguimiento, escribes `/meeting-prep` y te devuelve los asistentes, lo que se decidió y los temas pendientes, sin que tengas que buscar nada. + +Ese ciclo de segundo cerebro es el valor que obtienes de inmediato. Compabob es un andamio, no un asistente terminado: todo lo demás se multiplica a medida que lo haces tuyo. + +## Qué es (y qué no es) + +- **Es**: una plantilla con opinión propia para un solo usuario. Arquitectura, convenciones y un conjunto de agentes, hooks y skills que requirieron verdadera iteración para dar en el clavo, empaquetados para que no empieces desde cero. +- **No es**: un producto, un SaaS ni una caja mágica sin configuración. No hay registro, ni telemetría, ni ventas adicionales. Funciona completamente en tu máquina bajo tu propia suscripción de Claude Code. + +## Estado de mantenimiento + +**Mantenimiento activo.** Este kit se actualiza con el tiempo a medida que sus mantenedores aprenden qué funciona y conforme Claude Code evoluciona, así que espera que siga mejorando. Sigues siendo dueño absoluto de tu copia: haz fork, cámbialo todo, aléjate todo lo que quieras. Las issues y PRs son bienvenidas; las revisiones se hacen con el tiempo disponible, ya que el kit se mantiene junto con otros trabajos. + +## Verlo en acción + +![Ejecutando ./setup.sh: Compabob hace algunas preguntas y luego crea tu espacio de trabajo en aproximadamente un minuto.](../assets/demo.png) + +Un solo script. Te pregunta quién eres y qué haces, crea tu vault, memoria y configuración a partir de las semillas incluidas, aplica un preset de personalidad y ya estás listo para empezar. + +## Inicio rápido + +¿Te manejas con la terminal? Tres comandos: + +```bash +git clone https://github.com/chacosoldier/compabob.git +cd compabob +./setup.sh +``` + +> Esta es una traducción parcial (las primeras seis secciones) del [README en inglés](../../README.md). Para la guía completa, incluyendo el tutorial paso a paso, las skills, los módulos y el resto, sigue el documento original. Traducción inicial aportada por [@MD-Mushfiqur123](https://github.com/MD-Mushfiqur123) en [#17](https://github.com/chacosoldier/compabob/pull/17). From 737f964e08c78e11fc64fca66bb579f17f927dd5 Mon Sep 17 00:00:00 2001 From: Philipp Wenger Lebron Date: Mon, 8 Jun 2026 23:08:51 +0200 Subject: [PATCH 2/2] feat: RevOps lead pipeline + CRM merge backbone Two opt-in modules for the commercial/RevOps workflow: - crm-merge (+ /merge-contacts): folds Google Contacts, a LinkedIn export, and vault People/ notes into one local CRM (SQLite + JSON + offline HTML browser). Union-find identity resolution on email / LinkedIn slug / normalized name, with a common-name collision guard. No creds, stdlib, idempotent. - lead-pipeline (+ /build-list): discover -> clean -> dedup-against-CRM -> enrich -> score, one CSV per stage. Deterministic stages run as a stdlib script; discovery/enrichment run as assistant steps over optional MCP. Dedup tags each lead proceed/warm/skip so you never enrich or cold-email a contact you already know. ICP scoring via a tunable JSON rubric. Wired into README, modules/README, CHANGELOG; icp.json git-ignored. Co-Authored-By: Claude Opus 4.8 --- .claude/skills/build-list/SKILL.md | 84 +++ .claude/skills/merge-contacts/SKILL.md | 50 ++ .gitignore | 2 + CHANGELOG.md | 17 + README.md | 4 + modules/README.md | 2 + modules/crm-merge/README.md | 103 ++++ modules/crm-merge/build.py | 716 +++++++++++++++++++++++++ modules/lead-pipeline/README.md | 138 +++++ modules/lead-pipeline/icp.example.json | 21 + modules/lead-pipeline/pipeline.py | 355 ++++++++++++ 11 files changed, 1492 insertions(+) create mode 100644 .claude/skills/build-list/SKILL.md create mode 100644 .claude/skills/merge-contacts/SKILL.md create mode 100644 modules/crm-merge/README.md create mode 100644 modules/crm-merge/build.py create mode 100644 modules/lead-pipeline/README.md create mode 100644 modules/lead-pipeline/icp.example.json create mode 100644 modules/lead-pipeline/pipeline.py diff --git a/.claude/skills/build-list/SKILL.md b/.claude/skills/build-list/SKILL.md new file mode 100644 index 0000000..d44dc4a --- /dev/null +++ b/.claude/skills/build-list/SKILL.md @@ -0,0 +1,84 @@ +--- +name: build-list +description: Build a ranked, CRM-deduped outbound prospect list from an ICP description. Use for "/build-list", "find me leads matching X", "build a prospect list for Y", or when the user wants a scored outbound list. Orchestrates discovery + the lead-pipeline module + enrichment. +--- + +# Build List + +Turn an ICP description into a ranked, deduped, CRM-aware outbound list using the +`lead-pipeline` module. You run the judgement/IO steps (discover, enrich); the +script runs the deterministic steps (clean, dedup, score). The cardinal rule: + +> **Dedup against the CRM before enriching.** Never spend enrichment effort or +> credits on accounts the user already knows. Known accounts become warm intros. + +## Preconditions + +- A CRM to dedup against: `data/crm-merge/contacts.json`. If it is missing, tell + the user to run `/merge-contacts` first (dedup is the whole point); offer to + proceed without it only if they insist, noting every lead will be `proceed`. +- An ICP rubric at `modules/lead-pipeline/icp.json`. If only `icp.example.json` + exists, copy it and adjust it to the user's request before scoring. + +## Steps + +1. **Read the ICP from the request.** Pull out: industry/segment, geography, + company-size band, and target buyer titles. Confirm anything ambiguous in one + short question rather than guessing. + +2. **Discover candidates → raw CSV.** Use the web-search MCP (e.g. exa + `web_search_exa` / `linkedin_search_exa`) to find matching accounts. Write a + CSV to `reports/lead-pipeline/00-raw.csv` with whatever you found across these + columns: `company, domain, contact_name, contact_email, title, country` + (partial rows are fine; later stages fill gaps). If no search MCP is + available, ask the user for a raw CSV and skip to step 3. + +3. **Clean** (script): + ```bash + python3 modules/lead-pipeline/pipeline.py clean \ + --in reports/lead-pipeline/00-raw.csv \ + --out reports/lead-pipeline/01-cleaned.csv --country + ``` + +4. **Dedup against the CRM** (script): + ```bash + python3 modules/lead-pipeline/pipeline.py dedup \ + --in reports/lead-pipeline/01-cleaned.csv \ + --crm data/crm-merge/contacts.json \ + --out reports/lead-pipeline/02-deduped.csv + ``` + Report the split: how many `proceed` (cold), `warm` (known account, route via + `crm_known_contact`), `skip` (already known, drop from outbound). + +5. **Enrich the survivors only.** For rows tagged `proceed` or `warm` that lack a + domain / email / decision-maker, enrich with whatever you have, in this order: + - web search MCP for the company domain + LinkedIn (high hit-rate, cheap), + - an enrichment provider MCP (e.g. Clay) for verified emails — check remaining + credits first, and only for rows still missing an email, + - public business registries for named decision-makers at small firms when + enrichment misses. + Never enrich `skip` rows. Write the enriched result back to + `reports/lead-pipeline/02-deduped.csv` (or a `02b-enriched.csv`). + +6. **Score** (script): + ```bash + python3 modules/lead-pipeline/pipeline.py score \ + --in reports/lead-pipeline/02-deduped.csv \ + --icp modules/lead-pipeline/icp.json \ + --out reports/lead-pipeline/03-scored.csv --top 50 + ``` + +7. **Hand back** the ranked top-N (path `reports/lead-pipeline/04-top-*.csv`), + the disposition split, and the tier distribution. Call out the `warm` + accounts explicitly: those are the highest-value, lowest-risk plays. + +## Guardrails + +- Do not send anything. This builds a list; outreach is a separate, reviewed step + (see the `linkedin-outreach` module). +- Respect data-protection law and platform terms. Flag if the user's target list + looks like scraped personal data being repurposed for cold mail. +- If a stage's numbers look wrong (e.g. zero `skip`/`warm` with a populated CRM), + inspect the stage CSV before continuing — that is why each stage is a file. +- Be honest about enrichment misses; report the hit-rate rather than fabricating + emails. diff --git a/.claude/skills/merge-contacts/SKILL.md b/.claude/skills/merge-contacts/SKILL.md new file mode 100644 index 0000000..5eb3d5e --- /dev/null +++ b/.claude/skills/merge-contacts/SKILL.md @@ -0,0 +1,50 @@ +--- +name: merge-contacts +description: Merge the user's contact exports (Google Takeout, LinkedIn export, vault People notes) into one local CRM. Use for "/merge-contacts", "build my CRM", "dedup my contacts", or when another workflow needs a clean contacts source of truth. +--- + +# Merge Contacts + +Build the user's local CRM source of truth with the `crm-merge` module. The +module does the work; your job is to find their exports, run it with the right +flags, and report what merged. Never invent data, never call an external CRM. + +## Steps + +1. **Find the exports.** Check, in order: + - A Google Takeout zip in `~/Downloads` (name contains "takeout" or + "google"). The script auto-finds the newest one, but confirm it exists. + - A LinkedIn export folder (contains `Connections.csv`, ideally + `messages.csv`). Ask the user for the path if you cannot find it. + - `vault/People/` for per-person notes (optional). + If none exist, point the user at the setup steps in + `modules/crm-merge/README.md` (both exports are self-service downloads) and stop. + +2. **Get their name** for message-frequency counting. Ask how their name appears + in LinkedIn messages (the "me" side), or read it from `config/user.config.yaml`. + +3. **Run the build:** + ```bash + python3 modules/crm-merge/build.py \ + --takeout-zip \ + --linkedin-dir \ + --me "" \ + --out data/crm-merge + ``` + Omit any source they do not have; one source alone is fine. + +4. **Report** the counts the script prints: total unique people, how many appear + in 2+ sources (the high-confidence merges), and how many link to a vault note. + Flag if a source loaded zero records, since that usually means a wrong path. + +5. **Point them onward.** Tell them `data/crm-merge/browser.html` is openable in + any browser, and that `contacts.json` is what `/build-list` dedups against. + +## Watch out for + +- **Over-merging on a common name:** if the user says distinct people got fused, + lower `--name-collision-threshold` and re-run. +- **Zero LinkedIn messages counted:** the `--me` name probably does not match the + export. Check the `FROM`/`TO` values in `messages.csv` and pass the exact form. +- **Privacy:** outputs are git-ignored; do not commit them or paste contact data + into anything outward-facing. diff --git a/.gitignore b/.gitignore index 0fa768f..8b79bde 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,8 @@ logs/ .tmp/ reports/ modules/*/generated/ +# Your ICP rubric (copied from icp.example.json) is your market definition, not the kit's +modules/lead-pipeline/icp.json # Python __pycache__/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 61b1666..da9c552 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,23 @@ All notable changes to Compabob are recorded here. Format follows ### Added +- **`crm-merge` module + `/merge-contacts` skill** — folds Google Contacts (Takeout + vCards), a LinkedIn data export (Connections + messages), and your vault + `People/` notes into one local source of truth: a SQLite DB, a JSON file, and a + self-contained offline HTML browser. Identity resolution uses union-find on + three keys (verified email, LinkedIn slug, normalized name) with a common-name + collision guard, so duplicates collapse without over-merging generic names. + Relationship strength (LinkedIn message count) ranks records. No credentials, + all stdlib, idempotent. Outputs are git-ignored. +- **`lead-pipeline` module + `/build-list` skill** — turns a raw candidate list + into a ranked, CRM-aware outbound list through staged steps that write one CSV + per stage (discover → clean → **dedup-against-CRM** → enrich → score → top-N). + The deterministic stages (`clean`, `dedup`, `score`) run as a stdlib script; + discovery and enrichment run as assistant steps over optional MCP servers (exa + free tier minimum; an enrichment provider optional). Dedup tags each lead + `proceed` (cold), `warm` (you already know someone at the account, route the + intro), or `skip` (already a known contact), so you never enrich or cold-email + a relationship. ICP scoring is a tunable JSON rubric (`icp.example.json`). - **`/chart-tufte` skill** — self-grade rubric for any quantitative chart, grounded in Edward Tufte's *Visual Display of Quantitative Information*. Nine criteria, ten genres, seven remedies, plus a 114-line diff --git a/README.md b/README.md index 3f0121b..a02bead 100644 --- a/README.md +++ b/README.md @@ -225,6 +225,8 @@ Skills are slash-command workflows. Type the command; the assistant runs the pro | `/chart-tufte` | Self-grade any chart against Tufte's *Visual Display of Quantitative Information* before showing it | | `/mcp-debug` | Health-check, trace, or audit your MCP servers when tools fail silently | | `/memory-debt` | Resolve memory updates that earlier sessions proposed but never wrote | +| `/merge-contacts` | Fold your Google + LinkedIn + vault contacts into one local CRM, deduped | +| `/build-list` | Build a ranked outbound list: discover, dedup against your CRM, enrich, score | ## Modules (opt-in) @@ -236,6 +238,8 @@ The core runs with zero external services. Modules add capability when you want | `telegram` | Available | A Telegram bot: inbound messages drafted for your approval, never auto-sent | | `integrations` | Available | MCP tools: browser automation, web search, Gmail, Calendar, utilities | | `linkedin-outreach` | Available | Drafts one LinkedIn connection-invitation card a day from a queue, for your review and manual send | +| `crm-merge` | Available | Folds Google + LinkedIn + vault contacts into one local CRM (SQLite + offline HTML browser), deduped across sources. No creds | +| `lead-pipeline` | Available | Builds a ranked outbound list: discover, clean, dedup against your CRM, enrich, score. One CSV per stage | | `memory-search` | Available | Real index over `memory/` and `vault/` so retrieval ranks by relevance (FTS5 by default; semantic via Ollama if installed) | | `extra-agents` | Roadmap | An agent gallery: designer, evaluator, sales coach, project manager | | `whatsapp` | Roadmap | WhatsApp channel — not built (account-ban risk); use Telegram instead | diff --git a/modules/README.md b/modules/README.md index a6b07a7..4e1ceec 100644 --- a/modules/README.md +++ b/modules/README.md @@ -8,6 +8,8 @@ The core kit runs with zero external services. Modules add capability when you w | [`telegram/`](telegram/) | **Available** | A Telegram bot. Inbound messages are drafted for your approval; it never auto-sends. Bot-token based, polling. | | [`integrations/`](integrations/) | **Available** | An MCP integration picker: web + browser tools, Google Workspace, web search, and utilities. Run `scripts/install-integrations.sh`. | | [`linkedin-outreach/`](linkedin-outreach/) | **Available** | Daily LinkedIn connection-invitation drafting from a queue you keep: it decides hook vs. no-note and writes a review card. You send it; it never sends. | +| [`crm-merge/`](crm-merge/) | **Available** | Folds Google Contacts + LinkedIn export + vault notes into one local CRM (SQLite + JSON + offline HTML browser), deduping people across sources. No creds. Run by hand or via the `/merge-contacts` skill. | +| [`lead-pipeline/`](lead-pipeline/) | **Available** | Builds a ranked outbound list: discover, clean, **dedup against your CRM**, enrich, score. One CSV per stage. Run by hand or via the `/build-list` skill. exa MCP optional for discovery. | | [`memory-search/`](memory-search/) | **Available** | A search index over memory and the vault: ranked keyword search out of the box, semantic (search by meaning) with Ollama. Built by the `/index-memory` skill. | | [`extra-agents/`](extra-agents/) | Roadmap | A gallery of additional agents (designer, evaluator, sales coach, project manager) to copy into `.claude/agents/`. | | [`whatsapp/`](whatsapp/) | Roadmap | A WhatsApp channel. Not built: unofficial bridges risk an account ban. Use the Telegram module instead. | diff --git a/modules/crm-merge/README.md b/modules/crm-merge/README.md new file mode 100644 index 0000000..293cb11 --- /dev/null +++ b/modules/crm-merge/README.md @@ -0,0 +1,103 @@ +# Module: CRM merge + +Folds your scattered contact exports into **one local source of truth**: a +SQLite database, a JSON file, and a self-contained HTML browser you can open +offline. No external CRM, no API keys, no data leaving your machine. + +It merges three sources, deduping people who appear in more than one: + +- **Google Contacts** (from a Google Takeout zip) +- **LinkedIn Connections + message history** (from a LinkedIn data export) +- **Your vault `People/` notes** (optional) + +## Why own your contacts + +Your network lives split across LinkedIn, your phone, your inbox, and your CRM +vendor, each with a partial, slightly-wrong copy. The moment you want to "email +everyone I know at a fintech" or "check if we already know someone at this +account," you are stitching exports by hand. This module does that stitch once, +locally, and gives you a file you control. It is also the **dedup backbone** +the `lead-pipeline` module needs: you cannot "dedup a prospect list against your +CRM" until you have a clean CRM. + +## The hard part it gets right: identity resolution + +The same person shows up as `j.smith@acme.com` in Google, `linkedin.com/in/jsmith` +on LinkedIn, and `John Smith.md` in your notes. Naive dedup either misses these +(three rows for one person) or over-merges (every "John Smith" fused into one). + +This module uses **union-find on three identity keys** — verified email, +LinkedIn slug, and normalized name — so any shared key links records +transitively into one person. A **common-name guard** stops it merging on a name +that too many records share (the generic-name trap), while exact keys (email, +slug) always merge. This is the entity-resolution problem every CRM-hygiene +project hits, and the part most hand-rolled scripts get wrong. + +The merged record keeps the fullest name, the union of emails/phones, and the +strongest relationship signal (LinkedIn message count) to rank who you actually +know well. + +## Setup + +No credentials. You supply two self-service exports: + +1. **Google Takeout** — https://takeout.google.com → select **Contacts** + (vCard format) → download the zip. Leave it in `~/Downloads` and the script + finds it, or pass `--takeout-zip `. +2. **LinkedIn export** — LinkedIn → Settings & Privacy → Data privacy → + *Get a copy of your data* → include **Connections** and **Messages**. Unzip + it into a folder and pass `--linkedin-dir `. (LinkedIn emails the + archive; it can take minutes to a day.) +3. **Vault notes** — optional; defaults to `vault/People/`. + +Either source alone works; you do not need all three. + +## Use it + +```bash +python3 modules/crm-merge/build.py \ + --takeout-zip ~/Downloads/takeout-20260608.zip \ + --linkedin-dir ~/Downloads/linkedin-export \ + --me "Your Full Name" \ + --out data/crm-merge +``` + +`--me` is how the script tells you from the other party in your LinkedIn +messages, so it can count message frequency per contact. Pass the name(s) +exactly as they appear in the export, comma-separated if you have more than one. + +Outputs land in `data/crm-merge/` (git-ignored): + +- `contacts.db` — query it with any SQLite tool, or let the `crm-relationships` + agent read it. +- `contacts.json` — what the `lead-pipeline` module dedups against. +- `browser.html` — open in a browser: search, filter by source, click a person + for full detail. Fully offline, all data inlined. + +Re-run any time. It is idempotent: it rebuilds from scratch, so re-exporting and +re-running keeps your source of truth current. + +## Or just ask your assistant + +Run the `/merge-contacts` skill in a session and it will check which exports you +have, run the build with the right flags, and report what merged. + +## Tuning the merge + +If you have an unusually common name in your network and see over-merging, lower +`--name-collision-threshold` (default 30). If distinct people are not merging, +it is almost always because neither shares an email nor a LinkedIn slug and their +names differ; add the missing identifier to one source and re-run. + +## Pairs with + +- **`crm-relationships` agent** — reads `contacts.db` to answer "who do I know + at X" and to track interactions. +- **`lead-pipeline` module** — dedups new prospect lists against `contacts.json` + so you never enrich or cold-email someone you already know. + +## A note on privacy + +Everything stays local. The HTML browser inlines your data into one file with no +network calls; treat that file like the contact list it is. Outputs are +git-ignored by default so you never commit your network to a repo. diff --git a/modules/crm-merge/build.py b/modules/crm-merge/build.py new file mode 100644 index 0000000..ab47111 --- /dev/null +++ b/modules/crm-merge/build.py @@ -0,0 +1,716 @@ +"""CRM merge backbone. Folds Google Contacts (vCards), LinkedIn Connections, +LinkedIn message frequency, and vault People/ notes into a single local SQLite +database plus a JSON export plus a self-contained offline HTML browser. + +Design goals: +- You own the data (no external CRM dependency, no live API calls). +- Identity merge across sources (email > LinkedIn slug > normalized name), + using union-find so a person seen in three exports collapses to one record. +- A common-name-collision guard, so two different "John Smith" rows are not + fused just because they share a generic name. +- Relationship-strength signal (LinkedIn message count) ranks records. +- Idempotent: safe to re-run; rebuilds the DB from scratch each time. + +Everything is stdlib. Bring your own static exports: + - Google Takeout zip (Contacts as vCard) — https://takeout.google.com + - LinkedIn data export (Connections.csv, messages.csv) — Settings > Data privacy + - your vault's People/ notes (optional) + +Usage: + python3 build.py \ + --takeout-zip ~/Downloads/takeout-XXXX.zip \ + --linkedin-dir ~/Downloads/linkedin-export \ + --vault-people vault/People \ + --me "Your Full Name" \ + --out data/crm-merge +""" + +from __future__ import annotations + +import argparse +import csv +import json +import re +import sqlite3 +import sys +import unicodedata +import zipfile +from collections import defaultdict +from datetime import datetime, timezone +from pathlib import Path + +PROJECT_DIR = Path(__file__).resolve().parents[2] +DEFAULT_LINKEDIN_DIR = PROJECT_DIR / "data" / "linkedin" +DEFAULT_VAULT_PEOPLE = PROJECT_DIR / "vault" / "People" +DEFAULT_OUT = PROJECT_DIR / "data" / "crm-merge" + + +def find_takeout_zip() -> str: + """Best-effort: newest *.zip in ~/Downloads whose name mentions 'takeout'.""" + downloads = Path.home() / "Downloads" + if not downloads.exists(): + return "" + candidates = [ + p for p in downloads.glob("*.zip") + if "takeout" in p.name.lower() or "google" in p.name.lower() + ] + if not candidates: + return "" + return str(max(candidates, key=lambda p: p.stat().st_mtime)) + + +def normalize_name(s: str) -> str: + if not s: + return "" + s = unicodedata.normalize("NFKD", s).encode("ascii", "ignore").decode("ascii") + s = re.sub(r"[^a-zA-Z\s]", "", s).strip().lower() + return re.sub(r"\s+", " ", s) + + +def normalize_email(s: str) -> str: + return s.strip().lower() if s else "" + + +# ---------- vCard parser (no deps) ---------- + +def parse_vcards(blob: str) -> list[dict]: + cards = [] + for raw in blob.split("END:VCARD"): + if "BEGIN:VCARD" not in raw: + continue + card = {"emails": [], "phones": [], "orgs": [], "titles": [], + "addresses": [], "categories": [], "fn": None, "n": None, + "bday": None, "note": None, "url": None} + for line in raw.splitlines(): + line = line.strip() + if not line or line.startswith("BEGIN:") or line.startswith("VERSION"): + continue + if ":" not in line: + continue + key, value = line.split(":", 1) + key_upper = key.upper().split(";")[0] + if key_upper == "FN": + card["fn"] = value + elif key_upper == "N": + card["n"] = value + elif key_upper == "EMAIL": + e = normalize_email(value) + if e and e not in card["emails"]: + card["emails"].append(e) + elif key_upper == "TEL": + p = re.sub(r"[^\d+]", "", value) + if p and p not in card["phones"]: + card["phones"].append(p) + elif key_upper == "ORG": + card["orgs"].append(value) + elif key_upper == "TITLE": + card["titles"].append(value) + elif key_upper == "ADR": + card["addresses"].append(value) + elif key_upper == "BDAY": + card["bday"] = value + elif key_upper == "NOTE": + card["note"] = value + elif key_upper == "URL": + card["url"] = value + elif key_upper == "CATEGORIES": + card["categories"] = [c.strip() for c in value.split(",")] + if card["fn"] or card["emails"] or card["phones"]: + cards.append(card) + return cards + + +def load_google_contacts(zip_path: Path) -> list[dict]: + if not zip_path or not zip_path.exists(): + print(f"WARN: Takeout zip not found ({zip_path}), skipping Google Contacts", + file=sys.stderr) + return [] + contacts = [] + with zipfile.ZipFile(zip_path) as zf: + for info in zf.infolist(): + if not info.filename.endswith(".vcf") or "Contacts" not in info.filename: + continue + with zf.open(info.filename) as f: + blob = f.read().decode("utf-8", errors="replace") + for c in parse_vcards(blob): + c["_source"] = "google_contacts" + c["_source_file"] = info.filename.split("/")[-1] + contacts.append(c) + return contacts + + +# ---------- LinkedIn ---------- + +def load_linkedin_connections(folder: Path) -> list[dict]: + p = folder / "Connections.csv" + if not p.exists(): + return [] + text = p.read_text(encoding="utf-8", errors="replace") + lines = text.splitlines(keepends=True) + # LinkedIn prefixes the export with a "Notes:" preamble + a blank line. + if lines and lines[0].lstrip().startswith("Notes:"): + blank_at = next((i for i, line in enumerate(lines) if not line.strip()), 0) + body = "".join(lines[blank_at + 1:]) + else: + body = text + rows = list(csv.DictReader(body.splitlines())) + out = [] + for r in rows: + first = (r.get("First Name") or "").strip() + last = (r.get("Last Name") or "").strip() + fn = " ".join(filter(None, [first, last])).strip() + if not fn: + continue + out.append({ + "fn": fn, + "first_name": first, + "last_name": last, + "linkedin_url": (r.get("URL") or "").strip(), + "email": normalize_email(r.get("Email Address") or ""), + "company": (r.get("Company") or "").strip(), + "position": (r.get("Position") or "").strip(), + "connected_on": (r.get("Connected On") or "").strip(), + "_source": "linkedin", + }) + return out + + +def load_linkedin_message_stats(folder: Path, me_names: tuple[str, ...]) -> dict[str, int]: + """Returns correspondent-name -> message count, used as a relationship signal.""" + p = folder / "messages.csv" + if not p.exists(): + return {} + counts: dict[str, int] = defaultdict(int) + with p.open(encoding="utf-8", errors="replace", newline="") as f: + reader = csv.DictReader(f) + for r in reader: + frm = (r.get("FROM") or "").strip() + to = (r.get("TO") or "").strip() + other = to if frm in me_names else frm + if other and other not in me_names and other != "LinkedIn Member": + counts[other] += 1 + return dict(counts) + + +# ---------- Vault People/ ---------- + +def load_vault_people(folder: Path) -> list[dict]: + if not folder.exists(): + return [] + out = [] + for p in sorted(folder.glob("*.md")): + name = p.stem + try: + rel = str(p.relative_to(PROJECT_DIR)) + except ValueError: + rel = str(p) + out.append({ + "fn": name, + "vault_note": rel, + "_source": "vault", + }) + return out + + +# ---------- Merge engine ---------- + +def make_keys(record: dict) -> list[str]: + """Returns the set of merge keys this record exposes.""" + keys: list[str] = [] + if record.get("fn"): + n = normalize_name(record["fn"]) + # Require at least 2 word tokens AND >= 5 letters total, so we never + # collapse records on a single common first name ("Pedro", "Anna"). + if n and " " in n and len(n.replace(" ", "")) >= 5: + keys.append(f"name:{n}") + for e in record.get("emails", []) or [record.get("email")]: + if e and "@" in e: + keys.append(f"email:{e}") + if record.get("linkedin_url"): + m = re.search(r"linkedin\.com/in/([a-zA-Z0-9_-]+)", record["linkedin_url"]) + if m: + keys.append(f"linkedin:{m.group(1).lower()}") + return keys + + +def merge_records(google: list[dict], linkedin: list[dict], + msg_counts: dict[str, int], + vault: list[dict], + name_collision_threshold: int) -> list[dict]: + """Union-find merge across all sources. Returns merged list of dicts.""" + all_records: list[dict] = [] + + for r in google: + all_records.append({ + "fn": r.get("fn") or "", + "emails": list(r.get("emails", [])), + "phones": list(r.get("phones", [])), + "company": (r.get("orgs") or [""])[0], + "position": (r.get("titles") or [""])[0], + "address": (r.get("addresses") or [""])[0], + "bday": r.get("bday"), + "note": r.get("note"), + "url": r.get("url"), + "categories": r.get("categories", []), + "linkedin_url": "", + "connected_on": "", + "vault_note": "", + "linkedin_messages": 0, + "sources": ["google_contacts"], + "_source_files": [r.get("_source_file", "")], + }) + for r in linkedin: + all_records.append({ + "fn": r.get("fn") or "", + "emails": [r["email"]] if r.get("email") else [], + "phones": [], + "company": r.get("company") or "", + "position": r.get("position") or "", + "address": "", + "bday": None, + "note": None, + "url": "", + "categories": [], + "linkedin_url": r.get("linkedin_url") or "", + "connected_on": r.get("connected_on") or "", + "vault_note": "", + "linkedin_messages": msg_counts.get(r.get("fn", ""), 0), + "sources": ["linkedin"], + "_source_files": [], + }) + for r in vault: + all_records.append({ + "fn": r.get("fn") or "", + "emails": [], + "phones": [], + "company": "", + "position": "", + "address": "", + "bday": None, + "note": None, + "url": "", + "categories": [], + "linkedin_url": "", + "connected_on": "", + "vault_note": r.get("vault_note") or "", + "linkedin_messages": 0, + "sources": ["vault"], + "_source_files": [], + }) + + # Build key -> record-indices index, then union-find. + key_to_idx: dict[str, list[int]] = defaultdict(list) + for i, rec in enumerate(all_records): + for k in make_keys(rec): + key_to_idx[k].append(i) + + parent = list(range(len(all_records))) + + def find(x): + while parent[x] != x: + parent[x] = parent[parent[x]] + x = parent[x] + return x + + def union(a, b): + ra, rb = find(a), find(b) + if ra != rb: + parent[rb] = ra + + for label, indices in key_to_idx.items(): + # A name shared by many records is almost certainly a common/generic + # name (or a parsing artifact), not the same person. Above the + # threshold, do not merge on that name key. Email and LinkedIn slug + # keys are exact identifiers and always merge. + if label.startswith("name:") and len(indices) > name_collision_threshold: + continue + for j in range(1, len(indices)): + union(indices[0], indices[j]) + + # Group records by root and collapse. + groups: dict[int, list[int]] = defaultdict(list) + for i in range(len(all_records)): + groups[find(i)].append(i) + + merged = [] + for root, members in groups.items(): + merged_rec = { + "fn": "", + "emails": [], + "phones": [], + "company": "", + "position": "", + "address": "", + "bday": None, + "note": None, + "url": "", + "categories": [], + "linkedin_url": "", + "connected_on": "", + "vault_note": "", + "linkedin_messages": 0, + "sources": [], + "_source_files": [], + } + for idx in members: + r = all_records[idx] + for s in r["sources"]: + if s not in merged_rec["sources"]: + merged_rec["sources"].append(s) + for e in r["emails"]: + if e and e not in merged_rec["emails"]: + merged_rec["emails"].append(e) + for p in r["phones"]: + if p and p not in merged_rec["phones"]: + merged_rec["phones"].append(p) + for c in r["categories"]: + if c and c not in merged_rec["categories"]: + merged_rec["categories"].append(c) + for sf in r["_source_files"]: + if sf and sf not in merged_rec["_source_files"]: + merged_rec["_source_files"].append(sf) + # Prefer the longest fn (LinkedIn names are usually fuller). + if r["fn"] and len(r["fn"]) > len(merged_rec["fn"]): + merged_rec["fn"] = r["fn"] + for field in ("company", "position", "address", "url", + "linkedin_url", "connected_on", "vault_note"): + if r[field] and not merged_rec[field]: + merged_rec[field] = r[field] + if r["bday"] and not merged_rec["bday"]: + merged_rec["bday"] = r["bday"] + if r["note"] and not merged_rec["note"]: + merged_rec["note"] = r["note"] + merged_rec["linkedin_messages"] = max( + merged_rec["linkedin_messages"], r["linkedin_messages"]) + merged.append(merged_rec) + + # Sort by relationship signal: most-messaged first, then breadth of sources. + merged.sort( + key=lambda r: ( + -r["linkedin_messages"], + -len(r["sources"]), + r["fn"].lower(), + ) + ) + return merged + + +# ---------- SQLite + JSON output ---------- + +def write_outputs(merged: list[dict], out_dir: Path) -> None: + out_dir.mkdir(parents=True, exist_ok=True) + db_path = out_dir / "contacts.db" + if db_path.exists(): + db_path.unlink() + conn = sqlite3.connect(str(db_path)) + cur = conn.cursor() + cur.execute(""" + CREATE TABLE contacts ( + id INTEGER PRIMARY KEY, + fn TEXT, + emails TEXT, + phones TEXT, + company TEXT, + position TEXT, + address TEXT, + bday TEXT, + note TEXT, + url TEXT, + categories TEXT, + linkedin_url TEXT, + connected_on TEXT, + vault_note TEXT, + linkedin_messages INTEGER, + sources TEXT + )""") + cur.execute("CREATE INDEX idx_fn ON contacts(fn)") + cur.execute("CREATE INDEX idx_company ON contacts(company)") + cur.execute("CREATE INDEX idx_msg ON contacts(linkedin_messages DESC)") + for r in merged: + cur.execute( + "INSERT INTO contacts (fn, emails, phones, company, position, address, " + "bday, note, url, categories, linkedin_url, connected_on, vault_note, " + "linkedin_messages, sources) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + ( + r["fn"], + json.dumps(r["emails"], ensure_ascii=False), + json.dumps(r["phones"], ensure_ascii=False), + r["company"], r["position"], r["address"], r["bday"], r["note"], + r["url"], json.dumps(r["categories"], ensure_ascii=False), + r["linkedin_url"], r["connected_on"], r["vault_note"], + r["linkedin_messages"], json.dumps(r["sources"]), + ), + ) + conn.commit() + conn.close() + print(f" contacts.db: {db_path.stat().st_size:,} bytes") + + json_path = out_dir / "contacts.json" + json_path.write_text(json.dumps(merged, indent=2, ensure_ascii=False)) + print(f" contacts.json: {json_path.stat().st_size:,} bytes") + + +# ---------- HTML browser ---------- + +def write_html_browser(merged: list[dict], out_dir: Path) -> None: + """Single-file searchable browser with all data inlined. Open in any browser.""" + template = HTML_TEMPLATE + payload = json.dumps(merged, ensure_ascii=False) + template = template.replace("__DATA_PLACEHOLDER__", payload) + template = template.replace( + "__GENERATED_AT__", datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")) + template = template.replace("__TOTAL__", str(len(merged))) + out = out_dir / "browser.html" + out.write_text(template, encoding="utf-8") + print(f" browser.html: {out.stat().st_size:,} bytes") + + +HTML_TEMPLATE = """ + +CRM +
+

Your CRM

+

Merged from Google Contacts + LinkedIn + vault People/. Source of truth lives in data/crm-merge/contacts.db. Generated __GENERATED_AT__.

+
+
+ + All + In 2+ sources + In vault + LinkedIn + Google + Messaged +
+
+ +
+
+ +""" + + +def main(): + ap = argparse.ArgumentParser(description="Merge contact exports into one local CRM.") + ap.add_argument("--takeout-zip", default="", + help="Google Takeout zip (Contacts as vCard). " + "Default: newest takeout/google zip in ~/Downloads.") + ap.add_argument("--linkedin-dir", default=str(DEFAULT_LINKEDIN_DIR), + help="Folder with LinkedIn Connections.csv and messages.csv.") + ap.add_argument("--vault-people", default=str(DEFAULT_VAULT_PEOPLE), + help="Folder of per-person markdown notes (optional).") + ap.add_argument("--me", default="", + help="Your name(s) as they appear in LinkedIn messages, " + "comma-separated. Used to count message frequency.") + ap.add_argument("--name-collision-threshold", type=int, default=30, + help="Do not merge on a name shared by more than this many " + "records (guards against generic-name collisions).") + ap.add_argument("--out", default=str(DEFAULT_OUT), + help="Output directory for contacts.db / .json / browser.html.") + args = ap.parse_args() + + takeout = args.takeout_zip or find_takeout_zip() + me_names = tuple(n.strip() for n in args.me.split(",") if n.strip()) + + print("Loading sources...") + google = load_google_contacts(Path(takeout)) if takeout else [] + print(f" Google Contacts: {len(google)} vCards") + + linkedin = load_linkedin_connections(Path(args.linkedin_dir)) + print(f" LinkedIn Connections: {len(linkedin)}") + + msg_counts = load_linkedin_message_stats(Path(args.linkedin_dir), me_names) + print(f" LinkedIn message correspondents: {len(msg_counts)}") + + vault = load_vault_people(Path(args.vault_people)) + print(f" Vault People notes: {len(vault)}") + + if not (google or linkedin or vault): + print("\nNo sources found. Point --takeout-zip / --linkedin-dir / " + "--vault-people at your exports. See the module README.", + file=sys.stderr) + sys.exit(1) + + print("\nMerging by email + LinkedIn slug + normalized name...") + merged = merge_records(google, linkedin, msg_counts, vault, + args.name_collision_threshold) + print(f" Total unique people after merge: {len(merged)}") + + in_multi = sum(1 for r in merged if len(r["sources"]) >= 2) + print(f" Appearing in 2+ sources: {in_multi}") + in_vault = sum(1 for r in merged if "vault" in r["sources"]) + print(f" Linked to vault note: {in_vault}") + + print(f"\nWriting outputs to {args.out}/") + out_dir = Path(args.out) + write_outputs(merged, out_dir) + write_html_browser(merged, out_dir) + print("Done.") + + +if __name__ == "__main__": + main() diff --git a/modules/lead-pipeline/README.md b/modules/lead-pipeline/README.md new file mode 100644 index 0000000..ba3cd25 --- /dev/null +++ b/modules/lead-pipeline/README.md @@ -0,0 +1,138 @@ +# Module: Lead pipeline + +Turns a raw list of candidate accounts into a **ranked, deduped, CRM-aware +outbound list** through staged steps, writing one CSV per stage so you can +inspect or hand-fix the work at any point. + +``` +discover -> clean -> dedup-against-CRM -> enrich -> score -> top-N + (assistant) (script) (script) (assistant) (script) +``` + +The teachable insight, and the reason this beats a one-shot "find me leads" +prompt: + +> **Dedup against your CRM before you enrich.** You never spend enrichment +> credits on a company you already know, and you never cold-email a person who +> is already a relationship or an open deal. The known accounts become *warm* +> intros instead of cold ones. + +This is the workflow most teams do by hand in a spreadsheet. Here it is a +resumable pipeline. + +## How the work splits + +Two of the steps need judgement and live tools, so your **assistant** runs them +with MCP servers. Three steps are pure logic, so a **script** (`pipeline.py`) +runs them deterministically: + +| Stage | Who | What | +|-------|-----|------| +| discover | assistant (web search MCP) | find candidate accounts, write a raw CSV | +| **clean** | `pipeline.py clean` | normalize, drop junk, dedup within the list | +| **dedup** | `pipeline.py dedup` | tag each row `proceed` / `warm` / `skip` vs your CRM | +| enrich | assistant (enrichment MCP / public registry) | fill domains, emails, decision-maker names | +| **score** | `pipeline.py score` | ICP fit score + A/B/C tier + ranked top-N | + +The `/build-list` skill orchestrates the whole chain. You can also run the +script stages by hand. + +## The three dispositions + +`pipeline.py dedup` compares each lead against your CRM (the `contacts.json` +that the `crm-merge` module produces) and tags it: + +- **`skip`** — you already know this exact person (email match). Do not + cold-outreach; they are a relationship, not a lead. +- **`warm`** — you know *someone else* at this account (matched by corporate + email domain or company name). Do not cold-email a stranger here; route the + approach through the contact you already have. The matched contact's name is + written to the `crm_known_contact` column. +- **`proceed`** — genuinely cold, no match. Fair game for outbound. + +Free webmail domains (gmail, gmx, ...) are ignored as account signals so a +personal email never fakes a company match. + +## Setup + +1. **Build your CRM first** (so dedup has something to match against): + ```bash + python3 modules/crm-merge/build.py --out data/crm-merge # see that module's README + ``` + You can skip this, but then every lead is `proceed` and you lose the whole + point of the pipeline. + +2. **Define your ICP.** Copy the rubric and edit it for your market: + ```bash + cp modules/lead-pipeline/icp.example.json modules/lead-pipeline/icp.json + ``` + It is plain JSON: target countries, buyer-title keywords, and how many points + each signal is worth. Tune the numbers to your taste. + +3. **Optional, for richer discovery/enrichment**, enable MCP servers via + `bash scripts/install-integrations.sh`: + - **exa** (`search` category) — web/firmographic discovery. Free tier exists. + This is the minimum useful add-on. + - An enrichment provider (e.g. a Clay MCP) — verified emails and + decision-maker names. Optional; uses your own account/credits. + + No MCP at all still works: bring your own raw CSV and the script stages clean, + dedup, and score it. + +## Use it (by hand) + +Starting from a raw CSV with any of these columns (`company`, `domain`, +`contact_name`, `contact_email`, `title`, `country`; extras are preserved): + +```bash +python3 modules/lead-pipeline/pipeline.py clean \ + --in raw.csv --out reports/lead-pipeline/01-cleaned.csv --country DE,BR + +python3 modules/lead-pipeline/pipeline.py dedup \ + --in reports/lead-pipeline/01-cleaned.csv \ + --crm data/crm-merge/contacts.json \ + --out reports/lead-pipeline/02-deduped.csv + +python3 modules/lead-pipeline/pipeline.py score \ + --in reports/lead-pipeline/02-deduped.csv \ + --icp modules/lead-pipeline/icp.json \ + --out reports/lead-pipeline/03-scored.csv --top 50 +``` + +Each stage prints a summary (rows in/out, disposition split, tier distribution) +and writes its CSV to `reports/lead-pipeline/` (git-ignored). The enrich step +goes between `dedup` and `score`: enrich only the `proceed` and `warm` rows, so +you never pay to enrich a `skip`. + +## Or just ask your assistant + +``` +/build-list fintech companies in Germany, 10-200 employees, head of revenue +``` + +The skill runs discovery, calls the script for clean + dedup, enriches the +survivors with whatever MCP you have, scores, and hands you the ranked top-N +plus the disposition breakdown. + +## On enrichment quality (set expectations) + +Enrichment hit-rate varies sharply by market. In well-covered regions (DE, US, +UK, BR, CA) expect a strong majority of `proceed` rows to come back with a +verified email or a named decision-maker. In thinner markets, expect to fall +back to public business registries or manual research for the long tail. Scope +your list to the markets that enrich well first, or budget more time for the +rest. Check your enrichment provider's remaining credits before a bulk run. + +## A note on outbound responsibility + +This module builds a list; it does not send anything. Respect GDPR/CAN-SPAM and +platform terms when you act on it. The `warm` tag exists precisely so your best +accounts get a real, consented intro rather than a cold blast. + +## Pairs with + +- **`crm-merge` module** — produces the `contacts.json` this dedups against. +- **`linkedin-outreach` module** — feed `warm` and high-tier `proceed` rows into + your connection queue. +- **`crm-relationships` agent** — for `warm` rows, ask it who your existing + contact at the account is and how to route the intro. diff --git a/modules/lead-pipeline/icp.example.json b/modules/lead-pipeline/icp.example.json new file mode 100644 index 0000000..b2988d6 --- /dev/null +++ b/modules/lead-pipeline/icp.example.json @@ -0,0 +1,21 @@ +{ + "_comment": "Your Ideal Customer Profile, as a scoring rubric. Copy this file to icp.json and edit it for your market. Every points value is yours to tune; higher total = better fit. The numbers below are a starting point, not gospel.", + + "target_countries": ["DE", "BR"], + "country_points": 20, + + "title_keywords": ["head", "vp", "director", "owner", "founder", "ceo", "managing", "growth", "revenue", "sales"], + "title_points": 30, + + "has_domain_points": 15, + "has_email_points": 20, + + "_size_comment": "Optional. Name a numeric column from your lead CSV (e.g. 'employees' or 'headcount'); rows whose value falls in [size_min, size_max] earn size_points. Remove these four keys if you have no size data.", + "size_field": "employees", + "size_min": 10, + "size_max": 500, + "size_points": 15, + + "_tiers_comment": "Score thresholds for the A/B/C label. A row scores its tier if its fit_score is at least the threshold.", + "tiers": { "A": 70, "B": 45, "C": 0 } +} diff --git a/modules/lead-pipeline/pipeline.py b/modules/lead-pipeline/pipeline.py new file mode 100644 index 0000000..8c599ec --- /dev/null +++ b/modules/lead-pipeline/pipeline.py @@ -0,0 +1,355 @@ +"""Lead-list pipeline: the deterministic stages of an outbound list build. + +The full motion is: discover -> clean -> dedup-against-CRM -> enrich -> score. +Discovery and enrichment are judgement/IO steps your assistant runs with MCP +tools (web search, an enrichment provider, public registries) — see the +/build-list skill. THIS script owns the three deterministic stages, each of +which reads a CSV and writes the next, so the pipeline is debuggable and +resumable: you can inspect (or hand-fix) the output of any stage. + + clean raw candidates -> normalize, drop junk, dedup within the list + dedup against your CRM -> tag each row proceed / warm / skip + score against your ICP -> fit score + tier, write a ranked top-N + +The senior move this encodes: DEDUP AGAINST YOUR CRM BEFORE YOU ENRICH. You +never spend enrichment credits on a company you already know, and you never +cold-email a contact who is already a relationship or an open deal. + +All stdlib. Lead CSVs are free-form; these columns are used when present and +everything else is preserved untouched: + company, domain, contact_name, contact_email, title, country +plus any numeric size column you name in the ICP config. + +Examples: + python3 pipeline.py clean --in raw.csv --out reports/lead-pipeline/01-cleaned.csv --country DE,BR + python3 pipeline.py dedup --in reports/lead-pipeline/01-cleaned.csv --crm data/crm-merge/contacts.json --out reports/lead-pipeline/02-deduped.csv + python3 pipeline.py score --in reports/lead-pipeline/02-deduped.csv --icp modules/lead-pipeline/icp.json --out reports/lead-pipeline/03-scored.csv --top 50 +""" + +from __future__ import annotations + +import argparse +import csv +import json +import re +import sys +import unicodedata +from pathlib import Path + +PROJECT_DIR = Path(__file__).resolve().parents[2] +DEFAULT_REPORTS = PROJECT_DIR / "reports" / "lead-pipeline" + +# Legal-form suffixes stripped when normalizing a company name for matching. +LEGAL_SUFFIXES = { + "gmbh", "ug", "ag", "kg", "ohg", "mbh", "se", # DE + "ltda", "sa", "eireli", "me", "epp", # BR / LATAM + "inc", "llc", "ltd", "corp", "co", "company", "plc", # EN + "bv", "nv", "oy", "ab", "as", "sarl", "srl", "spa", # EU +} + + +def norm_text(s: str) -> str: + if not s: + return "" + s = unicodedata.normalize("NFKD", s).encode("ascii", "ignore").decode("ascii") + return re.sub(r"\s+", " ", s).strip() + + +def norm_company(s: str) -> str: + """Lowercased, punctuation- and legal-suffix-stripped company key.""" + s = norm_text(s).lower() + s = re.sub(r"[^a-z0-9\s]", " ", s) + tokens = [t for t in s.split() if t and t not in LEGAL_SUFFIXES] + return " ".join(tokens) + + +def norm_domain(s: str) -> str: + s = norm_text(s).lower() + s = re.sub(r"^https?://", "", s) + s = re.sub(r"^www\.", "", s) + s = s.split("/")[0].strip() + return s + + +def email_domain(email: str) -> str: + email = norm_text(email).lower() + return email.split("@", 1)[1] if "@" in email else "" + + +def read_csv(path: Path) -> tuple[list[dict], list[str]]: + if not path.exists(): + sys.exit(f"ERROR: input not found: {path}") + with path.open(encoding="utf-8", errors="replace", newline="") as f: + reader = csv.DictReader(f) + rows = list(reader) + fields = reader.fieldnames or [] + return rows, list(fields) + + +def write_csv(rows: list[dict], fields: list[str], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + # Union of declared fields and any keys present on rows, declared order first. + seen = list(fields) + for r in rows: + for k in r: + if k not in seen: + seen.append(k) + with path.open("w", encoding="utf-8", newline="") as f: + w = csv.DictWriter(f, fieldnames=seen, extrasaction="ignore") + w.writeheader() + for r in rows: + w.writerow(r) + print(f" wrote {len(rows)} rows -> {path}") + + +def get(row: dict, *names: str) -> str: + """First non-empty value among case-insensitive column aliases.""" + lower = {k.lower(): v for k, v in row.items()} + for n in names: + v = lower.get(n.lower()) + if v and str(v).strip(): + return str(v).strip() + return "" + + +# ---------- Stage: clean ---------- + +def stage_clean(args) -> None: + rows, fields = read_csv(Path(args.infile)) + countries = {c.strip().upper() for c in args.country.split(",") if c.strip()} if args.country else set() + + kept: list[dict] = [] + seen_keys: set[str] = set() + dropped_empty = dropped_country = dropped_dup = 0 + + for r in rows: + company = get(r, "company", "company name", "account") + domain = norm_domain(get(r, "domain", "website", "url")) + if not domain: + domain = email_domain(get(r, "contact_email", "email")) + # Must have something to identify the account. + if not company and not domain: + dropped_empty += 1 + continue + if countries: + country = get(r, "country", "country_code", "region").upper() + if country and country not in countries: + dropped_country += 1 + continue + # Dedup within the list on domain, falling back to normalized company. + key = f"d:{domain}" if domain else f"c:{norm_company(company)}" + if key in seen_keys: + dropped_dup += 1 + continue + seen_keys.add(key) + if domain: + r["domain"] = domain + kept.append(r) + + out = Path(args.out) if args.out else DEFAULT_REPORTS / "01-cleaned.csv" + print(f"clean: {len(rows)} in -> {len(kept)} kept " + f"(dropped {dropped_empty} empty, {dropped_country} off-country, {dropped_dup} dup)") + write_csv(kept, fields, out) + + +# ---------- Stage: dedup against CRM ---------- + +def load_crm(path: Path) -> dict: + """Build lookup indexes from a crm-merge contacts.json.""" + emails: set[str] = set() + email_domains: dict[str, str] = {} # domain -> a contact name (for the report) + companies: dict[str, str] = {} # normalized company -> contact name + if not path.exists(): + print(f"WARN: CRM file not found ({path}); every lead will be 'proceed'.", + file=sys.stderr) + return {"emails": emails, "email_domains": email_domains, "companies": companies} + data = json.loads(path.read_text(encoding="utf-8")) + for rec in data: + name = rec.get("fn") or "" + for e in rec.get("emails", []) or []: + e = e.lower().strip() + if "@" in e: + emails.add(e) + dom = e.split("@", 1)[1] + # Skip free webmail domains as account signals. + if dom and dom not in FREEMAIL: + email_domains.setdefault(dom, name) + comp = norm_company(rec.get("company") or "") + if comp: + companies.setdefault(comp, name) + return {"emails": emails, "email_domains": email_domains, "companies": companies} + + +FREEMAIL = { + "gmail.com", "googlemail.com", "yahoo.com", "hotmail.com", "outlook.com", + "live.com", "icloud.com", "gmx.de", "gmx.net", "web.de", "aol.com", + "proton.me", "protonmail.com", "t-online.de", +} + + +def stage_dedup(args) -> None: + rows, fields = read_csv(Path(args.infile)) + crm = load_crm(Path(args.crm)) + + counts = {"proceed": 0, "warm": 0, "skip": 0} + for r in rows: + lead_email = get(r, "contact_email", "email").lower() + lead_dom = norm_domain(get(r, "domain", "website", "url")) or email_domain(lead_email) + lead_comp = norm_company(get(r, "company", "company name", "account")) + + disposition, match_on, match_name = "proceed", "", "" + + # 1. Exact person we already know -> skip (do not cold-outreach). + if lead_email and lead_email in crm["emails"]: + disposition, match_on = "skip", "email" + # 2. We know someone at this account (by corporate email domain) -> warm. + elif lead_dom and lead_dom not in FREEMAIL and lead_dom in crm["email_domains"]: + disposition, match_on = "warm", "domain" + match_name = crm["email_domains"][lead_dom] + # 3. We know someone at this account (by company name) -> warm. + elif lead_comp and lead_comp in crm["companies"]: + disposition, match_on = "warm", "company" + match_name = crm["companies"][lead_comp] + + r["disposition"] = disposition + r["crm_match_on"] = match_on + r["crm_known_contact"] = match_name + counts[disposition] += 1 + + out = Path(args.out) if args.out else DEFAULT_REPORTS / "02-deduped.csv" + total = len(rows) or 1 + print(f"dedup against CRM: {len(rows)} leads") + print(f" proceed (cold, no match): {counts['proceed']:>5} " + f"({100*counts['proceed']//total}%)") + print(f" warm (known account, new person):{counts['warm']:>5} " + f"({100*counts['warm']//total}%) -> route via your existing contact") + print(f" skip (already a known contact): {counts['skip']:>5} " + f"({100*counts['skip']//total}%) -> do NOT cold-outreach") + if not (crm["emails"] or crm["companies"]): + print(" (CRM was empty — run modules/crm-merge first to make this stage useful.)") + write_csv(rows, fields, out) + + +# ---------- Stage: score against ICP ---------- + +def load_icp(path: Path) -> dict: + if not path.exists(): + sys.exit(f"ERROR: ICP config not found: {path}. " + f"Copy modules/lead-pipeline/icp.example.json to {path} and edit it.") + return json.loads(path.read_text(encoding="utf-8")) + + +def score_row(r: dict, icp: dict) -> tuple[int, list[str]]: + score, why = 0, [] + + countries = {c.upper() for c in icp.get("target_countries", [])} + if countries: + country = get(r, "country", "country_code", "region").upper() + if country and country in countries: + pts = icp.get("country_points", 0) + score += pts + why.append(f"+{pts} country") + + kws = [k.lower() for k in icp.get("title_keywords", [])] + title = get(r, "title", "position", "role").lower() + if kws and title and any(k in title for k in kws): + pts = icp.get("title_points", 0) + score += pts + why.append(f"+{pts} title") + + if get(r, "domain", "website", "url"): + pts = icp.get("has_domain_points", 0) + score += pts + if pts: + why.append(f"+{pts} domain") + + if get(r, "contact_email", "email"): + pts = icp.get("has_email_points", 0) + score += pts + if pts: + why.append(f"+{pts} email") + + size_field = icp.get("size_field") + if size_field: + raw = get(r, size_field) + digits = re.sub(r"[^\d]", "", raw) + if digits: + n = int(digits) + lo = icp.get("size_min", 0) + hi = icp.get("size_max", 10**9) + if lo <= n <= hi: + pts = icp.get("size_points", 0) + score += pts + why.append(f"+{pts} size") + + return score, why + + +def stage_score(args) -> None: + rows, fields = read_csv(Path(args.infile)) + icp = load_icp(Path(args.icp)) + tiers = icp.get("tiers", {"A": 70, "B": 45, "C": 0}) + + scored = [] + for r in rows: + # Never rank a contact we should not cold-outreach. + if r.get("disposition") == "skip": + r["fit_score"] = 0 + r["fit_tier"] = "skip" + r["fit_reasons"] = "already a known contact" + scored.append(r) + continue + s, why = score_row(r, icp) + tier = next((t for t, thr in sorted(tiers.items(), key=lambda kv: -kv[1]) if s >= thr), "C") + r["fit_score"] = s + r["fit_tier"] = tier + r["fit_reasons"] = ", ".join(why) + scored.append(r) + + scored.sort(key=lambda r: -int(r.get("fit_score") or 0)) + + out = Path(args.out) if args.out else DEFAULT_REPORTS / "03-scored.csv" + write_csv(scored, fields, out) + + dist: dict[str, int] = {} + for r in scored: + dist[r["fit_tier"]] = dist.get(r["fit_tier"], 0) + 1 + print("score: tier distribution " + ", ".join(f"{t}={dist[t]}" for t in sorted(dist))) + + if args.top: + actionable = [r for r in scored if r.get("disposition") != "skip"][: args.top] + top_path = out.with_name(out.stem.replace("03-scored", "04-top") + ".csv") \ + if "03-scored" in out.stem else out.with_name(f"top-{args.top}.csv") + write_csv(actionable, fields, top_path) + print(f" top {len(actionable)} actionable leads -> {top_path}") + + +def main(): + ap = argparse.ArgumentParser(description="Deterministic stages of a lead-list build.") + sub = ap.add_subparsers(dest="cmd", required=True) + + c = sub.add_parser("clean", help="normalize, drop junk, dedup within the list") + c.add_argument("--in", dest="infile", required=True) + c.add_argument("--out") + c.add_argument("--country", default="", help="allowlist, comma-separated (e.g. DE,BR)") + c.set_defaults(func=stage_clean) + + d = sub.add_parser("dedup", help="tag each lead proceed/warm/skip vs your CRM") + d.add_argument("--in", dest="infile", required=True) + d.add_argument("--crm", default=str(PROJECT_DIR / "data" / "crm-merge" / "contacts.json")) + d.add_argument("--out") + d.set_defaults(func=stage_dedup) + + s = sub.add_parser("score", help="ICP fit score + tier + ranked top-N") + s.add_argument("--in", dest="infile", required=True) + s.add_argument("--icp", default=str(Path(__file__).parent / "icp.json")) + s.add_argument("--out") + s.add_argument("--top", type=int, default=50) + s.set_defaults(func=stage_score) + + args = ap.parse_args() + args.func(args) + + +if __name__ == "__main__": + main()