feat: Phase 2A — product cost tracking

- Products: add cost_simple (flat €) and cost_breakdown (JSON line items)
- OrderItems: add unit_cost snapshot written at time of order creation
- Snapshot logic: breakdown sum takes priority over cost_simple; NULL if neither set
- Product schema: serialize/deserialize cost_breakdown as JSON; expose in API
- Product performance report: add trackable_profit, uncosted_revenue, has_gap per product
- Manager UI: cost section in product edit form (simple / detailed toggle with live margin %)
- Products list: show margin % or cost-error badge per product
- Product performance report: profit column + gap warning banner

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-06-07 20:46:33 +03:00
parent ac7ec45279
commit d0dfd63415
10 changed files with 287 additions and 31 deletions

View File

@@ -270,12 +270,25 @@ def add_items(
(o.price_delta or o.extra_cost or 0.0)
for o in (item_in.selected_options or [])
)
# Phase 2A: cost snapshot — breakdown takes priority over simple cost
unit_cost = None
if product.cost_breakdown:
try:
entries = json.loads(product.cost_breakdown)
total = sum(e.get("amount", 0.0) for e in entries if isinstance(e, dict))
if total > 0:
unit_cost = total
except Exception:
pass
if unit_cost is None and product.cost_simple:
unit_cost = product.cost_simple
item = OrderItem(
order_id=order_id,
product_id=item_in.product_id,
added_by=user.id,
quantity=item_in.quantity,
unit_price=product.base_price + extra_cost,
unit_cost=unit_cost,
selected_options=json.dumps([o.model_dump() for o in item_in.selected_options]) if item_in.selected_options else None,
removed_ingredients=json.dumps(item_in.removed_ingredients) if item_in.removed_ingredients else None,
notes=item_in.notes,

View File

@@ -235,9 +235,11 @@ def reorder_products(items: List[ProductReorderItem], db: Session = Depends(get_
@router.post("/", response_model=ProductOut, status_code=status.HTTP_201_CREATED)
def create_product(body: ProductCreate, db: Session = Depends(get_db), user: User = Depends(require_manager)):
data = body.model_dump(exclude={"quick_options", "options", "ingredients", "preference_sets"})
data = body.model_dump(exclude={"quick_options", "options", "ingredients", "preference_sets", "cost_breakdown"})
if data.get("sort_order") == 0:
data["sort_order"] = db.query(Product).count()
if body.cost_breakdown is not None:
data["cost_breakdown"] = json.dumps([item.model_dump() for item in body.cost_breakdown])
product = Product(**data)
db.add(product)
db.flush()
@@ -277,8 +279,18 @@ def update_product(product_id: int, body: ProductUpdate, db: Session = Depends(g
product = db.query(Product).filter(Product.id == product_id).first()
if not product:
raise HTTPException(status_code=404, detail="Product not found")
for field, value in body.model_dump(exclude_none=True, exclude={"quick_options", "options", "ingredients", "preference_sets"}).items():
scalar_fields = body.model_dump(
exclude_none=True,
exclude={"quick_options", "options", "ingredients", "preference_sets", "cost_breakdown"},
)
for field, value in scalar_fields.items():
setattr(product, field, value)
# cost_breakdown is a list of objects — serialize to JSON for storage
if body.cost_breakdown is not None:
product.cost_breakdown = json.dumps([item.model_dump() for item in body.cost_breakdown])
elif "cost_breakdown" in body.model_fields_set:
# explicitly set to null — clear it
product.cost_breakdown = None
if body.quick_options is not None:
_replace_quick_options(db, product, body.quick_options)
if body.options is not None:

View File

@@ -950,16 +950,29 @@ def product_performance(
"category_id": product.category_id if product else None,
"qty_sold": 0,
"revenue": 0.0,
"trackable_profit": 0.0,
"uncosted_revenue": 0.0,
"uncosted_items": 0,
"order_ids": set(),
}
summary[pid]["qty_sold"] += item.quantity
summary[pid]["revenue"] += item.unit_price * item.quantity
qty = item.quantity
revenue = item.unit_price * qty
summary[pid]["qty_sold"] += qty
summary[pid]["revenue"] += revenue
summary[pid]["order_ids"].add(item.order_id)
if item.unit_cost is not None:
summary[pid]["trackable_profit"] += revenue - (item.unit_cost * qty)
else:
summary[pid]["uncosted_revenue"] += revenue
summary[pid]["uncosted_items"] += qty
result = []
for entry in summary.values():
entry["order_count"] = len(entry.pop("order_ids"))
entry["revenue"] = round(entry["revenue"], 2)
entry["trackable_profit"] = round(entry["trackable_profit"], 2)
entry["uncosted_revenue"] = round(entry["uncosted_revenue"], 2)
entry["has_gap"] = entry["uncosted_items"] > 0
result.append(entry)
result.sort(key=lambda x: x["qty_sold"], reverse=True)