"""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 primary; dev fallback headers (X-User-Role: pm|admin + X-Project-Id for pm) when no Authorization header — 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, 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.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.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 single project; None for admin display_name: str async def arrivals_scope(request: Request, db: AsyncSession = Depends(get_db)) -> ArrivalScope: """Resolve the arrivals caller — Casdoor JWT first, dev headers as fallback. PM → scoped to his own project. Admin / procurement manager → all projects. """ 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": if ctx.project_id is None: raise ForbiddenError("شما به هیچ پروژه‌ای متصل نیستید (projects.pm_user_id)") return ArrivalScope(role="pm", user_id=ctx.user.id, project_id=ctx.project_id, display_name=ctx.display_name) if ctx.role in ("admin", "procurement_manager"): return ArrivalScope(role="admin", user_id=ctx.user.id, project_id=None, display_name=ctx.display_name) raise ForbiddenError("این بخش فقط برای مدیر پروژه یا مدیر تدارکات است") # Dev fallback: X-User-Role: pm | admin (procurement_manager maps to admin) if 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 را بفرستید") project_id = None if 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": if ctx.project_id is None: raise ForbiddenError("شما به هیچ پروژه‌ای متصل نیستید (projects.pm_user_id)") 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.role == "pm" 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) 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": if ctx.project_id is None: raise ForbiddenError("شما به هیچ پروژه‌ای متصل نیستید (projects.pm_user_id)") 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 ]}