298 lines
10 KiB
Python
298 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from opentelemetry import metrics, trace
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
|
|
from app.core.audit import append_audit
|
|
from app.domain.wwi.models import (
|
|
WWIReorderRecommendation,
|
|
WWISupplierScore,
|
|
WWIWhatIfScenario,
|
|
WWIBusinessEvent,
|
|
)
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
tracer = trace.get_tracer("otel-bi.domain.wwi")
|
|
meter = metrics.get_meter("otel-bi.domain.wwi")
|
|
|
|
_persist_counter = meter.create_counter(
|
|
"wwi_persist_writes_total",
|
|
description="Number of WWI PostgreSQL write operations",
|
|
)
|
|
_event_counter = meter.create_counter(
|
|
"wwi_business_events_generated_total",
|
|
description="Business events automatically generated",
|
|
)
|
|
|
|
|
|
def _current_span_context() -> tuple[str | None, str | None]:
|
|
ctx = trace.get_current_span().get_span_context()
|
|
if not ctx.is_valid:
|
|
return None, None
|
|
return f"{ctx.trace_id:032x}", f"{ctx.span_id:016x}"
|
|
|
|
|
|
def _actor_type(trigger_source: str) -> str:
|
|
return "scheduler" if trigger_source.startswith("scheduler") else "api"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Persist functions — called after Go service returns data
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def persist_reorder_recommendations(
|
|
factory: sessionmaker[Session],
|
|
data: list[dict],
|
|
trigger_source: str,
|
|
) -> None:
|
|
trace_id, span_id = _current_span_context()
|
|
try:
|
|
with factory() as session:
|
|
session.add(WWIReorderRecommendation(
|
|
item_count=len(data),
|
|
trigger_source=trigger_source,
|
|
trace_id=trace_id,
|
|
span_id=span_id,
|
|
payload=data,
|
|
))
|
|
session.commit()
|
|
_persist_counter.add(1, {"entity": "reorder_recommendations"})
|
|
except Exception as exc: # noqa: BLE001
|
|
LOGGER.warning("Failed to persist WWI reorder recommendations: %s", exc)
|
|
append_audit(
|
|
factory,
|
|
action="recommendations.generated",
|
|
actor_type=_actor_type(trigger_source),
|
|
actor_id=trigger_source,
|
|
domain="wwi",
|
|
service="otel-bi-backend",
|
|
entity_type="reorder_recommendations",
|
|
payload={"item_count": len(data)},
|
|
)
|
|
|
|
|
|
def persist_supplier_scores(
|
|
factory: sessionmaker[Session],
|
|
data: list[dict],
|
|
top_n: int,
|
|
trigger_source: str,
|
|
) -> None:
|
|
trace_id, span_id = _current_span_context()
|
|
try:
|
|
with factory() as session:
|
|
session.add(WWISupplierScore(
|
|
supplier_count=len(data),
|
|
top_n=top_n,
|
|
trigger_source=trigger_source,
|
|
trace_id=trace_id,
|
|
span_id=span_id,
|
|
payload=data,
|
|
))
|
|
session.commit()
|
|
_persist_counter.add(1, {"entity": "supplier_scores"})
|
|
except Exception as exc: # noqa: BLE001
|
|
LOGGER.warning("Failed to persist WWI supplier scores: %s", exc)
|
|
append_audit(
|
|
factory,
|
|
action="scores.generated",
|
|
actor_type=_actor_type(trigger_source),
|
|
actor_id=trigger_source,
|
|
domain="wwi",
|
|
service="otel-bi-backend",
|
|
entity_type="supplier_scores",
|
|
payload={"supplier_count": len(data), "top_n": top_n},
|
|
)
|
|
|
|
|
|
def persist_whatif_scenario(
|
|
factory: sessionmaker[Session],
|
|
result: dict,
|
|
) -> None:
|
|
trace_id, span_id = _current_span_context()
|
|
try:
|
|
with factory() as session:
|
|
session.add(WWIWhatIfScenario(
|
|
stock_item_key=result["stock_item_key"],
|
|
stock_item_name=result["stock_item_name"],
|
|
demand_multiplier=result["demand_multiplier"],
|
|
current_stock=result["current_stock"],
|
|
avg_daily_demand=result["adjusted_daily_demand"],
|
|
projected_days_until_stockout=result.get("projected_days_until_stockout"),
|
|
recommended_order_qty=float(result["recommended_order_qty"]),
|
|
trace_id=trace_id,
|
|
span_id=span_id,
|
|
result=result,
|
|
))
|
|
session.commit()
|
|
_persist_counter.add(1, {"entity": "whatif_scenario"})
|
|
except Exception as exc: # noqa: BLE001
|
|
LOGGER.warning("Failed to persist WWI what-if scenario: %s", exc)
|
|
append_audit(
|
|
factory,
|
|
action="scenario.submitted",
|
|
actor_type="user",
|
|
domain="wwi",
|
|
service="otel-bi-backend",
|
|
entity_type="whatif_scenario",
|
|
payload={
|
|
"stock_item_key": result["stock_item_key"],
|
|
"demand_multiplier": result["demand_multiplier"],
|
|
"projected_days_until_stockout": result.get("projected_days_until_stockout"),
|
|
},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Business events — generated from reorder data in Python (PostgreSQL writes)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def generate_stock_events(
|
|
factory: sessionmaker[Session],
|
|
recommendations: list[dict],
|
|
) -> None:
|
|
"""Write LOW_STOCK events for HIGH-urgency items, deduplicating within 24h."""
|
|
trace_id, span_id = _current_span_context()
|
|
cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
|
|
try:
|
|
with factory() as session:
|
|
for item in recommendations:
|
|
if item.get("urgency") != "HIGH":
|
|
continue
|
|
entity_key = str(item["stock_item_key"])
|
|
existing = (
|
|
session.query(WWIBusinessEvent)
|
|
.filter(
|
|
WWIBusinessEvent.event_type == "LOW_STOCK",
|
|
WWIBusinessEvent.entity_key == entity_key,
|
|
WWIBusinessEvent.occurred_at >= cutoff,
|
|
)
|
|
.first()
|
|
)
|
|
if existing:
|
|
continue
|
|
days_str = (
|
|
f"{item['days_until_stockout']:.1f} days"
|
|
if item.get("days_until_stockout") is not None
|
|
else "immediately"
|
|
)
|
|
session.add(WWIBusinessEvent(
|
|
event_type="LOW_STOCK",
|
|
severity="HIGH",
|
|
entity_key=entity_key,
|
|
entity_name=item["stock_item_name"],
|
|
message=(
|
|
f"Stock for '{item['stock_item_name']}' will be exhausted in {days_str}. "
|
|
f"Current stock: {item['current_stock']:.0f} units, "
|
|
f"daily demand: {item['avg_daily_demand']:.1f} units."
|
|
),
|
|
trace_id=trace_id,
|
|
span_id=span_id,
|
|
details={
|
|
"current_stock": item["current_stock"],
|
|
"avg_daily_demand": item["avg_daily_demand"],
|
|
"recommended_reorder_qty": item["recommended_reorder_qty"],
|
|
},
|
|
))
|
|
_event_counter.add(1, {"event_type": "LOW_STOCK"})
|
|
session.commit()
|
|
except Exception as exc: # noqa: BLE001
|
|
LOGGER.warning("Failed to persist WWI business events: %s", exc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Read functions — query PostgreSQL for stored results
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_business_events(factory: sessionmaker[Session], limit: int = 100) -> list[dict]:
|
|
with tracer.start_as_current_span("wwi.analytics.business_events"):
|
|
with factory() as session:
|
|
rows = (
|
|
session.query(WWIBusinessEvent)
|
|
.order_by(WWIBusinessEvent.occurred_at.desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return [
|
|
{
|
|
"id": r.id,
|
|
"occurred_at": r.occurred_at.isoformat(),
|
|
"event_type": r.event_type,
|
|
"severity": r.severity,
|
|
"entity_key": r.entity_key,
|
|
"entity_name": r.entity_name,
|
|
"message": r.message,
|
|
"trace_id": r.trace_id,
|
|
"details": r.details,
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
def list_reorder_recommendations(factory: sessionmaker[Session], limit: int = 50) -> list[dict]:
|
|
with factory() as session:
|
|
rows = (
|
|
session.query(WWIReorderRecommendation)
|
|
.order_by(WWIReorderRecommendation.created_at.desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return [
|
|
{
|
|
"id": r.id,
|
|
"created_at": r.created_at.isoformat(),
|
|
"item_count": r.item_count,
|
|
"trigger_source": r.trigger_source,
|
|
"trace_id": r.trace_id,
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
def list_supplier_scores(factory: sessionmaker[Session], limit: int = 50) -> list[dict]:
|
|
with factory() as session:
|
|
rows = (
|
|
session.query(WWISupplierScore)
|
|
.order_by(WWISupplierScore.computed_at.desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return [
|
|
{
|
|
"id": r.id,
|
|
"computed_at": r.computed_at.isoformat(),
|
|
"supplier_count": r.supplier_count,
|
|
"top_n": r.top_n,
|
|
"trigger_source": r.trigger_source,
|
|
"trace_id": r.trace_id,
|
|
"payload": r.payload,
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
def list_whatif_scenarios(factory: sessionmaker[Session], limit: int = 50) -> list[dict]:
|
|
with factory() as session:
|
|
rows = (
|
|
session.query(WWIWhatIfScenario)
|
|
.order_by(WWIWhatIfScenario.created_at.desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
return [
|
|
{
|
|
"id": r.id,
|
|
"created_at": r.created_at.isoformat(),
|
|
"stock_item_key": r.stock_item_key,
|
|
"stock_item_name": r.stock_item_name,
|
|
"demand_multiplier": r.demand_multiplier,
|
|
"projected_days_until_stockout": r.projected_days_until_stockout,
|
|
"recommended_order_qty": r.recommended_order_qty,
|
|
"result": r.result,
|
|
}
|
|
for r in rows
|
|
]
|