"""Contacts, and what a PERSON did and still has to do with them.
No database, no HTTP, no identity. The design is a general CRM activity model, held to the CRM spec (modules/crm: CRM-02, 12):
* A LOGGED CALL IS THE ONLY HOME OF A FACT. "Rang Anna, she wants a proposal by Friday" exists nowhere
until somebody writes it down, and it is most of what a salesperson knows about a deal.
* A TASK IS AN ACTIVITY WITH A DUE DATE — one record, not two. A CRM that splits them ends up with "log a
call" and "schedule a call" as objects that cannot become one another, and completing a task must not
destroy what it said.
kind NOTE · CALL · MEETING · EMAIL · TASK
occurred_at set → it happened (a record of the past)
due_on set → it is work; status open → done | cancelled, and back to open
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
from typing import Optional
NOTE, CALL, MEETING, EMAIL, TASK = "NOTE", "CALL", "MEETING", "EMAIL", "TASK"
KINDS = (NOTE, CALL, MEETING, EMAIL, TASK)
PAST = frozenset({NOTE, CALL, MEETING, EMAIL}) # things that already happened
DIRECTED = frozenset({CALL, EMAIL}) # these have a direction
OPEN, DONE, CANCELLED = "open", "done", "cancelled"
class ActivityRuleBroken(Exception):
"""→ 409"""
class ActivityIncomplete(ActivityRuleBroken):
"""→ 422: something has to be supplied first."""
def normalise_email(email: str) -> str:
e = (email or "").strip().lower()
local, _, domain = e.partition("@")
if not local or "." not in domain or " " in e:
raise ActivityIncomplete(f"{email!r} is not an email address")
return e
# ── contacts ───────────────────────────────────────────────────────────────────────────────
@dataclass(slots=True)
class Contact:
contact_id: str
bp_id: int # the company they work at: a business partner
full_name: str
email: str
owner: str
title: str = ""
phone: str = ""
subscribed: bool = False # consent to marketing email; 1:1 sales email does not need it
status: str = "active" # active | left (left the company — kept, because history points here)
version: int = 0 # 0.2: saves so far — two people editing at once, one wins (409)
@classmethod
def new(cls, contact_id: str, bp_id: int, full_name: str, email: str, owner: str, *,
title: str = "", phone: str = "") -> "Contact":
c = cls(contact_id, int(bp_id), "", normalise_email(email), owner)
c.describe(full_name=full_name, title=title, phone=phone)
return c
def describe(self, *, full_name: Optional[str] = None, title: Optional[str] = None,
phone: Optional[str] = None) -> dict:
before = {"full_name": self.full_name, "title": self.title, "phone": self.phone}
if full_name is not None:
if not full_name.strip():
raise ActivityIncomplete("a contact needs a name")
self.full_name = " ".join(full_name.split())
if title is not None:
self.title = title.strip()
if phone is not None:
self.phone = phone.strip()
after = {"full_name": self.full_name, "title": self.title, "phone": self.phone}
return {k: (before[k], after[k]) for k in after if before[k] != after[k]}
def left_company(self) -> None:
"""They moved on. The contact stays — every activity that mentions them still points here — but
nobody should be emailing them at the old address."""
self.status, self.subscribed = "left", False
# ── activities and tasks ───────────────────────────────────────────────────────────────────
@dataclass(slots=True)
class Activity:
activity_id: str
kind: str
subject: str
owner: str # who did it, or whose task it is
bp_id: int # every activity is about a company …
contact_id: Optional[str] = None # … and usually a person there …
opportunity_id: Optional[str] = None # … and sometimes a deal
note: str = ""
direction: Optional[str] = None # inbound | outbound, for calls and emails
occurred_at: Optional[str] = None
due_on: Optional[str] = None
status: Optional[str] = None # tasks only
done_at: Optional[str] = None
version: int = 0 # 0.2: saves so far (a task is edited: completed, moved, reopened)
@classmethod
def log(cls, activity_id: str, kind: str, subject: str, owner: str, bp_id: int, *,
contact_id: Optional[str] = None, opportunity_id: Optional[str] = None, note: str = "",
direction: Optional[str] = None, occurred_at: Optional[str] = None,
due_on: Optional[str] = None, now: Optional[datetime] = None) -> "Activity":
k = (kind or "").strip().upper()
if k not in KINDS:
raise ActivityIncomplete(f"an activity is one of {', '.join(KINDS)}, not {kind!r}")
if not (subject or "").strip():
raise ActivityIncomplete("an activity needs a subject — it is what the next person reads first")
now = now or datetime.now(timezone.utc)
a = cls(activity_id, k, subject.strip(), owner, int(bp_id), contact_id, opportunity_id,
(note or "").strip())
if k in DIRECTED:
d = (direction or "outbound").strip().lower()
if d not in ("inbound", "outbound"):
raise ActivityIncomplete("a call or email is inbound or outbound")
a.direction = d
if k in PAST:
if due_on:
raise ActivityRuleBroken(f"a {k.lower()} is something that happened — to plan one, log a "
f"TASK with a due date")
when = _aware(occurred_at) if occurred_at else now
if when > now + timedelta(minutes=5):
raise ActivityRuleBroken(f"a {k.lower()} cannot have happened in the future ({when.isoformat()})")
a.occurred_at = when.isoformat()
else:
if not due_on:
raise ActivityIncomplete("a task needs a date it is due, or it is a note")
a.due_on, a.status = _date(due_on), OPEN
return a
@property
def is_task(self) -> bool:
return self.kind == TASK
@property
def at(self) -> str:
"""Where it sits on a timeline: when it happened, or for a task when it was done, or is due."""
return self.occurred_at or self.done_at or f"{self.due_on}T00:00:00+00:00"
def is_overdue(self, today: str) -> bool:
return self.is_task and self.status == OPEN and self.due_on < today
def complete(self, when: datetime, note: Optional[str] = None) -> bool:
"""Done. What the task said is kept; a closing note is appended. False when it already was."""
self._task("completing")
if self.status == DONE:
return False
self.status, self.done_at = DONE, when.isoformat()
if note and note.strip():
self.note = f"{self.note}\n\n{note.strip()}".strip()
return True
def cancel(self) -> bool:
self._task("cancelling")
if self.status == CANCELLED:
return False
self.status, self.done_at = CANCELLED, None
return True
def reopen(self) -> bool:
self._task("reopening")
if self.status == OPEN:
return False
self.status, self.done_at = OPEN, None
return True
def reschedule(self, due_on: str) -> None:
self._task("rescheduling")
if self.status != OPEN:
raise ActivityRuleBroken(f"this task is {self.status}; reopen it to reschedule it")
self.due_on = _date(due_on)
def _task(self, doing: str) -> None:
if not self.is_task:
raise ActivityRuleBroken(f"a {self.kind.lower()} is a record of what happened; {doing} is "
f"for tasks")
def task_board(tasks, today: str) -> dict:
"""CRM-12: a person's open tasks as overdue, today and this week (the next seven days)."""
week_end = (date.fromisoformat(today) + timedelta(days=7)).isoformat()
board = {"overdue": [], "today": [], "this_week": []}
for t in sorted((t for t in tasks if t.is_task and t.status == OPEN), key=lambda t: (t.due_on, t.subject)):
if t.due_on < today:
board["overdue"].append(t)
elif t.due_on == today:
board["today"].append(t)
elif t.due_on <= week_end:
board["this_week"].append(t)
return board
def _aware(iso: str) -> datetime:
try:
d = datetime.fromisoformat(str(iso).replace("Z", "+00:00"))
except ValueError:
raise ActivityIncomplete(f"{iso!r} is not a time (ISO 8601)") from None
return d if d.tzinfo else d.replace(tzinfo=timezone.utc)
def _date(v: str) -> str:
try:
return date.fromisoformat(str(v)[:10]).isoformat()
except ValueError:
raise ActivityIncomplete(f"{v!r} is not a date (YYYY-MM-DD)") from None