"""The Project aggregate: one piece of delivery for one customer from one sale, how a sale becomes its items
(PJ-01, PJ-20, PJ-21), and the reconciliation of sold against delivered when its window closes (PJ-06, PJ-12).
No database, no HTTP, no identity.
"""
import json
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Optional
from brick_projects.domain.errors import RuleBroken
from brick_projects.domain.events import Moved
from brick_projects.domain.item import Item
from brick_projects.domain.values import ItemSource, ItemState, ProjectState, Settings
SOURCE_KINDS = ("contract", "sales_order", "manual")
def project_key(source_kind: str, source_id: str) -> str:
"""pj:<source_kind>:<source_id> — the same sale always names the same project."""
return f"pj:{source_kind}:{source_id}"
@dataclass(frozen=True)
class SoldLine:
"""One line of the sale as the seller sends it. This module never reads the seller's tables (the
seam rule): what it needs arrives in the call."""
line_id: str
title: str
quantity: int
billing_kind: str
billing_plan: str
product_id: str = ""
due_on: Optional[date] = None
@dataclass
class Project:
"""one piece of delivery for one customer from one sale"""
project_id: str
bp_id: int = 0
source_kind: str = ""
source_id: str = ""
quote_id: str = ""
owner: str = ""
starts_on: Optional[date] = None
ends_on: Optional[date] = None
reconciled_at: Optional[datetime] = None
reconciliation_json: str = ""
state: str = ProjectState.PLANNED.value
events: list = field(default_factory=list, repr=False, compare=False)
def _moved(self, kind: str, frm: str, by: str, detail: str = "") -> Moved:
e = Moved(self.project_id, kind, frm, self.state, by, detail)
self.events.append(e)
return e
@property
def reconciliation(self) -> Optional[dict]:
return json.loads(self.reconciliation_json) if self.reconciliation_json else None
def activate(self, *, by: str = "") -> Optional[Moved]:
"""activate: planned → active, when its first item starts (the use case calls it then)."""
if self.state != ProjectState.PLANNED:
return None # already active (or closed): starting another item changes nothing
self.state = ProjectState.ACTIVE.value
return self._moved("activate", ProjectState.PLANNED.value, by, "the first item started")
def reconcile(self, items: list, *, now: datetime, by: str = "") -> Optional[Moved]:
"""reconcile: active → closed (PJ-12). Stores, per sold line, sold, delivered, short, over and
descoped, once: a project already closed is left as it is."""
if self.state == ProjectState.CLOSED:
return None
if self.state != ProjectState.ACTIVE:
raise RuleBroken(f"a {self.state} project cannot be reconciled: only an active one closes")
result = reconcile_lines([i for i in items if i.project_id == self.project_id])
self.reconciliation_json = json.dumps(result, sort_keys=True)
self.reconciled_at = now
self.state = ProjectState.CLOSED.value
short = sum(line["short"] for line in result["lines"])
return self._moved("reconcile", ProjectState.ACTIVE.value, by,
f"{short} short" if short else "delivered in full")
def cancel(self, reason: str, *, by: str = "") -> Optional[Moved]:
"""cancel_project: planned or active → cancelled."""
if self.state == ProjectState.CANCELLED:
return None
if self.state not in (ProjectState.PLANNED, ProjectState.ACTIVE):
raise RuleBroken(f"a {self.state} project cannot be cancelled")
if not (reason or "").strip():
raise RuleBroken("say why the project is cancelled")
frm = self.state
self.state = ProjectState.CANCELLED.value
return self._moved("cancel_project", frm, by, reason.strip())
def is_short(self, item: Item) -> bool:
"""Derived, never stored: sold work still open when the window closed."""
return (self.state == ProjectState.CLOSED and item.source == ItemSource.SOLD.value and item.is_open)
def plan_sale(*, source_kind: str, source_id: str, bp_id: int, starts_on: Optional[date],
ends_on: Optional[date], lines: list, settings: Settings, owner: str, quote_id: str = "",
by: str = "") -> tuple:
"""A sale becomes (Project, [Item]) — or (None, []) when nothing in it is delivered as work.
Each line is expanded by the customer's rule for its billing kind and plan: per_unit (one item per unit
sold), one (a single item), none (it elapses and invoicing's schedule bills it). Keys are derived, so
the same sale planned again names the same project and items: the caller keeps those it already has.
"""
if source_kind not in SOURCE_KINDS:
raise RuleBroken(f"a sale is a {', '.join(SOURCE_KINDS)}, not {source_kind!r}")
if not (source_id or "").strip():
raise RuleBroken("say which sale this is (source_id)")
if not bp_id:
raise RuleBroken("say which customer this is for (bp_id)")
if starts_on and ends_on and ends_on < starts_on:
raise RuleBroken("the project cannot end before it starts")
seen = set()
for line in lines:
if not line.line_id.strip():
raise RuleBroken("every sold line needs its line_id")
if "|" in line.line_id:
raise RuleBroken(f"a line_id cannot contain '|': {line.line_id}")
if line.line_id in seen:
raise RuleBroken(f"line {line.line_id} is sent twice")
seen.add(line.line_id)
if line.quantity < 1:
raise RuleBroken(f"line {line.line_id} sells nothing (quantity {line.quantity})")
pid = project_key(source_kind, source_id.strip())
items = []
for line in lines:
how = settings.expansion(line.billing_kind, line.billing_plan)
count = line.quantity if how == "per_unit" else 1 if how == "one" else 0
for n in range(1, count + 1):
title = f"{line.title} ({n} of {count})" if count > 1 else line.title
items.append(Item.planned(
f"{pid}|{line.line_id}|{n}", by=by, project_id=pid, bp_id=bp_id, source_kind=source_kind,
source_id=source_id.strip(), line_id=line.line_id, product_id=line.product_id,
billing_kind=line.billing_kind, billing_plan=line.billing_plan, title=title, owner=owner,
due_on=line.due_on or ends_on))
if not items:
return None, []
project = Project(project_id=pid, bp_id=bp_id, source_kind=source_kind, source_id=source_id.strip(),
quote_id=quote_id or "", owner=owner, starts_on=starts_on, ends_on=ends_on)
project._moved("plan_project", "none", by, f"{len(items)} item(s) from {source_kind} {source_id}")
return project, items
def reconcile_lines(items: list) -> dict:
"""Sold against delivered, per sold line (PJ-06). sold = the line's sold items; delivered = done, plus
any done change order on the line (which is also `over`); descoped = cancelled; short = still open.
Always: sold = delivered - over + short + descoped. Makegoods are listed apart: they carry no line."""
lines: dict = {}
for it in items:
if it.source not in (ItemSource.SOLD.value, ItemSource.CHANGE_ORDER.value) or not it.line_id:
continue
row = lines.setdefault(it.line_id, {"line_id": it.line_id, "sold": 0, "delivered": 0, "short": 0,
"over": 0, "descoped": 0})
done = it.state == ItemState.DONE
if it.source == ItemSource.CHANGE_ORDER.value:
row["delivered"] += done
row["over"] += done
continue
row["sold"] += 1
if done:
row["delivered"] += 1
elif it.state == ItemState.CANCELLED:
row["descoped"] += 1
else:
row["short"] += 1
makegoods = [{"item_id": it.item_id, "for": it.makegood_for, "state": it.state}
for it in items if it.source == ItemSource.MAKEGOOD.value]
short_items = sorted(it.item_id for it in items if it.source == ItemSource.SOLD.value and it.is_open)
return {"lines": [lines[k] for k in sorted(lines)], "short_items": short_items, "makegoods": makegoods}