"""The tables, and the code that turns rows into leads, scores and the policy and back."""
import json
import uuid
from datetime import datetime, timedelta, timezone
from typing import Iterable, Optional
from pgvector.sqlalchemy import Vector
from sqlalchemy import BigInteger, Column, Integer, MetaData, String, Table, TIMESTAMP, cast, insert, select
from sqlalchemy.exc import IntegrityError
from erp_platform.repository import Aggregate, Conflict, Part, upsert
from brick_lead.domain.lead import (DEFAULT_POLICY, DIM, OPEN, Lead, Rule, ScoringPolicy, cosine_distance,
profile)
md = MetaData()
leads = Table(
"lead_leads", md,
Column("lead_id", String, primary_key=True), Column("kind", String), Column("name", String),
Column("email", String), Column("company", String), Column("domain", String), Column("title", String),
Column("country", String), Column("industry", String), Column("size_band", String), Column("source", String),
Column("owner", String), Column("status", String), Column("disqualify_reason", String),
Column("bp_id", BigInteger), Column("contact_id", String), Column("opportunity_id", String),
Column("converted_at", TIMESTAMP), Column("converted_by", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP),
Column("version", Integer)) # 0.2.0: the optimistic version, at the end as its ALTER says
profiles = Table("lead_profiles", md,
Column("lead_id", String, primary_key=True), Column("profile", Vector(DIM)),
Column("updated_at", TIMESTAMP))
events = Table("lead_events", md,
Column("lead_id", String, primary_key=True), Column("event_id", String, primary_key=True),
Column("action", String), Column("detail", String), Column("actor", String),
Column("event_at", TIMESTAMP))
signals = Table("lead_signals", md,
Column("lead_id", String, primary_key=True), Column("signal_id", String, primary_key=True),
Column("kind", String), Column("event_at", TIMESTAMP), Column("source", String),
Column("created_at", TIMESTAMP))
scores = Table("lead_scores", md,
Column("lead_id", String, primary_key=True), Column("score", Integer),
Column("fit_points", Integer), Column("engagement_points", Integer), Column("reasons", String),
Column("run_id", String), Column("scored_at", TIMESTAMP))
runs = Table("lead_score_runs", md,
Column("run_id", String, primary_key=True), Column("run_at", TIMESTAMP), Column("as_of", TIMESTAMP),
Column("run_by", String), Column("leads_scored", Integer), Column("scores_changed", Integer),
Column("qualified", Integer))
policy_t = Table("lead_scoring_policy", md,
Column("policy_id", String, primary_key=True), Column("qualify_threshold", Integer),
Column("signal_window_days", Integer), Column("updated_by", String),
Column("updated_at", TIMESTAMP))
rules_t = Table("lead_scoring_rules", md,
Column("policy_id", String, primary_key=True), Column("rule_kind", String, primary_key=True),
Column("attribute", String, primary_key=True), Column("value", String, primary_key=True),
Column("points", Integer), Column("max_count", Integer))
LEAD = Aggregate(Part(leads, ("lead_id",)), version="version")
POLICY = Aggregate(Part(policy_t, ("policy_id",)), Part(rules_t, ("policy_id", "rule_kind", "attribute", "value")))
CHUNK = 500 # rows per INSERT on a bulk import: one statement stays well under any limit
def _now() -> datetime:
return datetime.now(timezone.utc)
def _ts(v) -> Optional[datetime]:
if not v:
return None
d = v if isinstance(v, datetime) else datetime.fromisoformat(str(v).replace("Z", "+00:00"))
return d if d.tzinfo else d.replace(tzinfo=timezone.utc)
def _iso(v) -> Optional[str]:
d = _ts(v)
return None if d is None else d.isoformat()
def _knn_in_database(conn) -> bool:
"""Whether `<=>` can rank in the database. DataK3 speaks Postgres and has pgvector; SQLite (the unit
tests) has neither, so there the same candidates are ranked in Python by the same cosine distance. The
live-check harness's connection has no `.dialect` — it compiles every statement for Postgres — so a
missing dialect means Postgres."""
return getattr(getattr(conn, "dialect", None), "name", "postgresql") == "postgresql"
class LeadRepository:
def __init__(self, conn):
self.conn = conn
# ── leads ────────────────────────────────────────────────────────────────────────────
@staticmethod
def _lead(r) -> Lead:
return Lead(
lead_id=r["lead_id"], kind=r["kind"], name=r["name"], email=r["email"] or "",
company=r["company"] or "", domain=r["domain"] or "", owner=r["owner"], title=r["title"] or "",
country=r["country"] or "", industry=r["industry"] or "", size_band=r["size_band"] or "",
source=r["source"] or "", status=r["status"], disqualify_reason=r["disqualify_reason"] or "",
bp_id=None if r["bp_id"] is None else int(r["bp_id"]), # BIGINT; the harness hands it as text
contact_id=r["contact_id"], opportunity_id=r["opportunity_id"],
converted_at=_iso(r["converted_at"]), converted_by=r["converted_by"] or "",
version=int(r.get("version") or 0))
@staticmethod
def _row(l: Lead) -> dict:
return {"lead_id": l.lead_id, "kind": l.kind, "name": l.name, "email": l.email, "company": l.company,
"domain": l.domain, "title": l.title, "country": l.country, "industry": l.industry,
"size_band": l.size_band, "source": l.source, "owner": l.owner, "status": l.status,
"disqualify_reason": l.disqualify_reason, "bp_id": l.bp_id, "contact_id": l.contact_id,
"opportunity_id": l.opportunity_id, "converted_at": _ts(l.converted_at),
"converted_by": l.converted_by}
def get(self, lead_id: str) -> Optional[Lead]:
rows = LEAD.load(self.conn, {"lead_id": lead_id})
return None if rows is None else self._lead(rows["lead_leads"][0])
def by_ids(self, ids: Iterable[str]) -> dict:
ids = sorted(set(ids))
out: dict = {}
for i in range(0, len(ids), CHUNK):
for r in self.conn.execute(select(leads).where(leads.c.lead_id.in_(ids[i:i + CHUNK]))):
lead = self._lead(dict(r._mapping))
out[lead.lead_id] = lead
return out
def leads(self, owner: Optional[str] = None, statuses: Optional[Iterable[str]] = None) -> list:
stmt = select(leads)
if owner is not None:
stmt = stmt.where(leads.c.owner == owner)
if statuses is not None:
stmt = stmt.where(leads.c.status.in_(list(statuses)))
return sorted((self._lead(dict(r._mapping)) for r in self.conn.execute(stmt)),
key=lambda l: (l.name.lower(), l.lead_id))
def converted_from(self, bp_id: int) -> Optional[Lead]:
"""The lead a partner was converted from, if it was (CRM-14's 'the won account's profile')."""
rows = self.conn.execute(select(leads).where(leads.c.bp_id == int(bp_id)))
found = sorted((self._lead(dict(r._mapping)) for r in rows), key=lambda l: l.lead_id)
return found[0] if found else None
def converted_domains(self) -> set:
"""Every company domain a lead has been converted into a partner for — whoever owns it, and including
a conversion still finishing (bp_id is set from its first step: the partner exists as a customer). A
customer is a customer for everyone, so this is deliberately NOT scoped to the caller."""
rows = self.conn.execute(select(leads.c.domain).where(leads.c.bp_id.isnot(None)))
return {r[0] for r in rows if r[0]}
def save(self, lead: Lead, actor: str) -> None:
"""The lead, its profile (derived from the same attributes, so written with them) and the domain
events it raised — one aggregate, one transaction. Claims version n+1 against the version the lead was
loaded at (0 for a new one): a racer that saved in between makes this a Conflict, nothing written."""
lead.version = LEAD.save(self.conn, {"lead_leads": [self._row(lead)]}, expected_version=lead.version)
upsert(self.conn, profiles, [self._profile_row(lead)], ("lead_id",))
self.write_events([lead], actor)
def write_events(self, items: list, actor: str) -> None:
"""The events the leads raised, as their history; then they are no longer pending."""
now = _now()
# a microsecond apart, in the order raised, so one save's events read back in the order they happened
rows = [{"lead_id": l.lead_id, "event_id": uuid.uuid4().hex, "action": e.name, "detail": e.detail,
"actor": actor, "event_at": now + timedelta(microseconds=i)}
for l in items for i, e in enumerate(l.events)]
for i in range(0, len(rows), CHUNK):
upsert(self.conn, events, rows[i:i + CHUNK], ("lead_id", "event_id"))
for l in items:
l.events.clear()
def save_many(self, items: list, existing: set, actor: str) -> None:
"""A bulk import: one INSERT per CHUNK rows instead of two statements per lead. `existing` is the ids
already stored, so created_at is written for new leads only (repository.upsert).
0.2.0: each lead still CLAIMS its next version, in bulk — a plain INSERT of (lead_id, v+1) per lead into
the aggregate's claim table, before any row is written. A rep's edit that committed after this import
read the lead makes the claim collide (23505): the whole import is a Conflict, writes nothing, and is
re-run on the latest (use_cases.import_leads) — the file's change and the rep's both stand."""
now = _now()
claims = [{"lead_id": l.lead_id, "version": l.version + 1, "saved_at": now} for l in items]
for i in range(0, len(claims), CHUNK):
try:
self.conn.execute(insert(LEAD.versions).values(claims[i:i + CHUNK]))
except IntegrityError:
raise Conflict("lead_leads") from None
for l in items:
l.version += 1
rows = [{**self._row(l), "version": l.version, "created_at": now, "updated_at": now} for l in items]
for i in range(0, len(rows), CHUNK):
part = rows[i:i + CHUNK]
upsert(self.conn, leads, part, ("lead_id",), existing={(l,) for l in existing})
upsert(self.conn, profiles, [self._profile_row(l) for l in items[i:i + CHUNK]], ("lead_id",))
self.write_events(items, actor)
@staticmethod
def _profile_row(lead: Lead) -> dict:
return {"lead_id": lead.lead_id, "profile": profile(lead.industry, lead.size_band, lead.country),
"updated_at": _now()}
# ── CRM-14: nearest companies by profile ─────────────────────────────────────────────
def nearest(self, target: list, owner: Optional[str], window: int) -> list:
"""Open leads with a company domain, nearest first: [(Lead, distance)], at most `window` of them.
On DataK3 the database ranks: `ORDER BY profile <=> CAST($target AS VECTOR(31)) LIMIT window` — the
CAST makes the parameter's type explicit whatever the driver sends a string as. Elsewhere (SQLite) the
same filter is read and ranked here with the same cosine distance. Ties break by lead_id, so the
order is deterministic either way."""
j = leads.join(profiles, profiles.c.lead_id == leads.c.lead_id)
cond = [leads.c.status.in_(list(OPEN)), leads.c.domain != ""]
if owner is not None:
cond.append(leads.c.owner == owner)
if _knn_in_database(self.conn):
distance = profiles.c.profile.cosine_distance(cast(target, Vector(DIM))).label("distance")
stmt = (select(leads, distance).select_from(j).where(*cond)
.order_by(distance, leads.c.lead_id).limit(window))
return [(self._lead(dict(r._mapping)), float(r._mapping["distance"])) for r in self.conn.execute(stmt)]
stmt = select(leads, cast(profiles.c.profile, String).label("profile_text")).select_from(j).where(*cond)
ranked = []
for r in self.conn.execute(stmt):
m = dict(r._mapping)
ranked.append((self._lead(m), cosine_distance(json.loads(m["profile_text"]), target)))
ranked.sort(key=lambda x: (x[1], x[0].lead_id))
return ranked[:window]
# ── CRM-10: signals, scores, runs ────────────────────────────────────────────────────
def add_signal(self, lead_id: str, signal_id: str, kind: str, at: datetime, source: str) -> bool:
"""True when this event is new. A re-delivered one rewrites the same row with the same values."""
known = self.conn.execute(select(signals.c.signal_id).where(
(signals.c.lead_id == lead_id) & (signals.c.signal_id == signal_id))).first() is not None
upsert(self.conn, signals, [{"lead_id": lead_id, "signal_id": signal_id, "kind": kind, "event_at": at,
"source": source, "created_at": _now()}], ("lead_id", "signal_id"),
existing={(lead_id, signal_id)} if known else set())
return not known
def signals_since(self, since: datetime, lead_ids: Optional[Iterable[str]] = None) -> dict:
"""{lead_id: [(kind, aware datetime)]} for events after `since` — one SELECT for a whole run."""
stmt = select(signals.c.lead_id, signals.c.kind, signals.c.event_at).where(signals.c.event_at > since)
if lead_ids is not None:
stmt = stmt.where(signals.c.lead_id.in_(sorted(set(lead_ids))))
out: dict = {}
for r in self.conn.execute(stmt):
out.setdefault(r[0], []).append((r[1], _ts(r[2])))
return out
def scores(self, lead_ids: Optional[Iterable[str]] = None) -> dict:
stmt = select(scores)
if lead_ids is not None:
stmt = stmt.where(scores.c.lead_id.in_(sorted(set(lead_ids))))
return {r["lead_id"]: {"score": int(r["score"]), "fit_points": int(r["fit_points"]),
"engagement_points": int(r["engagement_points"]),
"reasons": [x for x in (r["reasons"] or "").split("; ") if x],
"run_id": r["run_id"], "scored_at": _iso(r["scored_at"])}
for r in (dict(x._mapping) for x in self.conn.execute(stmt))}
def save_scores(self, rows: list) -> None:
for i in range(0, len(rows), CHUNK):
upsert(self.conn, scores, rows[i:i + CHUNK], ("lead_id",))
def record_run(self, run: dict) -> None:
upsert(self.conn, runs, [run], ("run_id",))
def runs(self, limit: int = 20) -> list:
rows = [dict(r._mapping) for r in self.conn.execute(select(runs))]
rows.sort(key=lambda r: _ts(r["run_at"]), reverse=True)
return [{"run_id": r["run_id"], "run_at": _iso(r["run_at"]), "as_of": _iso(r["as_of"]),
"run_by": r["run_by"], "leads_scored": int(r["leads_scored"]),
"scores_changed": int(r["scores_changed"]), "qualified": int(r["qualified"])}
for r in rows[:limit]]
# ── configuration: the scoring policy ────────────────────────────────────────────────
def policy(self) -> ScoringPolicy:
"""The stored policy, or the documented default when none has been set."""
rows = POLICY.load(self.conn, {"policy_id": "default"})
if rows is None:
return DEFAULT_POLICY
p = rows["lead_scoring_policy"][0]
return ScoringPolicy(int(p["qualify_threshold"]), int(p["signal_window_days"]), tuple(
Rule(r["rule_kind"], r["attribute"], r["value"] or "", int(r["points"]), int(r["max_count"]))
for r in sorted(rows["lead_scoring_rules"], key=lambda r: (r["rule_kind"], r["attribute"], r["value"]))))
def set_policy(self, p: ScoringPolicy, actor: str) -> None:
POLICY.save(self.conn, {
"lead_scoring_policy": [{"policy_id": "default", "qualify_threshold": p.qualify_threshold,
"signal_window_days": p.signal_window_days, "updated_by": actor}],
"lead_scoring_rules": [{"policy_id": "default", "rule_kind": r.kind, "attribute": r.attribute,
"value": r.value, "points": r.points, "max_count": r.max_count}
for r in p.rules]})
# ── history ──────────────────────────────────────────────────────────────────────────
def history(self, lead_id: str) -> list:
rows = self.conn.execute(select(events.c.action, events.c.detail, events.c.actor, events.c.event_at)
.where(events.c.lead_id == lead_id))
return sorted(({"action": r[0], "detail": r[1], "actor": r[2], "at": _iso(r[3])} for r in rows),
key=lambda e: e["at"] or "")