from __future__ import annotations from datetime import date from decimal import Decimal from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, Query, status from pydantic import BaseModel, Field, ValidationError from sqlalchemy import String, cast, or_ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.api.deps import ( get_current_user, has_functionality_access, require_access, require_roles, require_stage_access, stage_functionality, ) from app.core.i18n import Language, get_language, translate from app.db.session import get_db from app.models.enums import ProcurementStatus, Role from app.models.user import User from app.models.procurement import ( DeliveryAcceptance, Demand, ETenderCase, EMD, FinancialEvaluation, GeMProcurementCase, GlobalTenderCase, ItemDatabase, LPCProcurementCase, OpenMarketCase, OtherProcurementCase, PaymentTracking, PerformanceSecurity, ProcurementCase, PurchaseOrder, Requisition, SampleRegister, TechnicalEvaluation, TenderBid, ) from app.schemas import procurement as schemas from app.schemas.common import Page router = APIRouter(dependencies=[Depends(get_current_user)]) WRITER = Depends(require_roles(Role.procurement_officer, Role.evaluator, Role.finance)) ACTIVE_PROCUREMENT_MODES = {"GEM", "E_TENDER", "GLOBAL_TENDER", "LPC", "OPEN_MARKET", "OTHER"} PROCUREMENT_REGISTER_STATUSES = {"In Tendering Process", "Closed", "Re-Tendered", "Cancelled"} PROCUREMENT_TYPE_STAGE_MODES = { "gem-cases": "GEM", "e-tender-cases": "E_TENDER", "global-tender-cases": "GLOBAL_TENDER", "lpc-cases": "LPC", "open-market-cases": "OPEN_MARKET", "other-procurement-cases": "OTHER", } PROCUREMENT_MODE_DETAIL_MODELS = { "GEM": GeMProcurementCase, "E_TENDER": ETenderCase, "GLOBAL_TENDER": GlobalTenderCase, "LPC": LPCProcurementCase, "OPEN_MARKET": OpenMarketCase, "OTHER": OtherProcurementCase, } PROCESS_STAGES = { "tenders", "emds", "samples", "technical-evaluations", "financial-evaluations", "purchase-orders", "performance-securities", "delivery-acceptance", "payments", } class StageConfig(BaseModel): model: Any create_schema: Any update_schema: Any read_schema: Any search_fields: list[str] parent_refs: dict[str, tuple[Any, str]] = Field(default_factory=dict) model_config = {"arbitrary_types_allowed": True} STAGES: dict[str, StageConfig] = { "demands": StageConfig( model=Demand, create_schema=schemas.DemandCreate, update_schema=schemas.DemandUpdate, read_schema=schemas.DemandRead, search_fields=["demand_no", "department", "item_description", "status"], ), "item-database": StageConfig( model=ItemDatabase, create_schema=schemas.ItemDatabaseCreate, update_schema=schemas.ItemDatabaseUpdate, read_schema=schemas.ItemDatabaseRead, search_fields=["sr_no", "name", "item_type"], ), "requisitions": StageConfig( model=Requisition, create_schema=schemas.RequisitionCreate, update_schema=schemas.RequisitionUpdate, read_schema=schemas.RequisitionRead, search_fields=[ "requisition_no", "received_from", "received_from_district", "received_from_range", "sender_name", "sender_designation", "item_description", "procurement_mode", "procurement_mode_other", "budget_head", "requested_by", "approval_status", ], parent_refs={"demand_id": (Demand, "Demand")}, ), "procurement-cases": StageConfig( model=ProcurementCase, create_schema=schemas.ProcurementCaseCreate, update_schema=schemas.ProcurementCaseUpdate, read_schema=schemas.ProcurementCaseRead, search_fields=[ "procurement_case_no", "procurement_mode", "procurement_mode_other", "case_status", "mode_selected_by", "file_no", "sanction_no", "approving_authority", "remarks", ], parent_refs={"requisition_id": (Requisition, "Requisition")}, ), "gem-cases": StageConfig( model=GeMProcurementCase, create_schema=schemas.GeMProcurementCaseCreate, update_schema=schemas.GeMProcurementCaseUpdate, read_schema=schemas.GeMProcurementCaseRead, search_fields=[ "gem_procurement_type", "gem_bid_no", "gem_contract_no", "delivery_status", "delivery_delay_reason", "delivery_delay_document_path", "payment_status", "status", ], parent_refs={"procurement_case_id": (ProcurementCase, "Procurement case")}, ), "e-tender-cases": StageConfig( model=ETenderCase, create_schema=schemas.ETenderCaseCreate, update_schema=schemas.ETenderCaseUpdate, read_schema=schemas.ETenderCaseRead, search_fields=[ "tender_ref_no", "technical_evaluation_status", "financial_evaluation_status", "loa_no", "contract_no", "status", ], parent_refs={"procurement_case_id": (ProcurementCase, "Procurement case")}, ), "global-tender-cases": StageConfig( model=GlobalTenderCase, create_schema=schemas.GlobalTenderCaseCreate, update_schema=schemas.GlobalTenderCaseUpdate, read_schema=schemas.GlobalTenderCaseRead, search_fields=[ "global_tender_ref_no", "country_of_vendor", "foreign_vendor_name", "foreign_currency", "technical_evaluation_status", "commercial_evaluation_status", "contract_no", "status", ], parent_refs={"procurement_case_id": (ProcurementCase, "Procurement case")}, ), "lpc-cases": StageConfig( model=LPCProcurementCase, create_schema=schemas.LPCProcurementCaseCreate, update_schema=schemas.LPCProcurementCaseUpdate, read_schema=schemas.LPCProcurementCaseRead, search_fields=[ "lpc_meeting_no", "committee_members", "comparative_statement_path", "status", ], parent_refs={"procurement_case_id": (ProcurementCase, "Procurement case")}, ), "open-market-cases": StageConfig( model=OpenMarketCase, create_schema=schemas.OpenMarketCaseCreate, update_schema=schemas.OpenMarketCaseUpdate, read_schema=schemas.OpenMarketCaseRead, search_fields=[ "procurement_justification", "comparative_statement_path", "status", ], parent_refs={"procurement_case_id": (ProcurementCase, "Procurement case")}, ), "other-procurement-cases": StageConfig( model=OtherProcurementCase, create_schema=schemas.OtherProcurementCaseCreate, update_schema=schemas.OtherProcurementCaseUpdate, read_schema=schemas.OtherProcurementCaseRead, search_fields=[ "other_procurement_type", "procurement_reference_no", "procurement_description", "status", ], parent_refs={"procurement_case_id": (ProcurementCase, "Procurement case")}, ), "tenders": StageConfig( model=TenderBid, create_schema=schemas.TenderBidCreate, update_schema=schemas.TenderBidUpdate, read_schema=schemas.TenderBidRead, search_fields=[ "serial_no", "tender_no", "item_name", "budget_head", "l1_bidder_name", "contract_number", "remarks", "title", "bid_type", "status", ], parent_refs={ "requisition_id": (Requisition, "Requisition"), "procurement_case_id": (ProcurementCase, "Procurement case"), }, ), "emds": StageConfig( model=EMD, create_schema=schemas.EMDCreate, update_schema=schemas.EMDUpdate, read_schema=schemas.EMDRead, search_fields=[ "serial_no", "emd_unique_id", "tender_number", "item_name", "bidder_name", "emd_form", "emd_no", "issuing_branch", "return_details", "remarks", "instrument_no", "refund_status", ], parent_refs={ "tender_id": (TenderBid, "Tender"), "procurement_case_id": (ProcurementCase, "Procurement case"), }, ), "samples": StageConfig( model=SampleRegister, create_schema=schemas.SampleRegisterCreate, update_schema=schemas.SampleRegisterUpdate, read_schema=schemas.SampleRegisterRead, search_fields=[ "serial_no", "tender_number", "item_name", "bidder_name", "make", "model", "make_model", "recipient_name", "remarks", "sample_description", "evaluation_status", ], parent_refs={ "tender_id": (TenderBid, "Tender"), "procurement_case_id": (ProcurementCase, "Procurement case"), }, ), "technical-evaluations": StageConfig( model=TechnicalEvaluation, create_schema=schemas.TechnicalEvaluationCreate, update_schema=schemas.TechnicalEvaluationUpdate, read_schema=schemas.TechnicalEvaluationRead, search_fields=["bidder_name", "evaluator", "result"], parent_refs={ "tender_id": (TenderBid, "Tender"), "procurement_case_id": (ProcurementCase, "Procurement case"), }, ), "financial-evaluations": StageConfig( model=FinancialEvaluation, create_schema=schemas.FinancialEvaluationCreate, update_schema=schemas.FinancialEvaluationUpdate, read_schema=schemas.FinancialEvaluationRead, search_fields=["bidder_name", "result"], parent_refs={ "tender_id": (TenderBid, "Tender"), "procurement_case_id": (ProcurementCase, "Procurement case"), }, ), "purchase-orders": StageConfig( model=PurchaseOrder, create_schema=schemas.PurchaseOrderCreate, update_schema=schemas.PurchaseOrderUpdate, read_schema=schemas.PurchaseOrderRead, search_fields=[ "po_no", "procurement_mode", "tender_reference_no", "budget_head", "vendor_name", "item_description", "unit", "status", ], parent_refs={ "tender_id": (TenderBid, "Tender"), "procurement_case_id": (ProcurementCase, "Procurement case"), }, ), "performance-securities": StageConfig( model=PerformanceSecurity, create_schema=schemas.PerformanceSecurityCreate, update_schema=schemas.PerformanceSecurityUpdate, read_schema=schemas.PerformanceSecurityRead, search_fields=[ "contract_number", "item_name", "vendor_name", "bidder_name", "security_form", "security_number", "issuing_branch", "warranty_period", "return_details", "remarks", "instrument_no", "release_status", ], parent_refs={ "po_id": (PurchaseOrder, "Purchase order"), "procurement_case_id": (ProcurementCase, "Procurement case"), }, ), "delivery-acceptance": StageConfig( model=DeliveryAcceptance, create_schema=schemas.DeliveryAcceptanceCreate, update_schema=schemas.DeliveryAcceptanceUpdate, read_schema=schemas.DeliveryAcceptanceRead, search_fields=["acceptance_status", "remarks"], parent_refs={ "po_id": (PurchaseOrder, "Purchase order"), "procurement_case_id": (ProcurementCase, "Procurement case"), }, ), "payments": StageConfig( model=PaymentTracking, create_schema=schemas.PaymentTrackingCreate, update_schema=schemas.PaymentTrackingUpdate, read_schema=schemas.PaymentTrackingRead, search_fields=["invoice_no", "payment_status"], parent_refs={ "po_id": (PurchaseOrder, "Purchase order"), "procurement_case_id": (ProcurementCase, "Procurement case"), }, ), } def get_stage(stage: str, language: Language = "en") -> StageConfig: config = STAGES.get(stage) if not config: raise HTTPException(status_code=404, detail=translate("procurement.unknown_stage", language)) return config def _validate(schema: Any, payload: dict[str, Any]) -> Any: try: return schema.model_validate(payload) except ValidationError as exc: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=exc.errors(), ) from exc def _validate_parent_refs( db: Session, config: StageConfig, data: BaseModel, *, exclude_unset: bool = False, ) -> None: values = data.model_dump(exclude_unset=exclude_unset) for field_name, (parent_model, label) in config.parent_refs.items(): parent_id = values.get(field_name) if parent_id is not None and db.get(parent_model, parent_id) is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"{label} ID {parent_id} does not exist.", ) def _validate_procurement_type_parent(db: Session, stage: str, data: BaseModel) -> None: expected_mode = PROCUREMENT_TYPE_STAGE_MODES.get(stage) if not expected_mode: return procurement_case_id = getattr(data, "procurement_case_id", None) if procurement_case_id is None: return case = db.get(ProcurementCase, procurement_case_id) if case and _normalize_mode(case.procurement_mode) != expected_mode: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Selected procurement case must be {expected_mode}.", ) def _validate_procurement_register_status(stage: str, values: dict[str, Any]) -> None: status_value = values.get("case_status") if stage == "procurement-cases" else values.get("status") if status_value is None or stage not in {"procurement-cases", *PROCUREMENT_TYPE_STAGE_MODES.keys()}: return if status_value not in PROCUREMENT_REGISTER_STATUSES: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Status must be In Tendering Process, Closed, Re-Tendered, or Cancelled.", ) def _requisition_quantity(values: dict[str, Any]) -> int | None: items = values.get("requisition_items") if isinstance(items, list): quantities: list[int] = [] for item in items: if not isinstance(item, dict) or item.get("quantity") is None: continue try: quantities.append(int(item["quantity"])) except (TypeError, ValueError): continue if quantities: return sum(quantities) quantity = values.get("quantity") if quantity is None: return None try: return int(quantity) except (TypeError, ValueError): return None def _validate_requisition_approval_quantity(values: dict[str, Any]) -> None: approval_quantity = values.get("administrative_approval_quantity") if approval_quantity in (None, ""): return requisition_quantity = _requisition_quantity(values) if requisition_quantity is None: return if int(approval_quantity) > requisition_quantity: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Administrative Approval Quantity cannot be higher than Requisition quantity.", ) def _validate_requisition_approval_status(values: dict[str, Any]) -> None: approval_status = values.get("approval_status") if approval_status not in {ProcurementStatus.approved, ProcurementStatus.rejected, "approved", "rejected"}: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Approval Status must be either Approved or Rejected.", ) def _normalize_mode(value: Any) -> str: return str(value or "").strip().upper().replace("-", "_").replace(" ", "_") def _apply_procurement_case_rules(payload: dict[str, Any], current_mode: str | None = None) -> dict[str, Any]: normalized = {**payload} mode = _normalize_mode(normalized.get("procurement_mode") or current_mode) if not mode: return normalized if mode not in ACTIVE_PROCUREMENT_MODES: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Procurement mode must be one of GEM, E_TENDER, GLOBAL_TENDER, LPC, OPEN_MARKET, or OTHER.", ) if "procurement_mode" in normalized: normalized["procurement_mode"] = mode if mode == "OTHER": other_mode = str(normalized.get("procurement_mode_other") or "").strip() if not other_mode: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Enter the other procurement mode.", ) normalized["procurement_mode_other"] = other_mode else: normalized["procurement_mode_other"] = None if mode in {"E_TENDER", "GLOBAL_TENDER"}: normalized["emd_applicable"] = True normalized["pbg_applicable"] = True elif mode == "GEM": normalized.setdefault("emd_applicable", False) normalized["pbg_applicable"] = True elif mode == "LPC": normalized["emd_applicable"] = False normalized.setdefault("pbg_applicable", False) elif mode == "OPEN_MARKET": normalized.setdefault("emd_applicable", False) normalized.setdefault("pbg_applicable", False) elif mode == "OTHER": normalized.setdefault("emd_applicable", False) normalized.setdefault("pbg_applicable", False) if normalized.get("emd_applicable") is None: normalized["emd_applicable"] = False if normalized.get("pbg_applicable") is None: normalized["pbg_applicable"] = False return normalized def _apply_requisition_mode_rules( payload: dict[str, Any], current_mode: str | None = None, ) -> dict[str, Any]: normalized = {**payload} if "procurement_mode" not in normalized: return normalized mode = _normalize_mode(normalized.get("procurement_mode") or current_mode) if mode and mode not in ACTIVE_PROCUREMENT_MODES: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Procurement mode must be GEM, E_TENDER, GLOBAL_TENDER, LPC, OPEN_MARKET, or OTHER.", ) normalized["procurement_mode"] = mode or None if mode == "OTHER": other_mode = str(normalized.get("procurement_mode_other") or "").strip() if not other_mode: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Enter the other procurement mode.", ) normalized["procurement_mode_other"] = other_mode else: normalized["procurement_mode_other"] = None return normalized def _is_blank(value: Any) -> bool: return value is None or value == "" def _resolve_manual_procurement_case( stage: str, payload: dict[str, Any], db: Session, ) -> dict[str, Any]: normalized = {**payload} manual_reference = normalized.pop("procurement_case_other", None) if manual_reference is None: return normalized manual_reference = str(manual_reference).strip() if not manual_reference: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Enter a manual procurement case number or reference.", ) if len(manual_reference) > 255: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Manual procurement case number or reference cannot exceed 255 characters.", ) expected_mode = PROCUREMENT_TYPE_STAGE_MODES.get(stage) case = db.query(ProcurementCase).filter( ProcurementCase.procurement_case_no == manual_reference ).first() if case and expected_mode and _normalize_mode(case.procurement_mode) != expected_mode: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Procurement case {manual_reference} belongs to a different procurement mode.", ) if case is None: mode = expected_mode or _normalize_mode(normalized.get("procurement_mode")) or "OTHER" if mode not in ACTIVE_PROCUREMENT_MODES: mode = "OTHER" case = ProcurementCase( procurement_case_no=manual_reference, procurement_mode=mode, procurement_mode_other=( str(normalized.get("procurement_mode_other") or "Manually Entered Procurement Case").strip() if mode == "OTHER" else None ), case_status="In Tendering Process", mode_selection_date=date.today(), remarks="Manually created from an Others selection.", ) db.add(case) _flush_or_conflict(db) normalized["procurement_case_id"] = case.id return normalized def _validate_procurement_case_reference(stage: str, data: BaseModel) -> None: if stage not in {*PROCUREMENT_TYPE_STAGE_MODES.keys(), *PROCESS_STAGES}: return if getattr(data, "procurement_case_id", None) is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Select a procurement case or use Others to enter one manually.", ) def _fill_blank(payload: dict[str, Any], key: str, value: Any) -> None: if key not in payload or _is_blank(payload.get(key)): payload[key] = value def _requisition_items_for_process(requisition: Requisition | None) -> list[dict[str, Any]]: if not requisition: return [] if isinstance(requisition.requisition_items, list) and requisition.requisition_items: items: list[dict[str, Any]] = [] for index, item in enumerate(requisition.requisition_items): if not isinstance(item, dict): continue items.append( { "serial_no": item.get("serial_no") or index + 1, "description": item.get("description") or requisition.item_description, "item_type": item.get("item_type"), "quantity": item.get("quantity"), "unit": item.get("unit"), } ) if items: return items if requisition.item_description: return [ { "serial_no": 1, "description": requisition.item_description, "quantity": requisition.quantity, } ] return [] def _sample_items_from_tender(tender: TenderBid | None, requisition: Requisition | None) -> list[dict[str, Any]]: source = tender.tender_items if tender and isinstance(tender.tender_items, list) else _requisition_items_for_process(requisition) return [ { "serial_no": item.get("serial_no") or index + 1, "description": item.get("description"), "quantity": item.get("quantity"), "unit": item.get("unit"), } for index, item in enumerate(source) if isinstance(item, dict) and item.get("description") ] def _first_item_name(items: list[dict[str, Any]], fallback: Any = None) -> Any: if items: return items[0].get("description") or fallback return fallback def _case_bid_type(mode: str | None) -> str: return { "GEM": "GeM Procurement", "E_TENDER": "Odisha e-Procurement", "GLOBAL_TENDER": "Global Tender", "LPC": "Local Purchase Committee", "OPEN_MARKET": "Open Market", "OTHER": "Other Procurement", }.get(_normalize_mode(mode), "Open Tender") def _auto_procurement_case_no(db: Session, requisition: Requisition, mode: str) -> str: requisition_part = (requisition.requisition_no or f"REQ-{requisition.id}").replace("/", "-") if requisition_part.upper().startswith("REQ-"): requisition_part = requisition_part[4:] base = f"PC-{mode}-{requisition_part}" case_no = base[:80] sequence = 2 while db.query(ProcurementCase).filter( ProcurementCase.procurement_case_no == case_no, ProcurementCase.requisition_id != requisition.id, ).first(): suffix = f"-{sequence}" case_no = f"{base[:80 - len(suffix)]}{suffix}" sequence += 1 return case_no def _mode_detail_defaults(mode: str, case: ProcurementCase, requisition: Requisition | None) -> dict[str, Any]: suffix = case.procurement_case_no.removeprefix("PC-") if mode == "GEM": return {"gem_procurement_type": "Product / Normal Category Bid"} if mode == "E_TENDER": return { "tender_ref_no": f"ET-{suffix}"[:120], "estimated_cost": case.approved_estimated_cost, } if mode == "GLOBAL_TENDER": return { "global_tender_ref_no": f"GT-{suffix}"[:120], "bid_value_inr": case.approved_estimated_cost, } if mode == "OPEN_MARKET": return {"procurement_justification": getattr(requisition, "remarks", None)} if mode == "OTHER": return { "other_procurement_type": case.procurement_mode_other, "procurement_reference_no": case.procurement_case_no, "procurement_description": getattr(requisition, "remarks", None), } return {} def _ensure_procurement_mode_detail( db: Session, case: ProcurementCase, requisition: Requisition | None = None, ) -> None: mode = _normalize_mode(case.procurement_mode) detail_model = PROCUREMENT_MODE_DETAIL_MODELS.get(mode) if not detail_model: return detail = db.query(detail_model).filter(detail_model.procurement_case_id == case.id).first() defaults = _mode_detail_defaults(mode, case, requisition) if detail is None: db.add(detail_model(procurement_case_id=case.id, **defaults)) return for key, value in defaults.items(): if hasattr(detail, key) and _is_blank(getattr(detail, key)): setattr(detail, key, value) def _sync_approved_requisition_procurement_case(db: Session, requisition: Requisition) -> None: if requisition.approval_status != ProcurementStatus.approved: return mode = _normalize_mode(requisition.procurement_mode) if mode not in ACTIVE_PROCUREMENT_MODES: return requisition.procurement_mode = mode case = ( db.query(ProcurementCase) .filter(ProcurementCase.requisition_id == requisition.id) .order_by(ProcurementCase.id.asc()) .first() ) if case is None: case = ProcurementCase( procurement_case_no=_auto_procurement_case_no(db, requisition, mode), requisition_id=requisition.id, procurement_mode=mode, procurement_mode_other=requisition.procurement_mode_other, ) db.add(case) case.requisition_id = requisition.id case.approved_quantity = requisition.administrative_approval_quantity or requisition.quantity case.approved_estimated_cost = requisition.estimated_cost case.procurement_mode = mode case.procurement_mode_other = requisition.procurement_mode_other if mode == "OTHER" else None if _is_blank(case.case_status): case.case_status = "In Tendering Process" if _is_blank(case.mode_selection_date): case.mode_selection_date = requisition.administrative_approval_date or date.today() if _is_blank(case.sanction_date): case.sanction_date = requisition.administrative_approval_date if _is_blank(case.remarks): case.remarks = f"Auto-created from approved requisition {requisition.requisition_no}." defaults = _apply_procurement_case_rules( { "procurement_mode": case.procurement_mode, "emd_applicable": case.emd_applicable, "pbg_applicable": case.pbg_applicable, } ) case.emd_applicable = bool(defaults["emd_applicable"]) case.pbg_applicable = bool(defaults["pbg_applicable"]) db.flush() _ensure_procurement_mode_detail(db, case, requisition) def _apply_requisition_to_procurement_case( payload: dict[str, Any], db: Session, current: ProcurementCase | None = None, ) -> dict[str, Any]: normalized = {**payload} if current is not None and "requisition_id" not in normalized: return normalized requisition_id = normalized.get("requisition_id") if _is_blank(requisition_id) and current: requisition_id = current.requisition_id if _is_blank(requisition_id): return normalized requisition = db.get(Requisition, int(requisition_id)) if not requisition: return normalized _fill_blank( normalized, "approved_quantity", requisition.administrative_approval_quantity or requisition.quantity, ) _fill_blank(normalized, "approved_estimated_cost", requisition.estimated_cost) _fill_blank(normalized, "procurement_mode", requisition.procurement_mode) _fill_blank(normalized, "procurement_mode_other", requisition.procurement_mode_other) _fill_blank(normalized, "case_status", "In Tendering Process") _fill_blank(normalized, "mode_selection_date", date.today()) return normalized def _procurement_context( payload: dict[str, Any], db: Session, current: Any = None, ) -> tuple[ProcurementCase | None, Requisition | None, TenderBid | None, PurchaseOrder | None]: procurement_case_id = payload.get("procurement_case_id") tender: TenderBid | None = None po: PurchaseOrder | None = None if _is_blank(procurement_case_id) and not _is_blank(payload.get("tender_id")): tender = db.get(TenderBid, int(payload["tender_id"])) procurement_case_id = tender.procurement_case_id if tender else None if _is_blank(procurement_case_id) and not _is_blank(payload.get("po_id")): po = db.get(PurchaseOrder, int(payload["po_id"])) procurement_case_id = po.procurement_case_id if po else None if _is_blank(procurement_case_id) and po and po.tender_id: tender = db.get(TenderBid, po.tender_id) procurement_case_id = tender.procurement_case_id if tender else None if _is_blank(procurement_case_id) and current is not None: procurement_case_id = getattr(current, "procurement_case_id", None) if _is_blank(procurement_case_id) and getattr(current, "tender_id", None): tender = db.get(TenderBid, current.tender_id) procurement_case_id = tender.procurement_case_id if tender else None if _is_blank(procurement_case_id) and getattr(current, "po_id", None): po = db.get(PurchaseOrder, current.po_id) procurement_case_id = po.procurement_case_id if po else None if _is_blank(procurement_case_id): return None, None, tender, po case = db.get(ProcurementCase, int(procurement_case_id)) requisition = db.get(Requisition, case.requisition_id) if case and case.requisition_id else None if tender is None and case: tender = db.query(TenderBid).filter(TenderBid.procurement_case_id == case.id).order_by(TenderBid.id.desc()).first() if po is None and case: po = db.query(PurchaseOrder).filter(PurchaseOrder.procurement_case_id == case.id).order_by(PurchaseOrder.id.desc()).first() return case, requisition, tender, po def _apply_procurement_case_to_process_payload( stage: str, payload: dict[str, Any], db: Session, current: Any = None, ) -> dict[str, Any]: normalized = {**payload} if current is not None and not {"procurement_case_id", "tender_id", "po_id"}.intersection(normalized): return normalized case, requisition, tender, po = _procurement_context(normalized, db, current) if not case: return normalized _fill_blank(normalized, "procurement_case_id", case.id) tender_items = _requisition_items_for_process(requisition) item_name = _first_item_name(tender_items, getattr(tender, "item_name", None) or getattr(requisition, "item_description", None)) quantity = case.approved_quantity or getattr(requisition, "administrative_approval_quantity", None) or getattr(requisition, "quantity", None) estimated_cost = case.approved_estimated_cost or getattr(requisition, "estimated_cost", None) budget_head = getattr(requisition, "budget_head", None) if stage == "tenders": _fill_blank(normalized, "requisition_id", case.requisition_id) _fill_blank(normalized, "title", f"{case.procurement_case_no} - {item_name}" if item_name else case.procurement_case_no) _fill_blank(normalized, "tender_items", tender_items) _fill_blank(normalized, "item_name", item_name) _fill_blank(normalized, "quantity", quantity) _fill_blank(normalized, "estimated_cost", estimated_cost) _fill_blank(normalized, "budget_head", budget_head) _fill_blank(normalized, "tender_date", case.mode_selection_date) _fill_blank(normalized, "bid_type", _case_bid_type(case.procurement_mode)) return normalized if tender: _fill_blank(normalized, "tender_id", tender.id) _fill_blank(normalized, "tender_number", tender.tender_no) _fill_blank(normalized, "tender_date", tender.tender_date) if stage == "emds": _fill_blank(normalized, "entry_date", date.today()) _fill_blank(normalized, "item_name", getattr(tender, "item_name", None) or item_name) if case.emd_applicable: _fill_blank(normalized, "amount", (Decimal(estimated_cost or 0) * Decimal("0.02")).quantize(Decimal("0.01"))) return normalized if stage == "samples": sample_items = _sample_items_from_tender(tender, requisition) _fill_blank(normalized, "sample_items", sample_items) _fill_blank(normalized, "item_name", _first_item_name(sample_items, item_name)) _fill_blank(normalized, "sample_description", _first_item_name(sample_items, item_name)) _fill_blank(normalized, "quantity", quantity) return normalized if stage in {"technical-evaluations", "financial-evaluations"}: if stage == "financial-evaluations": _fill_blank(normalized, "quoted_amount", estimated_cost) return normalized if stage == "purchase-orders": _fill_blank(normalized, "tender_id", getattr(tender, "id", None)) _fill_blank(normalized, "procurement_mode", case.procurement_mode) _fill_blank(normalized, "procurement_mode_other", case.procurement_mode_other) _fill_blank(normalized, "tender_reference_no", getattr(tender, "tender_no", None) or case.procurement_case_no) _fill_blank(normalized, "tender_reference_date", getattr(tender, "tender_date", None) or case.mode_selection_date) _fill_blank(normalized, "estimated_cost", estimated_cost) _fill_blank(normalized, "process_initiate_date", case.mode_selection_date) _fill_blank(normalized, "budget_head", budget_head) _fill_blank(normalized, "item_description", item_name) _fill_blank(normalized, "quantity", quantity) _fill_blank(normalized, "po_value", estimated_cost) return normalized if po: _fill_blank(normalized, "po_id", po.id) if stage == "performance-securities": _fill_blank(normalized, "entry_date", date.today()) _fill_blank(normalized, "contract_number", getattr(po, "po_no", None) or case.procurement_case_no) _fill_blank(normalized, "contract_date", getattr(po, "po_date", None) or case.sanction_date) _fill_blank(normalized, "item_name", getattr(po, "item_description", None) or item_name) _fill_blank(normalized, "quantity", getattr(po, "quantity", None) or quantity) _fill_blank(normalized, "vendor_name", getattr(po, "vendor_name", None)) _fill_blank(normalized, "bidder_name", getattr(po, "vendor_name", None)) if case.pbg_applicable: base_value = getattr(po, "po_value", None) or estimated_cost or 0 _fill_blank(normalized, "amount", (Decimal(base_value) * Decimal("0.10")).quantize(Decimal("0.01"))) return normalized if stage == "delivery-acceptance": _fill_blank(normalized, "accepted_quantity", getattr(po, "quantity", None) or quantity or 0) return normalized if stage == "payments": _fill_blank(normalized, "invoice_amount", getattr(po, "po_value", None) or estimated_cost) _fill_blank(normalized, "paid_amount", Decimal("0.00")) return normalized return normalized def _normalize_sample_register_payload(payload: dict[str, Any]) -> dict[str, Any]: items = payload.get("sample_items") if not isinstance(items, list): return payload cleaned_items: list[dict[str, Any]] = [] for index, item in enumerate(items): if not isinstance(item, dict): continue cleaned = { key: value for key, value in item.items() if value is not None and value != "" } if not any(key != "serial_no" for key in cleaned): continue make_model = cleaned.get("make_model") if not make_model: make_model = " ".join(str(value).strip() for value in (cleaned.get("make"), cleaned.get("model")) if value) if make_model: cleaned["make_model"] = make_model cleaned.setdefault("serial_no", index + 1) cleaned_items.append(cleaned) normalized = {**payload, "sample_items": cleaned_items} if not cleaned_items: return normalized first_item = cleaned_items[0] normalized.setdefault("item_name", first_item.get("description")) normalized.setdefault("sample_description", first_item.get("description")) normalized.setdefault("make", first_item.get("make")) normalized.setdefault("model", first_item.get("model")) normalized.setdefault("make_model", first_item.get("make_model")) quantities: list[int] = [] for item in cleaned_items: try: quantities.append(int(item.get("quantity") or 0)) except (TypeError, ValueError): continue total_quantity = sum(quantities) if total_quantity > 0 and normalized.get("quantity") in (None, ""): normalized["quantity"] = total_quantity return normalized def _normalize_tender_payload(payload: dict[str, Any]) -> dict[str, Any]: items = payload.get("tender_items") if not isinstance(items, list): return payload cleaned_items: list[dict[str, Any]] = [] for index, item in enumerate(items): if not isinstance(item, dict): continue cleaned = { key: value for key, value in item.items() if value is not None and value != "" } if not any(key != "serial_no" for key in cleaned): continue cleaned.setdefault("serial_no", index + 1) cleaned_items.append(cleaned) normalized = {**payload, "tender_items": cleaned_items} if not cleaned_items: return normalized first_item = cleaned_items[0] normalized.setdefault("item_name", first_item.get("description")) quantities: list[int] = [] for item in cleaned_items: try: quantities.append(int(item.get("quantity") or 0)) except (TypeError, ValueError): continue total_quantity = sum(quantities) if total_quantity > 0 and normalized.get("quantity") in (None, ""): normalized["quantity"] = total_quantity return normalized def _next_requisition_no(db: Session) -> str: year = date.today().year prefix = f"REQ-{year}-" existing = db.query(Requisition.requisition_no).filter( Requisition.requisition_no.like(f"{prefix}%") ).all() max_sequence = 0 for (requisition_no,) in existing: if not requisition_no or not requisition_no.startswith(prefix): continue sequence = requisition_no.removeprefix(prefix) if len(sequence) == 4 and sequence.isdigit(): max_sequence = max(max_sequence, int(sequence)) next_sequence = max_sequence + 1 if next_sequence > 9999: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Requisition number sequence for {year} has reached 9999.", ) return f"{prefix}{next_sequence:04d}" def _commit_or_conflict(db: Session) -> None: try: db.commit() except IntegrityError as exc: db.rollback() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Record could not be saved because it conflicts with existing data.", ) from exc def _flush_or_conflict(db: Session) -> None: try: db.flush() except IntegrityError as exc: db.rollback() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Record could not be saved because it conflicts with existing data.", ) from exc @router.get("/stages") def list_stages(current_user: Annotated[User, Depends(get_current_user)]): hidden_stages = {"demands", "item-database"} return [ {"key": k, "label": k.replace("-", " ").title()} for k in STAGES if k not in hidden_stages and has_functionality_access(current_user, stage_functionality(k)) ] @router.get("/requisitions/next-number", dependencies=[Depends(require_access("requisitions"))]) def next_requisition_number(db: Annotated[Session, Depends(get_db)]): return {"requisition_no": _next_requisition_no(db)} @router.get("/{stage}", response_model=Page, dependencies=[Depends(require_stage_access)]) def list_records( stage: str, db: Annotated[Session, Depends(get_db)], language: Annotated[Language, Depends(get_language)], page: int = Query(1, ge=1), size: int = Query(10, ge=1, le=1000), search: str | None = None, ): config = get_stage(stage, language) query = db.query(config.model) if stage == "procurement-cases": query = query.filter(ProcurementCase.procurement_mode.in_(ACTIVE_PROCUREMENT_MODES)) expected_mode = PROCUREMENT_TYPE_STAGE_MODES.get(stage) if expected_mode: query = query.join( ProcurementCase, config.model.procurement_case_id == ProcurementCase.id, ).filter(ProcurementCase.procurement_mode == expected_mode) if search: conditions = [ cast(getattr(config.model, f), String).like(f"%{search}%") for f in config.search_fields ] query = query.filter(or_(*conditions)) total = query.count() rows = query.order_by(config.model.id.desc()).offset((page - 1) * size).limit(size).all() items = [config.read_schema.model_validate(r).model_dump(mode="json") for r in rows] return Page(total=total, page=page, size=size, items=items) @router.get("/{stage}/{record_id}", dependencies=[Depends(require_stage_access)]) def get_record( stage: str, record_id: int, db: Annotated[Session, Depends(get_db)], language: Annotated[Language, Depends(get_language)], ): config = get_stage(stage, language) row = db.get(config.model, record_id) if not row: raise HTTPException(status_code=404, detail=translate("procurement.not_found", language)) if stage == "procurement-cases" and row.procurement_mode not in ACTIVE_PROCUREMENT_MODES: raise HTTPException(status_code=404, detail=translate("procurement.not_found", language)) expected_mode = PROCUREMENT_TYPE_STAGE_MODES.get(stage) if expected_mode: case = db.get(ProcurementCase, row.procurement_case_id) if not case or case.procurement_mode != expected_mode: raise HTTPException(status_code=404, detail=translate("procurement.not_found", language)) return config.read_schema.model_validate(row) @router.post("/{stage}", dependencies=[WRITER, Depends(require_stage_access)]) def create_record( stage: str, payload: dict[str, Any], db: Annotated[Session, Depends(get_db)], language: Annotated[Language, Depends(get_language)], ): config = get_stage(stage, language) if stage == "requisitions": payload = {**payload} if not str(payload.get("requisition_no") or "").strip(): payload["requisition_no"] = _next_requisition_no(db) payload = _apply_requisition_mode_rules(payload) if stage in {*PROCUREMENT_TYPE_STAGE_MODES.keys(), *PROCESS_STAGES}: payload = _resolve_manual_procurement_case(stage, payload, db) if stage == "procurement-cases": payload = _apply_requisition_to_procurement_case(payload, db) payload = _apply_procurement_case_rules(payload) if stage in PROCESS_STAGES: payload = _apply_procurement_case_to_process_payload(stage, payload, db) if stage == "samples": payload = _normalize_sample_register_payload(payload) if stage == "tenders": payload = _normalize_tender_payload(payload) data = _validate(config.create_schema, payload) _validate_procurement_case_reference(stage, data) _validate_parent_refs(db, config, data) _validate_procurement_type_parent(db, stage, data) _validate_procurement_register_status(stage, data.model_dump()) if stage == "requisitions": _validate_requisition_approval_quantity(data.model_dump()) _validate_requisition_approval_status(data.model_dump()) row = config.model(**data.model_dump()) db.add(row) if stage in {"requisitions", "procurement-cases"}: _flush_or_conflict(db) if stage == "requisitions": _sync_approved_requisition_procurement_case(db, row) if stage == "procurement-cases": requisition = db.get(Requisition, row.requisition_id) if row.requisition_id else None _ensure_procurement_mode_detail(db, row, requisition) _commit_or_conflict(db) db.refresh(row) return config.read_schema.model_validate(row) @router.put("/{stage}/{record_id}", dependencies=[WRITER, Depends(require_stage_access)]) def update_record( stage: str, record_id: int, payload: dict[str, Any], db: Annotated[Session, Depends(get_db)], language: Annotated[Language, Depends(get_language)], ): config = get_stage(stage, language) row = db.get(config.model, record_id) if not row: raise HTTPException(status_code=404, detail=translate("procurement.not_found", language)) if stage == "procurement-cases" and row.procurement_mode not in ACTIVE_PROCUREMENT_MODES: raise HTTPException(status_code=404, detail=translate("procurement.not_found", language)) if stage == "procurement-cases": payload = _apply_requisition_to_procurement_case(payload, db, row) payload = _apply_procurement_case_rules(payload, getattr(row, "procurement_mode", None)) if stage == "requisitions": payload = _apply_requisition_mode_rules(payload, getattr(row, "procurement_mode", None)) if stage in {*PROCUREMENT_TYPE_STAGE_MODES.keys(), *PROCESS_STAGES}: payload = _resolve_manual_procurement_case(stage, payload, db) if stage in PROCESS_STAGES: payload = _apply_procurement_case_to_process_payload(stage, payload, db, row) if stage == "samples": payload = _normalize_sample_register_payload(payload) if stage == "tenders": payload = _normalize_tender_payload(payload) data = _validate(config.update_schema, payload) if stage in {*PROCUREMENT_TYPE_STAGE_MODES.keys(), *PROCESS_STAGES}: effective_case_id = data.model_dump(exclude_unset=True).get( "procurement_case_id", getattr(row, "procurement_case_id", None) ) if effective_case_id is None: _validate_procurement_case_reference(stage, data) _validate_parent_refs(db, config, data, exclude_unset=True) _validate_procurement_type_parent(db, stage, data) if stage in {"procurement-cases", *PROCUREMENT_TYPE_STAGE_MODES.keys()}: current = config.read_schema.model_validate(row).model_dump() current.update(data.model_dump(exclude_unset=True)) _validate_procurement_register_status(stage, current) if stage == "requisitions": current = config.read_schema.model_validate(row).model_dump() current.update(data.model_dump(exclude_unset=True)) _validate_requisition_approval_quantity(current) _validate_requisition_approval_status(current) for key, value in data.model_dump(exclude_unset=True).items(): setattr(row, key, value) if stage in {"requisitions", "procurement-cases"}: _flush_or_conflict(db) if stage == "requisitions": _sync_approved_requisition_procurement_case(db, row) if stage == "procurement-cases": requisition = db.get(Requisition, row.requisition_id) if row.requisition_id else None _ensure_procurement_mode_detail(db, row, requisition) _commit_or_conflict(db) db.refresh(row) return config.read_schema.model_validate(row) @router.delete( "/{stage}/{record_id}", dependencies=[Depends(require_roles(Role.admin, Role.procurement_officer)), Depends(require_stage_access)], ) def delete_record( stage: str, record_id: int, db: Annotated[Session, Depends(get_db)], language: Annotated[Language, Depends(get_language)], ): config = get_stage(stage, language) row = db.get(config.model, record_id) if not row: raise HTTPException(status_code=404, detail=translate("procurement.not_found", language)) if stage == "procurement-cases" and row.procurement_mode not in ACTIVE_PROCUREMENT_MODES: raise HTTPException(status_code=404, detail=translate("procurement.not_found", language)) db.delete(row) _commit_or_conflict(db) return {"message": translate("procurement.deleted", language)}