331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""Чтение алертов feed_quality для API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app import db
|
|
from app.models import (
|
|
ComponentLoadingTime,
|
|
FeedingPeriod,
|
|
LoadingReport,
|
|
LoadingReportComponent,
|
|
PeriodRecipe,
|
|
UnloadingReport,
|
|
UnloadingReportGroup,
|
|
)
|
|
from app.models.feed_alert import FeedAlert
|
|
from app.services.feed_quality.evaluator import backfill_feed_alerts_for_period
|
|
from app.services.feed_quality.rules import EVENT_LEFT_IN_MIXER, EVENT_MIX_TIME
|
|
from app.timeutil import naive_utc_to_iso
|
|
|
|
|
|
def _parse_date(value: Optional[str], *, end_of_day: bool = False) -> Optional[datetime]:
|
|
if not value:
|
|
return None
|
|
try:
|
|
dt = datetime.strptime(str(value).strip()[:10], "%Y-%m-%d")
|
|
except ValueError:
|
|
return None
|
|
if end_of_day:
|
|
return dt.replace(hour=23, minute=59, second=59, microsecond=999999)
|
|
return dt.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
|
|
|
def _recipe_ids_for_dispenser(dispenser_id: str) -> List[str]:
|
|
period_ids = (
|
|
db.session.execute(
|
|
select(FeedingPeriod.id).where(
|
|
FeedingPeriod.dispenser_id == dispenser_id,
|
|
FeedingPeriod.is_deleted.is_(False),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
if not period_ids:
|
|
return []
|
|
return list(
|
|
db.session.execute(
|
|
select(PeriodRecipe.recipe_id).where(
|
|
PeriodRecipe.period_id.in_(period_ids),
|
|
PeriodRecipe.is_deleted.is_(False),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
|
|
def _parse_dispenser_ids(dispenser_id: Optional[str]) -> List[str]:
|
|
if not dispenser_id:
|
|
return []
|
|
return [part.strip() for part in str(dispenser_id).split(",") if part.strip()]
|
|
|
|
|
|
def _recipe_ids_for_dispensers(dispenser_ids: List[str]) -> List[str]:
|
|
recipe_ids: set[str] = set()
|
|
for dispenser_id in dispenser_ids:
|
|
recipe_ids.update(_recipe_ids_for_dispenser(dispenser_id))
|
|
return list(recipe_ids)
|
|
|
|
|
|
def _unload_duration_sec(report: Optional[UnloadingReport]) -> Optional[float]:
|
|
if report is None or not report.start_time or not report.end_time:
|
|
return None
|
|
delta = (report.end_time - report.start_time).total_seconds()
|
|
return float(delta) if delta > 0 else None
|
|
|
|
|
|
def _build_report_context(loading_report_ids: List[str]) -> Dict[str, Any]:
|
|
if not loading_report_ids:
|
|
return {
|
|
"component_weights": {},
|
|
"component_durations": {},
|
|
"group_weights": {},
|
|
"group_durations": {},
|
|
"mix_times": {},
|
|
"remaining_by_loading": {},
|
|
"unload_duration_by_loading": {},
|
|
}
|
|
|
|
component_weights: Dict[tuple, Dict[str, float]] = {}
|
|
component_durations: Dict[tuple, float] = {}
|
|
for row in (
|
|
db.session.execute(
|
|
select(LoadingReportComponent).where(
|
|
LoadingReportComponent.report_id.in_(loading_report_ids),
|
|
LoadingReportComponent.is_deleted.is_(False),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
):
|
|
key = (row.report_id, str(row.component_name or ""))
|
|
component_weights[key] = {
|
|
"targetKg": float(row.target_weight or 0),
|
|
"actualKg": float(row.actual_weight or 0),
|
|
}
|
|
|
|
for row in (
|
|
db.session.execute(
|
|
select(ComponentLoadingTime).where(
|
|
ComponentLoadingTime.report_id.in_(loading_report_ids),
|
|
ComponentLoadingTime.is_deleted.is_(False),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
):
|
|
key = (row.report_id, str(row.component_name or ""))
|
|
component_durations[key] = float(row.loading_duration or 0)
|
|
|
|
mix_times: Dict[str, Dict[str, Optional[int]]] = {}
|
|
for row in (
|
|
db.session.execute(
|
|
select(LoadingReport).where(
|
|
LoadingReport.id.in_(loading_report_ids),
|
|
LoadingReport.is_deleted.is_(False),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
):
|
|
mix_times[row.id] = {
|
|
"targetMixingSec": row.target_mixing_time,
|
|
"actualMixingSec": row.actual_mixing_time,
|
|
}
|
|
|
|
group_weights: Dict[tuple, Dict[str, float]] = {}
|
|
group_durations: Dict[tuple, Optional[float]] = {}
|
|
remaining_by_loading: Dict[str, float] = {}
|
|
unload_duration_by_loading: Dict[str, Optional[float]] = {}
|
|
unloading_rows = (
|
|
db.session.execute(
|
|
select(UnloadingReport).where(
|
|
UnloadingReport.loading_report_id.in_(loading_report_ids),
|
|
UnloadingReport.is_deleted.is_(False),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
unload_by_id = {row.id: row for row in unloading_rows}
|
|
unload_duration_by_id = {
|
|
row.id: _unload_duration_sec(row) for row in unloading_rows
|
|
}
|
|
for row in unloading_rows:
|
|
remaining_by_loading[row.loading_report_id] = float(row.remaining_weight or 0)
|
|
unload_duration_by_loading[row.loading_report_id] = _unload_duration_sec(row)
|
|
|
|
if unload_by_id:
|
|
for group in (
|
|
db.session.execute(
|
|
select(UnloadingReportGroup).where(
|
|
UnloadingReportGroup.report_id.in_(list(unload_by_id.keys())),
|
|
UnloadingReportGroup.is_deleted.is_(False),
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
):
|
|
unloading = unload_by_id.get(group.report_id)
|
|
if unloading is None:
|
|
continue
|
|
key = (unloading.loading_report_id, str(group.name or ""))
|
|
group_weights[key] = {
|
|
"targetKg": float(group.target_weight or 0),
|
|
"actualKg": float(group.unloaded_weight or 0),
|
|
}
|
|
group_durations[key] = unload_duration_by_id.get(group.report_id)
|
|
|
|
return {
|
|
"component_weights": component_weights,
|
|
"component_durations": component_durations,
|
|
"group_weights": group_weights,
|
|
"group_durations": group_durations,
|
|
"mix_times": mix_times,
|
|
"remaining_by_loading": remaining_by_loading,
|
|
"unload_duration_by_loading": unload_duration_by_loading,
|
|
}
|
|
|
|
|
|
def _attach_fact_and_duration(item: Dict[str, Any], ctx: Dict[str, Any]) -> Dict[str, Any]:
|
|
loading_id = item.get("loadingReportId") or ""
|
|
event_type = item.get("eventType") or ""
|
|
component_name = str(item.get("componentName") or "")
|
|
group_name = str(item.get("groupName") or "")
|
|
|
|
target_kg = None
|
|
actual_kg = None
|
|
duration_sec = None
|
|
|
|
if component_name:
|
|
weights = ctx["component_weights"].get((loading_id, component_name))
|
|
if weights:
|
|
target_kg = weights.get("targetKg")
|
|
actual_kg = weights.get("actualKg")
|
|
duration_sec = ctx["component_durations"].get((loading_id, component_name))
|
|
elif group_name:
|
|
weights = ctx["group_weights"].get((loading_id, group_name))
|
|
if weights:
|
|
target_kg = weights.get("targetKg")
|
|
actual_kg = weights.get("actualKg")
|
|
duration_sec = ctx["unload_duration_by_loading"].get(loading_id)
|
|
|
|
if event_type == EVENT_MIX_TIME:
|
|
mix = ctx["mix_times"].get(loading_id) or {}
|
|
target_kg = mix.get("targetMixingSec")
|
|
actual_kg = mix.get("actualMixingSec")
|
|
duration_sec = float(actual_kg) if actual_kg is not None else None
|
|
elif event_type == EVENT_LEFT_IN_MIXER:
|
|
actual_kg = ctx["remaining_by_loading"].get(loading_id)
|
|
duration_sec = ctx["unload_duration_by_loading"].get(loading_id)
|
|
|
|
item["targetKg"] = target_kg
|
|
item["actualKg"] = actual_kg
|
|
item["durationSec"] = duration_sec
|
|
return item
|
|
|
|
|
|
def _serialize_alert(row: FeedAlert) -> Dict[str, Any]:
|
|
return {
|
|
"id": row.id,
|
|
"eventType": row.event_type,
|
|
"severity": row.severity,
|
|
"loadingReportId": row.loading_report_id,
|
|
"unloadingReportId": row.unloading_report_id,
|
|
"recipeId": row.recipe_id,
|
|
"recipeName": row.recipe_name,
|
|
"componentName": row.component_name,
|
|
"groupName": row.group_name,
|
|
"detail": row.detail,
|
|
"deviationKg": row.deviation_kg,
|
|
"deviationPct": row.deviation_pct,
|
|
"costDeviationRub": row.cost_deviation_rub,
|
|
"clientId": row.client_id,
|
|
"createdAt": naive_utc_to_iso(row.created_at),
|
|
"linkKind": "report_loading",
|
|
"linkId": row.loading_report_id,
|
|
"targetKg": None,
|
|
"actualKg": None,
|
|
"durationSec": None,
|
|
}
|
|
|
|
|
|
def list_feed_alerts(
|
|
*,
|
|
date_from: Optional[str] = None,
|
|
date_to: Optional[str] = None,
|
|
severity: Optional[str] = None,
|
|
event_type: Optional[str] = None,
|
|
dispenser_id: Optional[str] = None,
|
|
limit: int = 200,
|
|
) -> Dict[str, Any]:
|
|
dt_from = _parse_date(date_from)
|
|
dt_to = _parse_date(date_to, end_of_day=True)
|
|
lim = max(1, min(int(limit), 500))
|
|
|
|
if date_from or date_to:
|
|
backfill_feed_alerts_for_period(date_from=date_from, date_to=date_to, limit=lim)
|
|
|
|
stmt = (
|
|
select(FeedAlert, LoadingReport.start_time)
|
|
.join(LoadingReport, LoadingReport.id == FeedAlert.loading_report_id)
|
|
.where(LoadingReport.is_deleted.is_(False))
|
|
)
|
|
if dt_from is not None:
|
|
stmt = stmt.where(LoadingReport.start_time >= dt_from)
|
|
if dt_to is not None:
|
|
stmt = stmt.where(LoadingReport.start_time <= dt_to)
|
|
if severity:
|
|
stmt = stmt.where(FeedAlert.severity == severity.strip().lower())
|
|
if event_type:
|
|
stmt = stmt.where(FeedAlert.event_type == event_type.strip().upper())
|
|
if dispenser_id:
|
|
dispenser_ids = _parse_dispenser_ids(dispenser_id)
|
|
recipe_ids = _recipe_ids_for_dispensers(dispenser_ids)
|
|
if recipe_ids:
|
|
stmt = stmt.where(FeedAlert.recipe_id.in_(recipe_ids))
|
|
else:
|
|
return {"items": [], "total": 0, "bySeverity": {"warning": 0, "error": 0, "info": 0}}
|
|
|
|
stmt = stmt.order_by(LoadingReport.start_time.desc(), FeedAlert.created_at.desc()).limit(lim)
|
|
rows = db.session.execute(stmt).all()
|
|
|
|
items = [_serialize_alert(alert) for alert, _ in rows]
|
|
loading_ids = list({item["loadingReportId"] for item in items if item.get("loadingReportId")})
|
|
ctx = _build_report_context(loading_ids)
|
|
items = [_attach_fact_and_duration(item, ctx) for item in items]
|
|
by_severity = {"warning": 0, "error": 0, "info": 0}
|
|
for item in items:
|
|
sev = item.get("severity") or "warning"
|
|
if sev in by_severity:
|
|
by_severity[sev] += 1
|
|
return {"items": items, "total": len(items), "bySeverity": by_severity}
|
|
|
|
|
|
def feed_alerts_summary(
|
|
*,
|
|
date_from: Optional[str] = None,
|
|
date_to: Optional[str] = None,
|
|
dispenser_id: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
data = list_feed_alerts(
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
dispenser_id=dispenser_id,
|
|
limit=500,
|
|
)
|
|
total = data["total"]
|
|
by = data["bySeverity"]
|
|
return {
|
|
"total": total,
|
|
"warning": by.get("warning", 0),
|
|
"error": by.get("error", 0),
|
|
"info": by.get("info", 0),
|
|
}
|