MMD
e17d1f0883
- /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
1017 lines
43 KiB
Python
1017 lines
43 KiB
Python
"""Quote lifecycle services — nested creation, acceptance, arrival, transitions.
|
||
|
||
Every mutation here runs in ONE ``async with db.begin()`` block; any failure
|
||
rolls the whole operation back. All state changes log a quote_event in the
|
||
same transaction.
|
||
"""
|
||
import asyncio
|
||
import logging
|
||
from datetime import date, datetime, timezone
|
||
from uuid import UUID
|
||
|
||
from sqlalchemy import delete, func, select, update
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.models.procurement import (
|
||
ItemDecision,
|
||
ItemReceipt,
|
||
ItemTermsheet,
|
||
LineItem,
|
||
LineItemProjects,
|
||
Project,
|
||
Quote,
|
||
QuoteEvent,
|
||
QuoteProjects,
|
||
Supplier,
|
||
Termsheet,
|
||
TermsheetScheduleStep,
|
||
)
|
||
from app.schemas.procurement import (
|
||
AcceptQuoteIn,
|
||
ArriveItemsIn,
|
||
QuoteChildrenReplaceIn,
|
||
QuoteCreate,
|
||
QuoteUpdate,
|
||
RejectQuoteIn,
|
||
)
|
||
from app.services import analytics
|
||
from app.services.material_service import ensure_brand_material_belong
|
||
from app.utils.errors import ConflictError, NotFoundError, ValidationError
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
QUOTE_STATUSES = ("offered", "accepted", "arrived", "rejected", "expired")
|
||
|
||
# Backward transitions allowed via PATCH /quotes/{id} (contract P0-17).
|
||
BACKWARD_TRANSITIONS: dict[str, set[str]] = {
|
||
"accepted": {"offered"},
|
||
"arrived": {"accepted", "offered"},
|
||
"rejected": {"offered"},
|
||
"expired": {"offered"},
|
||
"offered": set(),
|
||
}
|
||
|
||
|
||
def utcnow() -> datetime:
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
def _add_event(db: AsyncSession, quote_id: UUID, event_type: str, payload: dict | None = None, actor: str | None = None) -> None:
|
||
db.add(QuoteEvent(quote_id=quote_id, event_type=event_type, payload=payload or {}, actor=actor))
|
||
# Product analytics (fire-and-forget, never blocks or raises): the whole
|
||
# quote lifecycle funnels through here, so one hook covers every
|
||
# status transition for the OpenPanel dashboards.
|
||
try:
|
||
asyncio.get_running_loop().create_task(
|
||
analytics.track(
|
||
f"quote_{event_type}",
|
||
{"quote_id": str(quote_id), "actor": actor, **(payload or {})},
|
||
profile_id=actor,
|
||
)
|
||
)
|
||
except RuntimeError: # no running loop (defensive — callers are async)
|
||
pass
|
||
|
||
|
||
async def get_quote_or_404(db: AsyncSession, quote_id: UUID) -> Quote:
|
||
quote = await db.get(Quote, quote_id)
|
||
if quote is None:
|
||
raise NotFoundError("پیشنهاد (استعلام) یافت نشد")
|
||
return quote
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# POST /quotes — nested creation in ONE transaction (temp_id cross-referencing)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def _insert_children(
|
||
db: AsyncSession,
|
||
quote_id: UUID,
|
||
quote_supplier_id: UUID,
|
||
line_items,
|
||
termsheets,
|
||
item_termsheets,
|
||
) -> None:
|
||
"""Insert line items + termsheets + item_termsheets links + pending decisions.
|
||
|
||
Shared by create_quote and replace_quote_children. temp_id values are
|
||
cross-referenced between line_items and item_termsheets inside the payload.
|
||
quote_supplier_id drives the v2.5 supplier<->material auto-linking.
|
||
"""
|
||
# --- temp_id sanity (400) ---
|
||
li_temp_ids = [li.temp_id for li in line_items]
|
||
ts_temp_ids = [ts.temp_id for ts in termsheets]
|
||
if len(li_temp_ids) != len(set(li_temp_ids)):
|
||
raise ValidationError("temp_id خطوط آیتم باید یکتا باشند", field="line_items")
|
||
if len(ts_temp_ids) != len(set(ts_temp_ids)):
|
||
raise ValidationError("temp_id ترمشیتها باید یکتا باشند", field="termsheets")
|
||
|
||
# --- line items: material / unit / brand_material checks ---
|
||
from app.models.catalog import Material, Unit
|
||
|
||
for li in line_items:
|
||
if await db.get(Material, li.material_id) is None:
|
||
raise NotFoundError("متریال یافت نشد", field="line_items.material_id")
|
||
if await db.get(Unit, li.unit_id) is None:
|
||
raise NotFoundError("واحد یافت نشد", field="line_items.unit_id")
|
||
await ensure_brand_material_belong(db, li.brand_material_id, li.material_id)
|
||
|
||
# --- insert line items; map temp_id -> real id ---
|
||
li_id_map: dict[str, UUID] = {}
|
||
for li in line_items:
|
||
row = LineItem(
|
||
quote_id=quote_id,
|
||
material_id=li.material_id,
|
||
brand_material_id=li.brand_material_id,
|
||
unit_id=li.unit_id,
|
||
quantity=li.quantity,
|
||
notes=li.notes,
|
||
)
|
||
db.add(row)
|
||
await db.flush()
|
||
li_id_map[li.temp_id] = row.id
|
||
# پروژههای تخصیصیافته به این قلم
|
||
for pid in (li.project_ids or []):
|
||
if await db.get(Project, pid) is None:
|
||
raise NotFoundError("پروژه یافت نشد", field="line_items.project_ids")
|
||
db.add(LineItemProjects(line_item_id=row.id, project_id=pid))
|
||
# v2.5: auto-link the material to the quote's supplier (catalog edge).
|
||
from app.services.supplier_material_service import link_supplier_material
|
||
await link_supplier_material(db, quote_supplier_id, li.material_id)
|
||
|
||
# --- insert termsheets + schedule steps; map temp_id -> real id ---
|
||
ts_id_map: dict[str, UUID] = {}
|
||
for ts in termsheets:
|
||
row = Termsheet(
|
||
quote_id=quote_id,
|
||
label=ts.label,
|
||
status="offered",
|
||
discount_type=ts.discount_type,
|
||
discount_value=ts.discount_value,
|
||
)
|
||
db.add(row)
|
||
await db.flush()
|
||
ts_id_map[ts.temp_id] = row.id
|
||
await _insert_schedule_steps(db, row.id, ts.schedule_steps, existing_count=0)
|
||
|
||
# --- item_termsheets links (resolve temp_ids) ---
|
||
# اقلام بدون قیمت اصلاً لینک نمیشوند — ذخیرهی پیشنهاد ناقص مجاز است.
|
||
for link in item_termsheets:
|
||
li_id = li_id_map.get(link.line_item_temp_id)
|
||
ts_id = ts_id_map.get(link.termsheet_temp_id)
|
||
if li_id is None:
|
||
raise ValidationError("line_item_temp_id نامعتبر است", field="item_termsheets.line_item_temp_id")
|
||
if ts_id is None:
|
||
raise ValidationError("termsheet_temp_id نامعتبر است", field="item_termsheets.termsheet_temp_id")
|
||
db.add(
|
||
ItemTermsheet(
|
||
line_item_id=li_id, termsheet_id=ts_id,
|
||
unit_price=link.unit_price, min_order_qty=link.min_order_qty,
|
||
)
|
||
)
|
||
|
||
# --- one pending item_decision per line_item (1:1 UNIQUE) ---
|
||
for li_id in li_id_map.values():
|
||
db.add(ItemDecision(line_item_id=li_id, status="pending"))
|
||
|
||
|
||
async def _sync_quote_projects(db: AsyncSession, quote: Quote, project_ids: list[UUID]) -> None:
|
||
"""Full replacement of the quote's project junction rows."""
|
||
await db.execute(
|
||
delete(QuoteProjects)
|
||
.where(QuoteProjects.quote_id == quote.id)
|
||
.execution_options(synchronize_session=False)
|
||
)
|
||
for pid in project_ids:
|
||
if await db.get(Project, pid) is None:
|
||
raise NotFoundError("پروژه یافت نشد", field="project_ids")
|
||
db.add(QuoteProjects(quote_id=quote.id, project_id=pid))
|
||
# legacy single-project column = اولین پروژه (برای فیلترها و آمار قدیمی)
|
||
quote.project_id = project_ids[0] if project_ids else None
|
||
|
||
|
||
async def create_quote(db: AsyncSession, payload: QuoteCreate, actor: str | None = None) -> Quote:
|
||
async with db.begin():
|
||
# --- FK existence (404) ---
|
||
if await db.get(Supplier, payload.supplier_id) is None:
|
||
raise NotFoundError("تأمینکننده یافت نشد", field="supplier_id")
|
||
|
||
project_ids = list(payload.project_ids)
|
||
if payload.project_id is not None and payload.project_id not in project_ids:
|
||
project_ids.insert(0, payload.project_id)
|
||
|
||
# --- insert quote ---
|
||
quote = Quote(
|
||
supplier_id=payload.supplier_id,
|
||
project_id=project_ids[0] if project_ids else payload.project_id,
|
||
quote_date=payload.quote_date or utcnow(),
|
||
expiry_date=payload.expiry_date,
|
||
is_official=payload.is_official,
|
||
notes=payload.notes,
|
||
status="offered", # server-set; never client-settable
|
||
)
|
||
db.add(quote)
|
||
await db.flush()
|
||
|
||
await _insert_children(db, quote.id, quote.supplier_id, payload.line_items, payload.termsheets, payload.item_termsheets)
|
||
if project_ids:
|
||
await _sync_quote_projects(db, quote, project_ids)
|
||
|
||
# --- offered event ---
|
||
_add_event(db, quote.id, "offered", {"by": "system"}, actor)
|
||
|
||
return quote
|
||
|
||
|
||
async def _insert_schedule_steps(
|
||
db: AsyncSession, termsheet_id: UUID, steps, existing_count: int = 0
|
||
) -> None:
|
||
"""Insert schedule steps; omitted step_no is auto-numbered MAX+1 onward."""
|
||
next_no = existing_count
|
||
for index, step in enumerate(steps):
|
||
step_no = step.step_no if step.step_no is not None else next_no + index + 1
|
||
db.add(
|
||
TermsheetScheduleStep(
|
||
termsheet_id=termsheet_id, step_no=step_no,
|
||
due_date=step.due_date, amount=step.amount, note=step.note,
|
||
)
|
||
)
|
||
|
||
|
||
async def next_step_no(db: AsyncSession, termsheet_id: UUID) -> int:
|
||
max_no = await db.scalar(
|
||
select(func.max(TermsheetScheduleStep.step_no)).where(
|
||
TermsheetScheduleStep.termsheet_id == termsheet_id
|
||
)
|
||
)
|
||
return (max_no or 0) + 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# PATCH /quotes/{id} — metadata + backward status transitions
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def update_quote(db: AsyncSession, quote_id: UUID, payload: QuoteUpdate, actor: str | None = None) -> Quote:
|
||
async with db.begin():
|
||
quote = await get_quote_or_404(db, quote_id)
|
||
changes: list[str] = []
|
||
|
||
if payload.supplier_id is not None and payload.supplier_id != quote.supplier_id:
|
||
if await db.get(Supplier, payload.supplier_id) is None:
|
||
raise NotFoundError("تأمینکننده یافت نشد", field="supplier_id")
|
||
quote.supplier_id = payload.supplier_id
|
||
changes.append("supplier_id")
|
||
if payload.project_ids is not None:
|
||
project_ids = list(payload.project_ids)
|
||
if payload.project_id is not None and payload.project_id not in project_ids:
|
||
project_ids.insert(0, payload.project_id)
|
||
await _sync_quote_projects(db, quote, project_ids)
|
||
changes.append("project_ids")
|
||
elif payload.project_id is not None and payload.project_id != quote.project_id:
|
||
if await db.get(Project, payload.project_id) is None:
|
||
raise NotFoundError("پروژه یافت نشد", field="project_id")
|
||
quote.project_id = payload.project_id
|
||
changes.append("project_id")
|
||
if payload.quote_date is not None and payload.quote_date != quote.quote_date:
|
||
quote.quote_date = payload.quote_date
|
||
changes.append("quote_date")
|
||
if payload.expiry_date is not None and payload.expiry_date != quote.expiry_date:
|
||
quote.expiry_date = payload.expiry_date
|
||
changes.append("expiry_date")
|
||
if payload.is_official is not None and payload.is_official != quote.is_official:
|
||
quote.is_official = payload.is_official
|
||
changes.append("is_official")
|
||
if payload.notes is not None and payload.notes != quote.notes:
|
||
quote.notes = payload.notes
|
||
changes.append("notes")
|
||
|
||
if payload.status is not None and payload.status != quote.status:
|
||
await _apply_status_transition(db, quote, payload.status)
|
||
changes.append("status")
|
||
|
||
if changes:
|
||
_add_event(db, quote.id, "edited", {"changed_fields": changes}, actor)
|
||
return quote
|
||
|
||
|
||
async def quote_has_receipts(db: AsyncSession, quote_id: UUID) -> bool:
|
||
"""Returns True if any warehouse receipt has been posted for the quote's items."""
|
||
count = await db.scalar(
|
||
select(func.count(ItemReceipt.id))
|
||
.join(ItemDecision, ItemReceipt.item_decision_id == ItemDecision.id)
|
||
.join(LineItem, ItemDecision.line_item_id == LineItem.id)
|
||
.where(LineItem.quote_id == quote_id, ItemReceipt.received_qty > 0)
|
||
)
|
||
return (count or 0) > 0
|
||
|
||
|
||
async def cleanup_orphan_materials(db: AsyncSession, material_ids) -> list[UUID]:
|
||
"""Check candidate material_ids and purge any that have zero references.
|
||
|
||
A material is considered an orphan if:
|
||
1. It is not referenced by ANY line_items across all quotes.
|
||
2. It is not referenced by ANY inventory stock items in warehouses.
|
||
|
||
If count == 0:
|
||
- Auxiliary records in supplier_materials, brand_materials, material_units,
|
||
and material_attribute_values are cleaned up.
|
||
- The material row is deleted from the central materials table.
|
||
|
||
Returns the list of purged material UUIDs.
|
||
"""
|
||
purged: list[UUID] = []
|
||
unique_ids = {m for m in material_ids if m is not None}
|
||
if not unique_ids:
|
||
return purged
|
||
|
||
from app.models.catalog import (
|
||
BrandMaterial,
|
||
Material,
|
||
MaterialAttributeValue,
|
||
MaterialUnit,
|
||
)
|
||
from app.models.inventory import InventoryItem
|
||
from app.models.procurement import LineItem, SupplierMaterial
|
||
|
||
for mat_id in unique_ids:
|
||
# Check active usage in line_items across the whole system
|
||
line_item_count = await db.scalar(
|
||
select(func.count(LineItem.id)).where(LineItem.material_id == mat_id)
|
||
)
|
||
if (line_item_count or 0) > 0:
|
||
continue
|
||
|
||
# Check active usage in warehouse inventory
|
||
inv_count = await db.scalar(
|
||
select(func.count(InventoryItem.id)).where(InventoryItem.material_id == mat_id)
|
||
)
|
||
if (inv_count or 0) > 0:
|
||
continue
|
||
|
||
# Material is not used in ANY line item and not in ANY warehouse inventory: pure orphan!
|
||
await db.execute(
|
||
delete(SupplierMaterial).where(SupplierMaterial.material_id == mat_id)
|
||
)
|
||
await db.execute(
|
||
delete(BrandMaterial).where(BrandMaterial.material_id == mat_id)
|
||
)
|
||
await db.execute(
|
||
delete(MaterialUnit).where(MaterialUnit.material_id == mat_id)
|
||
)
|
||
await db.execute(
|
||
delete(MaterialAttributeValue).where(MaterialAttributeValue.material_id == mat_id)
|
||
)
|
||
await db.execute(
|
||
delete(Material).where(Material.id == mat_id)
|
||
)
|
||
purged.append(mat_id)
|
||
logger.info("Purged orphan material %s from central catalog", mat_id)
|
||
|
||
return purged
|
||
|
||
|
||
async def replace_quote_children(
|
||
db: AsyncSession, quote_id: UUID, payload: QuoteChildrenReplaceIn, actor: str | None = None
|
||
) -> Quote:
|
||
"""Full replacement of a quote's line items + termsheets (edit modal).
|
||
|
||
Only allowed while the quote is still offered and has no warehouse receipts.
|
||
Automatically purges any materials detached during this replacement if they become orphans.
|
||
"""
|
||
async with db.begin():
|
||
quote = await get_quote_or_404(db, quote_id)
|
||
if quote.status != "offered":
|
||
raise ConflictError(
|
||
"برای تغییر اقلام یا شرایط پرداخت، ابتدا وضعیت را به «پیشنهاد شده» برگردانید",
|
||
field="status",
|
||
)
|
||
if await quote_has_receipts(db, quote.id):
|
||
raise ConflictError(
|
||
"این پیشنهاد بهدلیل ثبت رسید در انبار، غیرقابل ویرایش یا بازگشت است.",
|
||
field="status",
|
||
)
|
||
|
||
# 1. Pre-update snapshot: record previous material_ids assigned to this quote's items
|
||
old_items = (await db.execute(select(LineItem).where(LineItem.quote_id == quote.id))).scalars().all()
|
||
old_material_ids = {li.material_id for li in old_items if li.material_id is not None}
|
||
|
||
# children cascade via DDL: item_termsheets, item_decisions, item_receipts, schedule_steps
|
||
await db.execute(
|
||
delete(LineItem).where(LineItem.quote_id == quote.id).execution_options(synchronize_session=False)
|
||
)
|
||
await db.execute(
|
||
delete(Termsheet).where(Termsheet.quote_id == quote.id).execution_options(synchronize_session=False)
|
||
)
|
||
await _insert_children(db, quote.id, quote.supplier_id, payload.line_items, payload.termsheets, payload.item_termsheets)
|
||
await db.flush()
|
||
|
||
# 2. Identify detached materials: previous materials no longer present in the updated line items
|
||
new_material_ids = {li.material_id for li in payload.line_items if li.material_id is not None}
|
||
detached_material_ids = old_material_ids - new_material_ids
|
||
|
||
# 3. Clean up any detached materials that are now orphans (0 usage)
|
||
if detached_material_ids:
|
||
await cleanup_orphan_materials(db, detached_material_ids)
|
||
|
||
_add_event(db, quote.id, "edited", {"changed_fields": ["line_items", "termsheets"]}, actor)
|
||
return quote
|
||
|
||
|
||
|
||
async def _apply_status_transition(db: AsyncSession, quote: Quote, new_status: str) -> None:
|
||
"""Backward transitions only; reverts bookkeeping when going back to offered."""
|
||
current = quote.status
|
||
if new_status not in BACKWARD_TRANSITIONS.get(current, set()):
|
||
raise ConflictError("انتقال وضعیت مجاز نیست — فقط انتقالهای عقبگرد پذیرفته میشود", field="status")
|
||
|
||
if await quote_has_receipts(db, quote.id):
|
||
raise ConflictError(
|
||
"این پیشنهاد بهدلیل ثبت رسید در انبار، غیرقابل ویرایش یا بازگشت است.",
|
||
field="status",
|
||
)
|
||
|
||
if new_status == "offered":
|
||
# Reopen: clear lifecycle dates, demote accepted termsheets, reset decisions.
|
||
quote.accepted_date = None
|
||
quote.arrived_date = None
|
||
await db.execute(
|
||
update(Termsheet)
|
||
.where(Termsheet.quote_id == quote.id, Termsheet.status == "accepted")
|
||
.values(status="offered")
|
||
.execution_options(synchronize_session=False)
|
||
)
|
||
li_ids = select(LineItem.id).where(LineItem.quote_id == quote.id)
|
||
await db.execute(
|
||
update(ItemDecision)
|
||
.where(ItemDecision.line_item_id.in_(li_ids))
|
||
.values(
|
||
status="pending",
|
||
decided_at=None,
|
||
chosen_item_termsheet_id=None,
|
||
adjusted_unit_price=None,
|
||
adjusted_notes=None,
|
||
project_id=None,
|
||
)
|
||
.execution_options(synchronize_session=False)
|
||
)
|
||
# LineItemProjects (candidate projects) are preserved untouched!
|
||
elif new_status == "accepted": # arrived -> accepted
|
||
quote.arrived_date = None
|
||
|
||
quote.status = new_status
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# PATCH /quotes/{id}/accept — the accept flow (ONE transaction, FOR UPDATE)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def accept_quote(db: AsyncSession, quote_id: UUID, payload: AcceptQuoteIn, actor: str | None = None) -> Quote:
|
||
async with db.begin():
|
||
quote = await db.scalar(select(Quote).where(Quote.id == quote_id).with_for_update())
|
||
if quote is None:
|
||
raise NotFoundError("پیشنهاد (استعلام) یافت نشد")
|
||
if quote.status != "offered":
|
||
raise ConflictError("فقط پیشنهادهای offered قابل پذیرش هستند")
|
||
|
||
# --- validate accepted_termsheets ---
|
||
existing_ids = {t.id for t in payload.accepted_termsheets if t.id is not None}
|
||
existing_ts: dict[UUID, Termsheet] = {}
|
||
if existing_ids:
|
||
rows = (
|
||
await db.execute(
|
||
select(Termsheet).where(Termsheet.id.in_(existing_ids), Termsheet.quote_id == quote.id)
|
||
)
|
||
).scalars().all()
|
||
existing_ts = {t.id: t for t in rows}
|
||
missing = existing_ids - set(existing_ts)
|
||
if missing:
|
||
raise NotFoundError("ترمشیت یافت نشد یا متعلق به این پیشنهاد نیست", field="accepted_termsheets")
|
||
|
||
# --- validate item_decisions belong to the quote ---
|
||
li_ids = {d.line_item_id for d in payload.item_decisions}
|
||
if li_ids:
|
||
rows = (
|
||
await db.execute(
|
||
select(LineItem.id).where(LineItem.id.in_(li_ids), LineItem.quote_id == quote.id)
|
||
)
|
||
).scalars().all()
|
||
found = set(rows)
|
||
missing = li_ids - found
|
||
if missing:
|
||
raise NotFoundError("خط آیتم یافت نشد یا متعلق به این پیشنهاد نیست", field="item_decisions")
|
||
|
||
# --- resolve/create termsheets (existing: replace steps; new: insert) ---
|
||
ts_ids_by_index: list[UUID] = []
|
||
for index, t in enumerate(payload.accepted_termsheets):
|
||
if t.id is not None:
|
||
ts = existing_ts[t.id]
|
||
# replace schedule steps (DELETE old + INSERT new)
|
||
old_steps = (
|
||
(await db.execute(select(TermsheetScheduleStep).where(TermsheetScheduleStep.termsheet_id == ts.id)))
|
||
).scalars().all()
|
||
for step in old_steps:
|
||
await db.delete(step)
|
||
await db.flush()
|
||
await _insert_schedule_steps(db, ts.id, t.schedule_steps, existing_count=0)
|
||
if t.label is not None:
|
||
ts.label = t.label
|
||
ts.discount_type = t.discount_type
|
||
ts.discount_value = t.discount_value
|
||
ts_ids_by_index.append(ts.id)
|
||
else:
|
||
if not t.label:
|
||
raise ValidationError("برای ترمشیت جدید label الزامی است", field="accepted_termsheets")
|
||
new_ts = Termsheet(
|
||
quote_id=quote.id,
|
||
label=t.label,
|
||
status="offered",
|
||
discount_type=t.discount_type,
|
||
discount_value=t.discount_value,
|
||
)
|
||
db.add(new_ts)
|
||
await db.flush()
|
||
await _insert_schedule_steps(db, new_ts.id, t.schedule_steps, existing_count=0)
|
||
ts_ids_by_index.append(new_ts.id)
|
||
|
||
# --- process item decisions (UPSERT) ---
|
||
chosen_ts_ids: set[UUID] = set()
|
||
# پروژههای ارسالی در payload.item_projects — منبع اصلی برای
|
||
# decision.project_id (مدیر تدارکات برای هر قلم یک پروژه انتخاب میکند)
|
||
proj_by_line: dict[UUID, list[UUID]] = {e.line_item_id: list(e.project_ids) for e in payload.item_projects}
|
||
for decision_in in payload.item_decisions:
|
||
decision = await db.scalar(
|
||
select(ItemDecision).where(ItemDecision.line_item_id == decision_in.line_item_id).with_for_update()
|
||
)
|
||
if decision_in.status == "chosen":
|
||
index = decision_in.accepted_termsheet_index
|
||
if index is None:
|
||
raise ValidationError(
|
||
"برای قلم انتخابشده accepted_termsheet_index الزامی است", field="item_decisions"
|
||
)
|
||
if not 0 <= index < len(ts_ids_by_index):
|
||
raise ValidationError(
|
||
"accepted_termsheet_index خارج از محدوده ترمشیتها است", field="item_decisions"
|
||
)
|
||
if decision_in.adjusted_unit_price is None:
|
||
raise ValidationError(
|
||
"برای قلم انتخابشده adjusted_unit_price الزامی است", field="item_decisions"
|
||
)
|
||
termsheet_id = ts_ids_by_index[index]
|
||
it = await db.scalar(
|
||
select(ItemTermsheet).where(
|
||
ItemTermsheet.line_item_id == decision_in.line_item_id,
|
||
ItemTermsheet.termsheet_id == termsheet_id,
|
||
)
|
||
)
|
||
if it is None:
|
||
it = ItemTermsheet(
|
||
line_item_id=decision_in.line_item_id,
|
||
termsheet_id=termsheet_id,
|
||
unit_price=decision_in.adjusted_unit_price,
|
||
)
|
||
db.add(it)
|
||
else:
|
||
it.unit_price = decision_in.adjusted_unit_price
|
||
await db.flush()
|
||
|
||
if decision is None:
|
||
decision = ItemDecision(line_item_id=decision_in.line_item_id, status="pending")
|
||
db.add(decision)
|
||
decision.status = "chosen"
|
||
decision.chosen_item_termsheet_id = it.id
|
||
decision.adjusted_unit_price = decision_in.adjusted_unit_price
|
||
decision.adjusted_notes = decision_in.adjusted_notes
|
||
decision.decided_at = utcnow()
|
||
# v2.6: the SINGLE purchased project (proc manager picks it).
|
||
# Derivation order:
|
||
# 1) explicit decision_in.project_id
|
||
# 2) payload.item_projects for this line item (exactly one)
|
||
# 3) existing line_item_projects (exactly one)
|
||
# otherwise NULL — the item stays visible only to the office
|
||
# until a project is assigned (never wrongly visible to a PM).
|
||
if decision_in.project_id is not None:
|
||
decision.project_id = decision_in.project_id
|
||
else:
|
||
accept_projs = proj_by_line.get(decision_in.line_item_id) or []
|
||
if len(accept_projs) == 1:
|
||
decision.project_id = accept_projs[0]
|
||
else:
|
||
line_projs = (
|
||
(await db.execute(
|
||
select(LineItemProjects.project_id).where(
|
||
LineItemProjects.line_item_id == decision_in.line_item_id
|
||
)
|
||
))
|
||
.scalars()
|
||
.all()
|
||
)
|
||
if len(line_projs) == 1:
|
||
decision.project_id = line_projs[0]
|
||
if decision.project_id is not None:
|
||
if await db.get(Project, decision.project_id) is None:
|
||
raise NotFoundError("پروژه یافت نشد", field="item_decisions.project_id")
|
||
chosen_ts_ids.add(termsheet_id)
|
||
else: # rejected
|
||
if decision is None:
|
||
decision = ItemDecision(line_item_id=decision_in.line_item_id, status="pending")
|
||
db.add(decision)
|
||
decision.status = "rejected"
|
||
decision.chosen_item_termsheet_id = None
|
||
decision.adjusted_unit_price = None
|
||
decision.adjusted_notes = decision_in.adjusted_notes
|
||
decision.decided_at = utcnow()
|
||
|
||
# LineItemProjects records in PostgreSQL remain 100% untouched. Only item_decisions.project_id is written.
|
||
|
||
# termsheets referenced by at least one chosen item -> accepted
|
||
if chosen_ts_ids:
|
||
await db.execute(
|
||
update(Termsheet)
|
||
.where(Termsheet.id.in_(chosen_ts_ids))
|
||
.values(status="accepted")
|
||
.execution_options(synchronize_session=False)
|
||
)
|
||
|
||
quote.status = "accepted"
|
||
quote.accepted_date = utcnow()
|
||
_add_event(db, quote.id, "accepted", {"accepted_by": actor}, actor)
|
||
|
||
return quote
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Arrival
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def _decision_for_line_item(db: AsyncSession, line_item_id: UUID) -> ItemDecision:
|
||
decision = await db.scalar(select(ItemDecision).where(ItemDecision.line_item_id == line_item_id))
|
||
if decision is None:
|
||
decision = ItemDecision(line_item_id=line_item_id, status="pending")
|
||
db.add(decision)
|
||
await db.flush()
|
||
return decision
|
||
|
||
|
||
async def _fallback_confirmer_id(db: AsyncSession) -> UUID:
|
||
"""item_receipts.confirmed_by is NOT NULL — office (unauthenticated)
|
||
receipt paths reuse the dev-user row (same pattern as pm.py)."""
|
||
from app.models.procurement import User
|
||
dev = await db.scalar(select(User).where(User.casdoor_id == "dev"))
|
||
if dev is None:
|
||
dev = User(casdoor_id="dev", first_name="کاربر", last_name="آزمایشی", role="pm")
|
||
db.add(dev)
|
||
await db.flush()
|
||
return dev.id
|
||
|
||
|
||
async def arrive_quote(db: AsyncSession, quote_id: UUID, actor: str | None = None) -> Quote:
|
||
"""Full-arrival shortcut: receipt for every non-rejected line item."""
|
||
async with db.begin():
|
||
quote = await db.scalar(select(Quote).where(Quote.id == quote_id).with_for_update())
|
||
if quote is None:
|
||
raise NotFoundError("پیشنهاد (استعلام) یافت نشد")
|
||
if quote.status != "accepted":
|
||
raise ConflictError("فقط پیشنهادهای پذیرفتهشده قابل تحویل هستند")
|
||
|
||
line_items = (await db.execute(select(LineItem).where(LineItem.quote_id == quote.id))).scalars().all()
|
||
for li in line_items:
|
||
decision = await _decision_for_line_item(db, li.id)
|
||
if decision.status == "rejected":
|
||
continue
|
||
existing = await db.scalar(select(ItemReceipt).where(ItemReceipt.item_decision_id == decision.id))
|
||
if existing is None:
|
||
db.add(
|
||
ItemReceipt(
|
||
item_decision_id=decision.id,
|
||
confirmed_by=await _fallback_confirmer_id(db),
|
||
received_qty=li.quantity,
|
||
)
|
||
)
|
||
|
||
quote.status = "arrived"
|
||
quote.arrived_date = utcnow()
|
||
_add_event(db, quote.id, "arrived", {"source": "arrive"}, actor)
|
||
return quote
|
||
|
||
|
||
async def arrive_items(db: AsyncSession, quote_id: UUID, payload: ArriveItemsIn, actor: str | None = None) -> Quote:
|
||
"""Per-item arrival with full replacement semantics for receipts."""
|
||
async with db.begin():
|
||
quote = await db.scalar(select(Quote).where(Quote.id == quote_id).with_for_update())
|
||
if quote is None:
|
||
raise NotFoundError("پیشنهاد (استعلام) یافت نشد")
|
||
if quote.status != "accepted":
|
||
raise ConflictError("فقط پیشنهادهای پذیرفتهشده قابل تحویل هستند")
|
||
|
||
line_items = (await db.execute(select(LineItem).where(LineItem.quote_id == quote.id))).scalars().all()
|
||
li_by_id = {li.id: li for li in line_items}
|
||
|
||
items_in = {item.line_item_id for item in payload.items}
|
||
missing = items_in - set(li_by_id)
|
||
if missing:
|
||
raise NotFoundError("خط آیتم یافت نشد یا متعلق به این پیشنهاد نیست", field="items")
|
||
|
||
# upsert receipts for the provided items
|
||
for item in payload.items:
|
||
li = li_by_id[item.line_item_id]
|
||
decision = await _decision_for_line_item(db, li.id)
|
||
if decision.status == "rejected":
|
||
raise ConflictError("قلم ردشده قابل دریافت نیست", field="items")
|
||
receipt = await db.scalar(select(ItemReceipt).where(ItemReceipt.item_decision_id == decision.id))
|
||
if receipt is None:
|
||
db.add(
|
||
ItemReceipt(
|
||
item_decision_id=decision.id,
|
||
confirmed_by=await _fallback_confirmer_id(db),
|
||
received_at=item.received_at or utcnow(),
|
||
received_qty=item.received_qty,
|
||
on_time=item.on_time,
|
||
quality_note=item.quality_note,
|
||
)
|
||
)
|
||
else:
|
||
receipt.received_at = item.received_at or receipt.received_at
|
||
receipt.received_qty = item.received_qty
|
||
receipt.on_time = item.on_time
|
||
receipt.quality_note = item.quality_note
|
||
|
||
# full replacement: delete receipts of non-rejected items not in the list
|
||
for li in line_items:
|
||
if li.id in items_in:
|
||
continue
|
||
decision = await db.scalar(select(ItemDecision).where(ItemDecision.line_item_id == li.id))
|
||
if decision is None or decision.status == "rejected":
|
||
continue
|
||
receipt = await db.scalar(select(ItemReceipt).where(ItemReceipt.item_decision_id == decision.id))
|
||
if receipt is not None:
|
||
await db.delete(receipt)
|
||
|
||
# auto-arrive when every non-rejected line item has a receipt
|
||
all_received = True
|
||
for li in line_items:
|
||
decision = await db.scalar(select(ItemDecision).where(ItemDecision.line_item_id == li.id))
|
||
if decision is None or decision.status == "rejected":
|
||
continue
|
||
receipt = await db.scalar(select(ItemReceipt).where(ItemReceipt.item_decision_id == decision.id))
|
||
if receipt is None:
|
||
all_received = False
|
||
break
|
||
|
||
if all_received:
|
||
quote.status = "arrived"
|
||
quote.arrived_date = utcnow()
|
||
_add_event(db, quote.id, "arrived", {"source": "arrive-items"}, actor)
|
||
return quote
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Reject / expire / delete
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def reject_quote(db: AsyncSession, quote_id: UUID, payload: RejectQuoteIn, actor: str | None = None) -> Quote:
|
||
async with db.begin():
|
||
quote = await db.scalar(select(Quote).where(Quote.id == quote_id).with_for_update())
|
||
if quote is None:
|
||
raise NotFoundError("پیشنهاد (استعلام) یافت نشد")
|
||
if quote.status != "offered":
|
||
raise ConflictError("فقط پیشنهادهای offered قابل رد شدن هستند")
|
||
quote.status = "rejected"
|
||
_add_event(
|
||
db, quote.id, "rejected",
|
||
{"reason": payload.reason} if payload.reason else {},
|
||
actor,
|
||
)
|
||
return quote
|
||
|
||
|
||
async def expire_quote(db: AsyncSession, quote_id: UUID, actor: str | None = None) -> Quote:
|
||
async with db.begin():
|
||
quote = await db.scalar(select(Quote).where(Quote.id == quote_id).with_for_update())
|
||
if quote is None:
|
||
raise NotFoundError("پیشنهاد (استعلام) یافت نشد")
|
||
if quote.status != "offered":
|
||
raise ConflictError("فقط پیشنهادهای offered قابل انقضا هستند")
|
||
quote.status = "expired"
|
||
_add_event(db, quote.id, "expired", {}, actor)
|
||
return quote
|
||
|
||
|
||
async def delete_quote(db: AsyncSession, quote_id: UUID) -> dict:
|
||
tx = db.begin_nested() if db.in_transaction() else db.begin()
|
||
async with tx:
|
||
quote = await get_quote_or_404(db, quote_id)
|
||
if quote.status in ("accepted", "arrived"):
|
||
raise ConflictError("پیشنهادهای پذیرفتهشده یا رسیده قابل حذف نیستند")
|
||
|
||
# count cascaded children for the response envelope
|
||
line_items = (await db.execute(select(LineItem).where(LineItem.quote_id == quote.id))).scalars().all()
|
||
li_ids = [li.id for li in line_items]
|
||
decisions = (
|
||
(await db.execute(select(ItemDecision).where(ItemDecision.line_item_id.in_(li_ids)))).scalars().all()
|
||
if li_ids else []
|
||
)
|
||
dec_ids = [d.id for d in decisions]
|
||
receipts = (
|
||
(await db.execute(select(ItemReceipt).where(ItemReceipt.item_decision_id.in_(dec_ids)))).scalars().all()
|
||
if dec_ids else []
|
||
)
|
||
termsheets = (await db.execute(select(Termsheet).where(Termsheet.quote_id == quote.id))).scalars().all()
|
||
events = (await db.execute(select(QuoteEvent).where(QuoteEvent.quote_id == quote.id))).scalars().all()
|
||
|
||
# collect candidate material_ids for orphan cleanup before deleting quote
|
||
candidate_material_ids = [li.material_id for li in line_items if li.material_id is not None]
|
||
|
||
# collect attachment keys before deleting quote row
|
||
from app.models.files import QuoteAttachment
|
||
from app.services import storage
|
||
|
||
attachments = (await db.execute(select(QuoteAttachment).where(QuoteAttachment.quote_id == quote.id))).scalars().all()
|
||
att_keys = [att.object_key for att in attachments]
|
||
quote_prefix = f"{quote.id}/"
|
||
|
||
await db.delete(quote)
|
||
await db.flush()
|
||
|
||
# Purge any materials that no longer have any references in the system
|
||
purged_material_ids = await cleanup_orphan_materials(db, candidate_material_ids)
|
||
|
||
# Post-commit: clean up attachments in MinIO (best-effort; DB is already committed)
|
||
for key in att_keys:
|
||
try:
|
||
await storage.remove_object(key)
|
||
except Exception:
|
||
logger.warning("Failed to remove MinIO object %s for quote %s", key, quote_id)
|
||
try:
|
||
await storage.remove_prefix(quote_prefix)
|
||
except Exception:
|
||
logger.warning("Failed to remove MinIO prefix %s for quote %s", quote_prefix, quote_id)
|
||
|
||
return {
|
||
"id": str(quote_id),
|
||
"deleted": True,
|
||
"cascaded": {
|
||
"line_items": len(line_items),
|
||
"termsheets": len(termsheets),
|
||
"events": len(events),
|
||
"decisions": len(decisions),
|
||
"receipts": len(receipts),
|
||
"attachments": len(attachments),
|
||
"purged_orphan_materials": len(purged_material_ids),
|
||
},
|
||
}
|
||
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Statistics
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def quote_stats(
|
||
db: AsyncSession,
|
||
supplier_id: UUID | None = None,
|
||
project_id: UUID | None = None,
|
||
quote_date_from=None,
|
||
quote_date_to=None,
|
||
category_id: UUID | None = None,
|
||
) -> dict:
|
||
conditions = []
|
||
if supplier_id is not None:
|
||
conditions.append(Quote.supplier_id == supplier_id)
|
||
if project_id is not None:
|
||
conditions.append(Quote.project_id == project_id)
|
||
if quote_date_from is not None:
|
||
conditions.append(Quote.quote_date >= quote_date_from)
|
||
if quote_date_to is not None:
|
||
conditions.append(Quote.quote_date <= quote_date_to)
|
||
if category_id is not None:
|
||
conditions.append(_quote_has_category(category_id))
|
||
|
||
rows = (
|
||
await db.execute(
|
||
select(Quote.status, func.count()).where(*conditions).group_by(Quote.status)
|
||
)
|
||
).all()
|
||
counts = {status: 0 for status in QUOTE_STATUSES}
|
||
for status, count in rows:
|
||
counts[status] = count
|
||
|
||
avg = await db.scalar(
|
||
select(
|
||
func.avg(func.extract("epoch", Quote.arrived_date - Quote.accepted_date) / 86400.0)
|
||
).where(
|
||
Quote.status == "arrived",
|
||
Quote.arrived_date.is_not(None),
|
||
Quote.accepted_date.is_not(None),
|
||
*conditions,
|
||
)
|
||
)
|
||
return {
|
||
"total": sum(counts.values()),
|
||
"offered": counts["offered"],
|
||
"accepted": counts["accepted"],
|
||
"arrived": counts["arrived"],
|
||
"rejected": counts["rejected"],
|
||
"expired": counts["expired"],
|
||
"average_lead_time_days": int(round(float(avg))) if avg is not None else None,
|
||
}
|
||
|
||
|
||
def _quote_has_category(category_id: UUID):
|
||
"""EXISTS: quote has a line item whose material belongs to category_id."""
|
||
from app.models.catalog import Material
|
||
|
||
sub = (
|
||
select(1)
|
||
.where(
|
||
LineItem.quote_id == Quote.id,
|
||
Material.id == LineItem.material_id,
|
||
Material.category_id == category_id,
|
||
)
|
||
)
|
||
return func.exists(sub)
|
||
|
||
|
||
async def project_stats(db: AsyncSession, project_id: UUID) -> dict:
|
||
"""Counts + total value per quote status for one project.
|
||
|
||
total_value: SUM(quantity × unit_price) over line items using the chosen
|
||
termsheet link first, else the first link. Rejected line items are
|
||
excluded from value sums (contract P1-14).
|
||
"""
|
||
project = await db.get(Project, project_id)
|
||
if project is None:
|
||
raise NotFoundError("پروژه یافت نشد")
|
||
|
||
quotes = (await db.execute(select(Quote).where(Quote.project_id == project_id))).scalars().all()
|
||
if not quotes:
|
||
empty = {status: {"count": 0, "total_value": 0} for status in QUOTE_STATUSES}
|
||
return {"project_id": str(project_id), "total_quotes": 0, **empty}
|
||
|
||
quote_ids = [q.id for q in quotes]
|
||
line_items = (await db.execute(select(LineItem).where(LineItem.quote_id.in_(quote_ids)))).scalars().all()
|
||
li_ids = [li.id for li in line_items]
|
||
|
||
decisions = {
|
||
d.line_item_id: d
|
||
for d in (await db.execute(select(ItemDecision).where(ItemDecision.line_item_id.in_(li_ids)))).scalars().all()
|
||
} if li_ids else {}
|
||
item_termsheets = (
|
||
(await db.execute(select(ItemTermsheet).where(ItemTermsheet.line_item_id.in_(li_ids)))).scalars().all()
|
||
) if li_ids else []
|
||
its_by_id = {it.id: it for it in item_termsheets}
|
||
its_by_line: dict[UUID, list[ItemTermsheet]] = {}
|
||
for it in item_termsheets:
|
||
its_by_line.setdefault(it.line_item_id, []).append(it)
|
||
|
||
per_status = {status: {"count": 0, "total_value": 0.0} for status in QUOTE_STATUSES}
|
||
for q in quotes:
|
||
per_status[q.status]["count"] += 1
|
||
|
||
for li in line_items:
|
||
decision = decisions.get(li.id)
|
||
if decision is not None and decision.status == "rejected":
|
||
continue # rejected items excluded from value sums
|
||
if decision is not None and decision.chosen_item_termsheet_id is not None:
|
||
price_it = its_by_id.get(decision.chosen_item_termsheet_id)
|
||
else:
|
||
links = its_by_line.get(li.id, [])
|
||
price_it = links[0] if links else None
|
||
if price_it is None:
|
||
continue
|
||
quote = next(q for q in quotes if q.id == li.quote_id)
|
||
per_status[quote.status]["total_value"] += float(price_it.unit_price * li.quantity)
|
||
|
||
result = {"project_id": str(project_id), "total_quotes": len(quotes)}
|
||
for status, values in per_status.items():
|
||
result[status] = {"count": values["count"], "total_value": values["total_value"]}
|
||
return result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Background auto-expiry
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def auto_expire_quotes(db: AsyncSession) -> int:
|
||
"""Expire quotes with expiry_date < today AND status = 'offered' (scheduler)."""
|
||
today = date.today()
|
||
rows = await db.execute(
|
||
select(Quote).where(
|
||
Quote.status == "offered",
|
||
Quote.expiry_date.is_not(None),
|
||
Quote.expiry_date < today,
|
||
)
|
||
)
|
||
quotes = rows.scalars().all()
|
||
if not quotes:
|
||
return 0
|
||
async with db.begin():
|
||
for q in quotes:
|
||
q.status = "expired"
|
||
_add_event(
|
||
db, q.id, "expired",
|
||
{"reason": "منقضی خودکار — تاریخ اعتبار گذشته است"}, "سیستم",
|
||
)
|
||
logger.info("Auto-expired %d quote(s)", len(quotes))
|
||
return len(quotes)
|