"""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 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.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)) 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)