Files
ERP/backend/app/routers/pm.py
T
MMD e17d1f0883 feat(monitoring): observability stack + app instrumentation
- /health/ready (DB+MinIO+Casdoor checks, webhook post-deploy gate),
  /metrics via prometheus-fastapi-instrumentator, Sentry/GlitchTip capture,
  X-Request-ID middleware + JSON access logging (PROCUREMENT_LOG_FORMAT)
- server-side analytics (OpenPanel Track API): login, quote lifecycle,
  receipts — fire-and-forget, disabled without env config
- monitoring/ compose project: GlitchTip(+pg), Prometheus(+alerts),
  Grafana(provisioned ERP dashboard), Loki+Promtail, node/postgres
  exporters, cAdvisor — mem-limited (~2.2GB), loopback admin UIs,
  separate from app deploys; erp_shared network for scraping
- frontend: @openpanel/web + @sentry/react (build-arg gated, no-op when
  unset), funnel events (login, wizard steps, quote submitted)
- compose: api healthcheck, observability env wiring, .env.example keys
2026-09-18 22:18:02 +03:30

504 lines
22 KiB
Python

"""PM arrival flow (v2.6) — on-the-way list + confirm arrival + warehouse split.
Flow: proc manager purchases (item_decisions.status='chosen' + project_id)
→ item appears on the arrivals dash as on-the-way → the PM (or the admin /
procurement manager, who is the same person) confirms each delivery
(POST /pm/receipts) and splits it across warehouses (free count, typically
3) → inventory posts PURCHASE transactions (idempotent via
receipt_allocation_id UNIQUE) → the procurement dashboard flips to arrived
once every chosen item is fully received.
Rules enforced here:
- Callers: role 'pm' (scoped to their single project via projects.pm_user_id)
or 'admin' / 'procurement_manager' (all projects — full visibility).
- Partial deliveries allowed (1 item : N receipts); three derived states:
on-the-way (sum=0) / partially received / arrived (sum >= ordered).
- SUM(allocations) == received_qty; every warehouse belongs to the
decision's project; quantity converts to the inventory row's unit via the
material's canonical unit (material_units) — inventory rows are stored in
the canonical unit for auto-posted purchases.
- quote.status flips to 'arrived' when ALL chosen items are fully received.
Auth: Casdoor JWT (a global dependency in main.py) resolves the caller;
optional dev-header fallback (X-User-Role: pm|admin + X-Project-Id for pm)
exists only when PROCUREMENT_DEV_LOGIN_ENABLED=true — same pattern as the
inventory router.
"""
import base64
import uuid
from contextlib import nullcontext
from dataclasses import dataclass
from decimal import Decimal
from fastapi import APIRouter, Depends, Query, Request
from sqlalchemy import func, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth import UserContext, get_optional_user
from app.config import settings
from app.db.engine import get_db
from app.models.catalog import BrandMaterial, Material, MaterialCategory, Unit
from app.models.inventory import InventoryItem, InventoryTransaction, Warehouse
from app.models.procurement import (
ItemDecision,
ItemReceipt,
ItemReceiptAllocation,
LineItem,
Project,
Quote,
Supplier,
)
from app.schemas.procurement import PMReceiptCreate
from app.services import analytics
from app.services.inventory_service import canonical_specs_hash
from app.utils.errors import (
ForbiddenError,
NotFoundError,
UnauthorizedError,
ValidationError,
)
from app.utils.serialization import to_iso
router = APIRouter(prefix="/pm", tags=["pm"])
NULL_BRAND_SENTINEL = "00000000-0000-0000-0000-000000000000"
@dataclass
class ArrivalScope:
role: str # "pm" | "admin" (admin = admin or procurement_manager)
user_id: uuid.UUID | None
project_id: uuid.UUID | None # PM's active project; admin: optional filter
display_name: str
async def arrivals_scope(
request: Request,
project_id: uuid.UUID | None = Query(default=None, description="پروژه فعال (PM: الزامی برای داده‌های پروژه)"),
db: AsyncSession = Depends(get_db),
) -> ArrivalScope:
"""Resolve the arrivals caller — Casdoor JWT first, dev headers as fallback.
PM → scoped to the requested project, which must be one they manage
(projects.pm_user_id, 1:N); without the param, their first project.
Admin / procurement manager → all projects (param = optional filter).
"""
if request.headers.get("authorization"):
ctx: UserContext | None = await get_optional_user(request, db)
if ctx is not None:
if ctx.role is None:
raise ForbiddenError("نقش شما هنوز در سیستم تعیین نشده است")
if ctx.role == "pm":
allowed = list(ctx.project_ids or [])
if project_id is not None:
if project_id not in allowed:
raise ForbiddenError("این پروژه به شما تخصیص داده نشده است")
return ArrivalScope(role="pm", user_id=ctx.user.id, project_id=project_id, display_name=ctx.display_name)
if not allowed:
raise ForbiddenError("شما به هیچ پروژه‌ای متصل نیستید (projects.pm_user_id)")
return ArrivalScope(role="pm", user_id=ctx.user.id, project_id=allowed[0], display_name=ctx.display_name)
if ctx.role in ("admin", "procurement_manager"):
return ArrivalScope(role="admin", user_id=ctx.user.id, project_id=project_id, display_name=ctx.display_name)
raise ForbiddenError("این بخش فقط برای مدیر پروژه یا مدیر تدارکات است")
# Dev fallback: X-User-Role: pm | admin (procurement_manager maps to
# admin) — ONLY when DEV_LOGIN_ENABLED (local testing without Casdoor).
# Disabled by default: an ungated header identity in production would
# let anyone confirm receipts as PM.
if not settings.DEV_LOGIN_ENABLED or not request.headers.get("x-user-role"):
raise UnauthorizedError("احراز هویت لازم است — توکن Casdoor ارسال کنید")
role = (request.headers.get("x-user-role") or "").strip().lower()
if role == "procurement_manager":
role = "admin"
if role not in ("pm", "admin"):
raise ForbiddenError("این بخش فقط برای مدیر پروژه یا مدیر تدارکات است — هدر X-User-Role را بفرستید")
# The explicit query param wins; otherwise the dev header carries it.
if project_id is None and role == "pm":
raw_project = request.headers.get("x-project-id")
if not raw_project:
raise ForbiddenError("هدر X-Project-Id (پروژه مسئول) الزامی است")
try:
project_id = uuid.UUID(str(raw_project))
except ValueError:
raise ValidationError("شناسه پروژه نامعتبر است", field="X-Project-Id")
actor = request.headers.get("x-user-name")
b64 = request.headers.get("x-user-name-b64")
if b64:
try:
actor = base64.b64decode(b64.encode()).decode("utf-8")
except Exception:
pass
# Dev fallback needs a real users row (item_receipts.confirmed_by FK).
from app.models.procurement import User
dev = await db.scalar(select(User).where(User.casdoor_id == "dev"))
if dev is None:
# Insert-or-ignore: concurrent dev-mode requests must not race on the
# unique casdoor_id ("dev") and 500 with a unique violation.
await db.execute(
pg_insert(User)
.values(casdoor_id="dev", first_name="کاربر", last_name="آزمایشی", role="pm")
.on_conflict_do_nothing(index_elements=["casdoor_id"])
)
dev = await db.scalar(select(User).where(User.casdoor_id == "dev"))
return ArrivalScope(role=role, user_id=dev.id, project_id=project_id, display_name=actor or "کاربر آزمایشی")
# ---------------------------------------------------------------------------
# Derived arrival state helpers
# ---------------------------------------------------------------------------
async def _receipted_quantities(db: AsyncSession, decision_ids: list) -> dict:
"""decision_id -> sum of confirmed received_qty (in line-item unit)."""
if not decision_ids:
return {}
rows = (
await db.execute(
select(ItemReceipt.item_decision_id, func.sum(ItemReceipt.received_qty))
.where(ItemReceipt.item_decision_id.in_(decision_ids))
.group_by(ItemReceipt.item_decision_id)
)
).all()
return {r[0]: r[1] for r in rows}
def arrival_state(ordered: Decimal, received: Decimal | None) -> str:
received = received or Decimal("0")
if received <= 0:
return "on_the_way"
if received < ordered:
return "partially_received"
return "arrived"
# ---------------------------------------------------------------------------
# GET /pm/on-the-way — chosen items not fully received
# (PM: own project — Admin: all projects)
# ---------------------------------------------------------------------------
@router.get("/on-the-way")
async def pm_on_the_way(
ctx: ArrivalScope = Depends(arrivals_scope),
db: AsyncSession = Depends(get_db),
):
stmt = (
select(ItemDecision, LineItem, Quote, Material, Supplier, Project)
.join(LineItem, LineItem.id == ItemDecision.line_item_id)
.join(Quote, Quote.id == LineItem.quote_id)
.join(Material, Material.id == LineItem.material_id)
.join(Supplier, Supplier.id == Quote.supplier_id)
.outerjoin(Project, Project.id == ItemDecision.project_id)
.where(ItemDecision.status == "chosen")
)
if ctx.role == "pm" and ctx.project_id is None:
raise ForbiddenError("شما به هیچ پروژه‌ای متصل نیستید (projects.pm_user_id)")
if ctx.project_id is not None:
# PM: active project; admin/manager: optional project filter
stmt = stmt.where(ItemDecision.project_id == ctx.project_id)
decisions = (await db.execute(stmt.order_by(ItemDecision.decided_at.desc().nulls_last()))).all()
received_map = await _receipted_quantities(db, [d.ItemDecision.id for d in decisions])
# unit / category lookups for display
unit_ids = {d.LineItem.unit_id for d in decisions}
units = {u.id: u for u in (await db.execute(select(Unit).where(Unit.id.in_(unit_ids)))).scalars()}
mat_ids = {d.Material.id for d in decisions}
cats = {}
if mat_ids:
rows = (
await db.execute(
select(Material.id, MaterialCategory)
.join(MaterialCategory, MaterialCategory.id == Material.category_id)
.where(Material.id.in_(mat_ids))
)
).all()
cats = {m_id: c for m_id, c in rows}
out = []
for d, li, q, m, sup, proj in decisions:
received = received_map.get(d.id)
out.append({
"item_decision_id": str(d.id),
"line_item_id": str(li.id),
"quote_id": str(q.id),
"quote_status": q.status,
"quote_date": to_iso(q.quote_date),
"supplier": {"id": str(sup.id), "name": sup.name},
"project": {"id": str(proj.id), "name": proj.name} if proj else None,
"decided_at": to_iso(d.decided_at),
"material": {"id": str(m.id), "name": m.name,
"category": {"id": str(c.id), "name": c.name} if (c := cats.get(m.id)) else None},
"unit": {"id": units[li.unit_id].id, "code": units[li.unit_id].code,
"name_fa": units[li.unit_id].name_fa} if li.unit_id in units else None,
"ordered_qty": str(li.quantity),
"received_qty": str(received) if received is not None else "0",
"state": arrival_state(li.quantity, received),
})
return {"data": out}
# ---------------------------------------------------------------------------
# POST /pm/receipts — confirm arrival + split into warehouses + post PURCHASE
# (PM: own project only — Admin: any project)
# ---------------------------------------------------------------------------
@router.post("/receipts", status_code=201)
async def pm_confirm_receipt(
payload: PMReceiptCreate,
ctx: ArrivalScope = Depends(arrivals_scope),
db: AsyncSession = Depends(get_db),
):
# No explicit db.begin(): the session is already inside an implicit
# transaction (arrivals_scope read from it above). get_db commits on
# success and rolls back on failure, so atomicity is preserved — opening
# another transaction here would raise "A transaction is already begun".
with nullcontext():
# --- decision must be chosen ---
decision = await db.get(ItemDecision, payload.item_decision_id)
if decision is None or decision.status != "chosen":
raise NotFoundError("تصمیم خرید یافت نشد", field="item_decision_id")
if ctx.project_id is not None and decision.project_id != ctx.project_id:
raise ForbiddenError("این قلم متعلق به پروژه فعال نیست")
li = await db.get(LineItem, decision.line_item_id)
quote = await db.get(Quote, li.quote_id)
material = await db.get(Material, li.material_id)
# --- limit: total received (previous receipts + this one) must never
# exceed the ordered quantity ---
already_received = (await _receipted_quantities(db, [decision.id])).get(decision.id) or Decimal("0")
if already_received + payload.received_qty > li.quantity:
raise ValidationError(
"تعداد کالای بیشتری نسبت به کالای رسیده انتخاب کردید",
field="received_qty",
)
# --- allocations sum + warehouse ownership ---
total_alloc = sum((a.quantity for a in payload.allocations), Decimal("0"))
if total_alloc != payload.received_qty:
raise ValidationError(
f"مجموع تخصیص انبارها ({total_alloc}) باید با مقدار رسید ({payload.received_qty}) برابر باشد",
field="allocations",
)
wh_ids = {a.warehouse_id for a in payload.allocations}
warehouses = {w.id: w for w in (await db.execute(select(Warehouse).where(Warehouse.id.in_(wh_ids)))).scalars()}
for a in payload.allocations:
wh = warehouses.get(a.warehouse_id)
if wh is None:
raise NotFoundError("انبار یافت نشد", field="allocations.warehouse_id")
if wh.project_id != decision.project_id:
raise ForbiddenError("انبار انتخاب‌شده متعلق به پروژه این قلم نیست", field="allocations.warehouse_id")
# --- create the receipt (1:N partial deliveries allowed) ---
receipt = ItemReceipt(
item_decision_id=decision.id,
confirmed_by=ctx.user_id,
received_at=payload.received_at,
received_qty=payload.received_qty,
on_time=payload.on_time,
quality_note=payload.quality_note,
)
db.add(receipt)
await db.flush()
# --- unit conversion: line-item unit -> material canonical unit ---
canonical_unit_id = material.canonical_unit_id
factor = Decimal("1")
if li.unit_id != canonical_unit_id:
from app.models.catalog import MaterialUnit
mu = await db.scalar(
select(MaterialUnit).where(
MaterialUnit.material_id == material.id,
MaterialUnit.unit_id == li.unit_id,
MaterialUnit.brand_material_id.is_(None),
)
)
if mu is None:
raise ValidationError(
"واحد قلم به واحد مرجع متریال قابل تبدیل نیست — ابتدا material_units تعریف کنید",
field="allocations.quantity",
)
factor = mu.conversion_factor
# --- brand + specs for the inventory row ---
brand_id = None
if li.brand_material_id:
bm = await db.get(BrandMaterial, li.brand_material_id)
brand_id = bm.brand_id if bm else None
specs = {}
from app.models.catalog import MaterialAttributeValue
mavs = (
await db.execute(
select(MaterialAttributeValue).where(MaterialAttributeValue.material_id == material.id)
)
).scalars().all()
for mav in mavs:
from app.models.catalog import AttributeDefinition
ad = await db.get(AttributeDefinition, mav.attribute_definition_id)
if ad:
key = ad.attribute_key
if mav.value_string is not None:
specs[key] = mav.value_string
elif mav.value_number is not None:
specs[key] = str(mav.value_number)
elif mav.value_boolean is not None:
specs[key] = mav.value_boolean
elif mav.value_date is not None:
specs[key] = mav.value_date.isoformat()
specs_hash = canonical_specs_hash(specs)
# --- post one PURCHASE transaction per allocation (idempotent) ---
posted = []
for a in payload.allocations:
qty_canonical = a.quantity * factor
inv = await db.scalar(
select(InventoryItem).where(
InventoryItem.warehouse_id == a.warehouse_id,
InventoryItem.material_id == material.id,
InventoryItem.brand_id.is_(None) if brand_id is None else InventoryItem.brand_id == brand_id,
InventoryItem.specs_hash == specs_hash,
).with_for_update()
)
if inv is None:
inv = InventoryItem(
warehouse_id=a.warehouse_id,
material_id=material.id,
brand_id=brand_id,
quantity=Decimal("0"),
unit_id=canonical_unit_id,
specs=specs,
specs_hash=specs_hash,
)
db.add(inv)
await db.flush()
else:
if inv.unit_id != canonical_unit_id:
raise ValidationError(
f"واحد موجودی این ردیف ({inv.unit_id}) با واحد مرجع متریال متفاوت است",
field="allocations.quantity",
)
inv.quantity += qty_canonical
tx = InventoryTransaction(
inventory_id=inv.id,
warehouse_id=a.warehouse_id,
receipt_allocation_id=None, # set after allocation flush below
transaction_type="PURCHASE",
quantity=qty_canonical,
performed_by=ctx.display_name,
note=f"رسید {receipt.id}",
)
db.add(tx)
await db.flush()
posted.append(tx)
# --- link the transactions to the allocations (idempotency key) ---
for a, tx in zip(payload.allocations, posted):
alloc = ItemReceiptAllocation(
receipt_id=receipt.id,
warehouse_id=a.warehouse_id,
quantity=a.quantity,
)
db.add(alloc)
await db.flush()
tx.receipt_allocation_id = alloc.id
# --- auto-flip the quote to arrived when all chosen items are in ---
chosen_ids = (
(
await db.execute(
select(ItemDecision.id).where(
ItemDecision.line_item_id.in_(
select(LineItem.id).where(LineItem.quote_id == quote.id)
),
ItemDecision.status == "chosen",
)
)
)
.scalars()
.all()
)
received_map = await _receipted_quantities(db, chosen_ids)
all_arrived = True
for d_id in chosen_ids:
d = await db.get(ItemDecision, d_id)
li_d = await db.get(LineItem, d.line_item_id)
if arrival_state(li_d.quantity, received_map.get(d_id)) != "arrived":
all_arrived = False
break
if all_arrived and quote.status not in ("arrived", "rejected", "expired"):
quote.status = "arrived"
quote.arrived_date = payload.received_at
item_state = arrival_state(li.quantity, already_received + payload.received_qty)
await analytics.track(
"receipt_confirmed",
{
"receipt_id": str(receipt.id),
"quote_id": str(quote.id),
"quantity": str(payload.received_qty),
"state": item_state,
},
profile_id=ctx.display_name,
)
return {"data": {
"receipt_id": str(receipt.id),
"state": item_state,
"posted_transactions": [str(tx.id) for tx in posted],
}}
# ---------------------------------------------------------------------------
# GET /pm/receipts — delivery history (PM: own project — Admin: all projects)
# ---------------------------------------------------------------------------
@router.get("/receipts")
async def pm_receipts(
ctx: ArrivalScope = Depends(arrivals_scope),
db: AsyncSession = Depends(get_db),
):
stmt = (
select(ItemReceipt, ItemDecision, LineItem, Material, Quote, Project)
.join(ItemDecision, ItemDecision.id == ItemReceipt.item_decision_id)
.join(LineItem, LineItem.id == ItemDecision.line_item_id)
.join(Material, Material.id == LineItem.material_id)
.join(Quote, Quote.id == LineItem.quote_id)
.join(Project, Project.id == ItemDecision.project_id)
)
if ctx.role == "pm" and ctx.project_id is None:
raise ForbiddenError("شما به هیچ پروژه‌ای متصل نیستید (projects.pm_user_id)")
if ctx.project_id is not None:
# PM: active project; admin/manager: optional project filter
stmt = stmt.where(ItemDecision.project_id == ctx.project_id)
rows = (await db.execute(stmt.order_by(ItemReceipt.received_at.desc()))).all()
receipts = [r.ItemReceipt for r in rows]
alloc_rows = (
(
await db.execute(
select(ItemReceiptAllocation, Warehouse)
.join(Warehouse, Warehouse.id == ItemReceiptAllocation.warehouse_id)
.where(ItemReceiptAllocation.receipt_id.in_([r.id for r in receipts]) if receipts else False)
)
)
.all()
)
allocs_by_receipt: dict = {}
for alloc, wh in alloc_rows:
allocs_by_receipt.setdefault(str(alloc.receipt_id), []).append(
{"warehouse_id": str(wh.id), "warehouse_name": wh.name, "quantity": str(alloc.quantity)}
)
return {"data": [
{
"receipt_id": str(r.ItemReceipt.id),
"quote_id": str(r.Quote.id),
"line_item_id": str(r.LineItem.id),
"project": {"id": str(r.Project.id), "name": r.Project.name},
"material": {"id": str(r.Material.id), "name": r.Material.name},
"received_at": to_iso(r.ItemReceipt.received_at),
"received_qty": str(r.ItemReceipt.received_qty),
"on_time": r.ItemReceipt.on_time,
"quality_note": r.ItemReceipt.quality_note,
"allocations": allocs_by_receipt.get(str(r.ItemReceipt.id), []),
}
for r in rows
]}