Files
ERP/backend/app/services/dashboard_service.py
T

659 lines
25 KiB
Python

"""Aggregation queries behind the MVP dashboard (warehouse + procurement).
Read-only service. Every function receives the resolved scope (PM → the list
of projects they manage; office roles → project_ids=None = whole
organization) plus an optional explicit project filter, and the optional
«What» filters (category_id / material_id / brand_id / specs_hash).
Quantities are NEVER mixed across units: inventory rows live in the
material's canonical unit, so stock aggregates carry their unit code and the
UI picks one unit at a time. Numeric results leave the service chart-ready
(floats); identifiers and labels are plain strings.
Response shapes mirror frontend/src/lib/types.ts (v2.8 dashboard DTOs)
exactly — this module is the server side of that contract.
"""
import uuid
from datetime import date as date_cls
from datetime import datetime, time, timedelta, timezone
from sqlalchemy import case, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.catalog import Brand, BrandMaterial, Material, MaterialCategory, Unit
from app.models.inventory import InventoryItem, InventoryTransaction, Warehouse
from app.models.procurement import (
ItemDecision,
ItemReceipt,
LineItem,
Project,
Quote,
QuoteProjects,
)
# Iran has been on a fixed UTC+03:30 (no DST) since 2022 — «امروز» in the
# daily widgets means today at this offset.
TEHRAN = timezone(timedelta(hours=3, minutes=30))
# Flow buckets: outbound = site consumption + shrinkage; waste is reported
# separately so the UI can stack it as a distinct red segment inside the
# outbound bar. Transfers are excluded from the buckets (they are internal
# moves) but DO count toward the running balance.
OUTBOUND_TYPES = ("WITHDRAWAL", "DAMAGE", "WASTE", "LOSS")
WASTE_TYPES = ("DAMAGE", "WASTE", "LOSS")
def _stock_filters(stmt, *, category_id=None, material_id=None, brand_id=None, specs_hash=None):
"""«What» filters for a statement that already joins inventory→material."""
if category_id is not None:
stmt = stmt.where(Material.category_id == category_id)
if material_id is not None:
stmt = stmt.where(InventoryItem.material_id == material_id)
if brand_id is not None:
stmt = stmt.where(InventoryItem.brand_id == brand_id)
if specs_hash:
stmt = stmt.where(InventoryItem.specs_hash == specs_hash)
return stmt
def _scope_filters(stmt, *, project_ids, project_id):
"""warehouse.project_id restriction for the resolved dashboard scope."""
if project_id is not None:
stmt = stmt.where(Warehouse.project_id == project_id)
elif project_ids is not None:
stmt = stmt.where(Warehouse.project_id.in_(project_ids))
return stmt
def _qty(value) -> float:
return round(float(value), 4) if value is not None else 0.0
# ---------------------------------------------------------------------------
# «What» filter cascade — Category → Material → Brand → Specs
# ---------------------------------------------------------------------------
async def filter_options(db: AsyncSession) -> dict:
"""Cascading options for the global filter bar.
Categories/materials come from the catalog (active only). Brand options
per material are the UNION of the catalog links (brand_materials — every
brand defined for that material) and the distinct brands actually present
in stock, so the cascade is complete even for materials whose current
stock rows carry no brand yet. Spec options come from stock rows
(inventory.specs_hash), since specs are a stock-level dimension.
Materials without a category land in the synthetic «بدون دسته» bucket.
"""
categories = (
await db.execute(
select(MaterialCategory)
.where(MaterialCategory.is_active.is_(True))
.order_by(MaterialCategory.name)
)
).scalars().all()
materials = (
await db.execute(select(Material).where(Material.status == "active").order_by(Material.name))
).scalars().all()
units = {u.id: u for u in (await db.execute(select(Unit))).scalars()}
inv_rows = (
await db.execute(
select(
InventoryItem.material_id,
InventoryItem.brand_id,
InventoryItem.specs_hash,
InventoryItem.specs,
Brand.name,
)
.select_from(InventoryItem)
.outerjoin(Brand, Brand.id == InventoryItem.brand_id)
.distinct()
)
).all()
# برندهای کاتالوگی هر متریال (brand_materials) — مستقل از موجودی فعلی
catalog_brand_rows = (
await db.execute(
select(BrandMaterial.material_id, Brand.id, Brand.name)
.select_from(BrandMaterial)
.join(Brand, Brand.id == BrandMaterial.brand_id)
.where(BrandMaterial.is_active.is_(True))
)
).all()
brands_by_material: dict[uuid.UUID, dict] = {}
specs_by_material: dict[uuid.UUID, dict[str, dict]] = {}
for material_id, brand_id, brand_name in catalog_brand_rows:
brands_by_material.setdefault(material_id, {})[brand_id] = brand_name
for material_id, brand_id, specs_hash, specs, brand_name in inv_rows:
if brand_id is not None:
brands_by_material.setdefault(material_id, {}).setdefault(brand_id, brand_name)
if specs_hash:
specs_by_material.setdefault(material_id, {})[specs_hash] = specs or {}
def material_entry(m: Material) -> dict:
unit = units.get(m.canonical_unit_id)
brands = brands_by_material.get(m.id, {})
spec_map = specs_by_material.get(m.id, {})
return {
"id": str(m.id),
"name": m.name,
"unit": {"id": unit.id, "code": unit.code, "name_fa": unit.name_fa} if unit else None,
"brands": [
{"id": str(bid), "name": bname}
for bid, bname in sorted(brands.items(), key=lambda kv: kv[1] or "")
],
"specs": [
{"hash": h, "specs": sp}
for h, sp in sorted(spec_map.items())
],
}
buckets: dict[uuid.UUID | None, list] = {c.id: [] for c in categories}
buckets.setdefault(None, [])
for m in materials:
buckets.setdefault(m.category_id, []).append(material_entry(m))
out = [
{"id": str(c.id), "name": c.name, "slug": c.slug, "materials": buckets.get(c.id, [])}
for c in categories
]
if buckets.get(None):
out.append({"id": None, "name": "بدون دسته", "slug": None, "materials": buckets[None]})
return {"data": {"categories": out}}
# ---------------------------------------------------------------------------
# Widget 1 — Drill-down matrix (organization → project → warehouse)
# ---------------------------------------------------------------------------
async def inventory_drilldown(
db: AsyncSession,
*,
project_ids,
project_id,
category_id=None,
material_id=None,
brand_id=None,
specs_hash=None,
) -> dict:
"""Stock quantity as a three-level tree; each aggregate keyed by unit code
so incompatible units never merge into one number."""
stmt = (
select(
Project.id,
Project.name,
Warehouse.id,
Warehouse.name,
Unit.code,
Unit.name_fa,
func.sum(InventoryItem.quantity).label("qty"),
func.count().label("rows"),
)
.select_from(InventoryItem)
.join(Warehouse, Warehouse.id == InventoryItem.warehouse_id)
.join(Project, Project.id == Warehouse.project_id)
.join(Material, Material.id == InventoryItem.material_id)
.join(Unit, Unit.id == InventoryItem.unit_id)
.where(InventoryItem.quantity > 0)
.group_by(Project.id, Project.name, Warehouse.id, Warehouse.name, Unit.code, Unit.name_fa)
.order_by(Project.name, Warehouse.name)
)
stmt = _stock_filters(
stmt, category_id=category_id, material_id=material_id, brand_id=brand_id, specs_hash=specs_hash
)
stmt = _scope_filters(stmt, project_ids=project_ids, project_id=project_id)
rows = (await db.execute(stmt)).all()
units_legend: dict[str, str] = {}
total_by_unit: dict[str, float] = {}
total_rows = 0
projects_tree: dict[uuid.UUID, dict] = {}
for pid, pname, wid, wname, ucode, uname_fa, qty, row_count in rows:
q = _qty(qty)
units_legend[ucode] = uname_fa
total_by_unit[ucode] = round(total_by_unit.get(ucode, 0.0) + q, 4)
total_rows += row_count
proj = projects_tree.setdefault(
pid,
{"id": str(pid), "name": pname, "qty_by_unit": {}, "rows": 0, "_warehouses": {}},
)
proj["qty_by_unit"][ucode] = round(proj["qty_by_unit"].get(ucode, 0.0) + q, 4)
proj["rows"] += row_count
wh = proj["_warehouses"].setdefault(wid, {"id": str(wid), "name": wname, "qty_by_unit": {}, "rows": 0})
wh["qty_by_unit"][ucode] = round(wh["qty_by_unit"].get(ucode, 0.0) + q, 4)
wh["rows"] += row_count
projects = []
for proj in projects_tree.values():
warehouses = sorted(proj.pop("_warehouses").values(), key=lambda w: w["name"])
projects.append({**proj, "warehouses": warehouses})
projects.sort(key=lambda p: p["name"])
return {
"data": {
"units": units_legend,
"total": {"qty_by_unit": total_by_unit, "rows": total_rows},
"projects": projects,
}
}
# ---------------------------------------------------------------------------
# Widget 2 — Dead stock (dormant capital)
# ---------------------------------------------------------------------------
async def dead_stock(
db: AsyncSession,
*,
project_ids,
project_id,
idle_days: int = 60,
category_id=None,
material_id=None,
brand_id=None,
specs_hash=None,
limit: int = 200,
) -> dict:
"""Stock rows with quantity > 0 and NO ledger activity within `idle_days`.
Last activity = MAX(inventory_transactions.transaction_date) per row —
served by idx_inventory_tx_inventory_date — falling back to
inventory.updated_at for rows that never posted a transaction.
"""
last_tx = (
select(func.max(InventoryTransaction.transaction_date))
.where(InventoryTransaction.inventory_id == InventoryItem.id)
.correlate(InventoryItem)
.scalar_subquery()
)
last_activity = func.coalesce(last_tx, InventoryItem.updated_at)
cutoff = datetime.now(TEHRAN) - timedelta(days=idle_days)
stmt = (
select(
InventoryItem.id,
InventoryItem.quantity,
InventoryItem.specs,
Material.name,
Brand.name,
Warehouse.name,
Project.name,
Unit.code,
Unit.name_fa,
last_activity.label("last_activity"),
)
.select_from(InventoryItem)
.join(Warehouse, Warehouse.id == InventoryItem.warehouse_id)
.join(Project, Project.id == Warehouse.project_id)
.join(Material, Material.id == InventoryItem.material_id)
.join(Unit, Unit.id == InventoryItem.unit_id)
.outerjoin(Brand, Brand.id == InventoryItem.brand_id)
.where(InventoryItem.quantity > 0, last_activity <= cutoff)
.order_by(last_activity.asc())
.limit(limit)
)
stmt = _stock_filters(
stmt, category_id=category_id, material_id=material_id, brand_id=brand_id, specs_hash=specs_hash
)
stmt = _scope_filters(stmt, project_ids=project_ids, project_id=project_id)
rows = (await db.execute(stmt)).all()
now = datetime.now(TEHRAN)
items = []
for inv_id, quantity, specs, mat_name, brand_name, wh_name, proj_name, ucode, uname_fa, last in rows:
items.append({
"inventory_id": str(inv_id),
"material": mat_name,
"brand": brand_name,
"specs": specs or {},
"warehouse": wh_name,
"project": proj_name,
"quantity": _qty(quantity),
"unit": {"code": ucode, "name_fa": uname_fa},
"last_activity": last.isoformat() if last else None,
"idle_days": (now - last).days if last else None,
})
return {"data": {"idle_days_threshold": idle_days, "items": items}}
# ---------------------------------------------------------------------------
# Widget 3 — Daily time-series flow (inbound / outbound / returns / balance)
# ---------------------------------------------------------------------------
async def daily_flow(
db: AsyncSession,
*,
project_ids,
project_id,
date_from: datetime,
date_to: datetime,
warehouse_id=None,
category_id=None,
material_id=None,
brand_id=None,
specs_hash=None,
) -> dict:
"""Per-day buckets from the signed ledger.
inbound = PURCHASE · returns = RETURN · outbound = WITHDRAWAL + waste
(DAMAGE/WASTE/LOSS, reported separately for the stacked red segment).
balance = running sum of EVERY signed movement (transfers/found/stock
counts included), seeded with the opening balance before `date_from`.
"""
def _with_joins(sel):
return (
sel.select_from(InventoryTransaction)
.join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
.join(InventoryItem, InventoryItem.id == InventoryTransaction.inventory_id)
.join(Material, Material.id == InventoryItem.material_id)
)
scope_conditions = []
if project_id is not None:
scope_conditions.append(Warehouse.project_id == project_id)
elif project_ids is not None:
scope_conditions.append(Warehouse.project_id.in_(project_ids))
if warehouse_id is not None:
scope_conditions.append(InventoryTransaction.warehouse_id == warehouse_id)
what_conditions = [Material.category_id == category_id] if category_id is not None else []
if material_id is not None:
what_conditions.append(InventoryItem.material_id == material_id)
if brand_id is not None:
what_conditions.append(InventoryItem.brand_id == brand_id)
if specs_hash:
what_conditions.append(InventoryItem.specs_hash == specs_hash)
day_expr = func.date_trunc("day", InventoryTransaction.transaction_date).label("day")
bucket_rows = (
await db.execute(
_with_joins(
select(day_expr, InventoryTransaction.transaction_type, func.sum(InventoryTransaction.quantity))
.where(
InventoryTransaction.transaction_date >= date_from,
InventoryTransaction.transaction_date < date_to,
*scope_conditions,
*what_conditions,
)
.group_by(day_expr, InventoryTransaction.transaction_type)
)
)
).all()
opening = _qty(
await db.scalar(
_with_joins(
select(func.coalesce(func.sum(InventoryTransaction.quantity), 0.0)).where(
InventoryTransaction.transaction_date < date_from,
*scope_conditions,
*what_conditions,
)
)
)
)
signed_by_day: dict[date_cls, dict[str, float]] = {}
for day, tx_type, signed in bucket_rows:
day_key = day.date() if isinstance(day, datetime) else day
signed_by_day.setdefault(day_key, {})[str(tx_type)] = _qty(signed)
days = []
running = opening
current = date_from.date()
while current < date_to.date():
types = signed_by_day.get(current, {})
waste = round(-sum(types.get(t, 0.0) for t in WASTE_TYPES), 4) or 0.0
outbound = round(waste - types.get("WITHDRAWAL", 0.0), 4) or 0.0
running = round(running + sum(types.values()), 4)
days.append({
"date": current.isoformat(),
"inbound": round(types.get("PURCHASE", 0.0), 4) or 0.0,
"outbound": outbound,
"waste": waste,
"returns": round(types.get("RETURN", 0.0), 4) or 0.0,
"balance": running,
})
current += timedelta(days=1)
return {"data": {"opening_balance": opening, "days": days}}
# ---------------------------------------------------------------------------
# Widget 4 — Daily control (today's site operations)
# ---------------------------------------------------------------------------
async def daily_control(
db: AsyncSession,
*,
project_ids,
project_id,
day: date_cls,
warehouse_id=None,
) -> dict:
"""One day (Tehran calendar): outbound / return / waste / purchase ticket
counts + quantities, the exact performer names, and the operation detail."""
day_start = datetime.combine(day, time.min, tzinfo=TEHRAN)
day_end = day_start + timedelta(days=1)
scope_conditions = []
if project_id is not None:
scope_conditions.append(Warehouse.project_id == project_id)
elif project_ids is not None:
scope_conditions.append(Warehouse.project_id.in_(project_ids))
if warehouse_id is not None:
scope_conditions.append(InventoryTransaction.warehouse_id == warehouse_id)
count_rows = (
await db.execute(
select(
InventoryTransaction.transaction_type,
func.count().label("tickets"),
func.sum(func.abs(InventoryTransaction.quantity)).label("qty"),
)
.select_from(InventoryTransaction)
.join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
.where(
InventoryTransaction.transaction_date >= day_start,
InventoryTransaction.transaction_date < day_end,
*scope_conditions,
)
.group_by(InventoryTransaction.transaction_type)
)
).all()
by_type = {str(tx_type): {"tickets": tickets, "qty": _qty(qty)} for tx_type, tickets, qty in count_rows}
performers_rows = (
await db.execute(
select(
InventoryTransaction.performed_by,
func.count().label("tickets"),
func.sum(func.abs(InventoryTransaction.quantity)).label("qty"),
)
.select_from(InventoryTransaction)
.join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
.where(
InventoryTransaction.transaction_date >= day_start,
InventoryTransaction.transaction_date < day_end,
*scope_conditions,
)
.group_by(InventoryTransaction.performed_by)
.order_by(func.count().desc())
)
).all()
detail_rows = (
await db.execute(
select(
InventoryTransaction,
Warehouse.name,
Material.name,
InventoryItem.specs,
Unit.code,
)
.select_from(InventoryTransaction)
.join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
.join(InventoryItem, InventoryItem.id == InventoryTransaction.inventory_id)
.join(Material, Material.id == InventoryItem.material_id)
.join(Unit, Unit.id == InventoryItem.unit_id)
.where(
InventoryTransaction.transaction_date >= day_start,
InventoryTransaction.transaction_date < day_end,
InventoryTransaction.transaction_type.in_(
("WITHDRAWAL", "RETURN", "DAMAGE", "WASTE", "LOSS", "PURCHASE")
),
*scope_conditions,
)
.order_by(InventoryTransaction.transaction_date.desc())
.limit(200)
)
).all()
zero = {"tickets": 0, "qty": 0.0}
items = []
for tx, wh_name, mat_name, specs, ucode in detail_rows:
items.append({
"id": str(tx.id),
"transaction_type": str(tx.transaction_type),
"quantity": _qty(tx.quantity),
"unit_code": ucode,
"performed_by": tx.performed_by,
"note": tx.note,
"warehouse": wh_name,
"material": mat_name,
"specs": specs or {},
"transaction_date": tx.transaction_date.isoformat(),
})
return {
"data": {
"date": day.isoformat(),
"outbound": by_type.get("WITHDRAWAL", zero),
"returns": by_type.get("RETURN", zero),
"waste": {
"tickets": sum(by_type.get(t, zero)["tickets"] for t in WASTE_TYPES),
"qty": round(sum(by_type.get(t, zero)["qty"] for t in WASTE_TYPES), 4),
},
"purchases": by_type.get("PURCHASE", zero),
"performers": [
{"name": name or "نامشخص", "tickets": tickets, "qty": _qty(qty)}
for name, tickets, qty in performers_rows
],
"items": items,
}
}
# ---------------------------------------------------------------------------
# Widget 8 — Procurement workload per project (project density)
# ---------------------------------------------------------------------------
async def procurement_workload(db: AsyncSession, *, project_ids, project_id) -> dict:
"""Per project: offered quotes still open + chosen purchase lines (open
vs fully received). Chosen lines attribute to item_decisions.project_id
(the purchased destination); offered quotes to any linked project."""
received_sq = (
select(ItemReceipt.item_decision_id.label("did"), func.sum(ItemReceipt.received_qty).label("received_qty"))
.group_by(ItemReceipt.item_decision_id)
.subquery()
)
order_conditions = [ItemDecision.status == "chosen"]
if project_id is not None:
order_conditions.append(ItemDecision.project_id == project_id)
elif project_ids is not None:
order_conditions.append(ItemDecision.project_id.in_(project_ids))
order_rows = (
await db.execute(
select(
Project.id,
Project.name,
func.count().label("chosen_items"),
func.sum(LineItem.quantity).label("ordered_qty"),
func.coalesce(func.sum(received_sq.c.received_qty), 0).label("received_qty"),
func.sum(
case(
(func.coalesce(received_sq.c.received_qty, 0) < LineItem.quantity, 1),
else_=0,
)
).label("open_items"),
)
.select_from(ItemDecision)
.join(LineItem, LineItem.id == ItemDecision.line_item_id)
.join(Quote, Quote.id == LineItem.quote_id)
.outerjoin(Project, Project.id == ItemDecision.project_id)
.outerjoin(received_sq, received_sq.c.did == ItemDecision.id)
.where(*order_conditions)
.group_by(Project.id, Project.name)
)
).all()
offered_by_project: dict[uuid.UUID | None, set] = {}
offered_primary = (
await db.execute(
select(Quote.id, Quote.project_id).where(
Quote.status == "offered", Quote.project_id.is_not(None)
)
)
).all()
for quote_id, pid in offered_primary:
offered_by_project.setdefault(pid, set()).add(quote_id)
offered_junction = (
await db.execute(
select(QuoteProjects.quote_id, QuoteProjects.project_id)
.join(Quote, Quote.id == QuoteProjects.quote_id)
.where(Quote.status == "offered")
)
).all()
for quote_id, pid in offered_junction:
offered_by_project.setdefault(pid, set()).add(quote_id)
project_ids_needed = {row[0] for row in order_rows if row[0] is not None}
project_ids_needed.update(pid for pid in offered_by_project if pid is not None)
names: dict[uuid.UUID, str] = {}
if project_ids_needed:
names = {
p.id: p.name
for p in (
await db.execute(select(Project).where(Project.id.in_(project_ids_needed)))
).scalars()
}
projects = []
for pid, _pname, chosen, ordered, received, open_items in order_rows:
if pid is None:
offered = 0
else:
offered = len(offered_by_project.pop(pid, set()))
if project_id is not None and pid != project_id:
offered = 0
elif project_ids is not None and pid not in project_ids:
offered = 0
projects.append({
"id": str(pid) if pid else None,
"name": names.get(pid, "بدون پروژه") if pid else "بدون پروژه",
"offered_quotes": offered,
"chosen_items": chosen,
"open_orders": open_items,
"ordered_qty": _qty(ordered),
"received_qty": _qty(received),
})
# offered-only projects that have no chosen lines yet
for pid, quote_ids in offered_by_project.items():
if not quote_ids:
continue
if project_id is not None and pid != project_id:
continue
if project_ids is not None and (pid is None or pid not in project_ids):
continue
projects.append({
"id": str(pid) if pid else None,
"name": names.get(pid, "بدون پروژه") if pid else "بدون پروژه",
"offered_quotes": len(quote_ids),
"chosen_items": 0,
"open_orders": 0,
"ordered_qty": 0.0,
"received_qty": 0.0,
})
projects.sort(key=lambda p: (-p["open_orders"], -p["offered_quotes"], p["name"]))
return {"data": {"projects": projects}}