"""Leads — people and companies not yet worked into a deal: how good each one is, and who they resemble.
THE LANGUAGE (the same words in classes, use cases, docstrings and the history):
lead a person (keyed by email) or a company (keyed by web domain) not yet worked into a deal.
The AGGREGATE ROOT: `Lead`. Its key is a value object, `LeadKey`, derived — never minted.
signal one engagement event (opened the pricing email, asked for a demo): `Signal`, a value object.
scoring policy the business's own weights for fit and signals, and the score that qualifies: a POLICY kept
as config rows (`ScoringPolicy` of `Rule`s) and evaluated here, never in SQL.
score 0–100, DERIVED from the lead, its signals and the policy by the domain service `score()`;
`Score` is the value object with its breakdown (fit, engagement, reasons).
qualification working → qualified, when the policy says the score qualifies (`ScoringPolicy.qualifies`).
disqualification new | working → disqualified, with a reason.
conversion a qualified lead becoming a partner + contact + deal; recorded step by step on the lead,
terminal once complete. The neighbours are reached through application/neighbours.py.
lookalike a lead company whose `ProfileVector` is near a won account's.
Domain events (past tense, raised by the aggregate, written to lead_events with the lead): Added,
Imported, Reimported, Described, WorkStarted, Qualified, Disqualified, Assigned, SignalRecorded,
ConversionStepRecorded, Converted.
No database, no HTTP, no identity. Held to the CRM spec (modules/crm/module.yaml: CRM-03, 10, 13, 14; the
`lead` workflow; the entities `lead` and `lead_score`; the config row `scoring_policy`):
* THE ID IS DERIVED from the normalised email (a person) or web domain (a company). Importing the same
trade-show list twice lands on the same rows, so "each person exists once" (CRM-13) is a property of the
key, not of a lookup someone can forget. A re-import fills in what a row says and never blanks what it
leaves out, never changes who owns a lead, and never moves it back through the workflow.
* THE WORKFLOW IS CODE, THE WEIGHTS ARE ROWS. Which transitions exist (new → working → qualified |
disqualified → converted) and that disqualifying needs a reason are rules; how many points "software" or
"opened the pricing email" is worth, and the score that qualifies, are the customer's (`ScoringPolicy`).
* THE SCORE IS DERIVED from the lead's attributes, its engagement signals and the policy — never typed. The
same inputs give the same score, which is what makes a re-scoring run that is repeated change nothing.
* A PROFILE for lookalikes (CRM-14) is a deterministic vector of the company's fit attributes — industry,
size band, region — so similarity is explainable ("same industry, one size band up") and costs nothing to
compute. An embedding of free text from a language model would find subtler likeness, but it needs the
billed `ignite.model-user` role and a model call per lead; it is a later option, not this one.
"""
from __future__ import annotations
import csv
import hashlib
import io
import math
import re
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Iterable, Optional
STATUSES = ("new", "working", "qualified", "disqualified", "converted")
OPEN = ("new", "working", "qualified") # re-scored, and lookalike candidates
KINDS = ("person", "company")
class LeadRuleBroken(Exception):
"""A lead rule said no. → 409."""
class LeadIncomplete(LeadRuleBroken):
"""Something has to be supplied first — a reason, an email, a domain. → 422."""
# ── normalising what people type ───────────────────────────────────────────────────────────
_EMAIL = re.compile(r"^[^@\s,;<>]+@[^@\s,;<>]+\.[^@\s,;<>]+$")
# A person's address at one of these says nothing about the company they work for, so it is not taken as
# the company's domain (a gmail.com "company" would make every gmail lead look alike and convert into one
# partner).
FREE_MAIL = frozenset({
"gmail.com", "googlemail.com", "yahoo.com", "yahoo.co.uk", "hotmail.com", "hotmail.co.uk", "outlook.com",
"live.com", "msn.com", "icloud.com", "me.com", "aol.com", "proton.me", "protonmail.com", "gmx.de",
"gmx.net", "web.de", "mail.com", "yandex.com", "yandex.ru", "zoho.com", "hey.com"})
@dataclass(frozen=True, slots=True)
class Email:
"""A value object: an address in the one form it is compared and stored in."""
value: str
def __post_init__(self):
object.__setattr__(self, "value", normalise_email(self.value))
@property
def domain(self) -> str:
return self.value.rsplit("@", 1)[1]
@dataclass(frozen=True, slots=True)
class Domain:
"""A value object: a company's web domain, `https://www.Acme.com/about` → `acme.com`."""
value: str
def __post_init__(self):
object.__setattr__(self, "value", normalise_domain(self.value))
@dataclass(frozen=True, slots=True)
class LeadKey:
"""A value object: what a lead IS, and so its id. A person by email, a company by domain — derived, so the
same person imported twice is the same lead (CRM-13)."""
kind: str # person | company
value: str # the normalised email or domain
@classmethod
def of(cls, email: str = "", domain: str = "") -> "LeadKey":
if (email or "").strip():
return cls("person", Email(email).value)
if (domain or "").strip():
return cls("company", Domain(domain).value)
raise LeadIncomplete("a lead needs an email (a person) or a web domain (a company)")
@property
def lead_id(self) -> str:
prefix = "email:" if self.kind == "person" else "domain:"
return "L-" + hashlib.blake2b((prefix + self.value).encode(), digest_size=8).hexdigest().upper()
def normalise_email(email: str) -> str:
"""Lower-case and trimmed. One `@`, a dot after it, no spaces — at least as strict as the activity brick,
so a lead that imports is a lead that can become a contact."""
e = (email or "").strip().lower()
if e.startswith("mailto:"):
e = e[7:]
if not _EMAIL.match(e):
raise LeadIncomplete(f"{email!r} is not an email address")
return e
def normalise_domain(domain: str) -> str:
"""`https://www.Acme.com/about` → `acme.com`. The company's web domain is its key and, on conversion,
its business-partner key, so `Acme.com` and `acme.com` must be one company."""
d = (domain or "").strip().lower()
if "@" in d:
d = d.rsplit("@", 1)[1]
d = re.sub(r"^[a-z][a-z0-9+.-]*://", "", d)
d = d.split("/", 1)[0].split("?", 1)[0].split(":", 1)[0].strip(".")
if d.startswith("www."):
d = d[4:]
if not re.match(r"^[a-z0-9-]+(\.[a-z0-9-]+)+$", d):
raise LeadIncomplete(f"{domain!r} is not a web domain")
return d
def company_domain(email: str, domain: str = "") -> str:
"""The company's domain: the one given, else the email's unless it is a free-mail address; '' if none."""
if (domain or "").strip():
return normalise_domain(domain)
if email:
d = Email(email).domain
return "" if d in FREE_MAIL else d
return ""
def lead_id_for(email: str = "", domain: str = "") -> str:
"""The lead's id, derived — a person by email, a company by domain — so a re-import is idempotent."""
return LeadKey.of(email, domain).lead_id
INDUSTRIES = ("software", "manufacturing", "finserv", "retail", "healthcare", "media", "real-estate",
"education", "public-sector", "professional-services", "logistics", "energy", "other")
_INDUSTRY_ALIASES = {
"saas": "software", "tech": "software", "technology": "software", "it": "software", "internet": "software",
"financial services": "finserv", "finance": "finserv", "banking": "finserv", "insurance": "finserv",
"fintech": "finserv", "real estate": "real-estate", "property": "real-estate", "proptech": "real-estate",
"government": "public-sector", "public sector": "public-sector", "public": "public-sector",
"consulting": "professional-services", "legal": "professional-services", "accounting": "professional-services",
"professional services": "professional-services", "transport": "logistics", "shipping": "logistics",
"health": "healthcare", "pharma": "healthcare", "publishing": "media", "e-commerce": "retail",
"ecommerce": "retail", "industrial": "manufacturing", "utilities": "energy"}
def normalise_industry(v: str) -> str:
"""One of INDUSTRIES; anything unrecognised is 'other' (so it can still be weighted); '' if blank."""
s = re.sub(r"\s+", " ", (v or "").strip().lower().replace("_", " "))
if not s:
return ""
s = _INDUSTRY_ALIASES.get(s, s.replace(" ", "-"))
return s if s in INDUSTRIES else "other"
SIZE_BANDS = ("1-10", "11-50", "51-200", "201-1000", "1001-5000", "5000+")
_SIZE_FLOORS = (1, 11, 51, 201, 1001, 5001)
def normalise_size(v: str) -> str:
"""A size band (employees). Accepts a band as written, '5000+', or a head count ('250', '250 staff')."""
s = (v or "").strip().lower().replace("–", "-").replace(" ", "")
if not s:
return ""
if s in SIZE_BANDS:
return s
if s in (">5000", "5001+", "5000plus"):
return "5000+"
m = re.match(r"^(\d[\d,]*)", s)
if not m or "-" in s:
raise LeadIncomplete(f"{v!r} is not a size: use {', '.join(SIZE_BANDS)} or a head count")
n = int(m.group(1).replace(",", ""))
return next(b for b, f in reversed(list(zip(SIZE_BANDS, _SIZE_FLOORS))) if n >= f) if n >= 1 else "1-10"
_COUNTRY_ALIASES = {"uk": "GB", "england": "GB", "scotland": "GB", "wales": "GB", "united kingdom": "GB",
"great britain": "GB", "usa": "US", "united states": "US", "america": "US",
"germany": "DE", "deutschland": "DE", "sweden": "SE", "sverige": "SE", "norway": "NO",
"denmark": "DK", "finland": "FI", "france": "FR", "spain": "ES", "italy": "IT",
"netherlands": "NL", "holland": "NL", "ireland": "IE", "canada": "CA", "india": "IN",
"australia": "AU", "japan": "JP", "switzerland": "CH", "austria": "AT", "belgium": "BE"}
def normalise_country(v: str) -> str:
"""ISO 3166 alpha-2, upper-case ('uk' → 'GB'); '' if blank."""
s = (v or "").strip()
if not s:
return ""
if s.lower() in _COUNTRY_ALIASES:
return _COUNTRY_ALIASES[s.lower()]
if re.match(r"^[A-Za-z]{2}$", s):
return s.upper()
raise LeadIncomplete(f"{v!r} is not a country: use the two-letter code (GB, SE, DE, US)")
REGIONS = ("uk-ie", "nordics", "dach", "benelux", "south-eu", "east-eu", "north-america", "latam", "apac",
"mea", "other")
_REGION_OF = {
**dict.fromkeys(("GB", "IE"), "uk-ie"), **dict.fromkeys(("SE", "NO", "DK", "FI", "IS"), "nordics"),
**dict.fromkeys(("DE", "AT", "CH", "LI"), "dach"), **dict.fromkeys(("NL", "BE", "LU"), "benelux"),
**dict.fromkeys(("FR", "ES", "IT", "PT", "GR", "MT", "CY", "MC"), "south-eu"),
**dict.fromkeys(("PL", "CZ", "SK", "HU", "RO", "BG", "HR", "SI", "EE", "LV", "LT", "UA", "RS"), "east-eu"),
**dict.fromkeys(("US", "CA"), "north-america"),
**dict.fromkeys(("MX", "BR", "AR", "CL", "CO", "PE", "UY"), "latam"),
**dict.fromkeys(("JP", "CN", "IN", "SG", "AU", "NZ", "KR", "HK", "TW", "ID", "MY", "TH", "VN", "PH"), "apac"),
**dict.fromkeys(("AE", "SA", "IL", "TR", "ZA", "EG", "NG", "KE", "MA", "QA"), "mea")}
def region_of(country: str) -> str:
return _REGION_OF.get((country or "").upper(), "other") if country else ""
# ── domain events: what happened to a lead, in the past tense ──────────────────────────────
@dataclass(frozen=True, slots=True)
class LeadEvent:
"""Raised by the aggregate as it changes; the repository writes it to the lead's history in the same
transaction as the change, so the history cannot say something the lead does not."""
detail: str = ""
@property
def name(self) -> str: # Qualified → "qualified", WorkStarted → "work started"
return re.sub(r"(?<!^)(?=[A-Z])", " ", type(self).__name__).lower()
class Added(LeadEvent): pass
class Imported(LeadEvent): pass
class Reimported(LeadEvent): pass
class Described(LeadEvent): pass
class WorkStarted(LeadEvent): pass
class Qualified(LeadEvent): pass
class Disqualified(LeadEvent): pass
class Assigned(LeadEvent): pass
class SignalRecorded(LeadEvent): pass
class ConversionStepRecorded(LeadEvent): pass
class Converted(LeadEvent): pass
# ── the lead: the aggregate root ───────────────────────────────────────────────────────────
DESCRIPTIVE = ("name", "company", "domain", "title", "country", "industry", "size_band", "source")
@dataclass(slots=True)
class Lead:
lead_id: str
kind: str # person (keyed by email) | company (keyed by domain)
name: str
email: str # '' for a company lead
company: str
domain: str # '' when unknown, or the person uses a free-mail address
owner: str # the rep's email: row visibility reads it
title: str = ""
country: str = ""
industry: str = ""
size_band: str = ""
source: str = ""
status: str = "new"
disqualify_reason: str = ""
bp_id: Optional[int] = None # set by conversion, step by step (CRM-03) — modules/shared.yaml: BIGINT
contact_id: Optional[str] = None
opportunity_id: Optional[str] = None
converted_at: Optional[str] = None
converted_by: str = ""
version: int = 0 # 0.2.0: saves so far — two people at once, one wins (409)
events: list = field(default_factory=list, repr=False, compare=False) # raised, not yet written
@classmethod
def new(cls, owner: str, *, email: str = "", domain: str = "", name: str = "", company: str = "",
title: str = "", country: str = "", industry: str = "", size_band: str = "",
source: str = "", imported: bool = False) -> "Lead":
email = Email(email).value if (email or "").strip() else ""
key = LeadKey.of(email, company_domain(email, domain))
dom = company_domain(email, domain)
lead = cls(key.lead_id, key.kind, "", email, "", dom, owner)
lead.describe(name=name, company=company, title=title, country=country, industry=industry,
size_band=size_band, source=source)
lead.events.clear() # describing a lead being born is part of its birth
if not lead.name:
lead.name = lead.company or email or dom
lead.events.append((Imported if imported else Added)(f"{lead.kind} lead from {lead.source or 'manual'}"))
return lead
def _raise(self, event: LeadEvent) -> None:
self.events.append(event)
@property
def is_open(self) -> bool:
return self.status in OPEN
@property
def region(self) -> str:
return region_of(self.country)
def describe(self, *, reimport: bool = False, **fields) -> list:
"""Fill in or correct what the lead says about itself. A value left out or blank changes nothing — a
re-import of a sparser list must not erase what a rep typed. Returns the fields that changed."""
if self.status == "converted":
raise LeadRuleBroken("this lead is converted: its company, contact and deal are the record now — "
"change them there")
if self.bp_id is not None or self.contact_id:
# Half-converted (0.2.0): the partner was made from this domain and the contact from this email, and
# converting again finishes the rest FROM them. A new domain would make the deal another company's.
for k in ("domain",):
v = fields.get(k)
if v is not None and str(v).strip() and normalise_domain(str(v)) != self.domain:
raise LeadRuleBroken(f"this lead is being converted — its partner was made from {self.domain}; "
f"finish converting it (or change the partner) rather than its domain")
clean = {"name": lambda v: v.strip(), "company": lambda v: v.strip(), "title": lambda v: v.strip(),
"source": lambda v: v.strip().lower(), "domain": normalise_domain, "country": normalise_country,
"industry": normalise_industry, "size_band": normalise_size}
changed = []
for k, v in fields.items():
if k not in DESCRIPTIVE:
raise ValueError(f"{k} is not something a lead describes")
if v is None or not str(v).strip():
continue
if k == "domain" and self.kind == "company":
if normalise_domain(v) != self.domain:
raise LeadRuleBroken(f"a company lead is keyed by its domain {self.domain}; a different "
"domain is a different lead")
continue
new = clean[k](str(v))
if new and new != getattr(self, k):
setattr(self, k, new)
changed.append(k)
if changed:
self._raise((Reimported if reimport else Described)(", ".join(changed)))
return changed
# ── the workflow (spec: workflows.lead) ──────────────────────────────────────────────
def start_working(self) -> bool:
"""new → working. True when it changed; working already is no change."""
if self.status == "working":
return False
if self.status != "new":
raise LeadRuleBroken(f"a {self.status} lead cannot go back to working")
self.status = "working"
self._raise(WorkStarted())
return True
def qualify(self, s: "Score", policy: "ScoringPolicy", by: str) -> bool:
"""working → qualified, when the POLICY says the score qualifies — by a rep or by the scoring run. The
gate takes a `Score`, which only the domain service `score()` makes, so 'qualified' means the same
thing whoever decides it and no caller can hand in a number of its own."""
if self.status == "qualified":
return False
if self.status != "working":
raise LeadRuleBroken(f"only a lead being worked can be qualified; this one is {self.status}"
+ (" — start working it first" if self.status == "new" else ""))
if not policy.qualifies(s):
raise LeadRuleBroken(f"the score is {s.score}, below the {policy.qualify_threshold} that qualifies a lead")
self.status = "qualified"
self._raise(Qualified(f"by {by}: score {s.score} ≥ {policy.qualify_threshold}"
+ (f" ({'; '.join(s.reasons)})" if s.reasons else "")))
return True
def disqualify(self, reason: str) -> bool:
"""new | working → disqualified, with a reason (so 'why do we lose leads?' can be answered)."""
why = (reason or "").strip()
if not why:
raise LeadIncomplete("disqualifying a lead needs a reason")
if self.status == "disqualified" and self.disqualify_reason == why:
return False
if self.status not in ("new", "working"):
raise LeadRuleBroken(f"a {self.status} lead cannot be disqualified")
self.status, self.disqualify_reason = "disqualified", why
self._raise(Disqualified(why))
return True
def assign(self, owner: str) -> bool:
if not self.is_open:
raise LeadRuleBroken(f"a {self.status} lead is closed; its owner is history")
if not (owner or "").strip() or "@" not in owner:
raise LeadIncomplete("a lead's owner is a person's email")
if owner.strip().lower() == self.owner:
return False
before, self.owner = self.owner, owner.strip().lower()
self._raise(Assigned(f"{before} → {self.owner}"))
return True
def engaged(self, signal: "Signal", new: bool) -> None:
"""A signal was recorded for this lead (it lives in its own table: signals arrive far more often than
the lead changes). A re-delivered one raises nothing."""
if new:
self._raise(SignalRecorded(f"{signal.kind} at {signal.at.isoformat()}"))
# ── conversion (CRM-03), recorded step by step ───────────────────────────────────────
def ready_to_convert(self) -> None:
"""What converting needs from the lead itself, checked before any neighbour is called."""
if self.status != "qualified":
raise LeadRuleBroken(f"only a qualified lead is converted; this one is {self.status}")
if not self.domain:
raise LeadIncomplete("converting needs the company's web domain — it is the business partner's key "
"(add it with edit_lead)")
if not self.email:
raise LeadIncomplete("converting needs the person's email — it is the contact's key")
def record_step(self, *, bp_id: Optional[int] = None, contact_id: Optional[str] = None,
opportunity_id: Optional[str] = None, at: Optional[str] = None, by: str = "",
detail: str = "") -> bool:
"""Keep what a conversion step produced. A lead with all three is converted; one with fewer is still
qualified and converting again finishes the rest. True when the lead became converted."""
if self.status == "converted":
return False
if self.status != "qualified":
raise LeadRuleBroken(f"only a qualified lead is converted; this one is {self.status}")
for k, v in (("bp_id", bp_id), ("contact_id", contact_id), ("opportunity_id", opportunity_id)):
if v is None:
continue
v = int(v) if k == "bp_id" else str(v)
if getattr(self, k) not in (None, v):
raise LeadRuleBroken(f"this lead's {k} is already {getattr(self, k)}, not {v}")
setattr(self, k, v)
if self.bp_id is not None and self.contact_id and self.opportunity_id:
self.status, self.converted_at, self.converted_by = "converted", at, by
self._raise(Converted(detail))
return True
self._raise(ConversionStepRecorded(detail))
return False
# ── CRM-10: the scoring policy, and the score it derives ───────────────────────────────────
FIT_ATTRIBUTES = ("industry", "size_band", "country", "region", "source")
_SIGNAL = re.compile(r"^[a-z][a-z0-9_]{1,62}$")
def normalise_signal(kind: str) -> str:
k = (kind or "").strip().lower().replace("-", "_").replace(" ", "_")
if not _SIGNAL.match(k):
raise LeadIncomplete(f"{kind!r} is not a signal name (lower_snake_case, e.g. email_opened_pricing)")
return k
def normalise_fit_value(attribute: str, value: str) -> str:
if attribute not in FIT_ATTRIBUTES:
raise LeadIncomplete(f"a fit rule weighs one of {', '.join(FIT_ATTRIBUTES)}, not {attribute!r}")
v = {"industry": normalise_industry, "size_band": normalise_size, "country": normalise_country,
"source": lambda s: s.strip().lower(), "region": lambda s: s.strip().lower()}[attribute](value)
if attribute == "region" and v not in REGIONS:
raise LeadIncomplete(f"a region is one of {', '.join(REGIONS)}, not {value!r}")
if not v:
raise LeadIncomplete(f"a fit rule on {attribute} needs a value")
return v
@dataclass(frozen=True, slots=True)
class Rule:
"""One weight. A FIT rule scores an attribute value (industry = software: +20); a SIGNAL rule scores each
engagement event of a kind inside the window, up to `max_count` of them (email_opened_pricing: +25, ×2)."""
kind: str # fit | signal
attribute: str # fit: one of FIT_ATTRIBUTES; signal: the signal's name
value: str # fit: the attribute's value; signal: ''
points: int
max_count: int = 1
def __post_init__(self):
if self.kind not in ("fit", "signal"):
raise LeadIncomplete("a scoring rule is a fit rule or a signal rule")
if not -100 <= int(self.points) <= 100:
raise LeadIncomplete("a rule is worth between -100 and 100 points")
if not 1 <= int(self.max_count) <= 20:
raise LeadIncomplete("a signal counts between 1 and 20 times")
@dataclass(frozen=True, slots=True)
class ScoringPolicy:
"""The customer's way of qualifying a lead, as rows (config_rows: scoring_policy). The defaults are a
starting point for a business that has not decided yet: a few engagement signals and size, nothing about
industry or country, which only the customer knows."""
qualify_threshold: int = 60
signal_window_days: int = 30
rules: tuple = ()
def __post_init__(self):
if not 0 <= int(self.qualify_threshold) <= 100:
raise LeadIncomplete("the qualify threshold is a score from 0 to 100")
if not 1 <= int(self.signal_window_days) <= 365:
raise LeadIncomplete("the signal window is 1 to 365 days")
keys = [(r.kind, r.attribute, r.value) for r in self.rules]
if len(set(keys)) != len(keys):
raise LeadIncomplete("two rules weigh the same thing")
def qualifies(self, s: "Score") -> bool:
"""Whether a score qualifies a lead under this policy — the one place that comparison is made."""
return s.score >= self.qualify_threshold
def weighs(self, signal_kind: str) -> bool:
return any(r.kind == "signal" and r.attribute == signal_kind for r in self.rules)
DEFAULT_POLICY = ScoringPolicy(60, 30, (
Rule("fit", "size_band", "51-200", 10), Rule("fit", "size_band", "201-1000", 15),
Rule("fit", "size_band", "1001-5000", 15), Rule("fit", "size_band", "5000+", 10),
Rule("signal", "email_opened", "", 5, 3), Rule("signal", "email_clicked", "", 10, 3),
Rule("signal", "email_opened_pricing", "", 25, 2), Rule("signal", "form_submitted", "", 15, 2),
Rule("signal", "webinar_attended", "", 20, 1), Rule("signal", "demo_requested", "", 40, 1)))
@dataclass(frozen=True, slots=True)
class Score:
score: int # 0–100
fit_points: int
engagement_points: int
reasons: tuple # ("size_band=201-1000 +15", "email_opened_pricing ×1 +25")
@dataclass(frozen=True, slots=True)
class Signal:
"""A value object: one engagement event. Its id is derived from what and when, so a re-delivered event
(a retried webhook) is the same signal."""
kind: str
at: datetime # aware, UTC
def __post_init__(self):
object.__setattr__(self, "kind", normalise_signal(self.kind))
if self.at.tzinfo is None:
raise LeadIncomplete("a signal's time needs a zone")
@property
def signal_id(self) -> str:
return signal_id_for(self.kind, self.at)
def score(lead: Lead, signals: Iterable[tuple], policy: ScoringPolicy, as_of: datetime) -> Score:
"""THE SCORING DOMAIN SERVICE: the lead's score, derived — never typed. A pure function of the lead, its
signals and the policy; nothing else (no stored score, no clock) goes in.
score = clamp(0, 100, Σ fit rules whose value the lead has
+ Σ signal rules: points × min(events of that kind in the window, max_count))
The window is (as_of − signal_window_days, as_of]: an email opened last quarter does not make a lead hot
today, and an event dated after `as_of` has not happened yet as far as this run knows. `signals` is
(kind, aware datetime) pairs. Same lead, signals, policy and as_of → the same score, every time.
"""
have = {"industry": lead.industry, "size_band": lead.size_band, "country": lead.country,
"region": lead.region, "source": lead.source}
fit, eng, reasons = 0, 0, []
for r in policy.rules:
if r.kind == "fit" and have.get(r.attribute) == r.value:
fit += r.points
reasons.append(f"{r.attribute}={r.value} {r.points:+d}")
since = as_of - timedelta(days=policy.signal_window_days)
counts: dict = {}
for kind, at in signals:
if since < at <= as_of:
counts[kind] = counts.get(kind, 0) + 1
for r in policy.rules:
n = min(counts.get(r.attribute, 0), r.max_count) if r.kind == "signal" else 0
if n:
eng += r.points * n
reasons.append(f"{r.attribute} ×{n} {r.points * n:+d}")
return Score(max(0, min(100, fit + eng)), fit, eng, tuple(reasons))
def signal_id_for(kind: str, at: datetime) -> str:
"""One event, one row: the same signal reported twice (a retried webhook) is recorded once."""
return hashlib.blake2b(f"{kind}|{at.isoformat()}".encode(), digest_size=8).hexdigest()
# ── CRM-14: the company's profile, for lookalikes ──────────────────────────────────────────
DIM = len(INDUSTRIES) + len(SIZE_BANDS) + len(REGIONS) + 1 # 31 — the VECTOR(31) column
W_INDUSTRY, W_SIZE, W_SIZE_NEXT, W_REGION, BIAS = 1.0, 0.8, 0.4, 0.6, 0.1
@dataclass(frozen=True, slots=True)
class ProfileVector:
"""A value object: a company's fit, as the unit vector lookalikes compare (see profile())."""
values: tuple
@classmethod
def of(cls, industry: str, size_band: str, country: str) -> "ProfileVector":
return cls(tuple(profile(industry, size_band, country)))
@classmethod
def of_lead(cls, lead: "Lead") -> "ProfileVector":
return cls.of(lead.industry, lead.size_band, lead.country)
def distance(self, other: "ProfileVector") -> float:
"""Cosine distance — what pgvector's `<=>` computes in the database."""
return cosine_distance(list(self.values), list(other.values))
def profile(industry: str, size_band: str, country: str) -> list:
"""A unit vector of the company's fit attributes: industry one-hot, size band one-hot with its
neighbouring bands at half weight (so 51-200 is nearer 201-1000 than 1-10), region one-hot, and a small
constant so that a company we know nothing about is still a vector (cosine of a zero vector is
undefined). Industry weighs most, then size, then region — "a software firm of our size" is the usual
meaning of "like this one". An unknown attribute contributes nothing rather than guessing."""
v = [0.0] * DIM
ind = normalise_industry(industry)
if ind:
v[INDUSTRIES.index(ind)] = W_INDUSTRY
size = normalise_size(size_band) if size_band else ""
if size:
i, base = SIZE_BANDS.index(size), len(INDUSTRIES)
v[base + i] = W_SIZE
for j in (i - 1, i + 1):
if 0 <= j < len(SIZE_BANDS):
v[base + j] = W_SIZE_NEXT
reg = region_of(normalise_country(country)) if country else ""
if reg:
v[len(INDUSTRIES) + len(SIZE_BANDS) + REGIONS.index(reg)] = W_REGION
v[-1] = BIAS
norm = math.sqrt(sum(x * x for x in v))
return [round(x / norm, 6) for x in v]
def cosine_distance(a: list, b: list) -> float:
"""1 − cosine similarity: what pgvector's `<=>` returns, for a database that has no `<=>`."""
dot = sum(x * y for x, y in zip(a, b))
na, nb = math.sqrt(sum(x * x for x in a)), math.sqrt(sum(y * y for y in b))
return 1.0 - dot / (na * nb) if na and nb else 1.0
# ── CRM-13: a spreadsheet of contacts → an import plan ─────────────────────────────────────
MAX_ROWS = 5000
_COLUMNS = {
"name": ("name", "full name", "full_name", "contact", "contact name", "person"),
"email": ("email", "e-mail", "email address", "e-mail address", "mail"),
"company": ("company", "company name", "organisation", "organization", "account", "employer"),
"domain": ("domain", "website", "web", "url", "company domain", "company website"),
"title": ("title", "job title", "role", "position"),
"country": ("country", "country code"),
"industry": ("industry", "sector", "vertical"),
"size_band": ("size", "size band", "size_band", "employees", "company size", "headcount"),
"source": ("source", "lead source", "list"),
"owner": ("owner", "rep", "assigned to", "owner email"),
}
_ALIAS = {a: col for col, names in _COLUMNS.items() for a in names}
@dataclass
class ImportPlan:
"""What an import will do — computed before anything is written, so the counts are the truth."""
create: list = field(default_factory=list) # new Lead objects
update: list = field(default_factory=list) # (Lead, [changed fields])
unchanged: list = field(default_factory=list) # lead ids
skipped: list = field(default_factory=list) # {"row", "reason"}
warnings: list = field(default_factory=list) # {"row", "warning"} — imported, but a field was dropped
ignored_columns: list = field(default_factory=list)
def read_csv(text: str) -> tuple:
"""(header → field mapping, ignored headers, [(row number, {field: value})]).
The first line is the header, matched case-insensitively against the names above (so a CRM export, a
LinkedIn list and a hand-made sheet all work); unknown columns are ignored and reported. Comma, semicolon
(a European Excel export) or tab, detected from the header. Row numbers are SPREADSHEET row numbers —
the header is row 1 — so a marketer can find the row the report names."""
if text.startswith(""):
text = text[1:] # Excel's UTF-8 byte-order mark
first = text.split("\n", 1)[0]
delimiter = max((",", ";", "\t"), key=first.count)
reader = csv.reader(io.StringIO(text), delimiter=delimiter)
try:
header = next(reader)
except StopIteration:
raise LeadIncomplete("the file is empty — the first row must name the columns (name, email, company)")
except csv.Error as e:
raise LeadIncomplete(f"row 1 could not be read as CSV ({e})") from None
mapping, ignored = {}, []
for i, h in enumerate(header):
col = _ALIAS.get(re.sub(r"\s+", " ", h.strip().lower()))
if col and col not in mapping.values():
mapping[i] = col
elif h.strip():
ignored.append(h.strip())
if "email" not in mapping.values():
raise LeadIncomplete("the first row must name an email column — each person is one lead per email; "
f"found {', '.join(h for h in header if h.strip()) or 'nothing'}")
rows = []
n = 1
while True:
n += 1
try:
cells = next(reader)
except StopIteration:
break
except csv.Error as e:
# A broken file is the importer's to fix, with the row to look at: a 500 told nobody anything (H12).
raise LeadIncomplete(f"row {n} could not be read as CSV ({e}) — fix that row and import again") from None
if not any(c.strip() for c in cells):
continue # a blank line is not a row anyone meant
rows.append((n, {col: (cells[i].strip() if i < len(cells) else "") for i, col in mapping.items()}))
if len(rows) > MAX_ROWS:
raise LeadIncomplete(f"more than {MAX_ROWS} rows — split the file")
return mapping, ignored, rows
def plan_import(text: str, existing: dict, default_owner: str, source: str = "import") -> ImportPlan:
"""CRM-13: each person exists once, however many times the list is imported.
`existing` is {lead_id: Lead} for the ids this file names (the caller looks them up first). A row with no
email, or one that is not an address, is SKIPPED and reported with its row number — never dropped in
silence. A second row for the same email in one file is skipped as a duplicate of the first. A row for a
person who is already a lead UPDATES what it says (blank cells change nothing) and never their owner or
status, nor its source (where it first came from — a second list does not re-attribute it); a converted
or disqualified lead is left as it is. A bad optional cell (a size of 'lots') drops
that one field with a warning rather than losing the person.
"""
_, ignored, rows = read_csv(text)
plan = ImportPlan(ignored_columns=ignored)
seen: dict = {}
for n, row in rows:
raw = row.get("email", "")
if not raw:
plan.skipped.append({"row": n, "reason": "no email"})
continue
try:
email = normalise_email(raw)
except LeadIncomplete as e:
plan.skipped.append({"row": n, "reason": str(e)})
continue
if email in seen:
plan.skipped.append({"row": n, "reason": f"same email as row {seen[email]}"})
continue
seen[email] = n
fields = {k: row.get(k, "") for k in DESCRIPTIVE}
fields["source"] = fields["source"] or source
for k in ("domain", "country", "size_band"):
try:
if fields[k]:
{"domain": normalise_domain, "country": normalise_country, "size_band": normalise_size}[k](fields[k])
except LeadIncomplete as e:
plan.warnings.append({"row": n, "warning": f"{k} not imported: {e}"})
fields[k] = ""
lead_id = lead_id_for(email)
lead = existing.get(lead_id)
if lead is None:
owner = (row.get("owner") or "").strip().lower()
if owner and "@" not in owner:
plan.warnings.append({"row": n, "warning": f"owner {owner!r} is not an email; owned by the importer"})
owner = ""
plan.create.append(Lead.new(owner or default_owner, email=email, imported=True, **fields))
elif not lead.is_open:
plan.unchanged.append(lead_id)
else:
if lead.domain:
fields["domain"] = "" # a person's company domain is settled once known
fields["source"] = "" # where the lead FIRST came from: attribution is not re-imported
changed = lead.describe(reimport=True, **fields)
if changed:
plan.update.append((lead, changed))
else:
plan.unchanged.append(lead_id)
return plan