"""Module 5 — reports: the live dashboard, trend charts, and exportable tables. Read-only across the board. Every figure is composed from the module services, so this router adds no business rule of its own — if a number here looks wrong, it is wrong on its own module's screen too. Date parameters are Asia/Dhaka calendar days and both ends are inclusive. """ from datetime import date from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from sqlalchemy.orm import Session from app.api.deps import get_current_user from app.db.session import get_db from app.models.user import User from app.schemas.report import ( ApplicationReportRow, ApplicationTrendPoint, CombinedReportRow, ComplaintReportRow, ComplaintTrendPoint, DashboardOut, EquipmentReportRow, StockMovementPoint, ) from app.services import exports, reports router = APIRouter(prefix="/reports", tags=["reports"]) Granularity = Query("day", pattern="^(day|week|month)$", description="Trend bucket size") DateFrom = Query(None, description="Inclusive start, Asia/Dhaka. Defaults to 30 days back.") DateTo = Query(None, description="Inclusive end, Asia/Dhaka. Defaults to today.") #: Rows a report may return before the dataset is refused outright. The exporters cap at #: the same number; asking for a decade of tickets should fail loudly, not silently #: hand back a truncated spreadsheet that looks complete. MAX_RANGE_DAYS = 1096 # three years def _checked_range(date_from: date | None, date_to: date | None) -> tuple[date, date]: start, end = reports.resolve_range(date_from, date_to) if (end - start).days > MAX_RANGE_DAYS: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Date range must not exceed {MAX_RANGE_DAYS} days", ) return start, end # --- live dashboard --- @router.get("/dashboard", response_model=DashboardOut) def dashboard( db: Session = Depends(get_db), _: User = Depends(get_current_user), ) -> DashboardOut: """Today's figures across every module. Safe to poll — it is a handful of counts.""" return DashboardOut(**reports.dashboard(db)) # --- trends --- @router.get("/trends/applications", response_model=list[ApplicationTrendPoint]) def applications_trend( db: Session = Depends(get_db), _: User = Depends(get_current_user), date_from: date | None = DateFrom, date_to: date | None = DateTo, granularity: str = Granularity, ) -> list[ApplicationTrendPoint]: """Intake against throughput. Outcomes are bucketed by when the move happened.""" start, end = _checked_range(date_from, date_to) rows = reports.applications_trend( db, date_from=start, date_to=end, granularity=granularity ) return [ApplicationTrendPoint(**row) for row in rows] @router.get("/trends/complaints", response_model=list[ComplaintTrendPoint]) def complaints_trend( db: Session = Depends(get_db), _: User = Depends(get_current_user), date_from: date | None = DateFrom, date_to: date | None = DateTo, granularity: str = Granularity, ) -> list[ComplaintTrendPoint]: """Volume plus resolution-time trend, bucketed by when each ticket was resolved.""" start, end = _checked_range(date_from, date_to) rows = reports.complaints_trend(db, date_from=start, date_to=end, granularity=granularity) return [ComplaintTrendPoint(**row) for row in rows] @router.get("/trends/stock-movement", response_model=list[StockMovementPoint]) def stock_movement_trend( db: Session = Depends(get_db), _: User = Depends(get_current_user), date_from: date | None = DateFrom, date_to: date | None = DateTo, granularity: str = Granularity, equipment_type_id: int | None = Query(None, description="Narrow to one equipment type"), ) -> list[StockMovementPoint]: """Units moving through the warehouse, read off the append-only movement ledger.""" start, end = _checked_range(date_from, date_to) rows = reports.stock_movement_trend( db, date_from=start, date_to=end, granularity=granularity, equipment_type_id=equipment_type_id, ) return [StockMovementPoint(**row) for row in rows] # --- tabular reports --- @router.get("/applications", response_model=list[ApplicationReportRow]) def applications_report( db: Session = Depends(get_db), _: User = Depends(get_current_user), date_from: date | None = DateFrom, date_to: date | None = DateTo, ) -> list[ApplicationReportRow]: start, end = _checked_range(date_from, date_to) rows = reports.applications_report(db, date_from=start, date_to=end) return [ApplicationReportRow(**row) for row in rows] @router.get("/complaints", response_model=list[ComplaintReportRow]) def complaints_report( db: Session = Depends(get_db), _: User = Depends(get_current_user), date_from: date | None = DateFrom, date_to: date | None = DateTo, ) -> list[ComplaintReportRow]: start, end = _checked_range(date_from, date_to) rows = reports.complaints_report(db, date_from=start, date_to=end) return [ComplaintReportRow(**row) for row in rows] @router.get("/equipment", response_model=list[EquipmentReportRow]) def equipment_report( db: Session = Depends(get_db), _: User = Depends(get_current_user), date_from: date | None = DateFrom, date_to: date | None = DateTo, ) -> list[EquipmentReportRow]: start, end = _checked_range(date_from, date_to) rows = reports.equipment_report(db, date_from=start, date_to=end) return [EquipmentReportRow(**row) for row in rows] @router.get("/combined", response_model=list[CombinedReportRow]) def combined_report( db: Session = Depends(get_db), _: User = Depends(get_current_user), date_from: date | None = DateFrom, date_to: date | None = DateTo, granularity: str = Granularity, ) -> list[CombinedReportRow]: """Every module's totals per bucket, side by side.""" start, end = _checked_range(date_from, date_to) rows = reports.combined_report( db, date_from=start, date_to=end, granularity=granularity ) return [CombinedReportRow(**row) for row in rows] # --- export --- @router.get( "/export/{dataset}", response_class=Response, responses={ 200: { "content": {exports.CSV_MEDIA_TYPE: {}, exports.XLSX_MEDIA_TYPE: {}}, "description": "The report as a downloadable file.", } }, ) def export_report( dataset: str, db: Session = Depends(get_db), _: User = Depends(get_current_user), date_from: date | None = DateFrom, date_to: date | None = DateTo, fmt: str = Query("csv", pattern="^(csv|xlsx)$", alias="format"), ) -> Response: """One endpoint for every report in both formats. Built whole in memory rather than streamed: the row cap keeps a worst-case file in the low megabytes, and a buffered body means a query that fails mid-way returns a JSON error instead of a half-written file the browser has already saved. """ spec = reports.DATASETS.get(dataset) if spec is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Unknown report '{dataset}'. Choose one of: {', '.join(reports.DATASETS)}", ) start, end = _checked_range(date_from, date_to) rows = reports.dataset_rows(db, dataset, date_from=start, date_to=end) columns = spec["columns"] stem = f"{dataset}-{start.isoformat()}_{end.isoformat()}" if fmt == "xlsx": body = exports.to_xlsx(rows, columns, sheet_name=spec["title"]) media_type = exports.XLSX_MEDIA_TYPE else: body = exports.to_csv(rows, columns) media_type = exports.CSV_MEDIA_TYPE return Response( content=body, media_type=media_type, headers={ "Content-Disposition": f'attachment; filename="{exports.filename(stem, fmt)}"', # The row count is the only way a caller can tell a capped export from a # complete one, and a browser download shows no response body to inspect. "X-Row-Count": str(len(rows)), }, )