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)