116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
"""Строки листа «Итоги» для Excel (экономист)."""
|
|
|
|
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.finance_summary import _report_filters
|
|
from app.services.analytics.prices import component_prices, price_for
|
|
from app.services.analytics.recipe_filter import parse_recipe_ids
|
|
from app.services.analytics.recipe_location import build_recipe_location_maps
|
|
|
|
|
|
def build_summary_table_rows(
|
|
*,
|
|
date_from: Optional[str] = None,
|
|
date_to: Optional[str] = None,
|
|
recipe_id: Optional[str] = None,
|
|
recipe_ids: Optional[str] = None,
|
|
) -> List[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 []
|
|
|
|
report_by_id = {r.id: r for r in reports}
|
|
recipe_ids_set = {r.recipe_id for r in reports if r.recipe_id}
|
|
farm_by_recipe, _ = build_recipe_location_maps(recipe_ids_set)
|
|
|
|
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()
|
|
|
|
prices = component_prices(
|
|
component_ids=[c.component_id for c in components if c.component_id],
|
|
names=[c.component_name for c in components],
|
|
)
|
|
|
|
agg: Dict[tuple[str, str], Dict[str, Any]] = defaultdict(
|
|
lambda: {
|
|
"farm": "",
|
|
"component": "",
|
|
"planKg": 0.0,
|
|
"factKg": 0.0,
|
|
"pricePerKg": 0.0,
|
|
"overloadRub": 0.0,
|
|
"underloadRub": 0.0,
|
|
}
|
|
)
|
|
|
|
for comp in components:
|
|
report = report_by_id.get(comp.report_id)
|
|
if not report:
|
|
continue
|
|
farm = farm_by_recipe.get(report.recipe_id, "—")
|
|
cname = str(comp.component_name or "—")
|
|
key = (farm, cname)
|
|
row = agg[key]
|
|
row["farm"] = farm
|
|
row["component"] = cname
|
|
|
|
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=cname,
|
|
)
|
|
row["planKg"] += target
|
|
row["factKg"] += actual
|
|
row["pricePerKg"] = price
|
|
|
|
dev_kg = actual - target
|
|
dev_rub = dev_kg * price
|
|
if dev_rub > 0:
|
|
row["overloadRub"] += dev_rub
|
|
elif dev_rub < 0:
|
|
row["underloadRub"] += abs(dev_rub)
|
|
|
|
out: List[Dict[str, Any]] = []
|
|
for row in agg.values():
|
|
if row["planKg"] <= 0 and row["factKg"] <= 0:
|
|
continue
|
|
plan = round(row["planKg"], 2)
|
|
fact = round(row["factKg"], 2)
|
|
out.append(
|
|
{
|
|
"farm": row["farm"],
|
|
"component": row["component"],
|
|
"planKg": plan,
|
|
"factKg": fact,
|
|
"deviationKg": round(fact - plan, 2),
|
|
"pricePerKg": round(row["pricePerKg"], 2),
|
|
"overloadRub": round(row["overloadRub"], 2),
|
|
"underloadRub": round(row["underloadRub"], 2),
|
|
}
|
|
)
|
|
|
|
out.sort(key=lambda r: (r["farm"], r["component"]))
|
|
return out
|