"""The Product aggregate — the rules, and nothing else.
Something the business sells (or buys), goods or a service, with how it is billed and taxed.
No database, no HTTP, no identity: a rule here is true whoever calls it and however it is stored.
Three decisions an implementer must not undo (module.yaml): the id is the customer's own code, fixed once set;
a product carries a TAX CATEGORY, never a tax code or a rate; a product is never deleted once a document names it.
"""
import json
import re
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from typing import Iterable, Optional
from brick_product.domain.errors import RuleBroken
from brick_product.domain.events import (
DetailsSuggested,
DuplicateChecked,
ProductActivated,
ProductAdded,
ProductArchived,
ProductCompleted,
ProductReactivated,
ProductRenamed,
ProductTypeChanged,
UnitAdded,
UnitsSet,
)
from brick_product.domain.values import (
ProductBillingKind,
ProductBillingPlan,
ProductInvoicingPolicy,
ProductKind,
ProductState,
)
CODE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,63}$")
PLAN_FOR = {ProductBillingKind.ONE_OFF.value: (ProductBillingPlan.SINGLE.value,),
ProductBillingKind.RECURRING.value: (ProductBillingPlan.PERIODIC.value, ProductBillingPlan.ARREARS.value),
ProductBillingKind.METERED.value: (ProductBillingPlan.ARREARS.value, ProductBillingPlan.PERIODIC.value)}
COST_LIMIT = Decimal("1e14")
STOP = {"a", "an", "and", "the", "of", "for", "per", "in", "to", "with", "on", "by"}
def valid_code(code: str) -> str:
"""A product code as the customer types it (a SKU): letters, digits and . _ / - ; fixed once set."""
code = (code or "").strip()
if not CODE.match(code):
raise RuleBroken("a product code is up to 64 letters, digits and . _ / - characters, starting with a letter "
"or digit")
return code
def _words(text: str) -> set:
"""The words that say what a product IS: lower case, no filler, a trailing -ly/-s/-ing/-ed folded away
(monthly ≈ month, hosting ≈ host), so 'Managed hosting, per month' and 'Monthly managed hosting' meet."""
out = set()
for w in re.findall(r"[a-z0-9]+", (text or "").lower()):
if w in STOP:
continue
for suffix in ("ing", "ly", "ed", "es", "s"):
if len(w) > len(suffix) + 3 and w.endswith(suffix):
w = w[:-len(suffix)]
break
out.add(w)
return out
def likely_duplicates(name: str, description: str, others, *, limit: int = 3, threshold: float = 0.5) -> list:
"""The fixed baseline of the duplicate check (PRD-12): products whose name + description share most of their
words with this one — Jaccard similarity, no model. `others`: (product_id, name, description) triples.
Up to `limit`, best first; it only ever SUGGESTS."""
mine = _words(f"{name} {description}")
found = []
for pid, other_name, other_description in others:
theirs = _words(f"{other_name} {other_description}")
if not mine or not theirs:
continue
score = len(mine & theirs) / len(mine | theirs)
if score >= threshold:
found.append({"product_id": pid, "name": other_name, "score": round(score, 2)})
return sorted(found, key=lambda f: (-f["score"], f["product_id"]))[:limit]
@dataclass
class Product:
"""something the business sells (or buys), goods or a service, with how it is billed and taxed"""
product_id: str
name: str = ""
description: str = ""
kind: str = ProductKind.GOODS.value
base_unit_code: str = "" # the unit it is stocked in; every other unit converts to it exactly (PRD-23)
sales_unit_code: str = "" # the unit a price is asked in by default, and a quote line counts in
purchase_unit_code: str = ""
issue_unit_code: str = "" # the unit production or a warehouse issues it in
category_code: str = ""
tax_category: str = ""
billing_kind: str = ""
billing_plan: str = ""
invoicing_policy: str = ""
cost_amount: Decimal = Decimal("0")
product_template_id: str = "" # the template it is a variant of ("" : a product with no variants is its own template)
attribute_values: str = "" # JSON {attribute: value} of this variant
product_type_code: str = "" # v4 (PRD-35): MANDATORY; the type it was added with. The type's values below are COPIED at creation
stocked_default: Optional[bool] = None
sellable: Optional[bool] = None # copied from the type (a later edit of the type reaches no existing product; change_product_type re-copies)
valuation_class: str = "" # copied from the type
required_roles: str = "" # copied from the type: JSON list of the unit roles activation needs (gate type_requirements)
commodity_code: str = "" # the 8-digit CN code (Intrastat, customs), read by tax
country_of_origin: str = "" # ISO 3166 alpha-2, read by tax and ar
status: str = ProductState.DRAFT.value
history: list = field(default_factory=list) # prod_product_log rows: what happened, by whom
events: list = field(default_factory=list, repr=False, compare=False)
def _record(self, event):
self.events.append(event)
return event
def _log(self, event: str, by: str, detail: str = "") -> None:
self.history.append({"product_id": self.product_id, "seq": len(self.history) + 1, "event": event,
"actor": by, "detail": detail,
"event_at": datetime.now(timezone.utc).replace(tzinfo=None)})
# ── what a value may be ───────────────────────────────────────────────────────────────────────────────
def _set(self, **given) -> list:
"""Apply the fields given (never the code), checking each; returns what changed as 'field: a → b'."""
enums = {"kind": ProductKind, "billing_kind": ProductBillingKind, "billing_plan": ProductBillingPlan,
"invoicing_policy": ProductInvoicingPolicy}
changes = []
old_base = self.base_unit_code
for key, value in given.items():
if value is None:
continue
if key in enums and value not in {m.value for m in enums[key]}:
raise RuleBroken(f"{key} is one of {', '.join(m.value for m in enums[key])}, not '{value}'")
if key == "name" and not str(value).strip():
raise RuleBroken("a product needs a name")
if key == "commodity_code" and value and not re.fullmatch(r"[0-9]{8}", str(value).strip()):
raise RuleBroken("a commodity code is the 8-digit CN code, for example 22021000")
if key == "country_of_origin" and value:
value = str(value).strip().upper()
if not re.fullmatch(r"[A-Z]{2}", value):
raise RuleBroken("a country of origin is an ISO 3166 two-letter code, for example GB")
if key == "cost_amount":
value = Decimal(value)
if value < 0 or value >= COST_LIMIT:
raise RuleBroken("the standard cost is zero or more")
if isinstance(value, str):
value = value.strip()
if getattr(self, key) != value:
changes.append(f"{key}: {getattr(self, key)} → {value}")
setattr(self, key, value)
if old_base and self.base_unit_code != old_base: # a unit that WAS the base follows it to the new one
for role in ("sales_unit_code", "purchase_unit_code", "issue_unit_code"):
if getattr(self, role) == old_base:
setattr(self, role, self.base_unit_code)
if self.base_unit_code: # a product counted in one unit counts in it everywhere, until told otherwise
for role in ("sales_unit_code", "purchase_unit_code", "issue_unit_code"):
if not getattr(self, role):
setattr(self, role, self.base_unit_code)
if self.billing_kind and not self.billing_plan:
self.billing_plan = PLAN_FOR[self.billing_kind][0] # recurring bills periodically, one-off once
if self.billing_kind and self.billing_plan not in PLAN_FOR[self.billing_kind]:
raise RuleBroken(f"a {self.billing_kind} product is billed {' or '.join(PLAN_FOR[self.billing_kind])}, "
f"not {self.billing_plan}")
return changes
def missing_for_sale(self) -> list:
"""What stops this product being offered to sales (PRD-02), in words."""
need = [("a base unit", self.base_unit_code), ("a sales unit", self.sales_unit_code),
("a tax category", self.tax_category),
("a billing type", self.billing_kind and self.billing_plan),
("an invoicing policy", self.invoicing_policy)]
return [what for what, value in need if not value]
# ── the lifecycle (module.yaml workflows) ──────────────────────────────────────────────────────────────
@classmethod
def add_product(cls, product_id: str, *, by: str = "", name: str = "", type_row: Optional[dict] = None, **attrs) -> "Product":
"""Add the product: none → draft (PRD-01). The code (a SKU) must be well formed and is fixed from now on;
that no other product has it is the repository's to say (the use case asks it first). A product always has a type
(PRD-35): `type_row` is the type's row, whose kind, stocked default, sellable, valuation class and required unit roles
are copied onto the product now."""
it = cls(product_id=valid_code(product_id))
if not str(name or "").strip():
raise RuleBroken("a product needs a name")
if not type_row:
raise RuleBroken("a product needs a product_type_code")
it._apply_type(type_row)
it._set(name=name, **attrs)
it.status = ProductState.DRAFT.value
it._log("added", by, it.name)
it._record(ProductAdded(product_id=it.product_id, by=by))
return it
def _apply_type(self, t: dict) -> None:
"""Copy the type's values onto the product (a later edit of the type reaches no existing product)."""
self.product_type_code, self.kind, self.stocked_default = t["product_type_code"], t["kind"], t["stocked_default"]
self.sellable, self.valuation_class = bool(t["sellable"]), t.get("valuation_class") or ""
self.required_roles = json.dumps(list(t.get("required_roles") or []))
def legacy_retype_ok(self, type_row: dict) -> bool:
"""v5: a LEGACY product (it pre-dates mandatory types) may take a real type even when active and used, once, as long as the
type's kind and stocked default equal the product's own (an unknown stocked default equals only an unknown one): nothing about
how modules treat it changes."""
return (self.product_type_code == "LEGACY" and type_row["product_type_code"] != "LEGACY"
and type_row["kind"] == self.kind and type_row["stocked_default"] == self.stocked_default)
def change_product_type(self, type_row: dict, reason: str, *, by: str = "") -> Optional[ProductTypeChanged]:
"""Change the product's type: draft → draft (PRD-35). The use case holds the gate (type_unused: nothing has used the
product). Effect: kind and stocked_default follow the new type; the code is kept; the change is recorded with its reason."""
if self.status != ProductState.DRAFT and not (self.status == ProductState.ACTIVE and self.legacy_retype_ok(type_row)):
raise RuleBroken(f"{self.product_id} is {self.status}: a type changes while the product is a draft (a LEGACY product may "
"take a type of its own kind and stocked default once)")
if not (reason or "").strip():
raise RuleBroken("say why the type changes: a reason is required")
if type_row["product_type_code"] == self.product_type_code:
return None
before = self.product_type_code or "no type"
self._apply_type(type_row)
self._log("type changed", by, f"{before} → {self.product_type_code}: {reason.strip()}")
return self._record(ProductTypeChanged(product_id=self.product_id, product_type_code=self.product_type_code, by=by))
def ai_check_duplicate(self, mode: str, *, by: str = "") -> DuplicateChecked:
"""AI checks for a likely duplicate (suggest): draft → draft. Gate: the duplicate check is on. Effect: a
likely duplicate is shown; nothing is merged — the product itself does not change."""
if self.status != ProductState.DRAFT:
raise RuleBroken(f"{self.product_id} is {self.status}: only a draft product is checked for duplicates")
if mode == "off":
raise RuleBroken("the duplicate check is switched off in the settings")
return self._record(DuplicateChecked(product_id=self.product_id, by=by))
def ai_suggest_details(self, mode: str, *, by: str = "") -> DetailsSuggested:
"""AI suggests category, unit and tax category (suggest): draft → draft. Gate: a field is empty and the
step is on. Effect: a suggestion the catalogue manager accepts or changes — nothing is filled in."""
if self.status != ProductState.DRAFT:
raise RuleBroken(f"{self.product_id} is {self.status}: details are suggested for a draft product")
if mode == "off":
raise RuleBroken("suggesting details is switched off in the settings")
if self.category_code and self.base_unit_code and self.tax_category:
raise RuleBroken(f"{self.product_id} has its category, unit and tax category: nothing left to suggest")
return self._record(DetailsSuggested(product_id=self.product_id, by=by))
def set_units(self, *, by: str = "", **fields) -> UnitsSet:
"""Set the base unit and the other units: draft → draft (PRD-23). Gate: every unit converts exactly to the base unit
(the use case holds the conversions and checks it). Effect: purchase, sales and issue units chosen."""
if self.status != ProductState.DRAFT:
raise RuleBroken(f"{self.product_id} is {self.status}: its units are set while it is a draft "
"(an active product gets a new unit with add_unit)")
allowed = {k: v for k, v in fields.items() if k in ("base_unit_code", "sales_unit_code", "purchase_unit_code",
"issue_unit_code")}
changes = self._set(**allowed)
self._log("units", by, "; ".join(changes))
return self._record(UnitsSet(product_id=self.product_id, by=by))
def add_unit(self, unit_code: str, *, by: str = "", roles: Iterable[str] = ()) -> UnitAdded:
"""Add a unit or a pack size: active → active (PRD-23). Gate: a new unit code with an exact conversion, and a
conversion already used is never changed (the use case holds the rows). Effect: the unit can be bought, sold or
issued from now — `roles` names which of purchase / sales / issue it becomes."""
if self.status != ProductState.ACTIVE:
raise RuleBroken(f"{self.product_id} is {self.status}: a unit is added to an active product")
roles = list(roles)
if [r for r in roles if r not in ("purchase", "sales", "issue")]:
raise RuleBroken("a unit becomes the purchase, sales or issue unit")
changes = self._set(**{f"{r}_unit_code": unit_code for r in roles})
self._log("unit added", by, "; ".join([unit_code] + changes))
return self._record(UnitAdded(product_id=self.product_id, unit_code=unit_code, by=by))
def complete_product(self, *, by: str = "", **fields) -> ProductCompleted:
"""Complete unit, category, tax category and billing: draft → draft."""
if self.status != ProductState.DRAFT:
raise RuleBroken(f"{self.product_id} is {self.status}: only a draft product is completed")
changes = self._set(**fields)
self._log("completed", by, "; ".join(changes))
return self._record(ProductCompleted(product_id=self.product_id, by=by))
def activate_product(self, *, by: str = "") -> Optional[ProductActivated]:
"""Activate the product: draft → active (PRD-02). Gate: unit, tax category, billing type and invoicing
policy set. Effect: offered to sales."""
if self.status == ProductState.ACTIVE:
return None # already active: a repeat changes nothing
if self.status != ProductState.DRAFT:
raise RuleBroken(f"a {self.status} product cannot be activated")
missing = self.missing_for_sale()
if missing:
raise RuleBroken(f"{self.product_id} cannot be offered to sales yet: it is missing "
+ ", ".join(missing))
self.status = ProductState.ACTIVE.value
self._log("activated", by)
return self._record(ProductActivated(product_id=self.product_id, by=by))
def rename_product(self, *, by: str = "", **fields) -> ProductRenamed:
"""Change name or description: active → active (PRD-04). Also the category and the standard cost, which
move without touching what a document copied. Effect: documents already raised keep their copy — they
hold their own."""
if self.status != ProductState.ACTIVE:
raise RuleBroken(f"{self.product_id} is {self.status}: only an active product is changed")
allowed = {k: v for k, v in fields.items() if k in ("name", "description", "category_code", "cost_amount")}
if not any(v is not None for v in allowed.values()):
raise RuleBroken("say what changes: the name, description, category or standard cost")
changes = self._set(**allowed)
self._log("changed", by, "; ".join(changes))
return self._record(ProductRenamed(product_id=self.product_id, by=by))
def archive_product(self, *, by: str = "") -> Optional[ProductArchived]:
"""Archive the product: active → archived (PRD-03). Effect: no new lines; existing documents unchanged
(they hold their own copy; nothing is deleted)."""
if self.status == ProductState.ARCHIVED:
return None # already archived: a repeat changes nothing
if self.status != ProductState.ACTIVE:
raise RuleBroken(f"a {self.status} product cannot be archived: only an active one")
self.status = ProductState.ARCHIVED.value
self._log("archived", by)
return self._record(ProductArchived(product_id=self.product_id, by=by))
def reactivate_product(self, reason: str, *, by: str = "") -> Optional[ProductReactivated]:
"""Reactivate the product: archived → active (PRD-03): allowed, and recorded with who and why."""
if self.status == ProductState.ACTIVE:
return None # already active: a repeat changes nothing
if self.status != ProductState.ARCHIVED:
raise RuleBroken(f"a {self.status} product cannot be reactivated: only an archived one")
if not (reason or "").strip():
raise RuleBroken("say why it is back on sale: a reason is required")
self.status = ProductState.ACTIVE.value
self._log("reactivated", by, reason.strip())
return self._record(ProductReactivated(product_id=self.product_id, reason=reason.strip(), by=by))
# ── what a document copies (PRD-21) ───────────────────────────────────────────────────────────────────
def line_copy(self) -> dict:
"""What a quote, order or invoice line copies at the moment it is added; it never asks again."""
return {"product_id": self.product_id, "name": self.name, "description": self.description,
"base_unit_code": self.base_unit_code, "sales_unit_code": self.sales_unit_code, "tax_category": self.tax_category, "billing_kind": self.billing_kind,
"billing_plan": self.billing_plan, "invoicing_policy": self.invoicing_policy}
def offered(self) -> bool:
return self.status == ProductState.ACTIVE