"""Reports & Analytics router (INDL-45). Exposes 8 report data endpoints plus PDF / Excel / bulk-CSV export endpoints, and keeps the 3 legacy summary endpoints (occupancy / revenue / records-summary). Security controls applied consistently (see ``security.py``): * server-side role enforcement per report type (RBAC, never client-trusted) * tenant isolation on every query (tenant_id from request context) * report_type allowlist + path regex (blocks traversal / header injection) * bounded date range (max 366 days) to prevent unbounded exports * tenant-scoped section resolution (blocks cross-tenant IDOR) * per-user export rate limiting * structured audit log entry for every export * no-store / nosniff headers on every streamed file """ from __future__ import annotations from datetime import date, datetime, timezone from typing import Optional from uuid import UUID from fastapi import APIRouter, Depends, Path, Query, Request from fastapi.responses import StreamingResponse from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from src.apps.auth.models.user import User from src.apps.billing.models.invoice import Invoice from src.apps.plots.models.plot import Plot from src.apps.records.models.record import Record from src.core.constants import UserRole from src.core.dependencies import require_min_role, require_tenant from src.core.schemas.response import success from src.database.session import get_db from .constants import BULK_CSV_MIN_ROLE from .security import ( EXPORT_SECURITY_HEADERS, assert_report_role, check_export_rate_limit, log_export, resolve_section, safe_filename, validate_date_range, validate_report_type, ) from .services.data import ReportDataService from .services.export import ReportExportService router = APIRouter(prefix="/reports", tags=["reports"]) _data = ReportDataService() _export = ReportExportService() # Path regex: only lowercase letters and hyphens — blocks CRLF / traversal in # {report_type} before it reaches the allowlist, filename or any header. _REPORT_TYPE_PATH = Path(..., pattern="^[a-z-]+$") # ── Shared helpers ──────────────────────────────────────────────────────────── async def _run_report( db: AsyncSession, report_type: str, current_user: User, *, tenant_id: str, from_date: Optional[date], to_date: Optional[date], section_id: Optional[UUID], ): # REPORT_MIN_ROLES is the single source of truth for per-report authz; the # per-endpoint require_min_role below is only a STAFF authentication floor. assert_report_role(current_user, report_type) validate_date_range(from_date, to_date) await resolve_section(db, tenant_id, section_id) return await _data.build( db, report_type, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id, ) # ── Report data endpoints ───────────────────────────────────────────────────── @router.get("/monthly-sales") async def monthly_sales_report( from_date: Optional[date] = Query(None), to_date: Optional[date] = Query(None), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): result = await _run_report(db, "monthly-sales", current_user, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id) return success(result.as_json()) @router.get("/burial-register") async def burial_register_report( from_date: Optional[date] = Query(None), to_date: Optional[date] = Query(None), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): result = await _run_report(db, "burial-register", current_user, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id) return success(result.as_json()) @router.get("/capacity") async def capacity_report( from_date: Optional[date] = Query(None), to_date: Optional[date] = Query(None), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): result = await _run_report(db, "capacity", current_user, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id) return success(result.as_json()) @router.get("/revenue-by-section") async def revenue_by_section_report( from_date: Optional[date] = Query(None), to_date: Optional[date] = Query(None), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.MANAGER)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): result = await _run_report(db, "revenue-by-section", current_user, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id) return success(result.as_json()) @router.get("/outstanding-balances") async def outstanding_balances_report( from_date: Optional[date] = Query(None), to_date: Optional[date] = Query(None), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.MANAGER)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): result = await _run_report(db, "outstanding-balances", current_user, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id) return success(result.as_json()) @router.get("/service-schedule") async def service_schedule_report( from_date: Optional[date] = Query(None), to_date: Optional[date] = Query(None), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): result = await _run_report(db, "service-schedule", current_user, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id) return success(result.as_json()) @router.get("/audit-log") async def audit_log_report( from_date: Optional[date] = Query(None), to_date: Optional[date] = Query(None), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.MANAGER)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): result = await _run_report(db, "audit-log", current_user, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id) return success(result.as_json()) @router.get("/memorial-status") async def memorial_status_report( from_date: Optional[date] = Query(None), to_date: Optional[date] = Query(None), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): result = await _run_report(db, "memorial-status", current_user, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id) return success(result.as_json()) # ── Export endpoints ────────────────────────────────────────────────────────── async def _prepare_export( request: Request, db: AsyncSession, report_type: str, current_user: User, tenant_id: str, from_date: Optional[date], to_date: Optional[date], section_id: Optional[UUID], fmt: str, ): """Run the full pre-export gauntlet, then build the report data.""" validate_report_type(report_type) # 422 — allowlist assert_report_role(current_user, report_type) # 403 — per-type RBAC validate_date_range(from_date, to_date) # 422 — bounded range await resolve_section(db, tenant_id, section_id) # 404 — cross-tenant IDOR check_export_rate_limit(f"{tenant_id}:{current_user.id}") # 429 — rate limit result = await _data.build( db, report_type, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id, ) await log_export( db, tenant_id=tenant_id, user=current_user, report_type=report_type, fmt=fmt, from_date=from_date, to_date=to_date, section_id=section_id, request=request, ) return result def _export_headers(report_type: str, ext: str) -> dict: today = datetime.now(timezone.utc).date() filename = safe_filename(report_type, ext, today) return { **EXPORT_SECURITY_HEADERS, "Content-Disposition": f'attachment; filename="{filename}"', } @router.get("/{report_type}/export/pdf") async def export_pdf( request: Request, report_type: str = _REPORT_TYPE_PATH, from_date: Optional[date] = Query(None), to_date: Optional[date] = Query(None), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): result = await _prepare_export( request, db, report_type, current_user, tenant_id, from_date, to_date, section_id, "pdf", ) buffer = _export.generate_pdf(result) return StreamingResponse( buffer, media_type="application/pdf", headers=_export_headers(report_type, "pdf"), ) @router.get("/{report_type}/export/excel") async def export_excel( request: Request, report_type: str = _REPORT_TYPE_PATH, from_date: Optional[date] = Query(None), to_date: Optional[date] = Query(None), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): result = await _prepare_export( request, db, report_type, current_user, tenant_id, from_date, to_date, section_id, "excel", ) buffer = _export.generate_excel(result) return StreamingResponse( buffer, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers=_export_headers(report_type, "xlsx"), ) @router.get("/export/bulk-csv") async def export_bulk_csv( request: Request, from_date: date = Query(..., description="Required. Start of period (ISO date)."), to_date: date = Query(..., description="Required. End of period (ISO date)."), section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(BULK_CSV_MIN_ROLE)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): # Bulk CSV mandates a bounded date range (OWASP API4). validate_date_range(from_date, to_date, require=True) await resolve_section(db, tenant_id, section_id) check_export_rate_limit(f"{tenant_id}:{current_user.id}") await log_export( db, tenant_id=tenant_id, user=current_user, report_type="bulk-csv", fmt="csv", from_date=from_date, to_date=to_date, section_id=section_id, request=request, ) generator = _export.generate_bulk_csv( db, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id, ) return StreamingResponse( generator, media_type="text/csv", headers=_export_headers("bulk-csv", "csv"), ) # ── Legacy summary endpoints (kept; audited for RBAC + tenant isolation) ─────── @router.get("/occupancy") async def occupancy_report( section_id: Optional[UUID] = Query(None), current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """Plot occupancy breakdown by status.""" filters = [Plot.tenant_id == tenant_id] if section_id: await resolve_section(db, tenant_id, section_id) filters.append(Plot.section_id == section_id) result = await db.execute( select(Plot.status, func.count(Plot.id).label("count")) .where(*filters) .group_by(Plot.status) ) rows = result.all() total = sum(r.count for r in rows) breakdown = {r.status: r.count for r in rows} occupied = breakdown.get("occupied", 0) vacant = breakdown.get("vacant", 0) reserved = breakdown.get("reserved", 0) return success({ "total_plots": total, "breakdown": breakdown, "occupancy_rate": round(occupied / total * 100, 2) if total else 0, "available_plots": vacant, "reserved_plots": reserved, }) @router.get("/revenue") async def revenue_report( year: Optional[int] = Query(None), month: Optional[int] = Query(None), current_user: User = Depends(require_min_role(UserRole.MANAGER)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """Revenue summary from invoices.""" filters = [Invoice.tenant_id == tenant_id] if year: filters.append(func.extract("year", Invoice.created_at) == year) if month: filters.append(func.extract("month", Invoice.created_at) == month) result = await db.execute( select( Invoice.status, func.count(Invoice.id).label("count"), func.coalesce(func.sum(Invoice.total_amount), 0).label("total"), func.coalesce(func.sum(Invoice.paid_amount), 0).label("paid"), func.coalesce(func.sum(Invoice.balance_due), 0).label("outstanding"), ) .where(*filters) .group_by(Invoice.status) ) rows = result.all() breakdown = [ { "status": r.status, "count": r.count, "total": float(r.total), "paid": float(r.paid), "outstanding": float(r.outstanding), } for r in rows ] grand_total = sum(r["total"] for r in breakdown) grand_paid = sum(r["paid"] for r in breakdown) grand_outstanding = sum(r["outstanding"] for r in breakdown) return success({ "breakdown": breakdown, "summary": { "total_invoiced": grand_total, "total_collected": grand_paid, "total_outstanding": grand_outstanding, "collection_rate": round(grand_paid / grand_total * 100, 2) if grand_total else 0, }, }) @router.get("/records-summary") async def records_summary( current_user: User = Depends(require_min_role(UserRole.STAFF)), tenant_id: str = Depends(require_tenant), db: AsyncSession = Depends(get_db), ): """Summary counts for records.""" total = ( await db.execute( select(func.count(Record.id)).where( Record.tenant_id == tenant_id, Record.deleted_at.is_(None), ) ) ).scalar_one() veterans = ( await db.execute( select(func.count(Record.id)).where( Record.tenant_id == tenant_id, Record.is_veteran.is_(True), Record.deleted_at.is_(None), ) ) ).scalar_one() return success({ "total_records": total, "veteran_records": veterans, "non_veteran_records": total - veterans, })