Add auth + CSV upload/process API

.dockerignore

@@ -0,0 +1,10 @@
+thematic_analysis/
+.git
+logs/
+downloads/
+outputs/
+.env
+tests/
+__pycache__/
+.pytest_cache/
+.DS_Store

.env.example

@@ -1,5 +1,10 @@
 # Example .env file for analytics_service
 
+# Logging Configuration
+LOG_DIR=logs
+LOG_LEVEL=INFO
+LOG_RETENTION_DAYS=14
+
 # Kafka Configuration
 KAFKA_BOOTSTRAP_SERVERS=localhost:9092
 KAFKA_TOPIC_INGESTION=analytics.ingestion.raw
@@ -23,6 +28,18 @@ BATCH_SIZE=100
 TEMPORAL_HOST=localhost:7233
 TEMPORAL_QUEUE=analytics-processing-queue
 
+# API Authentication — single shared Bearer token checked via secrets.compare_digest
+# in app/api/deps.py. Required (no default) — the app will not start without it.
+AUTH_TOKEN=your-secret-bearer-token-here
+
+# CSV Upload / Processing Configuration
+MAX_CSV_UPLOAD_BYTES=10485760
+CSV_BLOB_UPLOADS=mitra_dashboard_api_output
+CSV_SCHEDULE_CRON_TIME=40 15 * * *
+STORY_CSV_COLUMN=["id","Title","User name","Designation","Location","District","Organization","Report Created At","Objective","Challenges","Action Steps","Impact","Duration","Blurb","masked_blurb","Content","masked_content","Images","Pdf","Transcript Link","Session ID"]
+DISCUSSION_CSV_COLUMN=["id","Title","User name","User Location","District","Participant Count","Men","Women","Children","Date of Discussion","Organization","Challenges","Solutions","Author","Language","Report Created At","Transcript Link","Image Urls","PDF Urls","Session ID"]
+DISCUSSION_PARTICIPANTS_MAP={"men": "Men", "women": "Women", "children": "Children", "teacher": "Teacher", "participant count": "Participant Count"}
+
 # LLM / OpenRouter Configuration
 OPENROUTER_API_KEY=your_openrouter_api_key_here
 OPENROUTER_MODEL=nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free

Dockerfile

@@ -0,0 +1,36 @@
+# syntax=docker/dockerfile:1
+FROM python:3.9-slim
+
+# libgl1/libglib2.0-0: needed by opencv-python (pulled in transitively via `deface`).
+# confluent-kafka needs no extra apt packages — manylinux wheels cover this base image.
+RUN apt-get update && apt-get install -y --no-install-recommends \
+      libgl1 libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1 \
+    HF_HOME=/opt/model-cache/huggingface
+
+WORKDIR /app
+
+COPY requirements-prod.txt .
+RUN pip install -r requirements-prod.txt \
+      --extra-index-url https://download.pytorch.org/whl/cpu
+
+# Bake the embedding model so the running container never needs internet access
+# for it (HF_HOME set above makes the cache path stable between build and run).
+RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('all-MiniLM-L6-v2')"
+
+# deface's centerface.onnx ships inside the pip package (not a runtime download) —
+# this just fails the build loudly if a future deface release changes that.
+RUN python -c "import os, deface.centerface as c; assert os.path.exists(c.default_onnx_path)"
+
+COPY app/ app/
+COPY main.py schema.sql seed_prompts.sql seed_themes.sql run_all.sh ./
+RUN chmod +x run_all.sh
+
+RUN mkdir -p logs downloads
+
+ENTRYPOINT ["python", "main.py"]
+CMD ["--mode", "all"]

app/api/deps.py

@@ -0,0 +1,12 @@
+import secrets
+from fastapi import Depends, HTTPException
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from app.config import settings
+
+security = HTTPBearer()
+
+
+async def verify_auth_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
+    """Verifies that the incoming Bearer token matches AUTH_TOKEN in settings."""
+    if not secrets.compare_digest(credentials.credentials, settings.AUTH_TOKEN):
+        raise HTTPException(status_code=401, detail="Unauthorized: Invalid token")

app/api/exceptions.py

@@ -0,0 +1,111 @@
+import logging
+from typing import List
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+logger = logging.getLogger("analytics_service.api.exceptions")
+
+
+# ---------------------------------------------------------------------------
+# Domain Exceptions
+# ---------------------------------------------------------------------------
+
+class InvalidReportType(Exception):
+    """Raised when report type is not story or discussion."""
+    pass
+
+
+class InvalidFileType(Exception):
+    """Raised when file extension is not csv."""
+    pass
+
+
+class FileTooLarge(Exception):
+    """Raised when upload size exceeds configuration settings."""
+    pass
+
+
+class EmptyFile(Exception):
+    """Raised when the uploaded file contains zero bytes."""
+    pass
+
+
+class DuplicateFile(Exception):
+    """Raised when a CSV file with the same details is already uploaded."""
+    pass
+
+
+class InvalidCsvColumns(Exception):
+    """
+    Raised when the uploaded CSV's header row doesn't match the expected
+    schema (missing and/or unexpected columns). Raised before the file is
+    uploaded to GCS or a csv_uploads row is created — a column mismatch must
+    never leave cloud-storage or tracking-table clutter behind.
+    """
+    def __init__(self, errors: List[str]):
+        self.errors = errors
+        super().__init__("CSV column mismatch — please correct the file and re-upload.")
+
+
+class RecordNotFound(Exception):
+    """Raised when the CSV record cannot be found in database."""
+    pass
+
+
+class RecordAlreadyProcessing(Exception):
+    """Raised when the record is already in_progress."""
+    pass
+
+
+class RecordNotPending(Exception):
+    """Raised when trying to process a record that is not in pending status."""
+    pass
+
+
+# ---------------------------------------------------------------------------
+# FastAPI Exception Handler Registration
+# ---------------------------------------------------------------------------
+
+def register_exception_handlers(app: FastAPI) -> None:
+    """
+    Registers global exception handlers on the FastAPI application instance
+    to translate domain exceptions cleanly into HTTP JSON responses.
+    """
+
+    @app.exception_handler(InvalidReportType)
+    async def invalid_report_type_handler(request: Request, exc: InvalidReportType):
+        return JSONResponse(status_code=400, content={"detail": str(exc)})
+
+    @app.exception_handler(InvalidFileType)
+    async def invalid_file_type_handler(request: Request, exc: InvalidFileType):
+        return JSONResponse(status_code=400, content={"detail": str(exc)})
+
+    @app.exception_handler(EmptyFile)
+    async def empty_file_handler(request: Request, exc: EmptyFile):
+        return JSONResponse(status_code=400, content={"detail": str(exc)})
+
+    @app.exception_handler(DuplicateFile)
+    async def duplicate_file_handler(request: Request, exc: DuplicateFile):
+        return JSONResponse(status_code=400, content={"detail": str(exc)})
+
+    @app.exception_handler(InvalidCsvColumns)
+    async def invalid_csv_columns_handler(request: Request, exc: InvalidCsvColumns):
+        return JSONResponse(status_code=400, content={"detail": str(exc), "errors": exc.errors})
+
+    @app.exception_handler(RecordNotPending)
+    async def record_not_pending_handler(request: Request, exc: RecordNotPending):
+        # Same status class as RecordAlreadyProcessing (409) — both are "request
+        # conflicts with the resource's current status," not a client input error.
+        return JSONResponse(status_code=409, content={"detail": str(exc)})
+
+    @app.exception_handler(FileTooLarge)
+    async def file_too_large_handler(request: Request, exc: FileTooLarge):
+        return JSONResponse(status_code=413, content={"detail": str(exc)})
+
+    @app.exception_handler(RecordNotFound)
+    async def record_not_found_handler(request: Request, exc: RecordNotFound):
+        return JSONResponse(status_code=404, content={"detail": str(exc)})
+
+    @app.exception_handler(RecordAlreadyProcessing)
+    async def record_already_processing_handler(request: Request, exc: RecordAlreadyProcessing):
+        return JSONResponse(status_code=409, content={"detail": str(exc)})

app/api/models/uploads.py

@@ -0,0 +1,16 @@
+from typing import Optional
+from pydantic import BaseModel
+
+
+class UploadResponse(BaseModel):
+    message: str
+    id: int
+    status: str
+
+
+class CsvUploadStatus(BaseModel):
+    id: int
+    status: str
+    report_type: str
+    cloud_storage_path: str
+    created_at: Optional[str] = None

app/api/response.py

@@ -0,0 +1,20 @@
+from typing import Any, Optional
+from fastapi.responses import JSONResponse
+
+
+def response_builder(
+    data: Optional[Any] = None,
+    message: str = "Success",
+    errors: Optional[Any] = None,
+    status_code: int = 200,
+) -> JSONResponse:
+    """
+    Standard envelope response builder utility for API endpoints.
+    """
+    payload = {
+        "status_code": status_code,
+        "message": message,
+        "data": data,
+        "errors": errors,
+    }
+    return JSONResponse(status_code=status_code, content=payload)

app/api/router.py

@@ -0,0 +1,5 @@
+from fastapi import APIRouter
+from app.api.routes.uploads import uploads_router
+
+api_router = APIRouter()
+api_router.include_router(uploads_router)