"""Итоги в рублях: перерасход, недогруз, топ компонентов.""" from __future__ import annotations from collections import defaultdict from typing import Any, Dict, List, Optional from sqlalchemy import select from app import db from app.models import LoadingReport, LoadingReportComponent from app.services.analytics.date_range import resolve_date_bounds from app.services.analytics.prices import component_prices, price_for from app.services.analytics.recipe_filter import parse_recipe_ids def _report_filters( *, start_dt, end_dt, recipe_id: Optional[str] = None, recipe_ids: Optional[list[str]] = None, ): q = select(LoadingReport).where( LoadingReport.is_deleted.is_(False), LoadingReport.start_time >= start_dt, LoadingReport.start_time <= end_dt, ) ids = recipe_ids or parse_recipe_ids(recipe_id=recipe_id) if ids: q = q.where(LoadingReport.recipe_id.in_(ids) if len(ids) > 1 else LoadingReport.recipe_id == ids[0]) return q.order_by(LoadingReport.start_time.desc()) def build_finance_summary( *, date_from: Optional[str] = None, date_to: Optional[str] = None, recipe_id: Optional[str] = None, recipe_ids: Optional[str] = None, ) -> Dict[str, Any]: start_dt, end_dt = resolve_date_bounds(date_from, date_to) ids = parse_recipe_ids(recipe_id=recipe_id, recipe_ids=recipe_ids) reports = db.session.execute(_report_filters( start_dt=start_dt, end_dt=end_dt, recipe_ids=ids )).scalars().all() if not reports: return { "overloadRub": 0.0, "underloadRub": 0.0, "netRub": 0.0, "dominantIssue": "balanced", "topComponents": [], "reportCount": 0, } report_ids = [r.id for r in reports] components = db.session.execute( select(LoadingReportComponent).where( LoadingReportComponent.report_id.in_(report_ids), LoadingReportComponent.is_deleted.is_(False), ) ).scalars().all() comp_ids = [c.component_id for c in components if c.component_id] names = [c.component_name for c in components] prices = component_prices(component_ids=comp_ids, names=names) by_component: Dict[str, Dict[str, Any]] = defaultdict( lambda: {"name": "", "overloadRub": 0.0, "underloadRub": 0.0, "netRub": 0.0} ) overload_total = 0.0 underload_total = 0.0 for comp in components: target = float(comp.target_weight or 0) actual = float(comp.actual_weight or 0) if target <= 0 and actual <= 0: continue price = price_for( prices=prices, component_id=comp.component_id, name=str(comp.component_name or ""), ) dev_kg = actual - target dev_rub = dev_kg * price key = str(comp.component_id or comp.component_name or "unknown") row = by_component[key] row["name"] = str(comp.component_name or "—") row["componentId"] = comp.component_id if dev_rub > 0: row["overloadRub"] += dev_rub overload_total += dev_rub elif dev_rub < 0: row["underloadRub"] += abs(dev_rub) underload_total += abs(dev_rub) row["netRub"] += dev_rub top = sorted( by_component.values(), key=lambda x: max(x["overloadRub"], x["underloadRub"]), reverse=True, )[:3] top_out = [ { "name": r["name"], "componentId": r.get("componentId"), "overloadRub": round(r["overloadRub"], 2), "underloadRub": round(r["underloadRub"], 2), "netRub": round(r["netRub"], 2), } for r in top if max(r["overloadRub"], r["underloadRub"]) > 0 ] net = overload_total - underload_total if overload_total > underload_total: dominant = "overload" elif underload_total > overload_total: dominant = "underload" else: dominant = "balanced" return { "overloadRub": round(overload_total, 2), "underloadRub": round(underload_total, 2), "netRub": round(net, 2), "dominantIssue": dominant, "topComponents": top_out, "reportCount": len(reports), }