from __future__ import annotations from calendar import month_name from datetime import date, datetime, time, timedelta from io import BytesIO from pathlib import Path from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, Query, Response from sqlalchemy import func, or_ from sqlalchemy.orm import Session from app.api.dashboard import ACTIVE_PROCUREMENT_MODES, MODULES, MODULE_CONFIG, _analytics_period, _analytics_query, _breakdown from app.api.deps import require_access from app.db.session import get_db from app.models.enums import ProcurementStatus from app.models.master_data import ReportBudgetAllocation from app.models.procurement import ( DeliveryAcceptance, ETenderCase, EMD, FinancialEvaluation, GeMProcurementCase, GlobalTenderCase, LPCProcurementCase, OpenMarketCase, OtherProcurementCase, PaymentTracking, PerformanceSecurity, ProcurementCase, PurchaseOrder, Requisition, SampleRegister, TechnicalEvaluation, TenderBid, ) router = APIRouter(dependencies=[Depends(require_access("reports"))]) REPORT_TITLES = { "summary": "Summary Report", "daily": "Daily Procurement Status Report", "establishment": "Establishment-wise Requisition Record", "monthly": "Monthly Procurement Report", "vendor-payment-status": "Vendor Payment Status Report", "budget-utilization": "Budget Utilization Report", "program-expenditure": "Programme Expenditure Report", "cumulative-activity": "Cumulative Activity Report", "yearly": "Yearly Procurement Report", "gem-register": "GeM Procurement Register", "gem-order-status": "GeM Order Status Report", "e-tender-register": "e-Tender Register", "technical-evaluation-status": "Technical Evaluation Status Report", "financial-evaluation-status": "Financial Evaluation Status Report", "global-tender-register": "Global Tender Register", "foreign-vendor-procurement": "Foreign Vendor Procurement Report", "lpc-register": "LPC Register", "lpc-recommendation": "LPC Recommendation Report", "open-market-register": "Open Market Procurement Register", "open-market-justification": "Open Market Justification Report", "other-procurement-register": "Other Procurement Register", "emd-register": "EMD Register", "emd-pending-refund": "EMD Pending Refund", "emd-forfeiture-register": "EMD Forfeiture Register", "performance-security-register": "Performance Security Register", "pbg-expiry-alert": "PBG Expiry Alert", "contract-expiry-alert": "Contract Expiry Alert", "budget-head-expenditure": "Budget Head-wise Expenditure", "procurement-value-analysis": "Procurement Value Analysis", "vendor-procurement-value": "Vendor-wise Procurement Value", "financial-year-summary": "Financial Year-wise Procurement Summary", "district-procurement-summary": "District-wise Procurement Summary", "circle-procurement-summary": "Circle-wise Procurement Summary", "range-procurement-summary": "Range-wise Procurement Summary", "directorate-dashboard": "Directorate Procurement Dashboard", "cycle-time-analysis": "Procurement Cycle Time Analysis", "procurement-aging-analysis": "Procurement Aging Analysis", "top-vendors": "Top Vendors Report", "asset-procurement-distribution": "Asset Procurement & Distribution Report", } CUSTOM_REPORT_SOURCES: dict[str, dict[str, Any]] = { "requisitions": { "title": "Requisitions", "model": Requisition, "date_attr": "requisition_date", "fields": [ ("requisition_no", "Requisition No", "text"), ("requisition_date", "Requisition Date", "text"), ("received_from", "Received From", "text"), ("received_from_district", "District", "text"), ("received_from_range", "Range", "text"), ("item_description", "Item Description", "text"), ("quantity", "Quantity", "number"), ("estimated_cost", "Estimated Cost", "currency"), ("procurement_mode", "Procurement Mode", "text"), ("procurement_mode_other", "Other Procurement Mode", "text"), ("budget_head", "Budget Head", "text"), ("approval_status", "Approval Status", "text"), ], }, "procurement-cases": { "title": "Procurement Cases", "model": ProcurementCase, "date_attr": "mode_selection_date", "fields": [ ("procurement_case_no", "Procurement Case No", "text"), ("procurement_mode", "Procurement Mode", "text"), ("procurement_mode_other", "Other Procurement Mode", "text"), ("mode_selection_date", "Mode Selection Date", "text"), ("approved_quantity", "Approved Quantity", "number"), ("approved_estimated_cost", "Approved Estimated Cost", "currency"), ("case_status", "Case Status", "text"), ("sanction_no", "Sanction No", "text"), ("sanction_date", "Sanction Date", "text"), ("approving_authority", "Approving Authority", "text"), ], }, "tenders": { "title": "Tender / Bid Register", "model": TenderBid, "date_attr": "tender_date", "fields": [ ("tender_no", "Tender No", "text"), ("tender_date", "Tender Date", "text"), ("item_name", "Item Name", "text"), ("quantity", "Quantity", "number"), ("estimated_cost", "Estimated Cost", "currency"), ("budget_head", "Budget Head", "text"), ("bid_published_date", "Bid Published Date", "text"), ("bid_closed_date", "Bid Closed Date", "text"), ("l1_bidder_name", "L1 Bidder", "text"), ("amount", "Amount", "currency"), ("status", "Status", "text"), ], }, "emds": { "title": "EMD Register", "model": EMD, "date_attr": "entry_date", "fields": [ ("emd_unique_id", "EMD ID", "text"), ("entry_date", "Entry Date", "text"), ("tender_number", "Tender No", "text"), ("bidder_name", "Bidder", "text"), ("emd_no", "EMD No", "text"), ("amount", "Amount", "currency"), ("expiry_date", "Expiry Date", "text"), ("refund_status", "Refund Status", "text"), ], }, "samples": { "title": "Sample Register", "model": SampleRegister, "date_attr": "receipt_date", "fields": [ ("tender_number", "Tender No", "text"), ("item_name", "Item Name", "text"), ("quantity", "Quantity", "number"), ("bidder_name", "Bidder", "text"), ("receipt_date", "Receipt Date", "text"), ("recipient_name", "Recipient", "text"), ("evaluation_status", "Evaluation Status", "text"), ], }, "purchase-orders": { "title": "Purchase Orders", "model": PurchaseOrder, "date_attr": "po_date", "fields": [ ("po_no", "PO No", "text"), ("po_date", "PO Date", "text"), ("vendor_name", "Vendor", "text"), ("item_description", "Item Description", "text"), ("quantity", "Quantity", "number"), ("po_value", "PO Value", "currency"), ("budget_head", "Budget Head", "text"), ("delivery_due_date", "Delivery Due Date", "text"), ("status", "Status", "text"), ], }, "performance-securities": { "title": "Performance Securities", "model": PerformanceSecurity, "date_attr": "entry_date", "fields": [ ("contract_number", "Contract No", "text"), ("vendor_name", "Vendor", "text"), ("security_number", "Security No", "text"), ("security_date", "Security Date", "text"), ("amount", "Amount", "currency"), ("expiry_date", "Expiry Date", "text"), ("release_status", "Release Status", "text"), ], }, "delivery-acceptance": { "title": "Delivery / Acceptance", "model": DeliveryAcceptance, "date_attr": "delivery_date", "fields": [ ("po_id", "PO ID", "number"), ("delivery_date", "Delivery Date", "text"), ("accepted_quantity", "Accepted Quantity", "number"), ("acceptance_status", "Acceptance Status", "text"), ("remarks", "Remarks", "text"), ], }, "payments": { "title": "Payments", "model": PaymentTracking, "date_attr": "invoice_date", "fields": [ ("po_id", "PO ID", "number"), ("invoice_no", "Invoice No", "text"), ("invoice_date", "Invoice Date", "text"), ("invoice_amount", "Invoice Amount", "currency"), ("paid_amount", "Paid Amount", "currency"), ("payment_status", "Payment Status", "text"), ], }, } CUSTOM_CASE_REFERENCE_MODELS = { TenderBid, EMD, SampleRegister, PurchaseOrder, PerformanceSecurity, DeliveryAcceptance, PaymentTracking, } for custom_source in CUSTOM_REPORT_SOURCES.values(): if custom_source["model"] in CUSTOM_CASE_REFERENCE_MODELS: custom_source["fields"].insert(0, ("procurement_case_reference", "Procurement Case", "text")) def _logo_path() -> Path: return Path(__file__).resolve().parents[3] / "frontend" / "src" / "assets" / "ofs-logo.png" def _money(value: Any) -> float: return float(value or 0) def _column(key: str, label: str, kind: str = "text") -> dict[str, str]: return {"key": key, "label": label, "kind": kind} REPORT_SPECS: dict[str, dict[str, Any]] = { "gem-register": { "model": GeMProcurementCase, "date_attr": "gem_bid_date", "columns": [ ("id", "ID", "number"), ("procurement_case_id", "Case ID", "number"), ("gem_procurement_type", "GeM Type", "text"), ("gem_bid_no", "GeM Bid No", "text"), ("gem_bid_date", "Bid Date", "text"), ("gem_contract_no", "Contract No", "text"), ("gem_order_value", "Order Value", "currency"), ("status", "Status", "text"), ], }, "gem-order-status": { "model": GeMProcurementCase, "date_attr": "gem_contract_date", "columns": [ ("gem_contract_no", "Contract No", "text"), ("gem_contract_date", "Contract Date", "text"), ("gem_order_value", "Order Value", "currency"), ("delivery_status", "Delivery Status", "text"), ("payment_status", "Payment Status", "text"), ("status", "Status", "text"), ], }, "e-tender-register": { "model": ETenderCase, "date_attr": "tender_publication_date", "columns": [ ("tender_ref_no", "Tender Ref No", "text"), ("tender_publication_date", "Publication Date", "text"), ("bid_submission_end_date", "Submission End Date", "text"), ("estimated_cost", "Estimated Cost", "currency"), ("l1_vendor_id", "L1 Vendor ID", "number"), ("contract_no", "Contract No", "text"), ("status", "Status", "text"), ], }, "technical-evaluation-status": { "model": ETenderCase, "date_attr": "technical_bid_opening_date", "columns": [ ("tender_ref_no", "Tender Ref No", "text"), ("technical_bid_opening_date", "Technical Opening", "text"), ("technical_evaluation_status", "Technical Status", "text"), ("status", "Case Status", "text"), ], }, "financial-evaluation-status": { "model": ETenderCase, "date_attr": "financial_bid_opening_date", "columns": [ ("tender_ref_no", "Tender Ref No", "text"), ("financial_bid_opening_date", "Financial Opening", "text"), ("financial_evaluation_status", "Financial Status", "text"), ("l1_vendor_id", "L1 Vendor ID", "number"), ("status", "Case Status", "text"), ], }, "global-tender-register": { "model": GlobalTenderCase, "date_attr": "approval_date", "columns": [ ("global_tender_ref_no", "Global Tender Ref No", "text"), ("country_of_vendor", "Vendor Country", "text"), ("foreign_vendor_name", "Foreign Vendor", "text"), ("foreign_currency", "Currency", "text"), ("bid_value_inr", "Bid Value INR", "currency"), ("contract_no", "Contract No", "text"), ("status", "Status", "text"), ], }, "foreign-vendor-procurement": { "model": GlobalTenderCase, "date_attr": "approval_date", "columns": [ ("foreign_vendor_name", "Foreign Vendor", "text"), ("country_of_vendor", "Country", "text"), ("foreign_currency", "Currency", "text"), ("bid_value_foreign_currency", "Foreign Value", "number"), ("exchange_rate", "Exchange Rate", "number"), ("bid_value_inr", "INR Value", "currency"), ], }, "lpc-register": { "model": LPCProcurementCase, "date_attr": "lpc_meeting_date", "columns": [ ("lpc_meeting_no", "Meeting No", "text"), ("lpc_meeting_date", "Meeting Date", "text"), ("quotation_count", "Quotations", "number"), ("lowest_vendor_id", "Lowest Vendor ID", "number"), ("approval_date", "Approval Date", "text"), ("status", "Status", "text"), ], }, "lpc-recommendation": { "model": LPCProcurementCase, "date_attr": "recommendation_date", "columns": [ ("lpc_meeting_no", "Meeting No", "text"), ("committee_members", "Committee Members", "text"), ("quotation_count", "Quotations", "number"), ("comparative_statement_path", "Comparative Statement", "text"), ("recommendation_date", "Recommendation Date", "text"), ("status", "Status", "text"), ], }, "open-market-register": { "model": OpenMarketCase, "date_attr": "market_survey_date", "columns": [ ("market_survey_date", "Market Survey Date", "text"), ("quotation_count", "Quotations", "number"), ("selected_vendor_id", "Selected Vendor ID", "number"), ("approval_date", "Approval Date", "text"), ("status", "Status", "text"), ], }, "open-market-justification": { "model": OpenMarketCase, "date_attr": "approval_date", "columns": [ ("procurement_justification", "Justification", "text"), ("market_survey_date", "Market Survey Date", "text"), ("quotation_count", "Quotations", "number"), ("comparative_statement_path", "Comparative Statement", "text"), ("status", "Status", "text"), ], }, "other-procurement-register": { "model": OtherProcurementCase, "date_attr": "approval_date", "columns": [ ("other_procurement_type", "Procurement Type", "text"), ("procurement_reference_no", "Reference No", "text"), ("procurement_description", "Description", "text"), ("vendor_id", "Vendor ID", "number"), ("approval_date", "Approval Date", "text"), ("status", "Status", "text"), ], }, "emd-register": { "model": EMD, "date_attr": "entry_date", "columns": [ ("emd_unique_id", "EMD ID", "text"), ("tender_number", "Tender No", "text"), ("bidder_name", "Bidder", "text"), ("amount", "Amount", "currency"), ("expiry_date", "Expiry Date", "text"), ("refund_status", "Refund Status", "text"), ], }, "emd-pending-refund": { "model": EMD, "date_attr": "entry_date", "filters": [("refund_status", {"Pending", "Pending Return", "Held"})], "columns": [ ("emd_unique_id", "EMD ID", "text"), ("tender_number", "Tender No", "text"), ("bidder_name", "Bidder", "text"), ("amount", "Amount", "currency"), ("expiry_date", "Expiry Date", "text"), ("refund_status", "Refund Status", "text"), ], }, "emd-forfeiture-register": { "model": EMD, "date_attr": "entry_date", "filters": [("refund_status", {"Forfeited"})], "columns": [ ("emd_unique_id", "EMD ID", "text"), ("tender_number", "Tender No", "text"), ("bidder_name", "Bidder", "text"), ("amount", "Amount", "currency"), ("remarks", "Remarks", "text"), ], }, "performance-security-register": { "model": PerformanceSecurity, "date_attr": "entry_date", "columns": [ ("contract_number", "Contract No", "text"), ("vendor_name", "Vendor", "text"), ("security_number", "Security No", "text"), ("amount", "Amount", "currency"), ("expiry_date", "Expiry Date", "text"), ("release_status", "Release Status", "text"), ], }, "pbg-expiry-alert": { "model": PerformanceSecurity, "date_attr": "expiry_date", "columns": [ ("contract_number", "Contract No", "text"), ("vendor_name", "Vendor", "text"), ("security_number", "Security No", "text"), ("amount", "Amount", "currency"), ("expiry_date", "Expiry Date", "text"), ("release_status", "Release Status", "text"), ], }, "contract-expiry-alert": { "model": PurchaseOrder, "date_attr": "delivery_due_date", "columns": [ ("po_no", "PO No", "text"), ("vendor_name", "Vendor", "text"), ("item_description", "Item", "text"), ("po_value", "PO Value", "currency"), ("delivery_due_date", "Delivery Due Date", "text"), ("status", "Status", "text"), ], }, "budget-head-expenditure": { "model": PurchaseOrder, "date_attr": "po_date", "columns": [("budget_head", "Budget Head", "text"), ("po_no", "PO No", "text"), ("vendor_name", "Vendor", "text"), ("po_value", "PO Value", "currency"), ("status", "Status", "text")], }, "procurement-value-analysis": { "model": ProcurementCase, "date_attr": "mode_selection_date", "columns": [("procurement_case_no", "Case No", "text"), ("procurement_mode", "Mode", "text"), ("approved_estimated_cost", "Approved Cost", "currency"), ("case_status", "Status", "text")], }, "vendor-procurement-value": { "model": PurchaseOrder, "date_attr": "po_date", "columns": [("vendor_name", "Vendor", "text"), ("po_no", "PO No", "text"), ("procurement_mode", "Mode", "text"), ("po_value", "PO Value", "currency"), ("status", "Status", "text")], }, "financial-year-summary": { "model": PurchaseOrder, "date_attr": "po_date", "columns": [("po_date", "PO Date", "text"), ("po_no", "PO No", "text"), ("budget_head", "Budget Head", "text"), ("po_value", "PO Value", "currency"), ("status", "Status", "text")], }, "district-procurement-summary": { "model": Requisition, "date_attr": "requisition_date", "columns": [("received_from_district", "District", "text"), ("requisition_no", "Requisition No", "text"), ("procurement_mode", "Mode", "text"), ("estimated_cost", "Estimated Cost", "currency"), ("approval_status", "Status", "text")], }, "circle-procurement-summary": { "model": Requisition, "date_attr": "requisition_date", "columns": [("received_from", "Circle / Establishment", "text"), ("requisition_no", "Requisition No", "text"), ("procurement_mode", "Mode", "text"), ("estimated_cost", "Estimated Cost", "currency"), ("approval_status", "Status", "text")], }, "range-procurement-summary": { "model": Requisition, "date_attr": "requisition_date", "columns": [("received_from_range", "Range", "text"), ("requisition_no", "Requisition No", "text"), ("procurement_mode", "Mode", "text"), ("estimated_cost", "Estimated Cost", "currency"), ("approval_status", "Status", "text")], }, "directorate-dashboard": { "model": ProcurementCase, "date_attr": "mode_selection_date", "columns": [("procurement_case_no", "Case No", "text"), ("procurement_mode", "Mode", "text"), ("approving_authority", "Approving Authority", "text"), ("approved_estimated_cost", "Approved Cost", "currency"), ("case_status", "Status", "text")], }, "cycle-time-analysis": { "model": ProcurementCase, "date_attr": "mode_selection_date", "columns": [("procurement_case_no", "Case No", "text"), ("mode_selection_date", "Mode Selection Date", "text"), ("sanction_date", "Sanction Date", "text"), ("procurement_mode", "Mode", "text"), ("case_status", "Status", "text")], }, "procurement-aging-analysis": { "model": ProcurementCase, "date_attr": "mode_selection_date", "columns": [("procurement_case_no", "Case No", "text"), ("mode_selection_date", "Mode Selection Date", "text"), ("procurement_mode", "Mode", "text"), ("case_status", "Status", "text")], }, "top-vendors": { "model": PurchaseOrder, "date_attr": "po_date", "columns": [("vendor_name", "Vendor", "text"), ("po_no", "PO No", "text"), ("procurement_mode", "Mode", "text"), ("po_value", "PO Value", "currency"), ("status", "Status", "text")], }, "asset-procurement-distribution": { "model": PurchaseOrder, "date_attr": "po_date", "columns": [("item_description", "Asset / Item", "text"), ("quantity", "Quantity", "number"), ("unit", "Unit", "text"), ("vendor_name", "Vendor", "text"), ("po_value", "PO Value", "currency"), ("status", "Status", "text")], }, } REPORT_CASE_REFERENCE_MODELS = { GeMProcurementCase, ETenderCase, GlobalTenderCase, LPCProcurementCase, OpenMarketCase, OtherProcurementCase, TenderBid, EMD, SampleRegister, TechnicalEvaluation, FinancialEvaluation, PurchaseOrder, PerformanceSecurity, DeliveryAcceptance, PaymentTracking, } for report_spec in REPORT_SPECS.values(): if report_spec["model"] in REPORT_CASE_REFERENCE_MODELS: report_spec["columns"] = [ column for column in report_spec["columns"] if column[0] != "procurement_case_id" ] report_spec["columns"].insert(0, ("procurement_case_reference", "Procurement Case", "text")) def _report_filter( report_key: str, period: str, start_date: date | None, end_date: date | None, month: int | None, year: int | None, ) -> tuple[date | None, date | None, str]: if report_key == "cumulative-activity": effective_end = end_date or date.today() effective_start = start_date or (effective_end - timedelta(days=6)) if effective_start > effective_end: raise HTTPException(status_code=400, detail="From date cannot be later than To date.") return effective_start, effective_end, f"{effective_start:%d %b %Y} - {effective_end:%d %b %Y}" if report_key == "monthly": selected_year = year or date.today().year return date(selected_year, 1, 1), date(selected_year, 12, 31), f"Calendar Year {selected_year}" if report_key == "yearly": return None, None, "All Available Years" return _analytics_period(period, start_date, end_date, month, year) def _base_report(report_key: str, period_label: str) -> dict[str, Any]: return { "key": report_key, "organization": "Odisha Fire and Emergency Services", "title": REPORT_TITLES[report_key], "period_label": period_label, "generated_at": datetime.now().astimezone().isoformat(timespec="minutes"), "footer_lines": ["Computer Generated report", "For the use of Procurement Cell"], } def _summary_report(db: Session, start_date: date | None, end_date: date | None, period_label: str) -> dict[str, Any]: rows = [] total_records = 0 requisition_query = _analytics_query(db, Requisition, MODULE_CONFIG["requisitions"], start_date, end_date) total_estimated_cost = _money( requisition_query.with_entities(func.coalesce(func.sum(Requisition.estimated_cost), 0)).scalar() ) for key, label, model, _ in MODULES: config = MODULE_CONFIG[key] query = _analytics_query(db, model, config, start_date, end_date) count = query.count() status_items = _breakdown(query, model, config["status_column"]) metric_values = [] for metric_label, expression, kind, aggregate in config["metrics"]: if aggregate == "count": calculator = func.count(expression) else: calculator = func.avg(expression) if aggregate == "avg" else func.sum(expression) value = query.with_entities(func.coalesce(calculator, 0)).scalar() prefix = "Rs. " if kind == "currency" else "" metric_values.append(f"{metric_label}: {prefix}{float(value or 0):,.2f}") rows.append( { "module": label, "records": count, "status": ", ".join(f"{item['label']}: {item['count']}" for item in status_items) or "No records", "indicator": "; ".join(metric_values), } ) total_records += count return { **_base_report("summary", period_label), "columns": [ _column("module", "Module"), _column("records", "Total Records", "number"), _column("status", "Status / Result Breakdown"), _column("indicator", "Value / Quantity Indicator"), ], "rows": rows, "summary": [ {"label": "Total Workflow Entries", "value": total_records, "kind": "number"}, {"label": "Total Estimated Cost", "value": total_estimated_cost, "kind": "currency"}, ], } def _establishment_report(db: Session, start_date: date | None, end_date: date | None, period_label: str) -> dict[str, Any]: query = _analytics_query(db, Requisition, MODULE_CONFIG["requisitions"], start_date, end_date) records = query.order_by(Requisition.received_from.asc()).all() establishments: dict[str, dict[str, Any]] = {} for row in records: name = row.received_from or "Not set" item = establishments.setdefault( name, { "establishment": name, "district": row.received_from_district or "-", "range": row.received_from_range or "-", "requisitions": 0, "approved": 0, "other_status": 0, "estimated_cost": 0.0, }, ) item["requisitions"] += 1 if row.approval_status == ProcurementStatus.approved: item["approved"] += 1 else: item["other_status"] += 1 item["estimated_cost"] += _money(row.estimated_cost) rows = sorted(establishments.values(), key=lambda item: (-item["requisitions"], item["establishment"])) return { **_base_report("establishment", period_label), "columns": [ _column("establishment", "Establishment"), _column("district", "District"), _column("range", "Range"), _column("requisitions", "Requisitions", "number"), _column("approved", "Approved", "number"), _column("other_status", "Other Status", "number"), _column("estimated_cost", "Estimated Cost", "currency"), ], "rows": rows, "summary": [ {"label": "Total Establishments", "value": len(rows), "kind": "number"}, {"label": "Total Requisitions", "value": len(records), "kind": "number"}, {"label": "Estimated Cost", "value": sum(item["estimated_cost"] for item in rows), "kind": "currency"}, ], } def _period_totals(db: Session, start_date: date, end_date: date) -> dict[str, Any]: requisitions = _analytics_query(db, Requisition, MODULE_CONFIG["requisitions"], start_date, end_date) tenders = _analytics_query(db, TenderBid, MODULE_CONFIG["tenders"], start_date, end_date) orders = _analytics_query(db, PurchaseOrder, MODULE_CONFIG["purchase-orders"], start_date, end_date) payments = _analytics_query(db, PaymentTracking, MODULE_CONFIG["payments"], start_date, end_date) return { "requisitions": requisitions.count(), "estimated_cost": _money( requisitions.with_entities(func.coalesce(func.sum(Requisition.estimated_cost), 0)).scalar() ), "tenders": tenders.count(), "purchase_orders": orders.count(), "payments": payments.count(), "po_value": _money( orders.with_entities(func.coalesce(func.sum(func.coalesce(PurchaseOrder.po_value, PurchaseOrder.estimated_cost)), 0)).scalar() ), "paid_amount": _money(payments.with_entities(func.coalesce(func.sum(PaymentTracking.paid_amount), 0)).scalar()), } def _cumulative_activity_report( db: Session, start_date: date | None, end_date: date | None, ) -> dict[str, Any]: """Compare lifecycle record activity day by day for the selected range. Activity is based on each record's latest saved timestamp. Grouping related registers keeps the report compact enough for the screen, Excel, and PDF. """ effective_end = end_date or date.today() effective_start = start_date or (effective_end - timedelta(days=6)) period_label = f"{effective_start:%d %b %Y} - {effective_end:%d %b %Y}" activity_groups: list[tuple[str, str, tuple[Any, ...]]] = [ ("requisitions", "Requisitions", (Requisition,)), ("procurement_cases", "Procurement Cases", (ProcurementCase,)), ("tenders", "Tender / Bid", (TenderBid,)), ("evaluations", "Evaluations", (SampleRegister, TechnicalEvaluation, FinancialEvaluation)), ("securities", "EMD / Securities", (EMD, PerformanceSecurity)), ("orders_deliveries", "Orders / Deliveries", (PurchaseOrder, DeliveryAcceptance)), ("payments", "Payments", (PaymentTracking,)), ] daily_counts: dict[date, dict[str, int]] = {} current_day = effective_start while current_day <= effective_end: daily_counts[current_day] = {key: 0 for key, _, _ in activity_groups} current_day += timedelta(days=1) range_start = datetime.combine(effective_start, time.min) range_end = datetime.combine(effective_end, time.max) for group_key, _, models in activity_groups: for model in models: results = ( db.query(func.date(model.updated_at).label("activity_date"), func.count(model.id)) .filter(model.updated_at >= range_start, model.updated_at <= range_end) .group_by(func.date(model.updated_at)) .all() ) for activity_date, count in results: if isinstance(activity_date, datetime): normalized_date = activity_date.date() elif isinstance(activity_date, date): normalized_date = activity_date else: normalized_date = date.fromisoformat(str(activity_date)) if normalized_date in daily_counts: daily_counts[normalized_date][group_key] += int(count or 0) rows: list[dict[str, Any]] = [] cumulative_total = 0 active_days = 0 highest_daily_activity = 0 for activity_date, counts in daily_counts.items(): daily_total = sum(counts.values()) cumulative_total += daily_total active_days += int(daily_total > 0) highest_daily_activity = max(highest_daily_activity, daily_total) rows.append( { "activity_date": activity_date.isoformat(), "day": activity_date.strftime("%A"), **counts, "daily_total": daily_total, "cumulative_total": cumulative_total, } ) report = _base_report("cumulative-activity", period_label) report["footer_lines"] = [ "Activity counts reflect records created or most recently updated on each day", "For the use of Procurement Cell", ] return { **report, "columns": [ _column("activity_date", "Date"), _column("day", "Day"), *[_column(key, label, "number") for key, label, _ in activity_groups], _column("daily_total", "Daily Total", "number"), _column("cumulative_total", "Cumulative Total", "number"), ], "rows": rows, "summary": [ {"label": "Reporting Days", "value": len(rows), "kind": "number"}, {"label": "Active Days", "value": active_days, "kind": "number"}, {"label": "Total Activities", "value": cumulative_total, "kind": "number"}, {"label": "Highest Daily Activity", "value": highest_daily_activity, "kind": "number"}, ], } def _trend_report(db: Session, report_key: str, year: int | None = None) -> dict[str, Any]: if report_key == "monthly": selected_year = year or date.today().year periods = [] for month in range(1, 13): start = date(selected_year, month, 1) next_month = date(selected_year + int(month == 12), (month % 12) + 1, 1) periods.append((month_name[month], start, date.fromordinal(next_month.toordinal() - 1))) period_label = f"Calendar Year {selected_year}" period_key = "month" period_heading = "Month" else: years = {date.today().year} for key, _, model, _ in MODULES: date_column = MODULE_CONFIG[key].get("date_column") for row in db.query(model).all(): item_date = getattr(row, date_column.key, None) if date_column is not None else None item_date = item_date or row.created_at.date() years.add(item_date.year) periods = [(str(selected_year), date(selected_year, 1, 1), date(selected_year, 12, 31)) for selected_year in sorted(years, reverse=True)] period_label = "All Available Years" period_key = "year" period_heading = "Year" rows = [{period_key: label, **_period_totals(db, start, end)} for label, start, end in periods] return { **_base_report(report_key, period_label), "columns": [ _column(period_key, period_heading), _column("requisitions", "Requisitions", "number"), _column("tenders", "Tenders", "number"), _column("purchase_orders", "Purchase Orders", "number"), _column("payments", "Payments", "number"), _column("estimated_cost", "Estimated Cost", "currency"), _column("po_value", "PO Value", "currency"), _column("paid_amount", "Paid Amount", "currency"), ], "rows": rows, "summary": [ {"label": "Total Requisitions", "value": sum(item["requisitions"] for item in rows), "kind": "number"}, {"label": "Total Estimated Cost", "value": sum(item["estimated_cost"] for item in rows), "kind": "currency"}, {"label": "Total Purchase Order Value", "value": sum(item["po_value"] for item in rows), "kind": "currency"}, {"label": "Total Paid Amount", "value": sum(item["paid_amount"] for item in rows), "kind": "currency"}, ], } def _cell_value(value: Any) -> Any: if hasattr(value, "value"): return value.value if isinstance(value, (date, datetime)): return value.isoformat() return value def _procurement_case_reference(db: Session, record: Any) -> str | None: if isinstance(record, ProcurementCase): return record.procurement_case_no procurement_case_id = getattr(record, "procurement_case_id", None) if procurement_case_id is None: return None case = db.get(ProcurementCase, procurement_case_id) return case.procurement_case_no if case else f"Case ID {procurement_case_id}" def _report_record_value(db: Session, record: Any, key: str) -> Any: if key == "procurement_case_reference": return _procurement_case_reference(db, record) return _cell_value(getattr(record, key, None)) def _generic_report( report_key: str, db: Session, start_date: date | None, end_date: date | None, period_label: str, ) -> dict[str, Any]: spec = REPORT_SPECS[report_key] model = spec["model"] query = db.query(model) if model is ProcurementCase: query = query.filter(ProcurementCase.procurement_mode.in_(ACTIVE_PROCUREMENT_MODES)) date_attr = spec.get("date_attr") date_column = getattr(model, date_attr, None) if date_attr else None event_date = func.coalesce(date_column, func.date(model.created_at)) if date_column is not None else func.date(model.created_at) if start_date: query = query.filter(event_date >= start_date) if end_date: query = query.filter(event_date <= end_date) for attr, values in spec.get("filters", []): query = query.filter(getattr(model, attr).in_(values)) rows = [] for record in query.order_by(model.id.desc()).limit(500).all(): row = {} for key, _, _ in spec["columns"]: row[key] = _report_record_value(db, record, key) rows.append(row) currency_keys = [key for key, _, kind in spec["columns"] if kind == "currency"] summary = [{"label": "Records", "value": len(rows), "kind": "number"}] for key in currency_keys[:2]: summary.append( { "label": next(label for column_key, label, _ in spec["columns"] if column_key == key), "value": sum(_money(row.get(key)) for row in rows), "kind": "currency", } ) return { **_base_report(report_key, period_label), "columns": [_column(key, label, kind) for key, label, kind in spec["columns"]], "rows": rows, "summary": summary, } def _financial_year(value: date) -> str: start_year = value.year if value.month >= 4 else value.year - 1 return f"{start_year}-{str(start_year + 1)[-2:]}" def _allocation_report( report_key: str, db: Session, start_date: date | None, end_date: date | None, period_label: str, expenditure_group: str | None = None, ) -> dict[str, Any]: financial_year = _financial_year(end_date or date.today()) query = db.query(ReportBudgetAllocation).filter( ReportBudgetAllocation.financial_year == financial_year, ReportBudgetAllocation.is_active.is_(True), ) if expenditure_group: query = query.filter(ReportBudgetAllocation.expenditure_group == expenditure_group) rows = [] for index, allocation in enumerate(query.order_by(ReportBudgetAllocation.expenditure_group, ReportBudgetAllocation.id).all(), start=1): spend_query = db.query(func.coalesce(func.sum(func.coalesce(PurchaseOrder.po_value, PurchaseOrder.estimated_cost, 0)), 0)).filter( func.lower(func.coalesce(PurchaseOrder.budget_head, "")).in_( [allocation.head_code.lower(), allocation.head_name.lower()] ) ) if start_date: spend_query = spend_query.filter(func.coalesce(PurchaseOrder.po_date, func.date(PurchaseOrder.created_at)) >= start_date) if end_date: spend_query = spend_query.filter(func.coalesce(PurchaseOrder.po_date, func.date(PurchaseOrder.created_at)) <= end_date) utilized = _money(spend_query.scalar()) allocated = _money(allocation.allocation_amount) rows.append( { "serial_no": index, "expenditure_group": allocation.expenditure_group, "head_code": allocation.head_code, "head_name": allocation.head_name, "allocation_amount": allocated, "utilized_amount": utilized, "balance_amount": allocated - utilized, "utilization_percent": (utilized / allocated * 100) if allocated else 0, } ) return { **_base_report(report_key, f"{period_label} | Financial Year {financial_year}"), "columns": [ _column("serial_no", "Sl. No.", "number"), _column("expenditure_group", "Expenditure Group"), _column("head_code", "Head Code"), _column("head_name", "Head of Account"), _column("allocation_amount", "Allocation Amount", "currency"), _column("utilized_amount", "Utilized Amount", "currency"), _column("balance_amount", "Balance Amount", "currency"), _column("utilization_percent", "Utilization %", "number"), ], "rows": rows, "summary": [ {"label": "Total Allocation", "value": sum(row["allocation_amount"] for row in rows), "kind": "currency"}, {"label": "Total Utilized", "value": sum(row["utilized_amount"] for row in rows), "kind": "currency"}, {"label": "Available Balance", "value": sum(row["balance_amount"] for row in rows), "kind": "currency"}, ], } def _vendor_payment_report( db: Session, start_date: date | None, end_date: date | None, period_label: str, ) -> dict[str, Any]: event_date = func.coalesce(PaymentTracking.invoice_date, PurchaseOrder.po_date, func.date(PurchaseOrder.created_at)) query = db.query(PurchaseOrder, PaymentTracking).outerjoin(PaymentTracking, PaymentTracking.po_id == PurchaseOrder.id) if start_date: query = query.filter(event_date >= start_date) if end_date: query = query.filter(event_date <= end_date) rows = [] for order, payment in query.order_by(event_date.desc(), PurchaseOrder.id.desc()).limit(1000).all(): invoice_amount = _money(payment.invoice_amount if payment else 0) paid_amount = _money(payment.paid_amount if payment else 0) rows.append( { "vendor_name": order.vendor_name, "po_no": order.po_no, "po_date": _cell_value(order.po_date), "invoice_no": payment.invoice_no if payment else "Not Submitted", "invoice_date": _cell_value(payment.invoice_date) if payment else None, "invoice_amount": invoice_amount, "paid_amount": paid_amount, "outstanding_amount": max(invoice_amount - paid_amount, 0), "payment_status": payment.payment_status if payment else "Invoice Pending", } ) return { **_base_report("vendor-payment-status", period_label), "columns": [ _column("vendor_name", "Vendor"), _column("po_no", "PO No"), _column("po_date", "PO Date"), _column("invoice_no", "Invoice No"), _column("invoice_date", "Invoice Date"), _column("invoice_amount", "Invoice Amount", "currency"), _column("paid_amount", "Paid Amount", "currency"), _column("outstanding_amount", "Outstanding Amount", "currency"), _column("payment_status", "Payment Status"), ], "rows": rows, "summary": [ {"label": "Vendors / Orders", "value": len(rows), "kind": "number"}, {"label": "Total Invoiced", "value": sum(row["invoice_amount"] for row in rows), "kind": "currency"}, {"label": "Total Paid", "value": sum(row["paid_amount"] for row in rows), "kind": "currency"}, {"label": "Outstanding", "value": sum(row["outstanding_amount"] for row in rows), "kind": "currency"}, ], } def _daily_report(db: Session, as_of: date, period_label: str) -> dict[str, Any]: pending_approval = db.query(Requisition).filter( Requisition.approval_status.notin_([ProcurementStatus.approved, ProcurementStatus.rejected]) ).count() mode_rows = [ ("Open Tender", "E_TENDER"), ("GeM (Direct)", "GEM"), ("Local (Direct)", "OPEN_MARKET"), ("Local Purchase Committee", "LPC"), ] rows: list[dict[str, Any]] = [ {"serial_no": "1", "description": "Requisitions whose Administrative Approval is pending", "detail": "", "number": pending_approval, "balance_amount": None} ] for index, (label, mode) in enumerate(mode_rows): not_started = db.query(Requisition).filter( Requisition.approval_status == ProcurementStatus.approved, Requisition.procurement_mode == mode, ~db.query(ProcurementCase.id).filter(ProcurementCase.requisition_id == Requisition.id).exists(), ).count() rows.append({"serial_no": "2" if index == 0 else "", "description": "Administrative Approvals whose purchase process has not started" if index == 0 else "", "detail": label, "number": not_started, "balance_amount": None}) for index, (label, mode) in enumerate(mode_rows): undelivered = ( db.query(PurchaseOrder) .outerjoin(DeliveryAcceptance, DeliveryAcceptance.po_id == PurchaseOrder.id) .filter( PurchaseOrder.procurement_mode == mode, or_( DeliveryAcceptance.id.is_(None), DeliveryAcceptance.acceptance_status.notin_([ProcurementStatus.approved, ProcurementStatus.completed]), ), ) .distinct() .count() ) rows.append({"serial_no": "3" if index == 0 else "", "description": "Purchase Orders whose items have not been delivered" if index == 0 else "", "detail": label, "number": undelivered, "balance_amount": None}) acceptance_pending = db.query(DeliveryAcceptance).filter( DeliveryAcceptance.delivery_date.is_not(None), DeliveryAcceptance.acceptance_status.notin_([ProcurementStatus.approved, ProcurementStatus.completed]), ).count() accepted_bill_pending = ( db.query(PurchaseOrder) .join(DeliveryAcceptance, DeliveryAcceptance.po_id == PurchaseOrder.id) .filter( DeliveryAcceptance.acceptance_status.in_([ProcurementStatus.approved, ProcurementStatus.completed]), ~db.query(PaymentTracking.id).filter(PaymentTracking.po_id == PurchaseOrder.id).exists(), ) .distinct() .count() ) accounts_pending = db.query(PaymentTracking).filter( PaymentTracking.paid_amount < PaymentTracking.invoice_amount, func.lower(PaymentTracking.payment_status).in_(["processed", "provisioning processed", "pending with accounts"]), ).count() distribution_pending = ( db.query(PaymentTracking) .join(PurchaseOrder, PurchaseOrder.id == PaymentTracking.po_id) .filter(PaymentTracking.paid_amount >= PaymentTracking.invoice_amount, PurchaseOrder.status != ProcurementStatus.completed) .count() ) emd_pending = db.query(EMD).filter( EMD.expiry_date <= as_of, func.lower(EMD.refund_status).in_(["pending", "pending return", "held"]), ).count() pbg_pending = db.query(PerformanceSecurity).filter( func.coalesce(PerformanceSecurity.expiry_date, PerformanceSecurity.valid_until) <= as_of, func.lower(PerformanceSecurity.release_status).in_(["pending", "held", "pending release"]), ).count() rows.extend( [ {"serial_no": "4", "description": "Items received but acceptance pending", "detail": "", "number": acceptance_pending, "balance_amount": None}, {"serial_no": "5", "description": "Items accepted but bills have not been processed", "detail": "", "number": accepted_bill_pending, "balance_amount": None}, {"serial_no": "6", "description": "Bills processed by provisioning but pending with accounts section", "detail": "", "number": accounts_pending, "balance_amount": None}, {"serial_no": "7", "description": "Items already paid but pending at store for distribution (PO wise)", "detail": "", "number": distribution_pending, "balance_amount": None}, {"serial_no": "8", "description": "EMDs due to be released but pending", "detail": "", "number": emd_pending, "balance_amount": None}, {"serial_no": "9", "description": "Performance Securities due to be released but pending", "detail": "", "number": pbg_pending, "balance_amount": None}, ] ) financial_year = _financial_year(as_of) start_year = int(financial_year[:4]) utilization_rows = _allocation_report( "budget-utilization", db, date(start_year, 4, 1), as_of, period_label )["rows"] for group in ("Administrative", "Programme"): group_rows = [item for item in utilization_rows if item["expenditure_group"] == group] if not group_rows: continue for index, item in enumerate(group_rows): rows.append( { "serial_no": "10" if group == "Administrative" and index == 0 else ("11" if group == "Programme" and index == 0 else ""), "description": f"{group} Expenditure" if index == 0 else "", "detail": item["head_name"], "number": None, "allocation_amount": item["allocation_amount"], "expenditure_amount": item["utilized_amount"], "balance_amount": item["balance_amount"], } ) rows.append( { "serial_no": "", "description": f"Total {group} Expenditure", "detail": "", "number": None, "allocation_amount": sum(item["allocation_amount"] for item in group_rows), "expenditure_amount": sum(item["utilized_amount"] for item in group_rows), "balance_amount": sum(item["balance_amount"] for item in group_rows), } ) return { **_base_report("daily", f"As on {as_of:%d.%m.%Y} | {period_label}"), "columns": [ _column("serial_no", "Sl. No."), _column("description", "Descriptions"), _column("detail", "Mode / Head of Account"), _column("number", "Numbers", "number"), _column("allocation_amount", "Allotted Fund", "currency"), _column("expenditure_amount", "Expenditure", "currency"), _column("balance_amount", "Balance Amount", "currency"), ], "rows": rows, "summary": [ {"label": "Approval Pending", "value": pending_approval, "kind": "number"}, {"label": "Acceptance Pending", "value": acceptance_pending, "kind": "number"}, {"label": "Bills Pending", "value": accepted_bill_pending + accounts_pending, "kind": "number"}, {"label": "Security Releases Pending", "value": emd_pending + pbg_pending, "kind": "number"}, ], } def _custom_report( source_key: str, field_keys: list[str], db: Session, start_date: date | None, end_date: date | None, period_label: str, ) -> dict[str, Any]: source = CUSTOM_REPORT_SOURCES.get(source_key) if not source: raise HTTPException(status_code=404, detail="Unknown report data source.") available = {key: (label, kind) for key, label, kind in source["fields"]} selected = list(dict.fromkeys(field_keys)) if not selected: raise HTTPException(status_code=400, detail="Select at least one report field.") if len(selected) > 15 or any(key not in available for key in selected): raise HTTPException(status_code=400, detail="Invalid custom report field selection.") model = source["model"] query = db.query(model) date_column = getattr(model, source["date_attr"]) event_date = func.coalesce(date_column, func.date(model.created_at)) if start_date: query = query.filter(event_date >= start_date) if end_date: query = query.filter(event_date <= end_date) rows = [ {key: _report_record_value(db, record, key) for key in selected} for record in query.order_by(model.id.desc()).limit(2000).all() ] currency_keys = [key for key in selected if available[key][1] == "currency"] summary: list[dict[str, Any]] = [{"label": "Records", "value": len(rows), "kind": "number"}] for key in currency_keys[:3]: summary.append({"label": f"Total {available[key][0]}", "value": sum(_money(row.get(key)) for row in rows), "kind": "currency"}) return { "key": "custom", "organization": "Odisha Fire and Emergency Services", "title": f"Custom {source['title']} Report", "period_label": period_label, "generated_at": datetime.now().astimezone().isoformat(timespec="minutes"), "footer_lines": ["Computer Generated report", "For the use of Procurement Cell"], "columns": [_column(key, available[key][0], available[key][1]) for key in selected], "rows": rows, "summary": summary, } def _generate_report( report_key: str, db: Session, period: str, start_date: date | None, end_date: date | None, month: int | None, year: int | None, ) -> dict[str, Any]: if report_key not in REPORT_TITLES: raise HTTPException(status_code=404, detail="Unknown report type.") filtered_start, filtered_end, period_label = _report_filter(report_key, period, start_date, end_date, month, year) if report_key == "summary": return _summary_report(db, filtered_start, filtered_end, period_label) if report_key == "daily": return _daily_report(db, filtered_end or date.today(), period_label) if report_key == "vendor-payment-status": return _vendor_payment_report(db, filtered_start, filtered_end, period_label) if report_key == "budget-utilization": return _allocation_report(report_key, db, filtered_start, filtered_end, period_label) if report_key == "program-expenditure": return _allocation_report(report_key, db, filtered_start, filtered_end, period_label, "Programme") if report_key == "cumulative-activity": return _cumulative_activity_report(db, filtered_start, filtered_end) if report_key == "establishment": return _establishment_report(db, filtered_start, filtered_end, period_label) if report_key in REPORT_SPECS: return _generic_report(report_key, db, filtered_start, filtered_end, period_label) return _trend_report(db, report_key, year) def _formatted(value: Any, kind: str) -> str: if value is None or value == "": return "-" if kind == "currency": return f"Rs. {_money(value):,.2f}" if kind == "number": return f"{int(value or 0):,}" return str(value if value not in (None, "") else "-") def _xlsx_bytes(report: dict[str, Any]) -> bytes: from openpyxl import Workbook from openpyxl.drawing.image import Image from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter workbook = Workbook() sheet = workbook.active sheet.title = report["title"][:31] sheet.sheet_view.showGridLines = False column_count = len(report["columns"]) last_column = get_column_letter(column_count) primary = "163B63" border = Side(style="thin", color="CBD5E1") logo = _logo_path() if logo.exists(): image = Image(str(logo)) image.height = 52 image.width = 52 sheet.add_image(image, "A1") sheet.merge_cells(start_row=1, start_column=2, end_row=1, end_column=column_count) sheet["B1"] = report["organization"] sheet["B1"].font = Font(name="Arial", size=16, bold=True, color=primary) sheet["B1"].alignment = Alignment(vertical="center") sheet.row_dimensions[1].height = 42 sheet.merge_cells(start_row=2, start_column=1, end_row=2, end_column=column_count) sheet["A2"] = report["title"] sheet["A2"].font = Font(name="Arial", size=13, bold=True, color=primary) sheet["A2"].alignment = Alignment(horizontal="center") sheet.merge_cells(start_row=3, start_column=1, end_row=3, end_column=column_count) sheet["A3"] = f"Reporting Period: {report['period_label']} | Generated: {report['generated_at']}" sheet["A3"].font = Font(name="Arial", size=9, italic=True, color="475569") sheet["A3"].alignment = Alignment(horizontal="center") summary_row = 5 for index, item in enumerate(report["summary"], start=1): cell = sheet.cell(row=summary_row, column=index, value=f"{item['label']}: {_formatted(item['value'], item['kind'])}") cell.font = Font(name="Arial", size=10, bold=True, color=primary) cell.fill = PatternFill("solid", fgColor="E8F1F8") cell.alignment = Alignment(horizontal="center", vertical="center") cell.border = Border(top=border, bottom=border, left=border, right=border) header_row = 7 for column_index, column in enumerate(report["columns"], start=1): cell = sheet.cell(row=header_row, column=column_index, value=column["label"]) cell.font = Font(name="Arial", size=9, bold=True, color="FFFFFF") cell.fill = PatternFill("solid", fgColor=primary) cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) cell.border = Border(top=border, bottom=border, left=border, right=border) sheet.row_dimensions[header_row].height = 28 for row_index, row in enumerate(report["rows"], start=header_row + 1): for column_index, column in enumerate(report["columns"], start=1): cell = sheet.cell(row=row_index, column=column_index, value=row.get(column["key"], "")) cell.font = Font(name="Arial", size=9, color="1F2937") cell.border = Border(top=border, bottom=border, left=border, right=border) cell.fill = PatternFill("solid", fgColor="FFFFFF" if row_index % 2 == 0 else "F8FAFC") cell.alignment = Alignment( horizontal="right" if column["kind"] in {"currency", "number"} else "left", vertical="center", wrap_text=True, ) if column["kind"] == "currency": cell.number_format = '"Rs." #,##0.00' elif column["kind"] == "number": cell.number_format = "#,##0" footer_row = max(header_row + len(report["rows"]) + 3, 11) for offset, line in enumerate(report["footer_lines"]): sheet.merge_cells(start_row=footer_row + offset, start_column=1, end_row=footer_row + offset, end_column=column_count) footer_cell = sheet.cell(row=footer_row + offset, column=1, value=line) footer_cell.font = Font(name="Arial", size=9, italic=True, color="64748B") footer_cell.alignment = Alignment(horizontal="center") widths = {"module": 27, "status": 34, "indicator": 32, "establishment": 33, "district": 18, "range": 20} for index, column in enumerate(report["columns"], start=1): sheet.column_dimensions[get_column_letter(index)].width = widths.get(column["key"], 17) sheet.freeze_panes = f"A{header_row + 1}" sheet.auto_filter.ref = f"A{header_row}:{last_column}{header_row + len(report['rows'])}" sheet.page_setup.orientation = "landscape" sheet.page_setup.fitToWidth = 1 sheet.page_margins.left = 0.35 sheet.page_margins.right = 0.35 sheet.print_title_rows = f"1:{header_row}" sheet.oddFooter.center.text = "Computer Generated report | For the use of Procurement Cell" sheet.oddFooter.center.size = 8 sheet.oddFooter.center.font = "Arial,Italic" output = BytesIO() workbook.save(output) return output.getvalue() def _pdf_bytes(report: dict[str, Any]) -> bytes: from reportlab.lib import colors from reportlab.lib.enums import TA_CENTER from reportlab.lib.pagesizes import A4, landscape from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet from reportlab.lib.units import mm from reportlab.platypus import Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle output = BytesIO() page_size = landscape(A4) document = SimpleDocTemplate( output, pagesize=page_size, leftMargin=13 * mm, rightMargin=13 * mm, topMargin=34 * mm, bottomMargin=20 * mm, ) styles = getSampleStyleSheet() title_style = ParagraphStyle( "ReportTitle", parent=styles["Heading2"], fontName="Helvetica-Bold", fontSize=13, leading=17, textColor=colors.HexColor("#163B63"), alignment=TA_CENTER, spaceAfter=3 * mm ) meta_style = ParagraphStyle( "Meta", parent=styles["Normal"], fontName="Helvetica-Oblique", fontSize=8.5, leading=11, textColor=colors.HexColor("#475569"), alignment=TA_CENTER, spaceAfter=5 * mm ) body_style = ParagraphStyle("Body", parent=styles["Normal"], fontName="Helvetica", fontSize=7.5, leading=9.5) header_style = ParagraphStyle("Header", parent=body_style, fontName="Helvetica-Bold", textColor=colors.white, alignment=TA_CENTER) def draw_page(canvas, _doc): width, height = page_size logo = _logo_path() if logo.exists(): canvas.drawImage(str(logo), 14 * mm, height - 25 * mm, width=18 * mm, height=18 * mm, preserveAspectRatio=True, mask="auto") canvas.setFont("Helvetica-Bold", 15) canvas.setFillColor(colors.HexColor("#163B63")) canvas.drawCentredString(width / 2, height - 15 * mm, report["organization"]) canvas.setStrokeColor(colors.HexColor("#CBD5E1")) canvas.line(13 * mm, height - 29 * mm, width - 13 * mm, height - 29 * mm) canvas.line(13 * mm, 14 * mm, width - 13 * mm, 14 * mm) canvas.setFont("Helvetica-Oblique", 8) canvas.setFillColor(colors.HexColor("#64748B")) canvas.drawString(13 * mm, 8 * mm, report["footer_lines"][0]) canvas.drawRightString(width - 13 * mm, 8 * mm, report["footer_lines"][1]) story = [ Paragraph(report["title"], title_style), Paragraph(f"Reporting Period: {report['period_label']} | Generated: {report['generated_at']}", meta_style), ] summary_data = [[Paragraph(item["label"], body_style), Paragraph(_formatted(item["value"], item["kind"]), body_style)] for item in report["summary"]] summary_table = Table(summary_data, colWidths=[48 * mm, 38 * mm], hAlign="LEFT") summary_table.setStyle( TableStyle( [ ("BACKGROUND", (0, 0), (-1, -1), colors.HexColor("#E8F1F8")), ("BOX", (0, 0), (-1, -1), 0.5, colors.HexColor("#CBD5E1")), ("INNERGRID", (0, 0), (-1, -1), 0.3, colors.HexColor("#CBD5E1")), ("FONTNAME", (0, 0), (0, -1), "Helvetica-Bold"), ("FONTSIZE", (0, 0), (-1, -1), 8), ("TEXTCOLOR", (0, 0), (-1, -1), colors.HexColor("#163B63")), ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), ("LEFTPADDING", (0, 0), (-1, -1), 7), ("RIGHTPADDING", (0, 0), (-1, -1), 7), ("TOPPADDING", (0, 0), (-1, -1), 6), ("BOTTOMPADDING", (0, 0), (-1, -1), 6), ] ) ) story.extend([summary_table, Spacer(1, 6 * mm)]) table_data = [[Paragraph(column["label"], header_style) for column in report["columns"]]] table_data.extend( [Paragraph(_formatted(row.get(column["key"]), column["kind"]), body_style) for column in report["columns"]] for row in report["rows"] ) available_width = page_size[0] - document.leftMargin - document.rightMargin table = Table(table_data, colWidths=[available_width / len(report["columns"])] * len(report["columns"]), repeatRows=1) table.setStyle( TableStyle( [ ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#163B63")), ("GRID", (0, 0), (-1, -1), 0.35, colors.HexColor("#CBD5E1")), ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#F8FAFC")]), ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), ("LEFTPADDING", (0, 0), (-1, -1), 6), ("RIGHTPADDING", (0, 0), (-1, -1), 6), ("TOPPADDING", (0, 0), (-1, -1), 6), ("BOTTOMPADDING", (0, 0), (-1, -1), 6), ] ) ) story.append(table) document.build(story, onFirstPage=draw_page, onLaterPages=draw_page) return output.getvalue() @router.get("/formal/{report_key}") def formal_report( report_key: str, db: Annotated[Session, Depends(get_db)], period: str = Query(default="month"), start_date: date | None = Query(default=None), end_date: date | None = Query(default=None), month: int | None = Query(default=None, ge=1, le=12), year: int | None = Query(default=None, ge=2000, le=2100), ): return _generate_report(report_key, db, period, start_date, end_date, month, year) @router.get("/formal/{report_key}/export") def export_formal_report( report_key: str, db: Annotated[Session, Depends(get_db)], format: str = Query(default="pdf"), period: str = Query(default="month"), start_date: date | None = Query(default=None), end_date: date | None = Query(default=None), month: int | None = Query(default=None, ge=1, le=12), year: int | None = Query(default=None, ge=2000, le=2100), ): report = _generate_report(report_key, db, period, start_date, end_date, month, year) filename = f"{report_key}-report-{date.today():%Y%m%d}" if format == "xlsx": return Response( content=_xlsx_bytes(report), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f'attachment; filename="{filename}.xlsx"'}, ) if format == "pdf": return Response( content=_pdf_bytes(report), media_type="application/pdf", headers={"Content-Disposition": f'attachment; filename="{filename}.pdf"'}, ) raise HTTPException(status_code=400, detail="Export format must be pdf or xlsx.") @router.get("/custom/metadata") def custom_report_metadata(): return { "sources": [ { "key": key, "title": source["title"], "fields": [_column(field_key, label, kind) for field_key, label, kind in source["fields"]], } for key, source in CUSTOM_REPORT_SOURCES.items() ] } @router.get("/custom") def custom_report( db: Annotated[Session, Depends(get_db)], source: str = Query(...), fields: list[str] = Query(default=[]), start_date: date | None = Query(default=None), end_date: date | None = Query(default=None), ): period_label = _analytics_period("range" if start_date or end_date else "all", start_date, end_date, None, None)[2] return _custom_report(source, fields, db, start_date, end_date, period_label) @router.get("/custom/export") def export_custom_report( db: Annotated[Session, Depends(get_db)], source: str = Query(...), fields: list[str] = Query(default=[]), format: str = Query(default="pdf"), start_date: date | None = Query(default=None), end_date: date | None = Query(default=None), ): period_label = _analytics_period("range" if start_date or end_date else "all", start_date, end_date, None, None)[2] report = _custom_report(source, fields, db, start_date, end_date, period_label) filename = f"custom-{source}-report-{date.today():%Y%m%d}" if format == "xlsx": return Response( content=_xlsx_bytes(report), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f'attachment; filename="{filename}.xlsx"'}, ) if format == "pdf": return Response( content=_pdf_bytes(report), media_type="application/pdf", headers={"Content-Disposition": f'attachment; filename="{filename}.pdf"'}, ) raise HTTPException(status_code=400, detail="Export format must be pdf or xlsx.") @router.get("/status-summary") def status_summary(db: Annotated[Session, Depends(get_db)]): requisition_status = db.query(Requisition.approval_status, func.count(Requisition.id)).group_by(Requisition.approval_status).all() tender_status = db.query(TenderBid.status, func.count(TenderBid.id)).group_by(TenderBid.status).all() po_status = db.query(PurchaseOrder.status, func.count(PurchaseOrder.id)).group_by(PurchaseOrder.status).all() return { "requisitions": [{"status": status.value, "count": count} for status, count in requisition_status], "tenders": [{"status": status.value, "count": count} for status, count in tender_status], "purchase_orders": [{"status": status.value, "count": count} for status, count in po_status], } @router.get("/payment-aging") def payment_aging(db: Annotated[Session, Depends(get_db)]): rows = db.query(PaymentTracking).order_by(PaymentTracking.invoice_date.desc()).all() return [ { "invoice_no": row.invoice_no, "invoice_date": row.invoice_date, "invoice_amount": float(row.invoice_amount), "paid_amount": float(row.paid_amount), "outstanding": float(row.invoice_amount - row.paid_amount), "payment_status": row.payment_status, } for row in rows ]