"""The use cases — what reps, marketing, managers, other bricks and their agents can do with leads.
Each is declared once and served as an API route AND an MCP tool.
WHO (CRM spec, actors rep / manager / marketing / system; CRM-01): a rep sees and works their own leads; a
manager or analyst sees every lead, and a manager may change any. MARKETING sees every lead too — the spec's
marketing actor "sees campaigns, leads, attribution" and imports lists for the whole team — but changes a
lead only by importing (lead:lead:import). A lead the caller may not see is 404, never 403. From 0.3.0 these are
PERMISSIONS, not role names: seeing every lead is lead:lead:read_all, changing anyone's lead:lead:manage, naming
another owner lead:lead:assign (roles.yaml: the default roles).
NEIGHBOURS, called through their routes (erp_platform.seam), never their tables — and only through the
anti-corruption layer in application/neighbours.py, which translates their shapes into the lead's language.
Converting a lead (CRM-03) needs all three; the lead brick's service identity needs the permission named on
each:
business-partner POST /partners/promote (bp:partner:promote) — create-or-reuse the company by its domain, as a
customer. Idempotent by bp_key. Also asked, read-only, whether a lookalike is already a customer.
activity POST /contacts (activity:contact:write) — the person, idempotent by email.
pipeline POST /opportunities/from-source (pipeline:opportunity:link_source, a machine's — pipeline 0.3.0)
— the deal, in the first stage, owned by whoever converts, with source_ref "lead:<lead_id>" so a
retry returns the same deal. A ref whose deal is another partner's is refused 409: the conversion
stops there, with the partner and contact kept, and nothing written after it.
NOT INSTALLED (configured nowhere): converting is refused, 409, before anything is called — a lead
converted into a partner and a contact but no deal is the half-conversion the story exists to prevent,
and with a neighbour missing it could never be finished. Everything else in this brick works without them.
INSTALLED BUT DOWN: whatever succeeded is kept on the lead (bp_id, contact_id) in its own transaction, the
call is refused 503 saying what is done and what is not, and converting again finishes the rest. Every
step is idempotent at the neighbour, so nothing is created twice — including the deal whose reply was lost.
"""
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from typing import Literal, Optional
from uuid import uuid4
from erp_platform import auth
from erp_platform.bricks import Brick, Refused
from erp_platform.repository import Conflict, retry_on_conflict
from erp_platform.types import Amount, BigId, Email, IsoDate, IsoTime, Name, Note, text
from brick_lead.application import neighbours
from brick_lead.application.neighbours import NeighbourFailed
# The seams, named in THIS module too: erp-compose check / plan find a brick's seams in the module that
# holds the Brick, and without these the composition reported no lead → bp/activity/pipeline seams (and a
# tier-2 plan gave the lead app no BP_/ACTIVITY_/PIPELINE_BASE_URL). They are still called only in neighbours.
from brick_lead.application.neighbours import CONTACTS, DEALS, PARTNERS # noqa: F401
from brick_lead.domain.lead import (FIT_ATTRIBUTES, OPEN, Lead, LeadIncomplete, LeadRuleBroken, ProfileVector,
Rule, ScoringPolicy, Signal, lead_id_for, normalise_country,
normalise_fit_value, normalise_industry, normalise_signal, normalise_size,
plan_import, read_csv, score)
WRITE, IMPORT, ASSIGN = "lead:lead:write", "lead:lead:import", "lead:lead:assign"
SCORE, SIGNAL, POLICY = "lead:lead:score", "lead:signal:record", "lead:policy:write"
# 0.3.0: a brick decides by PERMISSION, never by role name (erp-platform 0.9.2). Seeing every owner's leads was the
# `manager`/`analyst`/`marketing` ROLES, changing them the `manager` role, naming another owner on a new lead the
# `manager` role (now ASSIGN, as reassigning one is); they are now these (roles.yaml: the default roles).
READ_ALL, MANAGE = "lead:lead:read_all", "lead:lead:manage"
Status = Literal["new", "working", "qualified", "disqualified", "converted"]
Ref, Code, Currency, Csv = text(64), text(64), text(8), text(4_000_000)
Score, Count = __import__("typing").Annotated[int, __import__("pydantic").Field(ge=0, le=100)], \
__import__("typing").Annotated[int, __import__("pydantic").Field(ge=1, le=200)]
# 0.2.1: SCORE and SIGNAL are MACHINE permissions (erp-platform 0.9.2) — passed on an exact grant only, so `*:*:*`
# never reaches them; and the process is declared (process.yaml → GET /process/bpmn).
LEAD = Brick("lead", version="0.3.1", ids_as_strings=True, machine_permissions=(SCORE, SIGNAL), instructions=(
"Leads: people (keyed by email) and companies (keyed by web domain) not yet worked into a deal. A lead "
"moves new → working → qualified (its score has reached the policy's threshold) or disqualified (with a "
"reason), and a qualified lead is converted into a business partner, a contact and a deal in one step. "
"Marketing imports lists as CSV; importing twice changes nothing. The score (0–100) is derived from the "
"lead's fit and its engagement signals by the business's own scoring policy, and explains itself. "
"Lookalikes ranks lead companies by similarity to a won account, excluding existing customers. bp_id is a "
"64-bit id, sent and returned as digits. A lead carries a `version`: pass it back with a change, and a 409 "
"means someone changed it since."))
LEAD.declare_process("process.yaml")
@contextmanager
def _repository_default():
from erp_platform import db
from brick_lead.infrastructure.tables import LeadRepository
with db.engine.begin() as conn:
yield LeadRepository(conn)
repository = _repository_default
def _now() -> datetime:
return datetime.now(timezone.utc)
def _aware(iso: str) -> datetime:
"""An ISO time in UTC; one without a zone is taken as UTC. Converted, not just labelled: TIMESTAMP columns
keep no zone, so 10:00+02:00 stored as written would be two hours late."""
try:
d = datetime.fromisoformat(str(iso).strip().replace("Z", "+00:00"))
except ValueError:
raise Refused(422, f"{iso!r} is not a time (ISO 8601)") from None
return d.astimezone(timezone.utc) if d.tzinfo else d.replace(tzinfo=timezone.utc)
@contextmanager
def _rules():
try:
yield
except LeadIncomplete as e:
raise Refused(422, str(e)) from None
except LeadRuleBroken as e:
raise Refused(409, str(e)) from None
def _scope(caller: dict, mine: bool = False) -> Optional[str]:
"""None = every lead; else the email the caller's reads are narrowed to. READ_ALL sees all (marketing, managers)."""
return auth.owner_scope(caller, mine, see_all=READ_ALL)
def _owner_of_new(caller: dict, requested: Optional[str]) -> Optional[str]:
"""Who a new lead belongs to: whoever may ASSIGN leads may name the owner; everyone else is forced-own (the row
is the caller's, whatever `owner` was sent — no rejection path, so imports and seeded forms keep working)."""
email = (caller.get("email") or "").strip() or None
if auth.has_permission(caller, ASSIGN):
return (requested or "").strip() or email
return email
def _sees(caller: dict, lead: Lead) -> bool:
scope = _scope(caller)
return scope is None or lead.owner == scope
def _load(repo, caller: dict, lead_id: str, *, write: bool = False) -> Lead:
lead = repo.get(lead_id)
if lead is None or not _sees(caller, lead):
raise Refused(404, f"no lead {lead_id}")
if write and not (lead.owner == caller.get("email") or auth.has_permission(caller, MANAGE)):
raise Refused(403, f"lead {lead_id} belongs to {lead.owner}; only they or a manager can change it")
return lead
def _at(lead: Lead, version: Optional[int]) -> Lead:
"""The lead, if it is still at the version the caller loaded; a Conflict (409) if not."""
if version is not None and lead.version != int(version):
raise Conflict("lead_leads", {"lead_id": lead.lead_id}, int(version), lead.version)
return lead
def _changing(version: Optional[int], fn):
"""With the caller's version, a race is their 409; with none, the change is re-run on the latest lead."""
return fn() if version is not None else retry_on_conflict(fn)
def _view(lead: Lead, s: Optional[dict] = None) -> dict:
return {"lead_id": lead.lead_id, "kind": lead.kind, "name": lead.name, "email": lead.email,
"company": lead.company, "domain": lead.domain, "title": lead.title, "country": lead.country,
"industry": lead.industry, "size_band": lead.size_band, "source": lead.source, "owner": lead.owner,
"status": lead.status, "disqualify_reason": lead.disqualify_reason or None,
"score": None if s is None else s["score"], "score_reasons": [] if s is None else s["reasons"],
"scored_by_run": None if s is None else s["run_id"],
"conversion": {"bp_id": lead.bp_id, "contact_id": lead.contact_id,
"opportunity_id": lead.opportunity_id, "converted_at": lead.converted_at,
"converted_by": lead.converted_by or None},
"version": lead.version}
def _saved(repo, caller: dict, lead: Lead) -> dict:
"""Save the aggregate with the events it raised — one lead, one transaction — and show it."""
repo.save(lead, caller["email"])
return _view(lead, repo.scores([lead.lead_id]).get(lead.lead_id))
def _text(v: Optional[str]) -> str:
return (v or "").strip()
# ── reading ────────────────────────────────────────────────────────────────────────────────
@LEAD.use_case(kind="query", route="GET /leads")
def list_leads(caller: dict, status: Optional[Status] = None, mine: bool = False,
min_score: Optional[Score] = None) -> dict:
"""The leads the caller may see, highest score first — the call list (CRM-10). A rep sees their own; a
manager, analyst or marketer sees everyone's. Filter by status, by a minimum score, or `mine`."""
with repository() as repo:
found = repo.leads(owner=_scope(caller, mine), statuses=[status] if status else None)
s = repo.scores([l.lead_id for l in found]) if found else {}
out = [_view(l, s.get(l.lead_id)) for l in found]
if min_score is not None:
out = [v for v in out if (v["score"] or 0) >= min_score]
out.sort(key=lambda v: (-(v["score"] if v["score"] is not None else -1), v["name"].lower(), v["lead_id"]))
return {"leads": out}
@LEAD.use_case(kind="query", route="GET /leads/{lead_id}", etag="version")
def get_lead(caller: dict, lead_id: Ref) -> dict:
"""One lead: who, which company, its status, its latest score and why, once converted the partner,
contact and deal it became, and its `version` (send it back with a change)."""
with repository() as repo:
lead = _load(repo, caller, lead_id)
return _view(lead, repo.scores([lead_id]).get(lead_id))
@LEAD.use_case(kind="query", route="GET /leads/{lead_id}/history")
def lead_history(caller: dict, lead_id: Ref) -> dict:
"""Every change to a lead — imported, worked, qualified (and by whom: a rep or the scoring run),
disqualified, each conversion step — with who and when."""
with repository() as repo:
_load(repo, caller, lead_id)
return {"lead_id": lead_id, "events": repo.history(lead_id)}
# ── leads ──────────────────────────────────────────────────────────────────────────────────
@LEAD.use_case(kind="command", route="POST /leads", permission=WRITE, idempotent=True)
def add_lead(caller: dict, email: Optional[Email] = None, domain: Optional[Name] = None, name: Optional[Name] = None,
company: Optional[Name] = None, title: Optional[Name] = None, country: Optional[Code] = None,
industry: Optional[Name] = None, size_band: Optional[Code] = None, source: Optional[Name] = None,
owner: Optional[Email] = None) -> dict:
"""Add a lead: a person (give `email`) or a company (give only `domain`). It is owned by the caller; a
manager may name another `owner`. One email (or one domain) is one lead: adding one that exists returns
it unchanged, and is refused if it belongs to a rep the caller cannot see."""
with repository() as repo, _rules():
lead = Lead.new(_owner_of_new(caller, owner), email=_text(email), domain=_text(domain),
name=_text(name), company=_text(company), title=_text(title), country=_text(country),
industry=_text(industry), size_band=_text(size_band), source=_text(source) or "manual")
existing = repo.get(lead.lead_id)
if existing is not None:
if not _sees(caller, existing):
raise Refused(409, f"{lead.email or lead.domain} is already a lead, owned by another rep")
return {**_view(existing, repo.scores([existing.lead_id]).get(existing.lead_id)), "existing": True}
return {**_saved(repo, caller, lead), "existing": False}
@LEAD.use_case(kind="command", route="PATCH /leads/{lead_id}", permission=WRITE, idempotent=True,
if_match="version")
def edit_lead(caller: dict, lead_id: Ref, name: Optional[Name] = None, company: Optional[Name] = None,
domain: Optional[Name] = None, title: Optional[Name] = None, country: Optional[Code] = None,
industry: Optional[Name] = None, size_band: Optional[Code] = None,
source: Optional[Name] = None, version: Optional[int] = None) -> dict:
"""Correct what a lead says: name, company, web domain, title, country, industry, size band, source.
Fields left out or blank are unchanged. A converted lead is not edited — its partner, contact and deal
are the record now; a lead part-way through converting keeps the domain its partner was made from.
`version` (or If-Match): the one you read."""
def run():
with repository() as repo, _rules():
lead = _at(_load(repo, caller, lead_id, write=True), version)
changed = lead.describe(name=name, company=company, domain=domain, title=title, country=country,
industry=industry, size_band=size_band, source=source)
if not changed:
return _view(lead, repo.scores([lead_id]).get(lead_id))
return _saved(repo, caller, lead)
return _changing(version, run)
@LEAD.use_case(kind="command", route="POST /leads/import", permission=IMPORT, idempotent=True)
def import_leads(caller: dict, csv_text: Csv, source: Name = "import") -> dict:
"""CRM-13: import a list of people from a spreadsheet saved as CSV, given as text.
The first row names the columns: `email` is required; `name`, `company`, and optionally `domain`
(website), `title`, `country` (two-letter code), `industry`, `size` (a band like 51-200, or a head
count), `source` and `owner` (a rep's email; otherwise the importer owns the lead). Comma, semicolon or
tab separated. Each person is ONE lead, however often the list is imported: a re-import updates what the
file says and never the owner, the status or where the lead first came from (its source). Rows with no email, or an address that is not one, are
reported with their spreadsheet row number (the header is row 1) and reason — never dropped silently.
The result counts created, updated, unchanged and skipped. A file that cannot be read as CSV is refused
with the row to fix."""
return retry_on_conflict(lambda: _import(caller, csv_text, source))
def _import(caller: dict, csv_text: str, source: str) -> dict:
with repository() as repo, _rules():
_, _, rows = read_csv(csv_text)
ids = set()
for _, row in rows:
try:
ids.add(lead_id_for(row.get("email", "")))
except LeadIncomplete:
pass # reported by plan_import, with its row number
existing = repo.by_ids(ids) if ids else {}
plan = plan_import(csv_text, existing, caller["email"], (source or "import").strip().lower())
# A batch, deliberately one transaction: the result's counts must be what was written, all or none.
# Each lead's own rules still ran in its aggregate (plan_import → Lead.new / Lead.describe).
repo.save_many(plan.create + [l for l, _ in plan.update], set(existing), caller["email"])
return {"created": len(plan.create), "updated": len(plan.update), "unchanged": len(plan.unchanged),
"skipped": plan.skipped, "warnings": plan.warnings, "ignored_columns": plan.ignored_columns,
"lead_ids": sorted({l.lead_id for l in plan.create} | {l.lead_id for l, _ in plan.update}
| set(plan.unchanged))}
@LEAD.use_case(kind="command", route="POST /leads/{lead_id}/working", permission=WRITE, idempotent=True,
if_match="version")
def start_working(caller: dict, lead_id: Ref, version: Optional[int] = None) -> dict:
"""A rep has started working a new lead (new → working). Doing it again changes nothing."""
def run():
with repository() as repo, _rules():
lead = _at(_load(repo, caller, lead_id, write=True), version)
if not lead.start_working():
return _view(lead, repo.scores([lead_id]).get(lead_id))
return _saved(repo, caller, lead)
return _changing(version, run)
@LEAD.use_case(kind="command", route="POST /leads/{lead_id}/qualify", permission=WRITE, idempotent=True,
if_match="version")
def qualify_lead(caller: dict, lead_id: Ref, version: Optional[int] = None) -> dict:
"""Qualify a lead being worked (working → qualified). The gate is the scoring policy's threshold, applied
to the score computed NOW from the lead's attributes and signals — not to the last run's figure, which
can be hours old. Below the threshold it is refused with the score and the threshold."""
def run():
with repository() as repo, _rules():
lead = _at(_load(repo, caller, lead_id, write=True), version)
policy, now = repo.policy(), _now()
s = score(lead, repo.signals_since(now - timedelta(days=policy.signal_window_days), [lead_id])
.get(lead_id, []), policy, now)
if not lead.qualify(s, policy, by=caller["email"]):
return _view(lead, repo.scores([lead_id]).get(lead_id))
return _saved(repo, caller, lead)
return _changing(version, run)
@LEAD.use_case(kind="command", route="POST /leads/{lead_id}/disqualify", permission=WRITE, idempotent=True,
if_match="version")
def disqualify_lead(caller: dict, lead_id: Ref, reason: Note, version: Optional[int] = None) -> dict:
"""Disqualify a new or worked lead, with a reason (no budget, wrong market, a competitor's customer) so
that "why do we lose leads?" has an answer. A qualified lead is converted, not disqualified."""
def run():
with repository() as repo, _rules():
lead = _at(_load(repo, caller, lead_id, write=True), version)
if not lead.disqualify(reason):
return _view(lead, repo.scores([lead_id]).get(lead_id))
return _saved(repo, caller, lead)
return _changing(version, run)
@LEAD.use_case(kind="command", route="PUT /leads/{lead_id}/owner", permission=ASSIGN, idempotent=True,
if_match="version")
def assign_lead(caller: dict, lead_id: Ref, owner: Email, version: Optional[int] = None) -> dict:
"""Give an open lead to a rep (their email) — how an imported list is shared out."""
def run():
with repository() as repo, _rules():
lead = _at(_load(repo, caller, lead_id), version)
if not lead.assign(owner):
return _view(lead, repo.scores([lead_id]).get(lead_id))
return _saved(repo, caller, lead)
return _changing(version, run)
# ── CRM-03: convert ────────────────────────────────────────────────────────────────────────
def _stopped(e: NeighbourFailed, done: list) -> Refused:
"""A neighbour's failure, in the words a rep (or their agent) acts on: what stopped, what is kept."""
so_far = (f" Done so far: {', '.join(done)} — kept, and converting again finishes the rest without "
f"creating them twice." if done else " Nothing was created.")
if e.answered:
return Refused(422 if e.status == 422 else 409, f"{e.what} refused: {e.said}.{so_far}")
return Refused(503, f"{e.what} did not answer ({e.said}).{so_far}")
def _step(caller: dict, lead_id: str, detail: str, **produced) -> Lead:
"""Keep one step's result on the lead, in its OWN transaction — so a later step failing cannot roll it
back, and the next attempt starts from it. Re-run on a lost version race (0.2.0): two conversions at once
each record their step on the latest lead, so neither step's result is lost."""
def run():
with repository() as repo, _rules():
lead = repo.get(lead_id)
lead.record_step(**produced, at=_now().isoformat(), by=caller["email"], detail=detail)
repo.save(lead, caller["email"])
return lead
return retry_on_conflict(run)
def _conversion(lead: Lead, note: str) -> dict:
return {"lead_id": lead.lead_id, "status": lead.status, "bp_id": lead.bp_id, "contact_id": lead.contact_id,
"opportunity_id": lead.opportunity_id, "converted_at": lead.converted_at, "note": note}
@LEAD.use_case(kind="command", route="POST /leads/{lead_id}/convert", permission=WRITE, idempotent=True)
def convert_lead(caller: dict, lead_id: Ref, currency: Currency, opportunity_name: Optional[Name] = None,
amount: Optional[Amount] = None, close_date: Optional[IsoDate] = None) -> dict:
"""CRM-03: the conversion of a qualified lead into a business partner (its company, reused if the domain
is already one), a contact (the person, reused by email) and a deal in the first stage, in one step.
`currency` is the deal's; `opportunity_name` defaults to the company's name. Converting again changes
nothing and returns the same partner, contact and deal. If a neighbour is down part-way, what was done is
kept and the call is refused (503) saying so; converting again finishes the rest and creates nothing
twice. Refused (409) before anything is created if the partner register, the contact list or the
pipeline is not installed. The deal is owned by whoever converts.
The orchestration lives here, in the application layer: it spans the lead and three neighbours, and
each step is its own transaction so that progress survives a failure. The lead aggregate decides what
each step means (record_step); application/neighbours.py speaks to the neighbours."""
with repository() as repo, _rules():
lead = _load(repo, caller, lead_id, write=True)
if lead.status == "converted":
return _conversion(lead, "already converted — nothing changed")
lead.ready_to_convert()
missing = neighbours.missing_for_conversion()
if missing:
raise Refused(409, f"converting a lead needs the {', '.join(missing)} brick(s) installed: a partner "
f"without its contact or deal is a half-conversion. Nothing was created.")
actor, done = caller.get("email", ""), []
try:
if lead.bp_id is None:
bp_id = neighbours.make_customer(lead, actor)
lead = _step(caller, lead_id, f"partner {bp_id} ({lead.domain})", bp_id=bp_id)
done.append(f"partner {lead.bp_id}")
if lead.contact_id is None:
contact_id = neighbours.add_contact(lead, lead.bp_id, actor)
lead = _step(caller, lead_id, f"contact {contact_id} ({lead.email})", contact_id=contact_id)
done.append(f"contact {lead.contact_id}")
if lead.opportunity_id is None:
opportunity_id = neighbours.open_deal(lead, lead.bp_id, actor, currency=currency,
name=opportunity_name, amount=amount, close_date=close_date)
lead = _step(caller, lead_id, f"deal {opportunity_id}", opportunity_id=opportunity_id)
except NeighbourFailed as e:
raise _stopped(e, done) from None
return _conversion(lead, "converted")
# ── CRM-10: signals and scoring ────────────────────────────────────────────────────────────
@LEAD.use_case(kind="command", route="POST /leads/{lead_id}/signals", permission=SIGNAL, idempotent=True)
def record_signal(caller: dict, lead_id: Ref, kind: Code, at: IsoTime, source: Name = "") -> dict:
"""Record that a lead engaged: opened the pricing email, clicked, attended a webinar, asked for a demo —
`kind` in lower_snake_case, `at` when it happened (ISO 8601). Called by the campaign brick or a website
hook. The same event reported twice is recorded once. It counts towards the score from the next scoring
run (or the next qualify) if the scoring policy weighs that kind."""
with repository() as repo, _rules():
lead = repo.get(lead_id)
if lead is None:
raise Refused(404, f"no lead {lead_id}")
sig = Signal(kind, _aware(at))
new = repo.add_signal(lead_id, sig.signal_id, sig.kind, sig.at, _text(source))
lead.engaged(sig, new)
repo.write_events([lead], caller["email"])
return {"lead_id": lead_id, "signal_id": sig.signal_id, "kind": sig.kind, "at": sig.at.isoformat(),
"recorded": new, "weighed_by_policy": repo.policy().weighs(sig.kind)}
@LEAD.use_case(kind="command", route="POST /leads/score", permission=SCORE, idempotent=True)
def score_leads(caller: dict, as_of: Optional[IsoTime] = None) -> dict:
"""CRM-10: re-score every open lead against the current scoring policy, as of `as_of` (default now).
The spec says every 6 hours, but the platform has NO scheduler: whatever runs your jobs calls this — an
external cron, or an always-on worker app pinned warm (--reserved 1 --max-replicas 1) with its own loop.
Each run is recorded (run id, time, how many scored, changed, qualified). A score is DERIVED from the
lead's fit, its signals in the policy's window and the policy, so a second run with nothing new changes
nothing: an unchanged score is not rewritten and keeps the run that computed it. A lead being worked
whose score reaches the threshold is qualified by the run (the spec lets the system do it); a score
falling later does not un-qualify anyone — qualifying was a decision."""
return retry_on_conflict(lambda: _score(caller, as_of))
def _score(caller: dict, as_of: Optional[str]) -> dict:
when = _aware(as_of) if as_of else _now()
with repository() as repo, _rules():
policy = repo.policy()
open_leads = repo.leads(statuses=OPEN)
signals = repo.signals_since(when - timedelta(days=policy.signal_window_days))
stored = repo.scores()
run_id = f"R-{uuid4().hex[:12].upper()}"
changed, qualified = [], []
for lead in open_leads:
s = score(lead, signals.get(lead.lead_id, []), policy, when)
before = stored.get(lead.lead_id)
if before is None or (before["score"], before["fit_points"], before["engagement_points"],
before["reasons"]) != (s.score, s.fit_points, s.engagement_points, list(s.reasons)):
changed.append({"lead_id": lead.lead_id, "score": s.score, "fit_points": s.fit_points,
"engagement_points": s.engagement_points, "reasons": "; ".join(s.reasons),
"run_id": run_id, "scored_at": _now()})
if lead.status == "working" and policy.qualifies(s):
lead.qualify(s, policy, by=f"scoring run {run_id}")
qualified.append(lead)
# A batch, deliberately one transaction (like the import): a run is recorded with all its scores or
# none, so "the run is recorded" and "its scores are stored" cannot disagree.
repo.save_scores(changed)
for lead in qualified:
repo.save(lead, caller["email"])
repo.record_run({"run_id": run_id, "run_at": _now(), "as_of": when, "run_by": caller["email"],
"leads_scored": len(open_leads), "scores_changed": len(changed),
"qualified": len(qualified)})
return {"run_id": run_id, "as_of": when.isoformat(), "leads_scored": len(open_leads),
"scores_changed": len(changed), "qualified": [l.lead_id for l in qualified]}
@LEAD.use_case(kind="query", route="GET /score-runs")
def list_score_runs(caller: dict, limit: Count = 20) -> dict:
"""The most recent scoring runs, newest first: when, as of when, by whom, and how many leads each
scored, changed and qualified — how you tell that the job is running."""
with repository() as repo:
return {"runs": repo.runs(max(1, min(limit, 200)))}
# ── configuration: the scoring policy (config_rows: scoring_policy) ────────────────────────
def _policy_view(p: ScoringPolicy) -> dict:
return {"qualify_threshold": p.qualify_threshold, "signal_window_days": p.signal_window_days,
"fit": [{"attribute": r.attribute, "value": r.value, "points": r.points}
for r in p.rules if r.kind == "fit"],
"signals": [{"kind": r.attribute, "points": r.points, "max_count": r.max_count}
for r in p.rules if r.kind == "signal"],
"fit_attributes": list(FIT_ATTRIBUTES)}
@LEAD.use_case(kind="query", route="GET /scoring-policy")
def get_scoring_policy(caller: dict) -> dict:
"""How this business scores and qualifies a lead: the points for fit (industry, size band, country,
region, source) and for each engagement signal, how many of a signal count, the window signals count
in, and the score that qualifies."""
with repository() as repo:
return _policy_view(repo.policy())
@LEAD.use_case(kind="command", route="PUT /scoring-policy", permission=POLICY, idempotent=True)
def set_scoring_policy(caller: dict, qualify_threshold: int, signal_window_days: int, fit: list[dict],
signals: list[dict]) -> dict:
"""Replace the scoring policy (BANT, MEDDIC-shaped or your own weights): `fit` is a list of {attribute,
value, points} — e.g. {"attribute": "industry", "value": "software", "points": 20}; `signals` is a list
of {kind, points, max_count} — e.g. {"kind": "email_opened_pricing", "points": 25, "max_count": 2}.
Points are −100..100, the threshold 0..100. Scores change at the next scoring run."""
with repository() as repo, _rules():
try:
rules = tuple(Rule("fit", str(f["attribute"]).strip().lower(),
normalise_fit_value(str(f["attribute"]).strip().lower(), str(f["value"])),
int(f["points"])) for f in fit) + \
tuple(Rule("signal", normalise_signal(str(s["kind"])), "", int(s["points"]),
int(s.get("max_count", 1))) for s in signals)
except (KeyError, ValueError, TypeError) as e:
raise Refused(422, f"fit rules need attribute, value, points; signal rules need kind, points "
f"({e})") from None
p = ScoringPolicy(int(qualify_threshold), int(signal_window_days), rules)
repo.set_policy(p, caller["email"])
return _policy_view(p)
# ── CRM-14: lookalikes ─────────────────────────────────────────────────────────────────────
@LEAD.use_case(kind="query", route="GET /lookalikes")
def find_lookalikes(caller: dict, bp_id: Optional[BigId] = None, industry: Optional[Name] = None,
size_band: Optional[Code] = None, country: Optional[Code] = None, limit: Count = 10) -> dict:
"""CRM-14: companies like one you just won — who to call next. A ranked list of lead companies the
caller may see, most similar first, excluding companies that are already customers.
Give the won account's `bp_id`: if it was converted from a lead, that lead's industry, size band and
country are its profile. Otherwise (or to override) pass `industry`, `size_band`, `country`. Similarity
is the cosine distance between fit profiles (industry, then size, then region), so it explains itself.
Excluded: the account itself, every company converted from a lead (by anyone), and — when the partner
register is installed — any company it already holds as a customer under its web domain."""
limit = max(1, min(int(limit), 50))
with repository() as repo, _rules():
base = repo.converted_from(bp_id) if bp_id is not None else None
attrs = {"industry": normalise_industry(_text(industry)) or (base.industry if base else ""),
"size_band": normalise_size(_text(size_band)) or (base.size_band if base else ""),
"country": normalise_country(_text(country)) or (base.country if base else "")}
if not any(attrs.values()):
raise Refused(422, ("no lead was converted into partner %s, so its profile is unknown: " % bp_id
if bp_id is not None else "") + "pass industry, size_band and/or country")
target = ProfileVector.of(**attrs)
window = max(50, limit * 5)
near = repo.nearest(list(target.values), _scope(caller), window)
customers = repo.converted_domains()
skip = customers | ({base.domain} if base and base.domain else set())
by_domain: dict = {}
for lead, distance in near:
if lead.domain in skip:
continue
c = by_domain.setdefault(lead.domain, {"domain": lead.domain, "company": lead.company,
"industry": lead.industry, "size_band": lead.size_band,
"country": lead.country, "distance": distance, "leads": []})
c["leads"].append({"lead_id": lead.lead_id, "name": lead.name, "owner": lead.owner, "status": lead.status})
ranked = sorted(by_domain.values(), key=lambda c: (c["distance"], c["domain"]))
out, note, register_customers, asked = [], "", 0, neighbours.PARTNERS.installed
if not asked:
note = "no partner register installed: only companies converted from leads were excluded"
for c in ranked:
if len(out) == limit:
break
if asked:
try:
if neighbours.is_customer(c["domain"], caller.get("email", "")):
register_customers += 1
continue
except NeighbourFailed as e:
asked, note = False, (f"the partner register did not answer ({e.said}): only companies "
f"converted from leads were excluded")
out.append({**c, "similarity": round(1 - c["distance"], 4), "distance": round(c["distance"], 6)})
return {"based_on": {**attrs, "from": f"lead {base.lead_id}" if base else "the attributes given"},
"lookalikes": out,
"excluded_customers": len({l.domain for l, _ in near if l.domain in customers}) + register_customers,
"note": note or None}