feat: implement periodic report generation API (#557)

app/api/router.py

@@ -1,6 +1,16 @@
 from fastapi import APIRouter
 
-from app.api.routes import forms, input, jobs, llm, system, templates, weather, zipcode
+from app.api.routes import (
+    forms,
+    input,
+    jobs,
+    llm,
+    reports,
+    system,
+    templates,
+    weather,
+    zipcode,
+)
 from app.core.config import API_PREFIX
 
 api_router = APIRouter()
@@ -11,4 +21,5 @@
 api_router.include_router(jobs.router, prefix=API_PREFIX)
 api_router.include_router(weather.router, prefix=API_PREFIX)
 api_router.include_router(zipcode.router, prefix=API_PREFIX)
-api_router.include_router(input.router, prefix=API_PREFIX)
+api_router.include_router(input.router, prefix=API_PREFIX)
+api_router.include_router(reports.router, prefix=API_PREFIX)

app/api/routes/reports.py

@@ -0,0 +1,149 @@
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import UUID
+
+from fastapi import APIRouter, Depends
+from fastapi.responses import FileResponse, JSONResponse
+from sqlmodel import Session
+
+from app.api.deps import get_db
+from app.api.schemas.common import AsyncJobResponse
+from app.api.schemas.enums import JobStatus, JobType, OutputFormat
+from app.api.schemas.reporting import (
+    GenerateReportRequest,
+    PeriodicReport,
+    ReportProgressResponse,
+)
+from app.core.config import API_PREFIX
+from app.core.errors.base import AppError
+from app.db.repositories import create_job, create_report, get_report, update_job, update_report
+from app.models import Job, Report
+from app.services.reporting import ReportingService
+from app.tasks.report import generate_periodic_report
+
+router = APIRouter(prefix="/reports", tags=["reporting"])
+
+
+@router.post(
+    "/generate",
+    response_model=AsyncJobResponse,
+    status_code=202,
+    responses={
+        202: {"description": "Report generation job accepted", "model": AsyncJobResponse},
+    },
+)
+def generate_report(body: GenerateReportRequest, db: Session = Depends(get_db)):
+    # Validate request bounds and check future date
+    ReportingService().validate_and_compute_bounds(body)
+
+    # 1. Create Job record
+    job = Job(
+        job_type=JobType.report_generation,
+        status=JobStatus.queued,
+    )
+    job.celery_task_id = job.job_id
+    job = create_job(db, job)
+
+    # 2. Create Report record
+    report = Report(
+        period_type=body.period_type,
+        year=body.year,
+        month=body.month,
+        quarter=body.quarter,
+        output_format=body.format,
+        status=JobStatus.queued,
+        job_id=UUID(job.job_id),
+    )
+    report = create_report(db, report)
+
+    # 3. Dispatch background task
+    try:
+        generate_periodic_report.apply_async(
+            (str(report.report_id),), task_id=job.celery_task_id
+        )
+    except Exception as e:
+        job.status = JobStatus.failed
+        job.error = {"error_code": "QUEUE_UNAVAILABLE", "message": str(e)}
+        job.updated_at = datetime.now(timezone.utc)
+        update_job(db, job)
+
+        report.status = JobStatus.failed
+        report.updated_at = datetime.now(timezone.utc)
+        update_report(db, report)
+
+        raise AppError(
+            f"Could not queue report generation job: {e}",
+            status_code=503,
+            error_code="QUEUE_UNAVAILABLE",
+        ) from e
+
+    return AsyncJobResponse(
+        job_id=job.job_id,
+        job_type=job.job_type,
+        status=JobStatus.processing,
+        estimated_seconds=30,
+        poll_url=f"{API_PREFIX}/reports/{report.report_id}",
+        report_id=str(report.report_id),
+    )
+
+
+@router.get(
+    "/{report_id}",
+    responses={
+        200: {
+            "description": "Generated report (PDF binary or JSON)",
+            "content": {
+                "application/pdf": {"schema": {"type": "string", "format": "binary"}},
+                "application/json": {"schema": {"$ref": "#/components/schemas/PeriodicReport"}},
+            },
+        },
+        202: {
+            "description": "Report still being generated",
+            "model": ReportProgressResponse,
+        },
+    },
+)
+def get_report_by_id(report_id: UUID, db: Session = Depends(get_db)):
+    report = get_report(db, report_id)
+    if not report:
+        raise AppError(
+            f"Report {report_id} not found",
+            status_code=404,
+            error_code="REPORT_NOT_FOUND",
+        )
+
+    if report.status in (JobStatus.queued, JobStatus.processing):
+        return JSONResponse(
+            status_code=202,
+            content={
+                "message": "Report generation is still in progress",
+                "status": "processing",
+                "retry_after_seconds": 10,
+            },
+        )
+
+    if report.status == JobStatus.failed:
+        raise AppError(
+            "Report generation failed",
+            status_code=500,
+            error_code="REPORT_FAILED",
+        )
+
+    # Completed
+    if report.output_format == OutputFormat.pdf:
+        if not report.pdf_path or not Path(report.pdf_path).exists():
+            raise AppError(
+                "Generated PDF report file not found",
+                status_code=500,
+                error_code="PDF_NOT_FOUND",
+            )
+        filename = Path(report.pdf_path).name
+        return FileResponse(
+            path=report.pdf_path,
+            media_type="application/pdf",
+            filename=filename,
+            headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+        )
+
+    # OutputFormat.json or OutputFormat.both
+    return PeriodicReport.model_validate(report.json_data)

app/api/schemas/common.py

@@ -41,6 +41,7 @@ class AsyncJobResponse(BaseModel):
     status: JobStatus
     estimated_seconds: int | None = None
     poll_url: str | None = None
+    report_id: str | None = None
 
 
 class Job(BaseModel):

app/api/schemas/reporting.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from datetime import date, datetime
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.api.schemas.enums import OutputFormat, PeriodType
+
+
+class GenerateReportRequest(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    period_type: PeriodType
+    year: int = Field(ge=2000, le=2100)
+    month: int | None = Field(default=None, ge=1, le=12)
+    quarter: int | None = Field(default=None, ge=1, le=4)
+    format: OutputFormat = Field(default=OutputFormat.pdf)
+
+
+class ReportSummaryGroup(BaseModel):
+    model_config = ConfigDict()
+
+    period: str
+    incident_count: int
+    forms_generated: int
+
+
+class ReportSummary(BaseModel):
+    model_config = ConfigDict()
+
+    date_from: date
+    date_to: date
+    total_incidents: int
+    by_type: dict[str, int]
+    by_status: dict[str, int]
+    forms_generated: int
+    avg_completeness_score: float
+    avg_processing_time_seconds: float
+    groups: list[ReportSummaryGroup]
+
+
+class IncidentSummaryItem(BaseModel):
+    model_config = ConfigDict()
+
+    incident_id: UUID
+    incident_number: str | None = None
+    incident_date: date | None = None
+    incident_type: str | None = None
+    status: str | None = None
+    forms_generated: int = 0
+
+
+class ReportCompliance(BaseModel):
+    model_config = ConfigDict()
+
+    neris_submission_rate: float
+    average_submission_delay_days: float
+    overdue_incidents: int
+
+
+class PeriodicReport(BaseModel):
+    model_config = ConfigDict()
+
+    report_id: UUID
+    period_type: PeriodType
+    period_label: str
+    generated_at: datetime
+    summary: ReportSummary
+    incidents: list[IncidentSummaryItem]
+    compliance: ReportCompliance
+
+
+class ReportProgressResponse(BaseModel):
+    model_config = ConfigDict()
+
+    message: str = "Report generation is still in progress"
+    status: str = "processing"
+    retry_after_seconds: int = 10

app/core/celery.py

@@ -18,7 +18,12 @@
     result_expires=86400,
 )
 
-celery_app.conf.include = ["app.tasks.fill", "app.tasks.purge", "app.tasks.transcribe"]
+celery_app.conf.include = [
+    "app.tasks.fill",
+    "app.tasks.purge",
+    "app.tasks.transcribe",
+    "app.tasks.report",
+]
 
 
 celery_app.conf.beat_schedule = {

app/core/config.py

@@ -104,6 +104,9 @@ def _provider_env(prefix: str, **defaults: str) -> dict[str, str]:
 # Voice input audio files land here: {AUDIO_DIR}/{input_id}.{ext}
 AUDIO_DIR = DATA_DIR / "audio"
 
+# Generated periodic reports land here: {REPORTS_DIR}/{filename}
+REPORTS_DIR = DATA_DIR / "reports"
+
 # Advisory estimate returned in VoiceInputResponse.estimated_processing_seconds.
 ESTIMATED_TRANSCRIPTION_SECONDS = int(os.getenv("ESTIMATED_TRANSCRIPTION_SECONDS", "30"))
 

app/db/repositories.py

@@ -1,9 +1,10 @@
-from datetime import datetime
+from datetime import date, datetime, timezone
 from uuid import UUID
 
+from sqlalchemy import and_, or_
 from sqlmodel import Session, select
 
-from app.models import FormSubmission, Input, Job, Template
+from app.models import Extraction, Form, FormSubmission, Incident, Input, Job, Report, Template
 
 
 # Templates
@@ -128,3 +129,61 @@ def update_input(session: Session, input_obj: Input) -> Input:
     session.refresh(input_obj)
     return input_obj
 
+
+# Reports
+def create_report(session: Session, report: Report) -> Report:
+    session.add(report)
+    session.commit()
+    session.refresh(report)
+    return report
+
+
+def get_report(session: Session, report_id: UUID) -> Report | None:
+    return session.get(Report, report_id)
+
+
+def update_report(session: Session, report: Report) -> Report:
+    session.add(report)
+    session.commit()
+    session.refresh(report)
+    return report
+
+
+# Incidents & Forms for Reporting
+def get_incidents_in_date_range(session: Session, start_date: date, end_date: date) -> list[Incident]:
+    statement = (
+        select(Incident)
+        .where(
+            Incident.deleted_at.is_(None),
+            or_(
+                Incident.incident_date.between(start_date, end_date),
+                and_(
+                    Incident.incident_date.is_(None),
+                    Incident.created_at >= datetime.combine(start_date, datetime.min.time(), tzinfo=timezone.utc),
+                    Incident.created_at <= datetime.combine(end_date, datetime.max.time(), tzinfo=timezone.utc),
+                ),
+            ),
+        )
+        .order_by(Incident.incident_date.asc(), Incident.created_at.asc())
+    )
+    return list(session.exec(statement).all())
+
+
+def get_forms_for_incident(session: Session, incident_id: UUID) -> list[Form]:
+    statement = select(Form).where(Form.incident_id == incident_id)
+    return list(session.exec(statement).all())
+
+
+def get_forms_for_incidents(session: Session, incident_ids: list[UUID]) -> list[Form]:
+    if not incident_ids:
+        return []
+    statement = select(Form).where(Form.incident_id.in_(incident_ids))
+    return list(session.exec(statement).all())
+
+
+def get_extractions_by_ids(session: Session, extract_ids: list[UUID]) -> list[Extraction]:
+    if not extract_ids:
+        return []
+    statement = select(Extraction).where(Extraction.extract_id.in_(extract_ids))
+    return list(session.exec(statement).all())
+