"""WESP-shaped recipe write/calculate for orchestrator static UI.""" from __future__ import annotations from datetime import UTC, datetime from typing import Any from uuid import uuid4 from fastapi import HTTPException, status from sqlalchemy import select from app.core.database import session_scope from app.modules.sync.engine import SyncEngine from app.modules.sync.schemas import ChangeEventIn from app.modules.zootech.catalog_apply import load_catalog_row from app.modules.zootech.catalog_models import ZootechFeedingPeriod, ZootechPeriodRecipe, ZootechUnloadingGroup from app.modules.zootech.models import ZootechComponent, ZootechIngredient, ZootechRecipe from app.modules.zootech.recipe_calculator import calculate_recipe from app.modules.zootech.sync_content_hash import compute_content_hash class RecipeWriteError(Exception): def __init__(self, message: str, status_code: int = 400): self.message = message self.status_code = status_code super().__init__(message) def _normalize_groups(raw: list[Any]) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] for group in raw: if not isinstance(group, dict): continue g = dict(group) if "distributionType" not in g and "distribution_type" in g: g["distributionType"] = g.get("distribution_type") out.append(g) return out def calculate_recipe_wesp(body: dict[str, Any]) -> dict[str, Any]: if not body: raise RecipeWriteError("Данные не предоставлены", 400) try: heads_count = int(body.get("headsCount") or body.get("heads_count") or body.get("headsPerTrip") or 0) trip_percent = float(body.get("tripPercent") or body.get("trip_percent") or 100) except (TypeError, ValueError) as exc: raise RecipeWriteError("Некорректные числовые параметры", 400) from exc ingredients = body.get("ingredients") or [] unloading_groups = _normalize_groups(body.get("unloadingGroups") or body.get("unloading_groups") or []) calculate_from_dry_matter = bool( body.get("calculateFromDryMatter") if body.get("calculateFromDryMatter") is not None else body.get("calculate_from_dry_matter", False) ) component_ids = [i.get("component_id") for i in ingredients if isinstance(i, dict) and i.get("component_id")] component_dry_matter_map: dict[str, float] = {} if component_ids: with session_scope() as db: for comp in db.scalars(select(ZootechComponent).where(ZootechComponent.id.in_(component_ids))): component_dry_matter_map[comp.id] = float(comp.dry_matter or 0) result = calculate_recipe( ingredients=[i for i in ingredients if isinstance(i, dict)], heads_count=heads_count, trip_percent=trip_percent, unloading_groups=unloading_groups, component_dry_matter_map=component_dry_matter_map or None, calculate_from_dry_matter=calculate_from_dry_matter, ) def _to_float2(x: Any) -> float: try: return round(float(x), 2) except (TypeError, ValueError): return 0.0 def _truncate2(x: Any) -> float: try: v = float(x) return float(int(v * 100)) / 100.0 except (TypeError, ValueError): return 0.0 for ing in result.get("ingredients") or []: for key in ("weightPerHead", "tripWeight", "totalWeight", "dryMatterPerHead"): if key in ing and ing[key] is not None: raw_val = ing[key] v = _truncate2(raw_val) if key == "weightPerHead" else _to_float2(raw_val) ing[key] = f"{v:.2f}" if key == "weightPerHead" else v totals = result.get("totals") or {} for key in ("totalWeight", "totalTripWeight", "totalDryMatterPerHead", "totalWeightPerHead"): if key in totals and totals[key] is not None: totals[key] = _to_float2(totals[key]) result["totals"] = totals return result def _push_change( enterprise_id: str, table: str, record_id: str, payload: dict[str, Any], *, action: str = "upsert", ) -> None: version = int(payload.get("version") or 1) content_hash = compute_content_hash(payload) payload = {**payload, "version": version, "content_hash": content_hash} event = ChangeEventIn( event_id=str(uuid4()), seq=0, domain="global", table=table, record_id=record_id, action=action, version=version, content_hash=content_hash, payload=payload, emitted_at=datetime.now(UTC), origin_site_id=SyncEngine.ORCHESTRATOR_SITE_ID, ) SyncEngine(enterprise_id, SyncEngine.ORCHESTRATOR_SITE_ID).process_push([event], origin="orchestrator") def _resolve_component(enterprise_id: str, ing: dict[str, Any]) -> dict[str, Any] | None: comp_id = ing.get("component_id") with session_scope() as db: row: ZootechComponent | None = None if comp_id: row = db.scalar( select(ZootechComponent).where( ZootechComponent.enterprise_id == enterprise_id, ZootechComponent.id == str(comp_id), ZootechComponent.is_deleted.is_(False), ) ) if row is None: name = (ing.get("name") or "").strip() if name: row = db.scalar( select(ZootechComponent).where( ZootechComponent.enterprise_id == enterprise_id, ZootechComponent.name == name, ZootechComponent.is_deleted.is_(False), ) ) if row is None: return None return { "id": row.id, "name": row.name, "dry_matter": float(row.dry_matter or 0), } def update_recipe_wesp(enterprise_id: str, recipe_id: str, data: dict[str, Any]) -> dict[str, Any]: if not data: raise RecipeWriteError("Данные не предоставлены", 400) for key in ("name", "heads_count", "mixing_time"): if key not in data: raise RecipeWriteError(f'Отсутствует обязательное поле "{key}"', 400) with session_scope() as db: recipe = db.scalar( select(ZootechRecipe).where( ZootechRecipe.enterprise_id == enterprise_id, ZootechRecipe.id == recipe_id, ZootechRecipe.is_deleted.is_(False), ) ) if not recipe: raise RecipeWriteError("Рецепт не найден", 404) existing_ings = list( db.scalars( select(ZootechIngredient).where( ZootechIngredient.enterprise_id == enterprise_id, ZootechIngredient.recipe_id == recipe_id, ZootechIngredient.is_deleted.is_(False), ) ) ) existing_groups = list( db.scalars( select(ZootechUnloadingGroup).where( ZootechUnloadingGroup.enterprise_id == enterprise_id, ZootechUnloadingGroup.recipe_id == recipe_id, ZootechUnloadingGroup.is_deleted.is_(False), ) ) ) recipe_trip_percent = float(recipe.trip_percent or 100) recipe_dry_matter_locked = bool(recipe.dry_matter_locked) recipe_unloading_link_broken = bool(recipe.unloading_link_broken) recipe_target_component_id = recipe.target_component_id recipe_version = int(recipe.version or 1) + 1 ing_by_id = { str(i.id): {"id": str(i.id), "name": i.name, "version": int(i.version or 1)} for i in existing_ings } grp_by_id = { str(g.id): {"id": str(g.id), "name": g.name, "version": int(g.version or 1)} for g in existing_groups } recipe_payload = { "id": recipe_id, "name": str(data["name"]), "heads_per_trip": int(data["heads_count"]), "mixing_time": int(data["mixing_time"]), "trip_percent": float(data.get("trip_percent") or recipe_trip_percent), "dry_matter_locked": bool(data.get("dry_matter_locked", recipe_dry_matter_locked)), "unloading_link_broken": bool(data.get("unloading_link_broken", recipe_unloading_link_broken)), "target_component_id": data.get("target_component_id", recipe_target_component_id), "version": recipe_version, } dispenser_id = str(data.get("dispenser_id") or "").strip() if dispenser_id: recipe_payload["dispenser_id"] = dispenser_id _push_change(enterprise_id, "recipe", recipe_id, recipe_payload) seen_ing_ids: set[str] = set() for idx, ing in enumerate([x for x in (data.get("ingredients") or []) if isinstance(x, dict)], start=1): component = _resolve_component(enterprise_id, ing) if not component: raise RecipeWriteError( f'Компонент не найден (id="{ing.get("component_id", "")}", name="{ing.get("name", "")}")', 400, ) ing_id = str(ing.get("id") or "").strip() or str(uuid4()) existing = ing_by_id.get(ing_id) order_value = int(ing.get("order") or idx) wph = float(ing.get("weight_per_head") or ing.get("weightPerHead") or 0) amount = float(ing.get("amount") or 0) dm = float(ing.get("dry_matter") or component.get("dry_matter") or 0) dm_ph = ing.get("dry_matter_per_head") if dm_ph is None: dm_ph = ing.get("dryMatterPerHead") dm_ph_f = float(dm_ph) if dm_ph not in (None, "") else (wph * (dm / 100.0) if wph and dm else 0.0) version = int(existing.get("version") or 1) + 1 if existing else 1 payload = { "id": ing_id, "recipe_id": recipe_id, "component_id": component["id"], "name": component["name"], "amount": amount, "weight_per_head": wph, "dry_matter": dm, "dry_matter_per_head": dm_ph_f, "order": order_value, "version": version, } _push_change(enterprise_id, "ingredient", ing_id, payload) seen_ing_ids.add(ing_id) for raw_id in data.get("deleted_ingredient_ids") or []: ing_id = str(raw_id or "").strip() if not ing_id or ing_id not in ing_by_id: continue existing = ing_by_id[ing_id] version = int(existing["version"] or 1) + 1 payload = { "id": ing_id, "recipe_id": recipe_id, "name": existing["name"], "version": version, } _push_change(enterprise_id, "ingredient", ing_id, payload, action="delete") for ing_id, existing in ing_by_id.items(): if ing_id not in seen_ing_ids and ing_id not in {str(x) for x in (data.get("deleted_ingredient_ids") or [])}: version = int(existing["version"] or 1) + 1 _push_change( enterprise_id, "ingredient", ing_id, {"id": ing_id, "recipe_id": recipe_id, "name": existing["name"], "version": version}, action="delete", ) seen_grp_ids: set[str] = set() groups_in = [x for x in (data.get("unloading_groups") or data.get("unloadingGroups") or []) if isinstance(x, dict)] for idx, group in enumerate(groups_in, start=1): try: gname = str(group["name"]) gdist = str(group.get("distribution_type") or group.get("distributionType") or "percent") gval = float(group.get("value") or 0) except (KeyError, TypeError, ValueError) as exc: raise RecipeWriteError(f"Некорректная группа выгрузки (order={idx}): {exc}", 400) from exc grp_id = str(group.get("id") or "").strip() or str(uuid4()) existing = grp_by_id.get(grp_id) order_value = int(group.get("order") or idx) weight_raw = group.get("weight") weight = float(weight_raw) if weight_raw not in (None, "") else None version = int(existing.get("version") or 1) + 1 if existing else 1 payload = { "id": grp_id, "recipe_id": recipe_id, "name": gname, "distribution_type": gdist, "value": gval, "weight": weight, "order": order_value, "version": version, } _push_change(enterprise_id, "unloading_group", grp_id, payload) seen_grp_ids.add(grp_id) for raw_id in data.get("deleted_unloading_group_ids") or []: grp_id = str(raw_id or "").strip() if not grp_id or grp_id not in grp_by_id: continue existing = grp_by_id[grp_id] version = int(existing["version"] or 1) + 1 _push_change( enterprise_id, "unloading_group", grp_id, {"id": grp_id, "recipe_id": recipe_id, "name": existing["name"], "version": version}, action="delete", ) for grp_id, existing in grp_by_id.items(): if grp_id not in seen_grp_ids and grp_id not in {str(x) for x in (data.get("deleted_unloading_group_ids") or [])}: version = int(existing["version"] or 1) + 1 _push_change( enterprise_id, "unloading_group", grp_id, {"id": grp_id, "recipe_id": recipe_id, "name": existing["name"], "version": version}, action="delete", ) with session_scope() as db: saved = db.scalar( select(ZootechRecipe).where( ZootechRecipe.enterprise_id == enterprise_id, ZootechRecipe.id == recipe_id, ZootechRecipe.is_deleted.is_(False), ) ) if not saved: raise RecipeWriteError("Рецепт не найден после сохранения", 404) return { "success": True, "message": "Рецепт обновлен", "id": recipe_id, "stats": { "ingredients": len(seen_ing_ids), "unloading_groups": len(seen_grp_ids), }, } def create_recipe_wesp( enterprise_id: str, data: dict[str, Any], *, period_id: str | None = None, ) -> dict[str, Any]: if not data: raise RecipeWriteError("Данные не предоставлены", 400) for key in ("name", "heads_count", "mixing_time"): if key not in data: raise RecipeWriteError(f'Отсутствует обязательное поле "{key}"', 400) if period_id: with session_scope() as db: period = db.scalar( select(ZootechFeedingPeriod).where( ZootechFeedingPeriod.enterprise_id == enterprise_id, ZootechFeedingPeriod.id == period_id, ZootechFeedingPeriod.is_deleted.is_(False), ) ) if not period: raise RecipeWriteError("Период не найден", 404) recipe_id = str(uuid4()) recipe_payload = { "id": recipe_id, "name": str(data["name"]), "heads_per_trip": int(data["heads_count"]), "mixing_time": int(data["mixing_time"]), "trip_percent": float(data.get("trip_percent") or 100), "dry_matter_locked": bool(data.get("dry_matter_locked", False)), "unloading_link_broken": bool(data.get("unloading_link_broken", False)), "target_component_id": data.get("target_component_id"), "version": 1, } dispenser_id = str(data.get("dispenser_id") or "").strip() if dispenser_id: recipe_payload["dispenser_id"] = dispenser_id _push_change(enterprise_id, "recipe", recipe_id, recipe_payload) if period_id: with session_scope() as db: links = list( db.scalars( select(ZootechPeriodRecipe).where( ZootechPeriodRecipe.enterprise_id == enterprise_id, ZootechPeriodRecipe.period_id == period_id, ZootechPeriodRecipe.is_deleted.is_(False), ) ) ) order = max((int(link.order or 0) for link in links), default=0) + 1 _push_change( enterprise_id, "period_recipes", f"{period_id}:{recipe_id}", { "period_id": period_id, "recipe_id": recipe_id, "order": order, "version": 1, }, ) result = update_recipe_wesp(enterprise_id, recipe_id, data) if result.get("success"): result["message"] = "Рецепт создан" return result def unlink_recipe_from_period_wesp( enterprise_id: str, dispenser_id: str, period_id: str, recipe_id: str, ) -> dict[str, Any]: with session_scope() as db: period = db.scalar( select(ZootechFeedingPeriod).where( ZootechFeedingPeriod.enterprise_id == enterprise_id, ZootechFeedingPeriod.id == period_id, ZootechFeedingPeriod.dispenser_id == dispenser_id, ZootechFeedingPeriod.is_deleted.is_(False), ) ) if not period: raise RecipeWriteError("Период не найден", 404) link_id = f"{period_id}:{recipe_id}" existing = load_catalog_row(enterprise_id, "period_recipes", link_id) if not existing or existing.get("is_deleted"): raise RecipeWriteError("Рейс не найден в периоде", 404) version = int(existing.get("version") or 0) + 1 payload = { "period_id": period_id, "recipe_id": recipe_id, "order": int(existing.get("order") or 0), "version": version, "is_deleted": True, } _push_change(enterprise_id, "period_recipes", link_id, payload, action="delete") return {"success": True, "message": "Рейс удалён из периода"} def _period_recipe_ids_ordered(enterprise_id: str, period_id: str) -> list[str]: with session_scope() as db: links = list( db.scalars( select(ZootechPeriodRecipe) .where( ZootechPeriodRecipe.enterprise_id == enterprise_id, ZootechPeriodRecipe.period_id == period_id, ZootechPeriodRecipe.is_deleted.is_(False), ) .order_by(ZootechPeriodRecipe.order.asc(), ZootechPeriodRecipe.recipe_id.asc()) ) ) return [link.recipe_id for link in links] def _set_period_recipe_order(enterprise_id: str, period_id: str, recipe_ids: list[str]) -> None: for order, rid in enumerate(recipe_ids): link_id = f"{period_id}:{rid}" existing = load_catalog_row(enterprise_id, "period_recipes", link_id) if existing and not existing.get("is_deleted") and int(existing.get("order") or -1) == order: continue version = int(existing.get("version") or 0) + 1 if existing else 1 _push_change( enterprise_id, "period_recipes", link_id, { "period_id": period_id, "recipe_id": rid, "order": order, "version": version, }, ) def _soft_delete_period_recipe_link(enterprise_id: str, period_id: str, recipe_id: str) -> None: link_id = f"{period_id}:{recipe_id}" existing = load_catalog_row(enterprise_id, "period_recipes", link_id) if not existing or existing.get("is_deleted"): return version = int(existing.get("version") or 0) + 1 _push_change( enterprise_id, "period_recipes", link_id, { "period_id": period_id, "recipe_id": recipe_id, "order": int(existing.get("order") or 0), "version": version, "is_deleted": True, }, action="delete", ) def transfer_recipe_between_periods_wesp( enterprise_id: str, to_dispenser_id: str, to_period_id: str, recipe_id: str, *, from_dispenser_id: str, from_period_id: str, to_index: int, ) -> dict[str, Any]: if not from_dispenser_id or not from_period_id: raise RecipeWriteError("Не указан исходный период", 400) try: insert_at = int(to_index) except (TypeError, ValueError) as exc: raise RecipeWriteError("Некорректная позиция вставки", 400) from exc if insert_at < 0: raise RecipeWriteError("Некорректная позиция вставки", 400) with session_scope() as db: from_period = db.scalar( select(ZootechFeedingPeriod).where( ZootechFeedingPeriod.enterprise_id == enterprise_id, ZootechFeedingPeriod.id == from_period_id, ZootechFeedingPeriod.dispenser_id == from_dispenser_id, ZootechFeedingPeriod.is_deleted.is_(False), ) ) to_period = db.scalar( select(ZootechFeedingPeriod).where( ZootechFeedingPeriod.enterprise_id == enterprise_id, ZootechFeedingPeriod.id == to_period_id, ZootechFeedingPeriod.dispenser_id == to_dispenser_id, ZootechFeedingPeriod.is_deleted.is_(False), ) ) recipe = db.scalar( select(ZootechRecipe).where( ZootechRecipe.enterprise_id == enterprise_id, ZootechRecipe.id == recipe_id, ZootechRecipe.is_deleted.is_(False), ) ) if not recipe: raise RecipeWriteError("Рецепт не найден", 404) if not from_period: raise RecipeWriteError("Исходный период не найден", 404) if not to_period: raise RecipeWriteError("Целевой период не найден", 404) if from_period_id == to_period_id: raise RecipeWriteError("Период назначения совпадает с исходным", 400) src_link = load_catalog_row(enterprise_id, "period_recipes", f"{from_period_id}:{recipe_id}") if not src_link or src_link.get("is_deleted"): raise RecipeWriteError("Рейс не найден в исходном периоде", 404) tgt_link = load_catalog_row(enterprise_id, "period_recipes", f"{to_period_id}:{recipe_id}") if tgt_link and not tgt_link.get("is_deleted"): raise RecipeWriteError("Рейс уже есть в целевом периоде", 409) src_ids = _period_recipe_ids_ordered(enterprise_id, from_period_id) if recipe_id not in src_ids: raise RecipeWriteError("Рейс не найден в исходном периоде", 404) tgt_ids = _period_recipe_ids_ordered(enterprise_id, to_period_id) insert_at = min(insert_at, len(tgt_ids)) _soft_delete_period_recipe_link(enterprise_id, from_period_id, recipe_id) _set_period_recipe_order(enterprise_id, from_period_id, [rid for rid in src_ids if rid != recipe_id]) tgt_ids.insert(insert_at, recipe_id) _set_period_recipe_order(enterprise_id, to_period_id, tgt_ids) return {"success": True, "message": "Рейс перенесён"} def move_recipe_in_period_wesp( enterprise_id: str, recipe_id: str, *, period_id: str, from_index: int, to_index: int, ) -> dict[str, Any]: if not period_id: raise RecipeWriteError("Период не указан", 400) try: src = int(from_index) dst = int(to_index) except (TypeError, ValueError) as exc: raise RecipeWriteError("Некорректные индексы", 400) from exc with session_scope() as db: period = db.scalar( select(ZootechFeedingPeriod).where( ZootechFeedingPeriod.enterprise_id == enterprise_id, ZootechFeedingPeriod.id == period_id, ZootechFeedingPeriod.is_deleted.is_(False), ) ) recipe = db.scalar( select(ZootechRecipe).where( ZootechRecipe.enterprise_id == enterprise_id, ZootechRecipe.id == recipe_id, ZootechRecipe.is_deleted.is_(False), ) ) if not period: raise RecipeWriteError("Период не найден", 404) if not recipe: raise RecipeWriteError("Рецепт не найден", 404) ids = _period_recipe_ids_ordered(enterprise_id, period_id) if recipe_id not in ids: raise RecipeWriteError("Рейс не найден в периоде", 404) if dst < 0 or dst >= len(ids): raise RecipeWriteError("Некорректные индексы", 400) actual_from = ids.index(recipe_id) if src != actual_from: src = actual_from if src == dst: return {"success": True, "message": "Порядок не изменился"} ids.pop(src) ids.insert(dst, recipe_id) _set_period_recipe_order(enterprise_id, period_id, ids) return {"success": True, "message": "Порядок обновлён"} def delete_recipe_wesp(enterprise_id: str, recipe_id: str) -> dict[str, Any]: existing = load_catalog_row(enterprise_id, "recipe", recipe_id) if not existing or existing.get("is_deleted"): raise RecipeWriteError("Рецепт не найден", 404) version = int(existing.get("version") or 0) + 1 payload = {**existing, "version": version, "is_deleted": True} _push_change(enterprise_id, "recipe", recipe_id, payload, action="delete") return {"success": True, "message": "Рейс удален"} def recipe_write_http_error(exc: RecipeWriteError) -> HTTPException: return HTTPException(status_code=exc.status_code, detail={"message": exc.message, "error": True})