Files
2026-07-17 12:57:18 +03:00

633 lines
22 KiB
Python

from __future__ import annotations
import os
from flask import Blueprint, jsonify, render_template, request, session
from sqlalchemy import select
from app import db
from app.lab.calc.diff import compare_master_execution
from app.lab.calc.gfe_norms import preview_dynamic_norms
from app.lab.calc.norms_resolver import NormsParams, NormsResolveRequest, normalize_norms_method, resolve_norms
from app.lab.calc.nutrients import parse_num
from app.lab.calc.feed_groups import list_feed_groups_api, parse_group_selections
from app.lab.calc.formulate import FormulateRequest, formulate, list_formulate_components
from app.lab.calc.formulate_validate import validate_components
from app.lab.commands import (
apply_from_master,
delete_animal_profile,
ensure_empty_master,
recalculate_ration,
seed_from_execution,
sync_from_execution,
upsert_animal_profile,
upsert_ration,
)
from app.lab.loaders.execution_loader import load_execution
from app.lab.loaders.profile_loader import list_animal_profiles, load_animal_profile
from app.lab.norm_catalog import list_norm_indicators
from app.lab.services.seed_norms_catalog import (
ReferenceNormsCatalogEmptyError,
get_seed_norm_entry,
list_seed_norm_catalog,
)
from app.lab.etl.lab_import_router import parse_lab_import
from app.lab.etl.agrostar_xml_import import (
apply_agrostar_import,
enrich_parse_with_matches,
parse_agrostar_xml,
)
from app.lab.loaders.ration_loader import load_ration, ration_to_calc_lines
from app.lab.serde.api import diff_to_api, ration_to_api
from app.models import Recipe
from app.routes.auth_decorators import can_access_lab, require_auth
_LAB_TEMPLATES = os.path.join(os.path.dirname(__file__), "..", "lab", "templates")
bp = Blueprint("lab", __name__, url_prefix="/api/lab", template_folder=_LAB_TEMPLATES)
@bp.before_request
def _lab_module_access_guard():
if request.endpoint == "lab.health":
return None
if not session.get("authenticated", False):
return None
if not can_access_lab():
return (
jsonify(
{
"error": True,
"status": "error",
"message": "Модуль Lab недоступен для этого пользователя",
}
),
403,
)
return None
def _user() -> str:
return str(session.get("user_login") or "system")
@bp.get("/health")
def health():
return jsonify({"ok": True, "module": "lab"})
@bp.get("/recipes")
@require_auth
def list_lab_recipes():
"""Список партий для UI lab — только контур зоотехника (сессия)."""
rows = db.session.execute(
select(Recipe.id, Recipe.name)
.where(Recipe.is_deleted.is_(False))
.order_by(Recipe.name)
.limit(500)
).all()
return jsonify({"recipes": [{"id": rid, "name": name} for rid, name in rows]})
def _read_agrostar_xml_from_request() -> tuple[str | None, tuple | None]:
"""Вернуть (xml_text, error_response) — error_response = (json, status) или None."""
upload = request.files.get("file")
if upload is not None:
raw = upload.read()
for encoding in ("utf-8", "utf-8-sig", "cp1251"):
try:
return raw.decode(encoding), None
except UnicodeDecodeError:
continue
return None, (jsonify({"error": True, "message": "Не удалось прочитать кодировку XML"}), 400)
data = request.get_json(silent=True) or {}
xml_text = (data.get("xml") or data.get("content") or "").strip()
if xml_text:
return xml_text, None
return None, (jsonify({"error": True, "message": "Передайте file (multipart) или json.xml"}), 400)
@bp.post("/import/parse")
@require_auth
def post_lab_import_parse():
"""Единый комбайн: AgroStar xml/pdf/xlsx, ПЛИНОР pdf."""
upload = request.files.get("file")
if upload is None:
return jsonify({"error": True, "message": "Передайте file (multipart)"}), 400
data = upload.read()
body = parse_lab_import(upload.filename or "upload.bin", data)
if body.get("error"):
return jsonify(body), 400
return jsonify(body)
@bp.post("/import/agrostar-xml")
@require_auth
def post_agrostar_parse():
"""Разбор AgroStar Standard_XML_Data + подсказки сопоставления с component."""
xml_text, err = _read_agrostar_xml_from_request()
if err:
return err
parsed = parse_agrostar_xml(xml_text or "")
if parsed.errors and not parsed.samples:
return jsonify({"error": True, "message": "; ".join(parsed.errors), **parsed.to_api_dict()}), 400
body = enrich_parse_with_matches(parsed)
if parsed.errors:
body["warnings"] = parsed.errors
return jsonify(body)
@bp.post("/import/agrostar-xml/apply")
@require_auth
def post_agrostar_apply():
"""Запись nutrients и dry_matter в component по сопоставленным пробам."""
data = request.get_json(silent=True) or {}
assignments = data.get("assignments") or []
if not assignments:
return jsonify({"error": True, "message": "assignments required"}), 400
dry_run = bool(data.get("dryRun") or data.get("dry_run"))
result = apply_agrostar_import(assignments, user_id=_user(), dry_run=dry_run)
return jsonify(result.to_api_dict())
@bp.get("/rations/diff")
@require_auth
def get_diff_batch():
raw = (request.args.get("recipe_ids") or "").strip()
if not raw:
return jsonify({"error": True, "message": "recipe_ids required"}), 400
ids = [x.strip() for x in raw.split(",") if x.strip()]
results = []
for recipe_id in ids:
try:
master = load_ration(recipe_id)
execution = load_execution(recipe_id)
except LookupError:
results.append({"recipeId": recipe_id, "error": "not_found"})
continue
master_lines = [
{
"component_id": l.component_id,
"daily_kg": l.daily_kg,
"in_ration": l.in_ration,
}
for l in master.lines
]
exec_lines = [
{
"component_id": l.component_id,
"daily_kg_total": l.daily_kg_total,
}
for l in execution.lines
]
results.append(
{
"recipeId": recipe_id,
**diff_to_api(compare_master_execution(master_lines, exec_lines)),
}
)
return jsonify({"results": results})
@bp.get("/rations/<recipe_id>")
@require_auth
def get_ration(recipe_id: str):
try:
snapshot = load_ration(recipe_id)
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
return jsonify(ration_to_api(snapshot))
@bp.put("/rations/<recipe_id>")
@require_auth
def put_ration(recipe_id: str):
data = request.get_json(silent=True) or {}
try:
upsert_ration(recipe_id, data, _user())
snapshot = load_ration(recipe_id)
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
except Exception as exc:
return jsonify({"error": True, "message": str(exc)}), 400
return jsonify(ration_to_api(snapshot))
@bp.post("/rations/<recipe_id>/recalculate")
@require_auth
def post_recalculate(recipe_id: str):
try:
result = recalculate_ration(recipe_id, _user())
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
except ValueError as exc:
return jsonify({"error": True, "message": str(exc)}), 400
return jsonify(result)
@bp.get("/rations/<recipe_id>/diff")
@require_auth
def get_diff(recipe_id: str):
try:
master = load_ration(recipe_id)
execution = load_execution(recipe_id)
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
master_lines = [
{
"component_id": l.component_id,
"daily_kg": l.daily_kg,
"in_ration": l.in_ration,
}
for l in master.lines
]
exec_lines = [
{
"component_id": l.component_id,
"daily_kg_total": l.daily_kg_total,
}
for l in execution.lines
]
return jsonify(diff_to_api(compare_master_execution(master_lines, exec_lines)))
@bp.post("/rations/<recipe_id>/seed-from-execution")
@require_auth
def post_seed(recipe_id: str):
try:
result = seed_from_execution(recipe_id, _user())
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
return jsonify(result)
@bp.post("/rations/<recipe_id>/ensure-empty")
@require_auth
def post_ensure_empty(recipe_id: str):
try:
result = ensure_empty_master(recipe_id, _user())
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
return jsonify(result)
@bp.post("/rations/<recipe_id>/sync-from-execution")
@require_auth
def post_sync_from_execution(recipe_id: str):
try:
result = sync_from_execution(recipe_id, _user())
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
return jsonify(result)
@bp.post("/rations/<recipe_id>/apply-from-master")
@require_auth
def post_apply(recipe_id: str):
try:
result = apply_from_master(recipe_id, _user())
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
except ValueError as exc:
return jsonify({"error": True, "message": str(exc)}), 400
return jsonify(result)
@bp.get("/rations/<recipe_id>/compound")
@require_auth
def get_compound(recipe_id: str):
try:
snapshot = load_ration(recipe_id)
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
return jsonify(snapshot.compound_results or {})
@bp.get("/norm-indicators")
@require_auth
def get_norm_indicators():
return jsonify({"indicators": list_norm_indicators()})
@bp.get("/norms-preview")
@require_auth
def get_norms_preview():
"""Preview суточных норм по выбранной методике."""
from app.lab.models import LabAnimalProfile
from app.lab.services.norms_params import load_norms_params
method = normalize_norms_method(request.args.get("method"))
mass = parse_num(request.args.get("mass_kg"))
milk = parse_num(request.args.get("milk_yield_kg"))
profile_id = (request.args.get("profile_id") or "").strip()
stored: dict = {}
params = NormsParams()
if profile_id:
profile = LabAnimalProfile.query.filter_by(id=profile_id, is_deleted=False).first()
if profile is not None:
from app.lab.services.profile_norms import load_norms_dict
stored = load_norms_dict(profile.id)
params = load_norms_params(profile)
if mass is None:
mass = profile.mass_kg
if milk is None:
milk = profile.milk_yield_kg
if not request.args.get("method"):
method = normalize_norms_method(profile.norms_method)
override = NormsParams.from_dict(
{
"milkFatPct": parse_num(request.args.get("milk_fat_pct")),
"lactationNo": request.args.get("lactation_no"),
"bodyCondition": request.args.get("body_condition"),
"housingSystem": request.args.get("housing_system"),
"koncOeSv": parse_num(request.args.get("konc_oe_sv")),
}
)
params = NormsParams(
milk_fat_pct=override.milk_fat_pct if override.milk_fat_pct is not None else params.milk_fat_pct,
lactation_no=override.lactation_no if override.lactation_no is not None else params.lactation_no,
lactation_stage=override.lactation_stage if override.lactation_stage is not None else params.lactation_stage,
body_condition=override.body_condition if override.body_condition is not None else params.body_condition,
housing_system=override.housing_system if override.housing_system is not None else params.housing_system,
konc_oe_sv=override.konc_oe_sv if override.konc_oe_sv is not None else params.konc_oe_sv,
)
try:
resolved, meta = resolve_norms(
NormsResolveRequest(
method=method,
stored=stored,
mass_kg=mass,
milk_yield_kg=milk,
force_dynamic=method == "wesp",
params=params,
)
)
except ValueError as exc:
return jsonify({"error": True, "message": str(exc)}), 400
dynamic = meta.get("dynamicNorms") or meta.get("dynamic") or {}
if method == "wesp" and not dynamic:
dynamic = preview_dynamic_norms(mass, milk)
body: dict = {
"normsMethod": method,
"resolvedIndicators": resolved,
"dynamicNorms": dynamic,
"normsMeta": meta.get("meta"),
}
if meta.get("coverage"):
body["coverage"] = meta["coverage"]
return jsonify(body)
@bp.get("/gfe-norms-preview")
@require_auth
def get_gfe_norms_preview():
"""Расчётные min по GfE 2001 (alias method=wesp)."""
mass = parse_num(request.args.get("mass_kg"))
milk = parse_num(request.args.get("milk_yield_kg"))
return jsonify({"dynamicNorms": preview_dynamic_norms(mass, milk), "normsMethod": "wesp"})
@bp.get("/seed-norms-catalog")
@require_auth
def get_seed_norms_catalog():
"""Строки справочника zootech (lab_animal_profile norm_*)."""
ration_type = (request.args.get("ration_type") or "DAIRY").strip().upper()
try:
entries = list_seed_norm_catalog(ration_type)
except ValueError as exc:
return jsonify({"error": True, "message": str(exc)}), 400
except ReferenceNormsCatalogEmptyError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
return jsonify({"rationType": ration_type, "entries": entries})
@bp.get("/seed-norms-catalog/<int:external_no>")
@require_auth
def get_seed_norms_catalog_row(external_no: int):
ration_type = (request.args.get("ration_type") or "DAIRY").strip().upper()
try:
entry = get_seed_norm_entry(ration_type, external_no)
except ValueError as exc:
return jsonify({"error": True, "message": str(exc)}), 400
if entry is None:
return jsonify({"error": True, "message": "Строка справочника не найдена"}), 404
return jsonify(entry)
@bp.get("/formulate/groups")
@require_auth
def get_formulate_groups():
return jsonify({"groups": list_feed_groups_api()})
@bp.get("/formulate/components")
@require_auth
def get_formulate_components():
return jsonify({"components": list_formulate_components(), "groups": list_feed_groups_api()})
@bp.get("/formulate/validate")
@require_auth
def get_formulate_validate():
raw = (request.args.get("ids") or "").strip()
if not raw:
return jsonify({"error": True, "message": "ids required"}), 400
ids = [x.strip() for x in raw.split(",") if x.strip()]
return jsonify({"components": validate_components(ids)})
@bp.post("/formulate")
@require_auth
def post_formulate():
data = request.get_json(silent=True) or {}
group_selections = parse_group_selections(data.get("groupSelections"))
main_feed_ids = data.get("mainFeedIds") or []
additional_ids = data.get("additionalIds") or []
candidate_ids = data.get("candidateIds") or data.get("componentIds") or []
if group_selections:
candidate_ids = []
elif main_feed_ids or additional_ids:
if not isinstance(main_feed_ids, list) or not isinstance(additional_ids, list):
return jsonify({"error": True, "message": "mainFeedIds/additionalIds must be lists"}), 400
group_selections = {
"rough": [str(x) for x in main_feed_ids],
"succulent": [],
"concentrate": [str(x) for x in additional_ids if str(x) not in main_feed_ids],
"other": [],
}
for x in additional_ids:
sx = str(x)
if sx not in group_selections["rough"] and sx not in group_selections["concentrate"]:
group_selections["concentrate"].append(sx)
candidate_ids = []
elif not isinstance(candidate_ids, list):
return jsonify({"error": True, "message": "candidateIds must be a list"}), 400
profile_id = (data.get("profileId") or "").strip()
if not profile_id:
return jsonify({"error": True, "message": "profileId required"}), 400
try:
result = formulate(
FormulateRequest(
profile_id=profile_id,
candidate_ids=[str(x) for x in candidate_ids],
group_selections=group_selections,
main_feed_ids=[str(x) for x in main_feed_ids],
mass_kg=parse_formulate_num(data.get("massKg")),
milk_yield_kg=parse_formulate_num(data.get("milkYieldKg")),
heads_per_trip=_parse_int(data.get("headsPerTrip")),
total_kg_per_head=float(data.get("totalKgPerHead") or 7.3),
optimize_keys=list(data.get("optimizeKeys") or []),
objective=str(data.get("objective") or "min_cost"),
cost_weight=float(data.get("costWeight") or 100),
grid_step=float(data.get("gridStep") or 0.1),
prefilter_k=int(data.get("prefilterK") or 18),
min_share=float(data.get("minShare") or 0.05),
norms_method=str(data.get("normsMethod") or "wesp"),
norms_params=dict(data.get("normsParams") or {}),
)
)
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
except ValueError as exc:
args = exc.args
if args and args[0] == "ineligible_components":
return jsonify(
{
"error": True,
"message": "Некоторые компоненты не подходят для авторациона",
"components": args[1],
}
), 400
if args and args[0] == "candidate_ids_min_3":
return jsonify({"error": True, "message": "Нужно минимум 3 кандидата"}), 400
if args and args[0] == "candidate_ids_duplicate":
return jsonify({"error": True, "message": "Дубликаты в candidateIds"}), 400
if args and args[0] == "no_feasible_solution":
return jsonify({"error": True, "message": "Не удалось подобрать рацион"}), 400
if args and args[0] == "group_selection_invalid":
return jsonify(
{
"error": True,
"message": "Некорректный выбор по группам кормов",
"errors": args[1],
}
), 400
if args and args[0] == "no_feasible_solution_groups":
return jsonify(
{
"error": True,
"message": "Не удалось подобрать рацион с учётом обязательных групп кормов",
}
), 400
return jsonify({"error": True, "message": str(exc)}), 400
return jsonify(result)
def parse_formulate_num(value) -> float | None:
if value is None or value == "":
return None
try:
return float(value)
except (TypeError, ValueError):
return None
def _parse_int(value) -> int | None:
if value is None or value == "":
return None
try:
return int(value)
except (TypeError, ValueError):
return None
@bp.get("/animal-profiles")
@require_auth
def get_profiles():
ration_type = (request.args.get("ration_type") or "").strip() or None
return jsonify({"profiles": list_animal_profiles(ration_type)})
@bp.post("/animal-profiles")
@require_auth
def post_profile():
data = request.get_json(silent=True) or {}
try:
result = upsert_animal_profile(None, data, _user())
except ValueError as exc:
return jsonify({"error": True, "message": str(exc)}), 400
return jsonify(result), 201
@bp.get("/animal-profiles/<profile_id>")
@require_auth
def get_profile(profile_id: str):
try:
return jsonify(load_animal_profile(profile_id))
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
@bp.put("/animal-profiles/<profile_id>")
@require_auth
def put_profile(profile_id: str):
data = request.get_json(silent=True) or {}
try:
result = upsert_animal_profile(profile_id, data, _user())
except ValueError as exc:
return jsonify({"error": True, "message": str(exc)}), 400
return jsonify(result)
@bp.delete("/animal-profiles/<profile_id>")
@require_auth
def delete_profile(profile_id: str):
try:
result = delete_animal_profile(profile_id, _user())
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
return jsonify(result)
@bp.get("/rations/<recipe_id>/print")
@require_auth
def get_print(recipe_id: str):
mode = (request.args.get("mode") or "ration").strip().lower()
if mode not in ("ration", "compound"):
return jsonify({"error": True, "message": "mode must be ration or compound"}), 400
try:
snapshot = load_ration(recipe_id)
except LookupError as exc:
return jsonify({"error": True, "message": str(exc)}), 404
if not snapshot.exists:
return jsonify({"error": True, "message": "Мастер рациона не найден"}), 404
lines = []
for line in snapshot.lines:
included = line.in_ration if mode == "ration" else line.in_compound
if not included:
continue
lines.append(
{
"name": line.ingredient_name or line.component_id or "—",
"daily_kg": line.daily_kg,
"included": True,
}
)
results = snapshot.ration_results if mode == "ration" else snapshot.compound_results
indicators = (results or {}).get("indicators") or []
title = "Рацион" if mode == "ration" else "Комбикорм"
return render_template(
"print_ration.html",
title=title,
recipe_name=snapshot.recipe_name,
ration_type=snapshot.ration_type,
heads=snapshot.heads_per_trip,
mode=mode,
lines=lines,
indicators=indicators[:12],
)