"""Daily plan service for WESP-shaped orchestrator API (enterprise-scoped).""" from __future__ import annotations import json from collections import defaultdict from datetime import UTC, date, datetime, timedelta from typing import Any from sqlalchemy import select from app.core.database import session_scope from app.modules.zootech.catalog_models import ( ZootechFeedDispenser, ZootechFeedingPeriod, ZootechPeriodRecipe, ZootechUnloadingGroup, ) from app.modules.zootech.models import ZootechComponent, ZootechIngredient, ZootechRecipe from app.modules.zootech.pdf_export import build_table_pdf from app.modules.zootech.recipe_calculator import calculate_ingredients, calculate_recipe from app.modules.zootech.wesp_daily_plan_store import ( find_daily_row, list_daily_rows, soft_delete_daily_row, upsert_daily_row, ) ALL_DISPENSERS_ID = "__all_dispensers__" ALL_MILLS_ID = "__all_mills__" _EPSILON = 1e-6 def _parse_payload_json(raw: str | None) -> dict[str, Any]: if not raw: return {} try: parsed = json.loads(raw) except json.JSONDecodeError: return {} return parsed if isinstance(parsed, dict) else {} def _parse_plan_date(value: str | None) -> date: if value: try: return date.fromisoformat(str(value).strip()[:10]) except ValueError: pass return date.today() def _iso_date(value: str | date | None) -> str: if isinstance(value, date): return value.isoformat() if value: try: return date.fromisoformat(str(value).strip()[:10]).isoformat() except ValueError: pass return date.today().isoformat() def resolve_skip_range( start: date, *, duration: str | None = None, until_date: str | None = None, ) -> tuple[date, date]: dur = (duration or "today").strip().lower() if dur == "week": days_to_sunday = 6 - start.weekday() return start, start + timedelta(days=days_to_sunday) if dur == "date" and until_date: try: end = date.fromisoformat(str(until_date).strip()[:10]) except ValueError: end = start return start, max(start, end) return start, start def _skip_end_date(row: dict[str, Any]) -> date: end_raw = row.get("valid_until") or row.get("plan_date") if isinstance(end_raw, date): return end_raw try: return date.fromisoformat(str(end_raw).strip()[:10]) except (TypeError, ValueError): return _parse_plan_date(None) def _is_skip_active(row: dict[str, Any], target: date) -> bool: if row.get("is_deleted"): return False try: start = date.fromisoformat(str(row.get("plan_date") or "").strip()[:10]) except ValueError: return False end = _skip_end_date(row) return start <= target <= end def _active_daily_rows(enterprise_id: str, table_name: str, target: date) -> list[dict[str, Any]]: return [row for row in list_daily_rows(enterprise_id, table_name) if _is_skip_active(row, target)] def _serialize_skip_dates(row: dict[str, Any], *, fallback: date) -> dict[str, str]: start = _iso_date(row.get("plan_date") or fallback.isoformat()) end = _iso_date(row.get("valid_until") or row.get("plan_date") or start) return {"date": start, "validUntil": end} def _valid_until_field(start: date, end: date) -> str | None: return None if end == start else end.isoformat() def _upsert_daily( enterprise_id: str, table_name: str, *, lookup: dict[str, Any], fields: dict[str, Any], user: str = "system", ) -> dict[str, Any]: existing = find_daily_row(enterprise_id, table_name, match=lookup) record_id = str(existing["id"]) if existing else None payload = { **fields, "updated_by": user, } if not existing: payload["created_by"] = user return upsert_daily_row(enterprise_id, table_name, record_id, payload) def _dispenser_meta(row: ZootechFeedDispenser) -> dict[str, Any]: extra = _parse_payload_json(row.payload_json) return { "name": row.name, "type": str(extra.get("type") or "dispenser"), "farm": str(extra.get("farm") or ""), "operator": str(extra.get("operator") or ""), } def _period_is_active(period: ZootechFeedingPeriod) -> bool: meta = _parse_payload_json(period.payload_json) return bool(meta.get("is_active", True)) def _unloading_group_fields(row: ZootechUnloadingGroup) -> dict[str, Any]: extra = _parse_payload_json(row.payload_json) distribution_type = str(extra.get("distribution_type") or extra.get("distributionType") or "percent") return { "distribution_type": distribution_type, "distributionType": distribution_type, "value": float(extra.get("value") or 0), "weight": float(extra.get("weight") or 0), "order": int(extra.get("order") or 0), } def _distribution_label(dist_type: str | None, value: float) -> str: if (dist_type or "percent") == "heads": rounded = int(value) if value == int(value) else value return f"{rounded} гол." rounded = int(value) if value == int(value) else value return f"{rounded}%" def _ingredient_display_name(ing: ZootechIngredient, comp_names: dict[str, str]) -> str: raw = (ing.name or "").strip() if raw: return raw if ing.component_id and ing.component_id in comp_names: return comp_names[ing.component_id] return "—" def ingredient_dry_matter_percent(ing: ZootechIngredient, components_by_id: dict[str, ZootechComponent]) -> float: dm = float(ing.dry_matter or 0) if dm > 0: return dm if ing.component_id and ing.component_id in components_by_id: return float(components_by_id[ing.component_id].dry_matter or 0) return 0.0 def ingredient_dry_matter_per_head( ing: ZootechIngredient, *, dry_matter_percent: float | None = None, ) -> float: if ing.dry_matter_per_head is not None and float(ing.dry_matter_per_head) > 0: return float(ing.dry_matter_per_head) wph = float(ing.weight_per_head or 0) dm = dry_matter_percent if dry_matter_percent is not None else float(ing.dry_matter or 0) if wph > 0 and dm > 0: return wph * (dm / 100.0) return 0.0 def _round_wph(value: float) -> float: if 0 < value < 0.01: return round(value, 3) return round(value, 2) def _total_kg(weight_per_head: float, heads: int, fallback_amount: float) -> float: if weight_per_head > 0: return round(weight_per_head * max(heads, 0), 2) return round(float(fallback_amount or 0), 2) def _apply_component_adjustment( *, recipe: ZootechRecipe, component_id: str | None, components_by_id: dict[str, ZootechComponent], master_wph: float, master_dm_ph: float, master_dm_pct: float, component_adjustment: dict[str, float | bool | None] | None, ) -> tuple[float, float, float, bool]: if not component_adjustment or not component_id: return master_wph, master_dm_ph, master_dm_pct, False adj_dm_pct = component_adjustment.get("dry_matter") if adj_dm_pct is not None: new_dm_pct = float(adj_dm_pct) locked = bool(recipe.dry_matter_locked) if locked: new_dm_ph = master_dm_ph if new_dm_pct > _EPSILON: new_wph = (master_dm_ph * 100.0) / new_dm_pct if new_wph < 0.01 and master_dm_ph >= 0.001: new_wph = 0.01 else: new_wph = 0.0 return _round_wph(new_wph), round(new_dm_ph, 4), new_dm_pct, True new_dm_ph = master_wph * (new_dm_pct / 100.0) if master_wph > 0 else master_dm_ph return master_wph, round(new_dm_ph, 4), new_dm_pct, True adj_wph = component_adjustment.get("weight_per_head") adj_dm_ph = component_adjustment.get("dry_matter_per_head") if adj_wph is None and adj_dm_ph is None: return master_wph, master_dm_ph, master_dm_pct, False comp = components_by_id.get(component_id) dm_pct = float(comp.dry_matter or 0) if comp else master_dm_pct locked = bool(recipe.dry_matter_locked) if locked and adj_dm_ph is not None: new_dm_ph = float(adj_dm_ph) new_wph = (new_dm_ph * 100.0) / dm_pct if dm_pct > _EPSILON else 0.0 return _round_wph(new_wph), new_dm_ph, dm_pct, True if not locked and adj_wph is not None: new_wph = float(adj_wph) new_dm_ph = new_wph * (dm_pct / 100.0) if dm_pct > 0 else 0.0 return _round_wph(new_wph), round(new_dm_ph, 4), dm_pct, True if adj_dm_ph is not None: new_dm_ph = float(adj_dm_ph) new_wph = (new_dm_ph * 100.0) / dm_pct if dm_pct > _EPSILON else master_wph return _round_wph(new_wph), new_dm_ph, dm_pct, True if adj_wph is not None: new_wph = float(adj_wph) new_dm_ph = new_wph * (dm_pct / 100.0) if dm_pct > 0 else master_dm_ph return _round_wph(new_wph), round(new_dm_ph, 4), dm_pct, True return master_wph, master_dm_ph, master_dm_pct, False def resolve_plan_ingredient_weights( ing: ZootechIngredient, recipe: ZootechRecipe, *, heads: int, components_by_id: dict[str, ZootechComponent], replacement_component_id: str | None = None, component_adjustment: dict[str, float | bool | None] | None = None, ) -> dict[str, Any]: master_wph = float(ing.weight_per_head or 0) master_dm_pct = ingredient_dry_matter_percent(ing, components_by_id) master_dm_ph = ingredient_dry_matter_per_head(ing, dry_matter_percent=master_dm_pct) effective_wph, effective_dm_ph, effective_dm_pct, adjusted_today = _apply_component_adjustment( recipe=recipe, component_id=ing.component_id, components_by_id=components_by_id, master_wph=master_wph, master_dm_ph=master_dm_ph, master_dm_pct=master_dm_pct, component_adjustment=component_adjustment, ) replacement = ( components_by_id.get(replacement_component_id) if replacement_component_id else None ) if replacement is None: wph = effective_wph return { "weightPerHead": wph, "totalKg": _total_kg(wph, heads, float(ing.amount or 0)), "dryMatterPct": effective_dm_pct, "dryMatterPerHead": round(effective_dm_ph, 4), "originalWeightPerHead": master_wph, "originalDryMatterPerHead": round(master_dm_ph, 4), "originalDryMatterPct": master_dm_pct, "adjustedToday": adjusted_today, "replacedToday": False, } repl_dm_pct = float(replacement.dry_matter or 0) locked = bool(recipe.dry_matter_locked) if locked: new_dm_ph = effective_dm_ph if repl_dm_pct > _EPSILON: new_wph = (new_dm_ph * 100.0) / repl_dm_pct else: new_wph = 0.0 mode = "dry_matter" else: new_wph = effective_wph new_dm_ph = new_wph * (repl_dm_pct / 100.0) if repl_dm_pct > 0 else 0.0 mode = "weight" new_wph = _round_wph(new_wph) return { "weightPerHead": new_wph, "totalKg": _total_kg(new_wph, heads, float(ing.amount or 0)), "dryMatterPct": repl_dm_pct, "dryMatterPerHead": round(new_dm_ph, 4), "originalWeightPerHead": master_wph, "originalDryMatterPerHead": round(master_dm_ph, 4), "originalDryMatterPct": master_dm_pct, "recalculationMode": mode, "adjustedToday": adjusted_today, "replacedToday": True, } def _load_components_by_id( db, enterprise_id: str, comp_ids: list[str], ) -> dict[str, ZootechComponent]: if not comp_ids: return {} rows = list( db.scalars( select(ZootechComponent).where( ZootechComponent.enterprise_id == enterprise_id, ZootechComponent.id.in_(set(comp_ids)), ZootechComponent.is_deleted.is_(False), ) ) ) return {c.id: c for c in rows} def _recipes_for_period_ordered(db, enterprise_id: str, period_id: str) -> list[ZootechRecipe]: 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.created_at.asc()) ) ) out: list[ZootechRecipe] = [] for link in links: recipe = db.scalar( select(ZootechRecipe).where( ZootechRecipe.enterprise_id == enterprise_id, ZootechRecipe.id == link.recipe_id, ZootechRecipe.is_deleted.is_(False), ) ) if recipe: out.append(recipe) return out def _mill_recipes( db, enterprise_id: str, *, dispenser_id: str | None = None, ) -> list[ZootechRecipe]: linked_ids = set( db.scalars( select(ZootechPeriodRecipe.recipe_id).where( ZootechPeriodRecipe.enterprise_id == enterprise_id, ZootechPeriodRecipe.is_deleted.is_(False), ) ) ) filters = [ ZootechRecipe.enterprise_id == enterprise_id, ZootechRecipe.is_deleted.is_(False), ] if linked_ids: filters.append(~ZootechRecipe.id.in_(linked_ids)) recipes = list( db.scalars( select(ZootechRecipe).where(*filters).order_by(ZootechRecipe.updated_at.desc()) ) ) if not dispenser_id or dispenser_id == ALL_MILLS_ID: return recipes filtered: list[ZootechRecipe] = [] for recipe in recipes: meta = _parse_payload_json(recipe.payload_json) rid = str(meta.get("dispenser_id") or "") if not rid or rid == str(dispenser_id): filtered.append(recipe) return filtered def get_skipped_recipe_ids(enterprise_id: str, plan_date: str | None = None) -> set[str]: target = _parse_plan_date(plan_date) return { str(row.get("recipe_id")) for row in _active_daily_rows(enterprise_id, "daily_trip_skip", target) if row.get("recipe_id") } def get_skipped_ingredient_ids(enterprise_id: str, plan_date: str | None = None) -> dict[str, set[str]]: target = _parse_plan_date(plan_date) out: dict[str, set[str]] = {} for row in _active_daily_rows(enterprise_id, "daily_ingredient_skip", target): recipe_id = str(row.get("recipe_id") or "") ingredient_id = str(row.get("ingredient_id") or "") if recipe_id and ingredient_id: out.setdefault(recipe_id, set()).add(ingredient_id) return out def get_skipped_unloading_group_ids(enterprise_id: str, plan_date: str | None = None) -> dict[str, set[str]]: target = _parse_plan_date(plan_date) out: dict[str, set[str]] = {} for row in _active_daily_rows(enterprise_id, "daily_unloading_group_skip", target): recipe_id = str(row.get("recipe_id") or "") group_id = str(row.get("unloading_group_id") or "") if recipe_id and group_id: out.setdefault(recipe_id, set()).add(group_id) return out def get_ingredient_replacement_map( enterprise_id: str, plan_date: str | None = None, ) -> dict[str, dict[str, str]]: target = _parse_plan_date(plan_date) out: dict[str, dict[str, str]] = {} for row in _active_daily_rows(enterprise_id, "daily_ingredient_replacement", target): recipe_id = str(row.get("recipe_id") or "") ingredient_id = str(row.get("ingredient_id") or "") replacement_id = str(row.get("replacement_component_id") or "") if recipe_id and ingredient_id and replacement_id: out.setdefault(recipe_id, {})[ingredient_id] = replacement_id return out def get_replaced_recipe_ids(enterprise_id: str, plan_date: str | None = None) -> set[str]: target = _parse_plan_date(plan_date) return { str(row.get("recipe_id")) for row in _active_daily_rows(enterprise_id, "daily_ingredient_replacement", target) if row.get("recipe_id") } def get_component_adjustment_map( enterprise_id: str, plan_date: str | None = None, ) -> dict[str, dict[str, float | bool | None]]: target = _parse_plan_date(plan_date) out: dict[str, dict[str, float | bool | None]] = {} for row in _active_daily_rows(enterprise_id, "daily_component_norm_adjustment", target): cid = str(row.get("component_id") or "") if not cid: continue dm = row.get("dry_matter") wph = row.get("weight_per_head") dm_ph = row.get("dry_matter_per_head") out[cid] = { "dry_matter": float(dm) if dm is not None else None, "dry_matter_locked": bool(row.get("dry_matter_locked")), "weight_per_head": float(wph) if wph is not None else None, "dry_matter_per_head": float(dm_ph) if dm_ph is not None else None, } return out def get_adjusted_recipe_ids(enterprise_id: str, plan_date: str | None = None) -> set[str]: adj_map = get_component_adjustment_map(enterprise_id, plan_date) if not adj_map: return set() skip_map = get_skipped_ingredient_ids(enterprise_id, plan_date) out: set[str] = set() with session_scope() as db: rows = db.execute( select( ZootechIngredient.recipe_id, ZootechIngredient.id, ZootechIngredient.component_id, ).where( ZootechIngredient.enterprise_id == enterprise_id, ZootechIngredient.is_deleted.is_(False), ZootechIngredient.component_id.isnot(None), ) ).all() for recipe_id, ing_id, component_id in rows: if str(component_id) not in adj_map: continue if ing_id in skip_map.get(str(recipe_id), set()): continue out.add(str(recipe_id)) return out def get_recipe_ids_with_any_skip(enterprise_id: str, plan_date: str | None = None) -> set[str]: target = _parse_plan_date(plan_date) ids: set[str] = set() for table in ( "daily_trip_skip", "daily_ingredient_skip", "daily_unloading_group_skip", ): for row in _active_daily_rows(enterprise_id, table, target): if row.get("recipe_id"): ids.add(str(row["recipe_id"])) ids.update(get_replaced_recipe_ids(enterprise_id, plan_date)) ids.update(get_adjusted_recipe_ids(enterprise_id, plan_date)) return ids def _recipe_name_map(db, enterprise_id: str, recipe_ids: set[str]) -> dict[str, str]: if not recipe_ids: return {} rows = list( db.scalars( select(ZootechRecipe).where( ZootechRecipe.enterprise_id == enterprise_id, ZootechRecipe.id.in_(recipe_ids), ZootechRecipe.is_deleted.is_(False), ) ) ) return {r.id: r.name for r in rows} def list_skips(enterprise_id: str, plan_date: str | None = None) -> list[dict[str, Any]]: target = _parse_plan_date(plan_date) rows = _active_daily_rows(enterprise_id, "daily_trip_skip", target) recipe_ids = {str(r.get("recipe_id")) for r in rows if r.get("recipe_id")} with session_scope() as db: names = _recipe_name_map(db, enterprise_id, recipe_ids) result = [] for row in rows: rid = str(row.get("recipe_id") or "") result.append( { "id": row.get("id"), "recipeId": rid, "recipeName": names.get(rid, "—"), **_serialize_skip_dates(row, fallback=target), } ) result.sort(key=lambda item: (item.get("recipeName") or "").lower()) return result def list_ingredient_skips(enterprise_id: str, plan_date: str | None = None) -> list[dict[str, Any]]: target = _parse_plan_date(plan_date) rows = _active_daily_rows(enterprise_id, "daily_ingredient_skip", target) recipe_ids = {str(r.get("recipe_id")) for r in rows if r.get("recipe_id")} ingredient_ids = {str(r.get("ingredient_id")) for r in rows if r.get("ingredient_id")} with session_scope() as db: recipe_names = _recipe_name_map(db, enterprise_id, recipe_ids) ingredients = list( db.scalars( select(ZootechIngredient).where( ZootechIngredient.enterprise_id == enterprise_id, ZootechIngredient.id.in_(ingredient_ids), ZootechIngredient.is_deleted.is_(False), ) ) ) ing_by_id = {i.id: i for i in ingredients} result = [] for row in rows: rid = str(row.get("recipe_id") or "") iid = str(row.get("ingredient_id") or "") ing = ing_by_id.get(iid) result.append( { "id": row.get("id"), "recipeId": rid, "recipeName": recipe_names.get(rid, "—"), "ingredientId": iid, "ingredientName": (ing.name if ing else None) or "—", **_serialize_skip_dates(row, fallback=target), } ) result.sort(key=lambda item: ((item.get("recipeName") or "").lower(), item.get("ingredientId") or "")) return result def list_unloading_group_skips(enterprise_id: str, plan_date: str | None = None) -> list[dict[str, Any]]: target = _parse_plan_date(plan_date) rows = _active_daily_rows(enterprise_id, "daily_unloading_group_skip", target) recipe_ids = {str(r.get("recipe_id")) for r in rows if r.get("recipe_id")} group_ids = {str(r.get("unloading_group_id")) for r in rows if r.get("unloading_group_id")} with session_scope() as db: recipe_names = _recipe_name_map(db, enterprise_id, recipe_ids) groups = list( db.scalars( select(ZootechUnloadingGroup).where( ZootechUnloadingGroup.enterprise_id == enterprise_id, ZootechUnloadingGroup.id.in_(group_ids), ZootechUnloadingGroup.is_deleted.is_(False), ) ) ) group_by_id = {g.id: g for g in groups} result = [] for row in rows: rid = str(row.get("recipe_id") or "") gid = str(row.get("unloading_group_id") or "") group = group_by_id.get(gid) result.append( { "id": row.get("id"), "recipeId": rid, "recipeName": recipe_names.get(rid, "—"), "unloadingGroupId": gid, "groupName": (group.name if group else None) or "—", **_serialize_skip_dates(row, fallback=target), } ) result.sort(key=lambda item: ((item.get("recipeName") or "").lower(), item.get("unloadingGroupId") or "")) return result def list_all_skips(enterprise_id: str, plan_date: str | None = None) -> dict[str, Any]: return { "trips": list_skips(enterprise_id, plan_date), "ingredients": list_ingredient_skips(enterprise_id, plan_date), "unloadingGroups": list_unloading_group_skips(enterprise_id, plan_date), } def list_ingredient_replacements(enterprise_id: str, plan_date: str | None = None) -> list[dict[str, Any]]: target = _parse_plan_date(plan_date) rows = _active_daily_rows(enterprise_id, "daily_ingredient_replacement", target) recipe_ids = {str(r.get("recipe_id")) for r in rows if r.get("recipe_id")} ingredient_ids = {str(r.get("ingredient_id")) for r in rows if r.get("ingredient_id")} replacement_ids = {str(r.get("replacement_component_id")) for r in rows if r.get("replacement_component_id")} with session_scope() as db: recipe_names = _recipe_name_map(db, enterprise_id, recipe_ids) ingredients = list( db.scalars( select(ZootechIngredient).where( ZootechIngredient.enterprise_id == enterprise_id, ZootechIngredient.id.in_(ingredient_ids), ZootechIngredient.is_deleted.is_(False), ) ) ) ing_by_id = {i.id: i for i in ingredients} comps = list( db.scalars( select(ZootechComponent).where( ZootechComponent.enterprise_id == enterprise_id, ZootechComponent.id.in_(replacement_ids), ZootechComponent.is_deleted.is_(False), ) ) ) comp_by_id = {c.id: c for c in comps} result = [] for row in rows: rid = str(row.get("recipe_id") or "") iid = str(row.get("ingredient_id") or "") repl_id = str(row.get("replacement_component_id") or "") ing = ing_by_id.get(iid) repl = comp_by_id.get(repl_id) result.append( { "id": row.get("id"), "recipeId": rid, "recipeName": recipe_names.get(rid, "—"), "ingredientId": iid, "ingredientName": (ing.name if ing else None) or "—", "replacementComponentId": repl_id, "replacementName": (repl.name if repl else None) or "—", **_serialize_skip_dates(row, fallback=target), } ) result.sort(key=lambda item: ((item.get("recipeName") or "").lower(), item.get("ingredientId") or "")) return result def list_component_norm_adjustments(enterprise_id: str, plan_date: str | None = None) -> list[dict[str, Any]]: target = _parse_plan_date(plan_date) rows = _active_daily_rows(enterprise_id, "daily_component_norm_adjustment", target) comp_ids = {str(r.get("component_id")) for r in rows if r.get("component_id")} with session_scope() as db: comps = list( db.scalars( select(ZootechComponent).where( ZootechComponent.enterprise_id == enterprise_id, ZootechComponent.id.in_(comp_ids), ZootechComponent.is_deleted.is_(False), ) ) ) comp_by_id = {c.id: c for c in comps} result = [] for row in rows: cid = str(row.get("component_id") or "") comp = comp_by_id.get(cid) result.append( { "id": row.get("id"), "componentId": cid, "componentName": (comp.name if comp else None) or "—", "dryMatter": row.get("dry_matter"), "dryMatterLocked": bool(row.get("dry_matter_locked")), "weightPerHead": row.get("weight_per_head"), "dryMatterPerHead": row.get("dry_matter_per_head"), **_serialize_skip_dates(row, fallback=target), } ) result.sort(key=lambda item: (item.get("componentName") or "").lower()) return result def _serialize_trip( db, enterprise_id: str, recipe: ZootechRecipe, *, order: int, skipped_ingredient_ids: set[str] | None = None, skipped_group_ids: set[str] | None = None, replacement_by_ingredient: dict[str, str] | None = None, component_adjustment_map: dict[str, dict[str, float | bool | None]] | None = None, ) -> dict[str, Any]: ingredients = list( db.scalars( select(ZootechIngredient) .where( ZootechIngredient.enterprise_id == enterprise_id, ZootechIngredient.recipe_id == recipe.id, ZootechIngredient.is_deleted.is_(False), ) .order_by(ZootechIngredient.order.asc()) ) ) comp_ids = [i.component_id for i in ingredients if i.component_id] repl_ids = list((replacement_by_ingredient or {}).values()) comp_ids.extend(repl_ids) components_by_id = _load_components_by_id(db, enterprise_id, [c for c in comp_ids if c]) comp_names = {cid: c.name for cid, c in components_by_id.items()} heads = int(recipe.heads_per_trip or 0) skip_ings = skipped_ingredient_ids or set() skip_groups = skipped_group_ids or set() repl_map = replacement_by_ingredient or {} adj_map = component_adjustment_map or {} dm_map = {cid: float(c.dry_matter or 0) for cid, c in components_by_id.items()} prepared: list[dict[str, Any]] = [] calc_inputs: list[dict[str, Any]] = [] calc_index_by_ing: dict[str, int] = {} for ing in ingredients: is_skipped = ing.id in skip_ings replacement_id = repl_map.get(ing.id) if not is_skipped else None component_adj = adj_map.get(str(ing.component_id)) if ing.component_id else None weights = resolve_plan_ingredient_weights( ing, recipe, heads=heads, components_by_id=components_by_id, replacement_component_id=replacement_id, component_adjustment=component_adj, ) original_name = _ingredient_display_name(ing, comp_names) replaced_today = bool(replacement_id) adjusted_today = bool(weights.get("adjustedToday")) display_name = original_name replacement_name = None if replaced_today: replacement_name = comp_names.get(replacement_id or "", "—") display_name = f"{original_name} → {replacement_name}" trip_kg = 0.0 if not is_skipped: calc_index_by_ing[ing.id] = len(calc_inputs) calc_inputs.append( { "weightPerHead": weights["weightPerHead"], "dryMatter": weights["dryMatterPct"], "component_id": replacement_id or ing.component_id, } ) prepared.append( { "ing": ing, "is_skipped": is_skipped, "weights": weights, "original_name": original_name, "display_name": display_name, "replacement_id": replacement_id, "replacement_name": replacement_name, "replaced_today": replaced_today, "adjusted_today": adjusted_today, "trip_kg": trip_kg, } ) groups = list( db.scalars( select(ZootechUnloadingGroup) .where( ZootechUnloadingGroup.enterprise_id == enterprise_id, ZootechUnloadingGroup.recipe_id == recipe.id, ZootechUnloadingGroup.is_deleted.is_(False), ) ) ) groups.sort(key=lambda g: _unloading_group_fields(g)["order"]) active_groups = [g for g in groups if g.id not in skip_groups] calc_groups_payload = [ { "distributionType": _unloading_group_fields(g)["distribution_type"], "value": _unloading_group_fields(g)["value"], } for g in active_groups ] calc_result = calculate_recipe( calc_inputs, heads_count=heads, trip_percent=float(recipe.trip_percent or 100), unloading_groups=calc_groups_payload, component_dry_matter_map=dm_map, ) calc_ingredients = calc_result.get("ingredients") or [] calc_unloading = calc_result.get("unloadingGroups") or [] total_weight = float(calc_result.get("totals", {}).get("totalTripWeight") or 0) unloading_total = float(calc_result.get("unloadingTotals", {}).get("totalWeightKg") or 0) baseline_inputs = [ { "weightPerHead": item["weights"]["weightPerHead"], "dryMatter": item["weights"]["dryMatterPct"], "component_id": item["replacement_id"] or item["ing"].component_id, } for item in prepared ] baseline_ingredients = calculate_ingredients( baseline_inputs, heads_count=heads, trip_percent=float(recipe.trip_percent or 100), component_dry_matter_map=dm_map, ) group_weight_by_id = { g.id: float(calc_unloading[idx].get("calculatedWeight") or 0) for idx, g in enumerate(active_groups) if idx < len(calc_unloading) } ing_rows = [] for idx, item in enumerate(prepared): ing = item["ing"] weights = item["weights"] if not item["is_skipped"]: calc_idx = calc_index_by_ing.get(ing.id) if calc_idx is not None and calc_idx < len(calc_ingredients): item["trip_kg"] = float(calc_ingredients[calc_idx].get("tripWeight") or 0) row: dict[str, Any] = { "id": ing.id, "name": item["display_name"], "originalName": item["original_name"], "weightPerHead": weights["weightPerHead"], "totalKg": item["trip_kg"], "dryMatterPct": weights["dryMatterPct"], "dryMatterPerHead": weights["dryMatterPerHead"], "componentId": ing.component_id, "originalComponentId": ing.component_id, "replacementComponentId": item["replacement_id"], "replacementName": item["replacement_name"], "skippedToday": item["is_skipped"], "replacedToday": item["replaced_today"], "adjustedToday": item["adjusted_today"], } if item["replaced_today"] or item["adjusted_today"]: row.update( { "originalWeightPerHead": weights.get("originalWeightPerHead"), "originalDryMatterPerHead": weights.get("originalDryMatterPerHead"), "originalDryMatterPct": weights.get("originalDryMatterPct"), } ) if item["replaced_today"]: row["recalculationMode"] = weights.get("recalculationMode") if item["is_skipped"] and idx < len(baseline_ingredients): baseline = baseline_ingredients[idx] row["baselineWeightPerHead"] = float(baseline.get("weightPerHead") or 0) row["baselineTotalKg"] = float(baseline.get("tripWeight") or 0) ing_rows.append(row) return { "order": order, "recipeId": recipe.id, "recipeName": recipe.name, "headsPerTrip": heads, "mixingTimeSec": int(recipe.mixing_time or 0), "tripPercent": float(recipe.trip_percent or 100), "dryMatterLocked": bool(recipe.dry_matter_locked), "totalWeightKg": round(total_weight, 2), "unloadingTotalKg": round(unloading_total, 2), "ingredients": ing_rows, "unloadingGroups": [ { "id": g.id, "name": g.name, "weightKg": 0.0 if g.id in skip_groups else float(group_weight_by_id.get(g.id, 0)), "distributionType": _unloading_group_fields(g)["distribution_type"], "distributionLabel": _distribution_label( _unloading_group_fields(g)["distribution_type"], _unloading_group_fields(g)["value"], ), "value": _unloading_group_fields(g)["value"], "skippedToday": g.id in skip_groups, } for g in groups ], } def _periods_for_dispenser( db, enterprise_id: str, dispenser: ZootechFeedDispenser, *, skipped_ids: set[str], skipped_ingredients: dict[str, set[str]], skipped_groups: dict[str, set[str]], replacement_map: dict[str, dict[str, str]], component_adjustment_map: dict[str, dict[str, float | bool | None]], ) -> list[dict[str, Any]]: periods = list( db.scalars( select(ZootechFeedingPeriod) .where( ZootechFeedingPeriod.enterprise_id == enterprise_id, ZootechFeedingPeriod.dispenser_id == dispenser.id, ZootechFeedingPeriod.is_deleted.is_(False), ) .order_by(ZootechFeedingPeriod.created_at.asc()) ) ) payload = [] for period in periods: if not _period_is_active(period): continue recipes = [ r for r in _recipes_for_period_ordered(db, enterprise_id, period.id) if r.id not in skipped_ids ] trips = [ _serialize_trip( db, enterprise_id, r, order=idx + 1, skipped_ingredient_ids=skipped_ingredients.get(r.id, set()), skipped_group_ids=skipped_groups.get(r.id, set()), replacement_by_ingredient=replacement_map.get(r.id, {}), component_adjustment_map=component_adjustment_map, ) for idx, r in enumerate(recipes) ] payload.append({"id": period.id, "name": period.name, "trips": trips}) return payload def _mill_trips_fallback( db, enterprise_id: str, dispenser_id: str, *, skipped_ids: set[str], skipped_ingredients: dict[str, set[str]], skipped_groups: dict[str, set[str]], replacement_map: dict[str, dict[str, str]], component_adjustment_map: dict[str, dict[str, float | bool | None]], ) -> list[dict[str, Any]]: recipes = _mill_recipes(db, enterprise_id, dispenser_id=dispenser_id) recipes = [r for r in recipes if r.id not in skipped_ids] trips = [ _serialize_trip( db, enterprise_id, r, order=idx + 1, skipped_ingredient_ids=skipped_ingredients.get(r.id, set()), skipped_group_ids=skipped_groups.get(r.id, set()), replacement_by_ingredient=replacement_map.get(r.id, {}), component_adjustment_map=component_adjustment_map, ) for idx, r in enumerate(recipes) ] if not trips: return [] return [{"id": None, "name": "Рейсы", "trips": trips}] def _aggregate_ingredient_totals(periods: list[dict[str, Any]]) -> dict[str, Any]: totals: dict[str, float] = {} grand_total = 0.0 for period in periods: for trip in period.get("trips") or []: for ing in trip.get("ingredients") or []: if ing.get("skippedToday"): continue if ing.get("replacedToday") and ing.get("replacementName"): name = str(ing.get("replacementName") or "—") else: name = str(ing.get("originalName") or ing.get("name") or "—") kg = float(ing.get("totalKg") or 0) totals[name] = totals.get(name, 0.0) + kg grand_total += kg rows = [ {"name": name, "totalKg": round(kg, 2)} for name, kg in sorted(totals.items(), key=lambda x: x[0].lower()) ] return {"rows": rows, "grandTotalKg": round(grand_total, 2)} def _periods_for_dispenser_named( db, enterprise_id: str, dispenser: ZootechFeedDispenser, *, prefix_name: bool = False, skipped_ids: set[str], skipped_ingredients: dict[str, set[str]], skipped_groups: dict[str, set[str]], replacement_map: dict[str, dict[str, str]], component_adjustment_map: dict[str, dict[str, float | bool | None]], ) -> list[dict[str, Any]]: meta = _dispenser_meta(dispenser) if meta["type"] == "mill": periods = _mill_trips_fallback( db, enterprise_id, dispenser.id, skipped_ids=skipped_ids, skipped_ingredients=skipped_ingredients, skipped_groups=skipped_groups, replacement_map=replacement_map, component_adjustment_map=component_adjustment_map, ) else: periods = _periods_for_dispenser( db, enterprise_id, dispenser, skipped_ids=skipped_ids, skipped_ingredients=skipped_ingredients, skipped_groups=skipped_groups, replacement_map=replacement_map, component_adjustment_map=component_adjustment_map, ) if not prefix_name: return periods prefixed: list[dict[str, Any]] = [] for period in periods: item = dict(period) item["name"] = f"{dispenser.name} · {period.get('name') or 'Период'}" item["dispenserId"] = dispenser.id prefixed.append(item) return prefixed def _plan_extras( enterprise_id: str, periods: list[dict[str, Any]], *, iso_date: str, ) -> dict[str, Any]: totals = _aggregate_ingredient_totals(periods) return { "ingredientTotals": totals["rows"], "ingredientGrandTotalKg": totals["grandTotalKg"], "ingredientReplacements": list_ingredient_replacements(enterprise_id, iso_date), "componentNormAdjustments": list_component_norm_adjustments(enterprise_id, iso_date), } def _plan_payload_base( enterprise_id: str, plan_date: str | None, ) -> tuple[str, set[str], list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: iso_date = _iso_date(plan_date) skipped_ids = get_skipped_recipe_ids(enterprise_id, iso_date) skipped_trips = list_skips(enterprise_id, iso_date) skipped_ingredients = list_ingredient_skips(enterprise_id, iso_date) skipped_groups = list_unloading_group_skips(enterprise_id, iso_date) return iso_date, skipped_ids, skipped_trips, skipped_ingredients, skipped_groups def _build_all_dispensers_plan(enterprise_id: str, plan_date: str | None = None) -> dict[str, Any]: iso_date, skipped_ids, skipped_trips, skipped_ingredients, skipped_groups = _plan_payload_base( enterprise_id, plan_date ) skip_ing_map = get_skipped_ingredient_ids(enterprise_id, iso_date) skip_grp_map = get_skipped_unloading_group_ids(enterprise_id, iso_date) replacement_map = get_ingredient_replacement_map(enterprise_id, iso_date) component_adjustment_map = get_component_adjustment_map(enterprise_id, iso_date) with session_scope() as db: dispensers = list( db.scalars( select(ZootechFeedDispenser) .where( ZootechFeedDispenser.enterprise_id == enterprise_id, ZootechFeedDispenser.is_deleted.is_(False), ) .order_by(ZootechFeedDispenser.name.asc()) ) ) dispensers = [d for d in dispensers if _dispenser_meta(d)["type"] == "dispenser"] periods: list[dict[str, Any]] = [] for dispenser in dispensers: periods.extend( _periods_for_dispenser_named( db, enterprise_id, dispenser, prefix_name=True, skipped_ids=skipped_ids, skipped_ingredients=skip_ing_map, skipped_groups=skip_grp_map, replacement_map=replacement_map, component_adjustment_map=component_adjustment_map, ) ) return { "date": iso_date, "generatedAt": datetime.now(UTC).isoformat(timespec="seconds"), "dispenserId": ALL_DISPENSERS_ID, "dispenserName": "Все кормораздатчики", "farm": "", "dispenserType": "dispenser", "periods": periods, **_plan_extras(enterprise_id, periods, iso_date=iso_date), "skippedTrips": skipped_trips, "skippedIngredients": skipped_ingredients, "skippedUnloadingGroups": skipped_groups, } def _build_all_mills_plan(enterprise_id: str, plan_date: str | None = None) -> dict[str, Any]: iso_date, skipped_ids, skipped_trips, skipped_ingredients, skipped_groups = _plan_payload_base( enterprise_id, plan_date ) skip_ing_map = get_skipped_ingredient_ids(enterprise_id, iso_date) skip_grp_map = get_skipped_unloading_group_ids(enterprise_id, iso_date) replacement_map = get_ingredient_replacement_map(enterprise_id, iso_date) component_adjustment_map = get_component_adjustment_map(enterprise_id, iso_date) with session_scope() as db: mills = [ d for d in db.scalars( select(ZootechFeedDispenser).where( ZootechFeedDispenser.enterprise_id == enterprise_id, ZootechFeedDispenser.is_deleted.is_(False), ) ) if _dispenser_meta(d)["type"] == "mill" ] periods = _mill_trips_fallback( db, enterprise_id, ALL_MILLS_ID, skipped_ids=skipped_ids, skipped_ingredients=skip_ing_map, skipped_groups=skip_grp_map, replacement_map=replacement_map, component_adjustment_map=component_adjustment_map, ) if mills: mill_names = ", ".join(d.name for d in mills) farm = _dispenser_meta(mills[0])["farm"] if len(mills) == 1 else "" else: mill_names = "" farm = "" return { "date": iso_date, "generatedAt": datetime.now(UTC).isoformat(timespec="seconds"), "dispenserId": ALL_MILLS_ID, "dispenserName": "Все кормоцеха", "farm": farm, "dispenserType": "mill", "scopeNote": mill_names, "periods": periods, **_plan_extras(enterprise_id, periods, iso_date=iso_date), "skippedTrips": skipped_trips, "skippedIngredients": skipped_ingredients, "skippedUnloadingGroups": skipped_groups, } def build_daily_plan( enterprise_id: str, dispenser_id: str, plan_date: str | None = None, ) -> dict[str, Any]: if dispenser_id == ALL_DISPENSERS_ID: return _build_all_dispensers_plan(enterprise_id, plan_date) if dispenser_id == ALL_MILLS_ID: return _build_all_mills_plan(enterprise_id, plan_date) iso_date, skipped_ids, skipped_trips, skipped_ingredients, skipped_groups = _plan_payload_base( enterprise_id, plan_date ) skip_ing_map = get_skipped_ingredient_ids(enterprise_id, iso_date) skip_grp_map = get_skipped_unloading_group_ids(enterprise_id, iso_date) replacement_map = get_ingredient_replacement_map(enterprise_id, iso_date) component_adjustment_map = get_component_adjustment_map(enterprise_id, iso_date) with session_scope() as db: dispenser = db.scalar( select(ZootechFeedDispenser).where( ZootechFeedDispenser.enterprise_id == enterprise_id, ZootechFeedDispenser.id == dispenser_id, ZootechFeedDispenser.is_deleted.is_(False), ) ) if dispenser is None: raise LookupError("Кормораздатчик не найден") meta = _dispenser_meta(dispenser) periods = _periods_for_dispenser_named( db, enterprise_id, dispenser, skipped_ids=skipped_ids, skipped_ingredients=skip_ing_map, skipped_groups=skip_grp_map, replacement_map=replacement_map, component_adjustment_map=component_adjustment_map, ) return { "date": iso_date, "generatedAt": datetime.now(UTC).isoformat(timespec="seconds"), "dispenserId": dispenser.id, "dispenserName": dispenser.name, "farm": meta["farm"], "dispenserType": meta["type"], "periods": periods, **_plan_extras(enterprise_id, periods, iso_date=iso_date), "skippedTrips": skipped_trips, "skippedIngredients": skipped_ingredients, "skippedUnloadingGroups": skipped_groups, } def trip_overlay_for_recipe( enterprise_id: str, recipe: ZootechRecipe, plan_date: str, ) -> dict[str, Any]: iso_date = _iso_date(plan_date) skipped_ing_by_recipe = get_skipped_ingredient_ids(enterprise_id, iso_date) skipped_grp_by_recipe = get_skipped_unloading_group_ids(enterprise_id, iso_date) replacement_map = get_ingredient_replacement_map(enterprise_id, iso_date) component_adjustment_map = get_component_adjustment_map(enterprise_id, iso_date) with session_scope() as db: return _serialize_trip( db, enterprise_id, recipe, order=1, skipped_ingredient_ids=skipped_ing_by_recipe.get(recipe.id, set()), skipped_group_ids=skipped_grp_by_recipe.get(recipe.id, set()), replacement_by_ingredient=replacement_map.get(recipe.id, {}), component_adjustment_map=component_adjustment_map, ) def _require_recipe(enterprise_id: str, recipe_id: str) -> None: 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 recipe is None: raise LookupError("Рецепт не найден") def _require_ingredient(enterprise_id: str, recipe_id: str, ingredient_id: str) -> None: with session_scope() as db: ingredient = db.scalar( select(ZootechIngredient).where( ZootechIngredient.enterprise_id == enterprise_id, ZootechIngredient.id == ingredient_id, ZootechIngredient.recipe_id == recipe_id, ZootechIngredient.is_deleted.is_(False), ) ) if ingredient is None: raise LookupError("Компонент не найден") def _require_unloading_group(enterprise_id: str, recipe_id: str, group_id: str) -> None: with session_scope() as db: group = db.scalar( select(ZootechUnloadingGroup).where( ZootechUnloadingGroup.enterprise_id == enterprise_id, ZootechUnloadingGroup.id == group_id, ZootechUnloadingGroup.recipe_id == recipe_id, ZootechUnloadingGroup.is_deleted.is_(False), ) ) if group is None: raise LookupError("Группа выгрузки не найдена") def skip_trip( enterprise_id: str, recipe_id: str, plan_date: str | None = None, *, duration: str | None = None, until_date: str | None = None, user: str = "system", ) -> dict[str, Any]: start = _parse_plan_date(plan_date) _, valid_until = resolve_skip_range(start, duration=duration, until_date=until_date) _require_recipe(enterprise_id, recipe_id) return _upsert_daily( enterprise_id, "daily_trip_skip", lookup={"recipe_id": recipe_id}, fields={ "recipe_id": recipe_id, "plan_date": start.isoformat(), "valid_until": _valid_until_field(start, valid_until), }, user=user, ) def unskip_trip( enterprise_id: str, recipe_id: str, plan_date: str | None = None, *, user: str = "system", ) -> bool: target = _parse_plan_date(plan_date) for row in _active_daily_rows(enterprise_id, "daily_trip_skip", target): if str(row.get("recipe_id")) == recipe_id: return soft_delete_daily_row(enterprise_id, "daily_trip_skip", str(row["id"])) return False def skip_ingredient( enterprise_id: str, recipe_id: str, ingredient_id: str, plan_date: str | None = None, *, duration: str | None = None, until_date: str | None = None, user: str = "system", ) -> dict[str, Any]: start = _parse_plan_date(plan_date) _, valid_until = resolve_skip_range(start, duration=duration, until_date=until_date) _require_recipe(enterprise_id, recipe_id) _require_ingredient(enterprise_id, recipe_id, ingredient_id) return _upsert_daily( enterprise_id, "daily_ingredient_skip", lookup={"recipe_id": recipe_id, "ingredient_id": ingredient_id}, fields={ "recipe_id": recipe_id, "ingredient_id": ingredient_id, "plan_date": start.isoformat(), "valid_until": _valid_until_field(start, valid_until), }, user=user, ) def unskip_ingredient( enterprise_id: str, recipe_id: str, ingredient_id: str, plan_date: str | None = None, *, user: str = "system", ) -> bool: target = _parse_plan_date(plan_date) for row in _active_daily_rows(enterprise_id, "daily_ingredient_skip", target): if str(row.get("recipe_id")) == recipe_id and str(row.get("ingredient_id")) == ingredient_id: return soft_delete_daily_row(enterprise_id, "daily_ingredient_skip", str(row["id"])) return False def unskip_all_ingredient_parts( enterprise_id: str, recipe_id: str, plan_date: str | None = None, *, user: str = "system", ) -> int: target = _parse_plan_date(plan_date) count = 0 for row in _active_daily_rows(enterprise_id, "daily_ingredient_skip", target): if str(row.get("recipe_id")) == recipe_id: if soft_delete_daily_row(enterprise_id, "daily_ingredient_skip", str(row["id"])): count += 1 return count def skip_unloading_group( enterprise_id: str, recipe_id: str, unloading_group_id: str, plan_date: str | None = None, *, duration: str | None = None, until_date: str | None = None, user: str = "system", ) -> dict[str, Any]: start = _parse_plan_date(plan_date) _, valid_until = resolve_skip_range(start, duration=duration, until_date=until_date) _require_recipe(enterprise_id, recipe_id) _require_unloading_group(enterprise_id, recipe_id, unloading_group_id) return _upsert_daily( enterprise_id, "daily_unloading_group_skip", lookup={"recipe_id": recipe_id, "unloading_group_id": unloading_group_id}, fields={ "recipe_id": recipe_id, "unloading_group_id": unloading_group_id, "plan_date": start.isoformat(), "valid_until": _valid_until_field(start, valid_until), }, user=user, ) def unskip_unloading_group( enterprise_id: str, recipe_id: str, unloading_group_id: str, plan_date: str | None = None, *, user: str = "system", ) -> bool: target = _parse_plan_date(plan_date) for row in _active_daily_rows(enterprise_id, "daily_unloading_group_skip", target): if ( str(row.get("recipe_id")) == recipe_id and str(row.get("unloading_group_id")) == unloading_group_id ): return soft_delete_daily_row(enterprise_id, "daily_unloading_group_skip", str(row["id"])) return False def unskip_all_unloading_group_parts( enterprise_id: str, recipe_id: str, plan_date: str | None = None, *, user: str = "system", ) -> int: target = _parse_plan_date(plan_date) count = 0 for row in _active_daily_rows(enterprise_id, "daily_unloading_group_skip", target): if str(row.get("recipe_id")) == recipe_id: if soft_delete_daily_row(enterprise_id, "daily_unloading_group_skip", str(row["id"])): count += 1 return count def _component_payload(component: ZootechComponent, *, similar: bool = False) -> dict[str, Any]: return { "id": component.id, "name": component.name, "type": component.type or "", "dryMatter": float(component.dry_matter or 0), "protein": float(component.protein or 0), "energy": float(component.energy or 0), "similar": similar, } def _similarity_score(source: ZootechComponent, candidate: ZootechComponent) -> float: if source.id == candidate.id: return -1.0 score = 0.0 src_type = (source.type or "").strip().lower() cand_type = (candidate.type or "").strip().lower() if src_type and cand_type and src_type == cand_type: score += 100.0 dm_diff = abs(float(source.dry_matter or 0) - float(candidate.dry_matter or 0)) score += max(0.0, 50.0 - dm_diff * 2.0) prot_diff = abs(float(source.protein or 0) - float(candidate.protein or 0)) score += max(0.0, 20.0 - prot_diff) return score def find_component_alternatives( enterprise_id: str, component_id: str, query: str = "", limit: int = 20, ) -> dict[str, Any]: with session_scope() as db: source = db.scalar( select(ZootechComponent).where( ZootechComponent.enterprise_id == enterprise_id, ZootechComponent.id == component_id, ZootechComponent.is_deleted.is_(False), ) ) if source is None: raise LookupError("Компонент не найден") q = (query or "").strip().lower() try: limit_val = max(1, min(int(limit), 100)) except (TypeError, ValueError): limit_val = 20 candidates = list( db.scalars( select(ZootechComponent).where( ZootechComponent.enterprise_id == enterprise_id, ZootechComponent.is_active.is_(True), ZootechComponent.is_deleted.is_(False), ZootechComponent.id != component_id, ) ) ) scored: list[tuple[float, ZootechComponent]] = [] for candidate in candidates: if q and q not in (candidate.name or "").lower(): continue score = _similarity_score(source, candidate) if score < 0: continue scored.append((score, candidate)) scored.sort(key=lambda item: (-item[0], (item[1].name or "").lower())) similar_cutoff = 80.0 if (source.type or "").strip() else 40.0 similar: list[dict[str, Any]] = [] all_items: list[dict[str, Any]] = [] for score, candidate in scored: is_similar = score >= similar_cutoff payload = _component_payload(candidate, similar=is_similar) all_items.append(payload) if is_similar and len(similar) < 8: similar.append(payload) if len(all_items) >= limit_val: break return { "sourceComponent": _component_payload(source), "similar": similar, "results": all_items, } def replace_ingredient( enterprise_id: str, recipe_id: str, ingredient_id: str, replacement_component_id: str, plan_date: str | None = None, *, duration: str | None = None, until_date: str | None = None, user: str = "system", ) -> dict[str, Any]: start = _parse_plan_date(plan_date) _, valid_until = resolve_skip_range(start, duration=duration, until_date=until_date) _require_recipe(enterprise_id, recipe_id) _require_ingredient(enterprise_id, recipe_id, ingredient_id) with session_scope() as db: ingredient = db.scalar( select(ZootechIngredient).where( ZootechIngredient.enterprise_id == enterprise_id, ZootechIngredient.id == ingredient_id, ZootechIngredient.recipe_id == recipe_id, ZootechIngredient.is_deleted.is_(False), ) ) replacement = db.scalar( select(ZootechComponent).where( ZootechComponent.enterprise_id == enterprise_id, ZootechComponent.id == replacement_component_id, ZootechComponent.is_deleted.is_(False), ZootechComponent.is_active.is_(True), ) ) if replacement is None: raise LookupError("Компонент-замена не найден") if ingredient and ingredient.component_id and ingredient.component_id == replacement_component_id: raise LookupError("Выберите другой компонент") return _upsert_daily( enterprise_id, "daily_ingredient_replacement", lookup={"recipe_id": recipe_id, "ingredient_id": ingredient_id}, fields={ "recipe_id": recipe_id, "ingredient_id": ingredient_id, "replacement_component_id": replacement_component_id, "plan_date": start.isoformat(), "valid_until": _valid_until_field(start, valid_until), }, user=user, ) def collect_component_replacement_targets( enterprise_id: str, component_id: str, plan_date: str | None = None, *, dispenser_id: str | None = None, ) -> list[tuple[str, str]]: iso_date = _iso_date(plan_date) if not dispenser_id: return [] try: plan = build_daily_plan(enterprise_id, dispenser_id, plan_date=iso_date) except LookupError: return [] seen: set[tuple[str, str]] = set() targets: list[tuple[str, str]] = [] for period in plan.get("periods") or []: for trip in period.get("trips") or []: recipe_id = str(trip.get("recipeId") or "") if not recipe_id: continue for ing in trip.get("ingredients") or []: if ing.get("skippedToday"): continue orig = str(ing.get("originalComponentId") or ing.get("componentId") or "") if orig != component_id: continue ingredient_id = str(ing.get("id") or "") if not ingredient_id: continue key = (recipe_id, ingredient_id) if key in seen: continue seen.add(key) targets.append(key) return targets def replace_component_in_plan( enterprise_id: str, component_id: str, replacement_component_id: str, plan_date: str | None = None, *, dispenser_id: str | None = None, duration: str | None = None, until_date: str | None = None, user: str = "system", ) -> list[dict[str, Any]]: if component_id == replacement_component_id: raise LookupError("Выберите другой компонент") targets = collect_component_replacement_targets( enterprise_id, component_id, plan_date, dispenser_id=dispenser_id, ) if not targets: raise LookupError("Компонент не найден в плане на эту дату") rows: list[dict[str, Any]] = [] for recipe_id, ingredient_id in targets: rows.append( replace_ingredient( enterprise_id, recipe_id, ingredient_id, replacement_component_id, plan_date, duration=duration, until_date=until_date, user=user, ) ) return rows def undo_ingredient_replacement( enterprise_id: str, recipe_id: str, ingredient_id: str, plan_date: str | None = None, *, user: str = "system", ) -> bool: target = _parse_plan_date(plan_date) for row in _active_daily_rows(enterprise_id, "daily_ingredient_replacement", target): if str(row.get("recipe_id")) == recipe_id and str(row.get("ingredient_id")) == ingredient_id: return soft_delete_daily_row(enterprise_id, "daily_ingredient_replacement", str(row["id"])) return False def _recipe_usages_summary(usages: list[dict[str, Any]]) -> list[dict[str, Any]]: by_recipe: dict[str, dict[str, Any]] = {} for usage in usages: rid = str(usage.get("recipeId") or "") if not rid: continue if rid not in by_recipe: by_recipe[rid] = { "recipeId": rid, "recipeName": usage.get("recipeName") or "—", "masterWeightPerHead": usage.get("masterWeightPerHead"), "masterDryMatterPerHead": usage.get("masterDryMatterPerHead"), "masterDryMatterPct": usage.get("masterDryMatterPct"), "dryMatterLocked": bool(usage.get("dryMatterLocked")), "tripCount": 1, } else: by_recipe[rid]["tripCount"] = int(by_recipe[rid].get("tripCount") or 0) + 1 return sorted(by_recipe.values(), key=lambda row: (row.get("recipeName") or "").lower()) def _master_norm_snapshot(component_id: str, usages: list[dict[str, Any]]) -> dict[str, Any]: wph_vals = [float(u["masterWeightPerHead"]) for u in usages if u.get("masterWeightPerHead") is not None] dm_vals = [float(u["masterDryMatterPerHead"]) for u in usages if u.get("masterDryMatterPerHead") is not None] dm_pct_vals = [float(u["masterDryMatterPct"]) for u in usages if u.get("masterDryMatterPct") is not None] return { "masterWeightPerHeadMin": min(wph_vals) if wph_vals else None, "masterWeightPerHeadMax": max(wph_vals) if wph_vals else None, "masterDryMatterPerHeadMin": min(dm_vals) if dm_vals else None, "masterDryMatterPerHeadMax": max(dm_vals) if dm_vals else None, "masterDryMatterPctMin": min(dm_pct_vals) if dm_pct_vals else None, "masterDryMatterPctMax": max(dm_pct_vals) if dm_pct_vals else None, "masterWeightPerHead": round(sum(wph_vals) / len(wph_vals), 3) if wph_vals else None, "masterDryMatterPerHead": round(sum(dm_vals) / len(dm_vals), 4) if dm_vals else None, "masterDryMatterPct": round(sum(dm_pct_vals) / len(dm_pct_vals), 2) if dm_pct_vals else None, } def list_component_norms_for_plan( enterprise_id: str, plan_date: str | None = None, *, dispenser_id: str | None = None, ) -> list[dict[str, Any]]: iso_date = _iso_date(plan_date) if not dispenser_id: return [] try: plan = build_daily_plan(enterprise_id, dispenser_id, plan_date=iso_date) except LookupError: return [] skip_map = get_skipped_ingredient_ids(enterprise_id, iso_date) adj_map = get_component_adjustment_map(enterprise_id, iso_date) comp_ids: set[str] = set() usages: dict[str, list[dict[str, Any]]] = defaultdict(list) comp_names: dict[str, str] = {} comp_dm: dict[str, float] = {} plan_effective: dict[str, dict[str, float | None]] = {} for period in plan.get("periods") or []: for trip in period.get("trips") or []: recipe_id = trip.get("recipeId") dry_matter_locked = bool(trip.get("dryMatterLocked")) for ing in trip.get("ingredients") or []: if ing.get("skippedToday"): continue cid = str(ing.get("componentId") or ing.get("originalComponentId") or "") if not cid: continue if cid not in plan_effective: plan_effective[cid] = { "weightPerHead": ( float(ing["weightPerHead"]) if ing.get("weightPerHead") is not None else None ), "dryMatterPerHead": ( float(ing["dryMatterPerHead"]) if ing.get("dryMatterPerHead") is not None else None ), "dryMatterPct": ( float(ing["dryMatterPct"]) if ing.get("dryMatterPct") is not None else None ), } comp_ids.add(cid) comp_names[cid] = str(ing.get("originalName") or ing.get("name") or "—") if ing.get("dryMatterPct") is not None: comp_dm[cid] = float(ing["dryMatterPct"]) master_wph = ing.get("originalWeightPerHead") master_dm = ing.get("originalDryMatterPerHead") if master_wph is None: master_wph = ing.get("weightPerHead") if master_dm is None: master_dm = ing.get("dryMatterPerHead") master_dm_pct = ing.get("originalDryMatterPct") if master_dm_pct is None: master_dm_pct = ing.get("dryMatterPct") usages[cid].append( { "recipeId": recipe_id, "recipeName": trip.get("recipeName"), "ingredientId": ing.get("id"), "masterWeightPerHead": master_wph, "masterDryMatterPerHead": master_dm, "masterDryMatterPct": master_dm_pct, "dryMatterLocked": dry_matter_locked, } ) if comp_ids: with session_scope() as db: comps = list( db.scalars( select(ZootechComponent).where( ZootechComponent.enterprise_id == enterprise_id, ZootechComponent.id.in_(comp_ids), ZootechComponent.is_deleted.is_(False), ) ) ) for comp in comps: comp_names[comp.id] = comp.name comp_dm[comp.id] = float(comp.dry_matter or 0) rows: list[dict[str, Any]] = [] for cid in sorted(comp_ids, key=lambda x: (comp_names.get(x) or x).lower()): usage_list = usages.get(cid, []) adj = adj_map.get(cid) snap = _master_norm_snapshot(cid, usage_list) eff = plan_effective.get(cid) or {} rows.append( { "componentId": cid, "componentName": comp_names.get(cid, "—"), "dryMatterPct": comp_dm.get(cid, 0), "usageCount": len(usage_list), "recipes": sorted({u.get("recipeName") or "" for u in usage_list if u.get("recipeName")}), "usages": _recipe_usages_summary(usage_list), **snap, "planDryMatterPct": adj.get("dry_matter") if adj else eff.get("dryMatterPct"), "planDryMatterLocked": bool(adj.get("dry_matter_locked")) if adj else None, "planWeightPerHead": eff.get("weightPerHead"), "planDryMatterPerHead": eff.get("dryMatterPerHead"), "adjustedToday": bool(adj), } ) return rows def adjust_component_norm( enterprise_id: str, component_id: str, plan_date: str | None = None, *, dry_matter: float | None = None, dry_matter_locked: bool = False, weight_per_head: float | None = None, dry_matter_per_head: float | None = None, duration: str | None = None, until_date: str | None = None, user: str = "system", ) -> dict[str, Any]: if dry_matter is None and weight_per_head is None and dry_matter_per_head is None: raise ValueError("Укажите dryMatter (СВ%)") start = _parse_plan_date(plan_date) _, valid_until = resolve_skip_range(start, duration=duration, until_date=until_date) with session_scope() as db: component = db.scalar( select(ZootechComponent).where( ZootechComponent.enterprise_id == enterprise_id, ZootechComponent.id == component_id, ZootechComponent.is_deleted.is_(False), ) ) if component is None: raise LookupError("Компонент не найден") if dry_matter is not None: weight_per_head = None dry_matter_per_head = None return _upsert_daily( enterprise_id, "daily_component_norm_adjustment", lookup={"component_id": component_id}, fields={ "component_id": component_id, "plan_date": start.isoformat(), "valid_until": _valid_until_field(start, valid_until), "dry_matter": float(dry_matter) if dry_matter is not None else None, "dry_matter_locked": bool(dry_matter_locked), "weight_per_head": float(weight_per_head) if weight_per_head is not None else None, "dry_matter_per_head": float(dry_matter_per_head) if dry_matter_per_head is not None else None, }, user=user, ) def undo_component_norm_adjustment( enterprise_id: str, component_id: str, plan_date: str | None = None, *, user: str = "system", ) -> bool: target = _parse_plan_date(plan_date) for row in _active_daily_rows(enterprise_id, "daily_component_norm_adjustment", target): if str(row.get("component_id")) == component_id: return soft_delete_daily_row(enterprise_id, "daily_component_norm_adjustment", str(row["id"])) return False def build_daily_plan_pdf(plan: dict[str, Any]) -> bytes: subtitle = [ f"Дата: {plan.get('date', '—')} · {plan.get('dispenserName', '—')} · {plan.get('farm', '')}", ] rows: list[list[Any]] = [] periods = plan.get("periods") or [] if not periods: rows.append(["Нет периодов или рейсов", "", ""]) else: for period in periods: rows.append([str(period.get("name") or "Период"), "", ""]) for trip in period.get("trips") or []: rows.append( [ f"Рейс {trip.get('order', '')}: {trip.get('recipeName', '—')}", f"{trip.get('headsPerTrip', 0)} гол.", f"смеш. {trip.get('mixingTimeSec', 0)} с", ] ) for ing in trip.get("ingredients") or []: rows.append( [ str(ing.get("name") or "—"), str(ing.get("weightPerHead") or ""), str(ing.get("totalKg") or ""), ] ) for group in trip.get("unloadingGroups") or []: rows.append( [ f"Группа: {group.get('name') or '—'}", str(group.get("weightKg") or ""), str(group.get("distributionLabel") or group.get("distributionType") or ""), ] ) rows.append(["", "", ""]) footer: list[str] = [] totals = plan.get("ingredientTotals") or [] if totals: footer.append("Итого по компонентам:") for row in totals: footer.append(f" {row.get('name', '—')}: {row.get('totalKg', 0)} кг") grand = plan.get("ingredientGrandTotalKg") if grand is not None: footer.append(f"Итого: {grand} кг") return build_table_pdf( title="План на день", subtitle_lines=subtitle, headers=["Компонент / рейс", "кг/гол или кг", "Всего / распределение"], rows=rows, footer_lines=footer or None, )