- Mix Calculator: searchable Mix Name picker (mirrors Throughput search) - Ingredients Editor: add manual Category column; used to order Mix Calculator output - Mix Calculator: surface formula-only mixes (no product yet) via -mix_id sentinel - Throughput: remove unused For order / For stock destination controls from composer - Editor change history: show timestamps in local time (stored UTC) instead of raw UTC Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
653 lines
27 KiB
Python
653 lines
27 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session, joinedload, selectinload
|
|
|
|
from app.api.deps import AuthSession
|
|
from app.models.mix import Mix, MixIngredient
|
|
from app.models.mix_calculator import MixCalculatorSession, MixCalculatorSessionLine
|
|
from app.models.product import Product, ProductIngredient
|
|
from app.schemas.mix_calculator import MixCalculatorSessionCreate, MixCalculatorSessionUpdate
|
|
from app.services.costing_engine import extract_unit_quantity_kg
|
|
|
|
|
|
def can_view_all_mix_calculator_sessions(session: AuthSession) -> bool:
|
|
return session.client_role in {"superadmin", "admin"}
|
|
|
|
|
|
def _build_session_access_query(session: AuthSession):
|
|
query = select(MixCalculatorSession).where(MixCalculatorSession.tenant_id == session.tenant_id)
|
|
if can_view_all_mix_calculator_sessions(session):
|
|
return query
|
|
return query.where(MixCalculatorSession.prepared_by_user_id == session.user_id)
|
|
|
|
|
|
def _load_product_for_calculation(db: Session, tenant_id: str, product_id: int) -> Product | None:
|
|
return db.scalar(
|
|
select(Product)
|
|
.where(Product.id == product_id, Product.tenant_id == tenant_id, Product.visible.is_(True))
|
|
.options(
|
|
selectinload(Product.ingredients).selectinload(ProductIngredient.raw_material),
|
|
selectinload(Product.mix).selectinload(Mix.ingredients).selectinload(MixIngredient.raw_material),
|
|
)
|
|
)
|
|
|
|
|
|
def _category_sort_key(category: str | None) -> tuple[int, str]:
|
|
"""Order ingredients by their manual category; uncategorised rows sort last."""
|
|
cleaned = (category or "").strip()
|
|
if not cleaned:
|
|
return (1, "")
|
|
return (0, cleaned.lower())
|
|
|
|
|
|
def _order_formula_rows(rows: list[dict]) -> list[dict]:
|
|
"""Sort rows by category (then their original order/name) and renumber.
|
|
|
|
Category is the primary key so the Mix Calculator groups ingredients by their
|
|
manually-assigned category. `sort_order` is reassigned sequentially after the
|
|
sort so every downstream consumer (lines, PDF) follows the same order.
|
|
"""
|
|
rows.sort(
|
|
key=lambda row: (
|
|
_category_sort_key(row.get("category")),
|
|
row.get("sort_order") or 0,
|
|
row["raw_material_name"].lower(),
|
|
)
|
|
)
|
|
for index, row in enumerate(rows, start=1):
|
|
row["sort_order"] = index
|
|
return rows
|
|
|
|
|
|
def _resolved_formula_rows(product: Product) -> tuple[list[dict], float]:
|
|
if product.ingredients:
|
|
rows = [
|
|
{
|
|
"raw_material_id": ingredient.raw_material_id,
|
|
"raw_material_name": ingredient.raw_material.name,
|
|
"quantity_kg": ingredient.quantity_kg,
|
|
"unit": ingredient.raw_material.unit_of_measure,
|
|
"rounding_decimals": ingredient.raw_material.rounding_decimals,
|
|
"category": ingredient.raw_material.category,
|
|
"sort_order": ingredient.sort_order,
|
|
}
|
|
for ingredient in product.ingredients
|
|
if ingredient.raw_material is not None
|
|
]
|
|
elif product.mix is not None:
|
|
rows = [
|
|
{
|
|
"raw_material_id": ingredient.raw_material_id,
|
|
"raw_material_name": ingredient.raw_material.name if ingredient.raw_material is not None else f"Raw material {ingredient.raw_material_id}",
|
|
"quantity_kg": ingredient.quantity_kg,
|
|
"unit": ingredient.raw_material.unit_of_measure if ingredient.raw_material is not None else "kg",
|
|
"rounding_decimals": ingredient.raw_material.rounding_decimals if ingredient.raw_material is not None else 2,
|
|
"category": ingredient.raw_material.category if ingredient.raw_material is not None else None,
|
|
"sort_order": index,
|
|
}
|
|
for index, ingredient in enumerate(product.mix.ingredients, start=1)
|
|
]
|
|
else:
|
|
rows = []
|
|
|
|
_order_formula_rows(rows)
|
|
return rows, round(sum(row["quantity_kg"] for row in rows), 4)
|
|
|
|
|
|
def _mix_formula_rows(mix: Mix) -> tuple[list[dict], float]:
|
|
"""Resolve a mix's own (mix-master) formula rows, category-ordered.
|
|
|
|
Used by the Mix Calculator for mixes that have a formula but no representative
|
|
product yet — the formula lives directly on the mix.
|
|
"""
|
|
rows = [
|
|
{
|
|
"raw_material_id": ingredient.raw_material_id,
|
|
"raw_material_name": ingredient.raw_material.name if ingredient.raw_material is not None else f"Raw material {ingredient.raw_material_id}",
|
|
"quantity_kg": ingredient.quantity_kg,
|
|
"unit": ingredient.raw_material.unit_of_measure if ingredient.raw_material is not None else "kg",
|
|
"rounding_decimals": ingredient.raw_material.rounding_decimals if ingredient.raw_material is not None else 2,
|
|
"category": ingredient.raw_material.category if ingredient.raw_material is not None else None,
|
|
"sort_order": index,
|
|
}
|
|
for index, ingredient in enumerate(mix.ingredients, start=1)
|
|
]
|
|
_order_formula_rows(rows)
|
|
return rows, round(sum(row["quantity_kg"] for row in rows), 4)
|
|
|
|
|
|
def _fractional_bag_warning(batch_size_kg: float, total_bags: float, unit_of_measure: str) -> str | None:
|
|
rounded_bags = round(total_bags)
|
|
if abs(total_bags - rounded_bags) < 1e-9:
|
|
return None
|
|
return (
|
|
f"Batch size {batch_size_kg:g}kg produces {total_bags:.2f} bags for {unit_of_measure}. "
|
|
"This is not a whole-bag quantity."
|
|
)
|
|
|
|
|
|
def _mix_calculator_label(product: Product) -> str:
|
|
return product.mix.name if product.mix else product.name
|
|
|
|
|
|
def _mix_calculator_option_rank(product: Product) -> tuple[int, int, float, int]:
|
|
unit_label = (product.unit_of_measure or "").lower()
|
|
unit_size = extract_unit_quantity_kg(product.unit_of_measure)
|
|
return (
|
|
0 if abs(unit_size - 20) < 1e-9 and "bag" in unit_label and "bulka" not in unit_label else 1,
|
|
0 if "bulka" not in unit_label else 1,
|
|
unit_size if unit_size > 0 else 999999,
|
|
product.id,
|
|
)
|
|
|
|
|
|
def resolve_representative_product(db: Session, *, tenant_id: str, mix_id: int) -> Product | None:
|
|
"""The single product the Mix Calculator surfaces for a given mix.
|
|
|
|
The calculator lists one representative product per (client, mix) and reads
|
|
its formula. The Mix Editor reuses this so it edits exactly what the
|
|
calculator shows. Preference order mirrors `build_mix_calculator_options`:
|
|
visible products that already have a product-specific formula, ranked by
|
|
`_mix_calculator_option_rank`; then any visible product; then any product.
|
|
"""
|
|
products = db.scalars(
|
|
select(Product)
|
|
.where(Product.tenant_id == tenant_id, Product.mix_id == mix_id)
|
|
.options(
|
|
selectinload(Product.ingredients).selectinload(ProductIngredient.raw_material),
|
|
selectinload(Product.mix).selectinload(Mix.ingredients).selectinload(MixIngredient.raw_material),
|
|
)
|
|
).all()
|
|
if not products:
|
|
return None
|
|
with_formula = [product for product in products if product.visible and product.ingredients]
|
|
pool = with_formula or [product for product in products if product.visible] or list(products)
|
|
return min(pool, key=_mix_calculator_option_rank)
|
|
|
|
|
|
def resolve_editor_mix_formula(db: Session, *, tenant_id: str, mix: Mix) -> dict:
|
|
"""Resolve a mix's formula the way the calculator does, for the editor.
|
|
|
|
Returns the resolved ingredient rows (with each row's share of the total as
|
|
`mix_percentage`), the total kg, and where the formula lives:
|
|
`source='product'` (a representative product's own formula) or `source='mix'`
|
|
(the shared mix master fallback). `product_id` names the product that owns the
|
|
formula when `source='product'`. The save path writes back to that same source.
|
|
"""
|
|
product = resolve_representative_product(db, tenant_id=tenant_id, mix_id=mix.id)
|
|
if product is not None and product.ingredients:
|
|
rows, total_kg = _resolved_formula_rows(product)
|
|
# Carry each ingredient's note through so saving doesn't wipe it.
|
|
notes_by_raw_material = {
|
|
ingredient.raw_material_id: ingredient.notes for ingredient in product.ingredients
|
|
}
|
|
for row in rows:
|
|
row["notes"] = notes_by_raw_material.get(row["raw_material_id"])
|
|
source = "product"
|
|
product_id = product.id
|
|
else:
|
|
# No product-specific formula: the calculator reads the shared mix master,
|
|
# so the editor shows and edits that.
|
|
rows = [
|
|
{
|
|
"raw_material_id": ingredient.raw_material_id,
|
|
"raw_material_name": ingredient.raw_material.name
|
|
if ingredient.raw_material is not None
|
|
else f"Raw material {ingredient.raw_material_id}",
|
|
"quantity_kg": ingredient.quantity_kg,
|
|
"unit": ingredient.raw_material.unit_of_measure if ingredient.raw_material is not None else "kg",
|
|
"sort_order": index,
|
|
"notes": ingredient.notes,
|
|
}
|
|
for index, ingredient in enumerate(
|
|
sorted(mix.ingredients, key=lambda item: item.raw_material.name if item.raw_material else ""),
|
|
start=1,
|
|
)
|
|
]
|
|
total_kg = round(sum(row["quantity_kg"] for row in rows), 4)
|
|
source = "mix"
|
|
product_id = product.id if product is not None else None
|
|
|
|
ingredients = [
|
|
{
|
|
"raw_material_id": row["raw_material_id"],
|
|
"raw_material_name": row["raw_material_name"],
|
|
"quantity_kg": round(row["quantity_kg"], 4),
|
|
"mix_percentage": round((row["quantity_kg"] / total_kg) * 100, 4) if total_kg > 0 else 0.0,
|
|
"unit": row["unit"],
|
|
"notes": row.get("notes"),
|
|
}
|
|
for row in rows
|
|
]
|
|
return {
|
|
"id": mix.id,
|
|
"tenant_id": mix.tenant_id,
|
|
"client_name": mix.client_name,
|
|
"name": mix.name,
|
|
"source": source,
|
|
"product_id": product_id,
|
|
"ingredients": ingredients,
|
|
"total_kg": total_kg,
|
|
}
|
|
|
|
|
|
def _scale_preview(
|
|
*,
|
|
values: dict,
|
|
formula_rows: list[dict],
|
|
source_total_kg: float,
|
|
client_name: str,
|
|
product_id: int,
|
|
mix_label: str,
|
|
mix_id: int,
|
|
unit_of_measure: str,
|
|
) -> dict:
|
|
"""Scale a resolved formula to the requested batch size and shape the preview.
|
|
|
|
Shared by the product-backed path and the formula-only mix path; only the
|
|
inputs (where the formula and unit come from) differ.
|
|
"""
|
|
batch_size_kg = float(values["batch_size_kg"])
|
|
scale_factor = batch_size_kg / source_total_kg
|
|
unit_size_kg = extract_unit_quantity_kg(unit_of_measure)
|
|
total_bags = round(batch_size_kg / unit_size_kg, 4) if unit_size_kg > 0 else 0.0
|
|
|
|
warnings: list[str] = []
|
|
# A bag warning only makes sense when the unit resolves to a bag size; a
|
|
# formula-only mix sells in bulk kg, so there's nothing to round to whole bags.
|
|
if unit_size_kg > 0:
|
|
bag_warning = _fractional_bag_warning(batch_size_kg, total_bags, unit_of_measure)
|
|
if bag_warning:
|
|
warnings.append(bag_warning)
|
|
|
|
lines = []
|
|
for index, ingredient in enumerate(formula_rows, start=1):
|
|
mix_percentage = round((ingredient["quantity_kg"] / source_total_kg) * 100, 4)
|
|
required_kg = round(ingredient["quantity_kg"] * scale_factor, 4)
|
|
lines.append(
|
|
{
|
|
"raw_material_id": ingredient["raw_material_id"],
|
|
"raw_material_name": ingredient["raw_material_name"],
|
|
"required_kg": required_kg,
|
|
"mix_percentage": mix_percentage,
|
|
"unit": ingredient["unit"],
|
|
"rounding_decimals": ingredient.get("rounding_decimals", 2),
|
|
"category": ingredient.get("category"),
|
|
"sort_order": ingredient["sort_order"] or index,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"client_name": client_name,
|
|
"product_id": product_id,
|
|
# The source workbook labels this as Product, but for the calculator
|
|
# it is the mix/formula being produced.
|
|
"product_name": mix_label,
|
|
"mix_id": mix_id,
|
|
"mix_name": mix_label,
|
|
"mix_date": values["mix_date"],
|
|
"batch_size_kg": round(batch_size_kg, 4),
|
|
"total_bags": total_bags,
|
|
"total_kg": round(batch_size_kg, 4),
|
|
"product_unit_of_measure": unit_of_measure,
|
|
"product_unit_size_kg": round(unit_size_kg, 4),
|
|
"prepared_by_name": values["prepared_by_name"],
|
|
"status": values.get("status") or "saved",
|
|
"notes": values.get("notes"),
|
|
"warnings": warnings,
|
|
"lines": lines,
|
|
}
|
|
|
|
|
|
def _calculate_mix_only_preview(db: Session, *, tenant_id: str, mix_id: int, values: dict) -> dict:
|
|
"""Preview for a mix that has a formula but no representative product.
|
|
|
|
The Mix Calculator surfaces these via a negative `product_id` sentinel
|
|
(`-mix_id`); the formula is read straight off the mix master and there's no
|
|
product unit, so output is bulk kg with no bag split.
|
|
"""
|
|
mix = db.scalar(
|
|
select(Mix)
|
|
.where(Mix.id == mix_id, Mix.tenant_id == tenant_id)
|
|
.options(selectinload(Mix.ingredients).selectinload(MixIngredient.raw_material))
|
|
)
|
|
if mix is None:
|
|
raise ValueError("Mix not found")
|
|
if mix.client_name != values["client_name"]:
|
|
raise ValueError("Selected mix does not belong to the chosen client")
|
|
formula_rows, source_total_kg = _mix_formula_rows(mix)
|
|
if source_total_kg <= 0:
|
|
raise ValueError("Mix has no formula to scale")
|
|
return _scale_preview(
|
|
values=values,
|
|
formula_rows=formula_rows,
|
|
source_total_kg=source_total_kg,
|
|
client_name=mix.client_name,
|
|
product_id=-mix.id,
|
|
mix_label=mix.name,
|
|
mix_id=mix.id,
|
|
unit_of_measure="kg",
|
|
)
|
|
|
|
|
|
def calculate_mix_calculator_preview(
|
|
db: Session,
|
|
*,
|
|
tenant_id: str,
|
|
payload: MixCalculatorSessionCreate | MixCalculatorSessionUpdate | dict,
|
|
):
|
|
values = payload if isinstance(payload, dict) else payload.model_dump(exclude_unset=False)
|
|
product_id = int(values["product_id"])
|
|
# Negative ids are the sentinel for a formula-only mix (no product yet).
|
|
if product_id < 0:
|
|
return _calculate_mix_only_preview(db, tenant_id=tenant_id, mix_id=-product_id, values=values)
|
|
|
|
product = _load_product_for_calculation(db, tenant_id, product_id)
|
|
if product is None:
|
|
raise ValueError("Product not found")
|
|
if product.client_name != values["client_name"]:
|
|
raise ValueError("Selected product does not belong to the chosen client")
|
|
formula_rows, source_total_kg = _resolved_formula_rows(product)
|
|
if source_total_kg <= 0:
|
|
raise ValueError("Product has no source kilograms to scale")
|
|
|
|
return _scale_preview(
|
|
values=values,
|
|
formula_rows=formula_rows,
|
|
source_total_kg=source_total_kg,
|
|
client_name=product.client_name,
|
|
product_id=product.id,
|
|
mix_label=_mix_calculator_label(product),
|
|
mix_id=product.mix_id,
|
|
unit_of_measure=product.unit_of_measure,
|
|
)
|
|
|
|
|
|
def build_mix_calculator_options(db: Session, *, tenant_id: str) -> dict:
|
|
# Prefer product-specific formulas where present; fall back to the shared
|
|
# mix master for legacy rows that have not been migrated yet.
|
|
product_totals_rows = db.execute(
|
|
select(ProductIngredient.product_id, func.coalesce(func.sum(ProductIngredient.quantity_kg), 0.0))
|
|
.join(Product, Product.id == ProductIngredient.product_id)
|
|
.where(Product.tenant_id == tenant_id)
|
|
.group_by(ProductIngredient.product_id)
|
|
).all()
|
|
product_totals: dict[int, float] = {product_id: round(total or 0.0, 4) for product_id, total in product_totals_rows}
|
|
|
|
mix_totals_rows = db.execute(
|
|
select(MixIngredient.mix_id, func.coalesce(func.sum(MixIngredient.quantity_kg), 0.0))
|
|
.join(Mix, Mix.id == MixIngredient.mix_id)
|
|
.where(Mix.tenant_id == tenant_id)
|
|
.group_by(MixIngredient.mix_id)
|
|
).all()
|
|
mix_totals: dict[int, float] = {mix_id: round(total or 0.0, 4) for mix_id, total in mix_totals_rows}
|
|
|
|
product_ids_with_formulas = select(ProductIngredient.product_id).where(ProductIngredient.tenant_id == tenant_id)
|
|
products = db.scalars(
|
|
select(Product)
|
|
.where(
|
|
Product.tenant_id == tenant_id,
|
|
Product.visible.is_(True),
|
|
Product.id.in_(product_ids_with_formulas),
|
|
)
|
|
.options(joinedload(Product.mix))
|
|
.order_by(Product.client_name, Product.name)
|
|
).all()
|
|
|
|
representative_products: dict[tuple[str, str], Product] = {}
|
|
for product in products:
|
|
mix_label = _mix_calculator_label(product)
|
|
key = (product.client_name, mix_label)
|
|
current = representative_products.get(key)
|
|
if current is None:
|
|
representative_products[key] = product
|
|
continue
|
|
|
|
if _mix_calculator_option_rank(product) < _mix_calculator_option_rank(current):
|
|
representative_products[key] = product
|
|
|
|
products = sorted(
|
|
representative_products.values(),
|
|
key=lambda product: (product.client_name, _mix_calculator_label(product), product.id),
|
|
)
|
|
|
|
product_rows = [
|
|
{
|
|
"product_id": product.id,
|
|
"client_name": product.client_name,
|
|
"product_name": _mix_calculator_label(product),
|
|
"mix_id": product.mix_id,
|
|
"mix_name": _mix_calculator_label(product),
|
|
"unit_of_measure": product.unit_of_measure,
|
|
"unit_size_kg": round(extract_unit_quantity_kg(product.unit_of_measure), 4),
|
|
"mix_total_kg": product_totals.get(product.id, mix_totals.get(product.mix_id, 0.0)),
|
|
}
|
|
for product in products
|
|
]
|
|
|
|
# Surface mixes that have a formula but no product at all yet (e.g. a freshly
|
|
# created mix). They're selected via a negative `product_id` sentinel (-mix_id)
|
|
# and calculated straight off the mix master — bulk kg, no bag split. A mix
|
|
# whose only product is hidden is intentionally excluded (it HAS a product),
|
|
# so check every product, not just the visible representatives.
|
|
covered_mix_ids = set(
|
|
db.scalars(
|
|
select(Product.mix_id).where(Product.tenant_id == tenant_id).distinct()
|
|
).all()
|
|
)
|
|
formula_only_mix_ids = [
|
|
mix_id for mix_id, total in mix_totals.items() if total > 0 and mix_id not in covered_mix_ids
|
|
]
|
|
if formula_only_mix_ids:
|
|
formula_only_mixes = db.scalars(
|
|
select(Mix).where(
|
|
Mix.tenant_id == tenant_id,
|
|
Mix.id.in_(formula_only_mix_ids),
|
|
Mix.status == "active",
|
|
)
|
|
).all()
|
|
product_rows.extend(
|
|
{
|
|
"product_id": -mix.id,
|
|
"client_name": mix.client_name,
|
|
"product_name": mix.name,
|
|
"mix_id": mix.id,
|
|
"mix_name": mix.name,
|
|
"unit_of_measure": "kg",
|
|
"unit_size_kg": 0.0,
|
|
"mix_total_kg": mix_totals.get(mix.id, 0.0),
|
|
}
|
|
for mix in formula_only_mixes
|
|
)
|
|
|
|
product_rows.sort(key=lambda row: (row["client_name"], row["product_name"], row["product_id"]))
|
|
clients = sorted({row["client_name"] for row in product_rows})
|
|
|
|
return {"clients": clients, "products": product_rows}
|
|
|
|
|
|
def serialize_mix_calculator_session(session_record: MixCalculatorSession, auth_session: AuthSession) -> dict:
|
|
total_bags = round(session_record.total_bags, 4)
|
|
warnings: list[str] = []
|
|
bag_warning = _fractional_bag_warning(session_record.batch_size_kg, total_bags, session_record.product_unit_of_measure)
|
|
if bag_warning:
|
|
warnings.append(bag_warning)
|
|
|
|
return {
|
|
"id": session_record.id,
|
|
"tenant_id": session_record.tenant_id,
|
|
"session_number": session_record.session_number,
|
|
"client_name": session_record.client_name,
|
|
"product_id": session_record.product_id,
|
|
"product_name": session_record.product_name,
|
|
"mix_id": session_record.mix_id,
|
|
"mix_name": session_record.mix_name,
|
|
"mix_date": session_record.mix_date,
|
|
"batch_size_kg": round(session_record.batch_size_kg, 4),
|
|
"total_bags": total_bags,
|
|
"total_kg": round(session_record.total_kg, 4),
|
|
"product_unit_of_measure": session_record.product_unit_of_measure,
|
|
"product_unit_size_kg": round(session_record.product_unit_size_kg, 4),
|
|
"prepared_by_user_id": session_record.prepared_by_user_id,
|
|
"prepared_by_name": session_record.prepared_by_name,
|
|
"created_by": session_record.created_by,
|
|
"status": session_record.status,
|
|
"notes": session_record.notes,
|
|
"created_at": session_record.created_at,
|
|
"updated_at": session_record.updated_at,
|
|
"warnings": warnings,
|
|
"is_owner": session_record.prepared_by_user_id == auth_session.user_id,
|
|
"lines": [
|
|
{
|
|
"id": line.id,
|
|
"raw_material_id": line.raw_material_id,
|
|
"raw_material_name": line.raw_material_name,
|
|
"required_kg": round(line.required_kg, 4),
|
|
"mix_percentage": round(line.mix_percentage, 4),
|
|
"unit": line.unit,
|
|
"rounding_decimals": line.rounding_decimals,
|
|
"sort_order": line.sort_order,
|
|
}
|
|
for line in session_record.lines
|
|
],
|
|
}
|
|
|
|
|
|
def list_mix_calculator_sessions(db: Session, *, auth_session: AuthSession, limit: int = 100) -> list[dict]:
|
|
sessions = db.scalars(
|
|
_build_session_access_query(auth_session)
|
|
.options(selectinload(MixCalculatorSession.lines))
|
|
.order_by(MixCalculatorSession.created_at.desc(), MixCalculatorSession.id.desc())
|
|
.limit(limit)
|
|
).all()
|
|
return [serialize_mix_calculator_session(session_record, auth_session) for session_record in sessions]
|
|
|
|
|
|
def get_mix_calculator_session(db: Session, *, auth_session: AuthSession, session_id: int) -> MixCalculatorSession | None:
|
|
return db.scalar(
|
|
_build_session_access_query(auth_session)
|
|
.where(MixCalculatorSession.id == session_id)
|
|
.options(selectinload(MixCalculatorSession.lines))
|
|
)
|
|
|
|
|
|
def _next_session_number(db: Session, *, tenant_id: str, mix_date: date) -> str:
|
|
prefix = f"HPP-{mix_date.strftime('%Y%m%d')}-"
|
|
existing = db.scalars(
|
|
select(MixCalculatorSession.session_number)
|
|
.where(
|
|
MixCalculatorSession.tenant_id == tenant_id,
|
|
MixCalculatorSession.mix_date == mix_date,
|
|
MixCalculatorSession.session_number.like(f"{prefix}%"),
|
|
)
|
|
).all()
|
|
sequence = 1
|
|
if existing:
|
|
sequence = max(int(value.rsplit("-", 1)[-1]) for value in existing) + 1
|
|
return f"{prefix}{sequence:04d}"
|
|
|
|
|
|
def create_mix_calculator_session(db: Session, *, auth_session: AuthSession, payload: MixCalculatorSessionCreate) -> dict:
|
|
if payload.product_id < 0:
|
|
# Sessions reference a real product (FK). A formula-only mix has none yet —
|
|
# it can still be previewed and printed, just not saved as a session.
|
|
raise ValueError("Add a product to this mix before saving a calculator session.")
|
|
preview = calculate_mix_calculator_preview(db, tenant_id=auth_session.tenant_id or "", payload=payload)
|
|
session_record = MixCalculatorSession(
|
|
tenant_id=auth_session.tenant_id or "default",
|
|
session_number=_next_session_number(db, tenant_id=auth_session.tenant_id or "default", mix_date=payload.mix_date),
|
|
client_name=preview["client_name"],
|
|
product_id=preview["product_id"],
|
|
product_name=preview["product_name"],
|
|
mix_id=preview["mix_id"],
|
|
mix_name=preview["mix_name"],
|
|
mix_date=preview["mix_date"],
|
|
batch_size_kg=preview["batch_size_kg"],
|
|
total_bags=preview["total_bags"],
|
|
total_kg=preview["total_kg"],
|
|
product_unit_of_measure=preview["product_unit_of_measure"],
|
|
product_unit_size_kg=preview["product_unit_size_kg"],
|
|
prepared_by_user_id=auth_session.user_id,
|
|
prepared_by_name=preview["prepared_by_name"],
|
|
created_by=auth_session.email,
|
|
status=preview["status"],
|
|
notes=preview["notes"],
|
|
)
|
|
session_record.lines = [
|
|
MixCalculatorSessionLine(
|
|
tenant_id=auth_session.tenant_id or "default",
|
|
raw_material_id=line["raw_material_id"],
|
|
raw_material_name=line["raw_material_name"],
|
|
required_kg=line["required_kg"],
|
|
mix_percentage=line["mix_percentage"],
|
|
unit=line["unit"],
|
|
rounding_decimals=line.get("rounding_decimals", 2),
|
|
sort_order=line["sort_order"],
|
|
)
|
|
for line in preview["lines"]
|
|
]
|
|
db.add(session_record)
|
|
db.commit()
|
|
db.refresh(session_record)
|
|
db.refresh(session_record, attribute_names=["lines"])
|
|
return serialize_mix_calculator_session(session_record, auth_session)
|
|
|
|
|
|
def update_mix_calculator_session(
|
|
db: Session,
|
|
*,
|
|
auth_session: AuthSession,
|
|
session_record: MixCalculatorSession,
|
|
payload: MixCalculatorSessionUpdate,
|
|
) -> dict:
|
|
merged_values = {
|
|
"mix_date": session_record.mix_date,
|
|
"client_name": session_record.client_name,
|
|
"product_id": session_record.product_id,
|
|
"batch_size_kg": session_record.batch_size_kg,
|
|
"prepared_by_name": session_record.prepared_by_name,
|
|
"status": session_record.status,
|
|
"notes": session_record.notes,
|
|
}
|
|
merged_values.update(payload.model_dump(exclude_unset=True))
|
|
preview = calculate_mix_calculator_preview(db, tenant_id=auth_session.tenant_id or "", payload=merged_values)
|
|
|
|
session_record.client_name = preview["client_name"]
|
|
session_record.product_id = preview["product_id"]
|
|
session_record.product_name = preview["product_name"]
|
|
session_record.mix_id = preview["mix_id"]
|
|
session_record.mix_name = preview["mix_name"]
|
|
session_record.mix_date = preview["mix_date"]
|
|
session_record.batch_size_kg = preview["batch_size_kg"]
|
|
session_record.total_bags = preview["total_bags"]
|
|
session_record.total_kg = preview["total_kg"]
|
|
session_record.product_unit_of_measure = preview["product_unit_of_measure"]
|
|
session_record.product_unit_size_kg = preview["product_unit_size_kg"]
|
|
session_record.prepared_by_name = preview["prepared_by_name"]
|
|
session_record.status = preview["status"]
|
|
session_record.notes = preview["notes"]
|
|
|
|
session_record.lines.clear()
|
|
session_record.lines.extend(
|
|
[
|
|
MixCalculatorSessionLine(
|
|
tenant_id=auth_session.tenant_id or "default",
|
|
raw_material_id=line["raw_material_id"],
|
|
raw_material_name=line["raw_material_name"],
|
|
required_kg=line["required_kg"],
|
|
mix_percentage=line["mix_percentage"],
|
|
unit=line["unit"],
|
|
sort_order=line["sort_order"],
|
|
)
|
|
for line in preview["lines"]
|
|
]
|
|
)
|
|
|
|
db.commit()
|
|
db.refresh(session_record)
|
|
db.refresh(session_record, attribute_names=["lines"])
|
|
return serialize_mix_calculator_session(session_record, auth_session)
|