From 9fb449f7e078eec28b6b486dd4c8c4b90ed81ee1 Mon Sep 17 00:00:00 2001 From: Vivek M <125434153+Vivek-M-08@users.noreply.github.com> Date: Wed, 22 Jul 2026 15:20:04 +0530 Subject: [PATCH 1/3] FIX: Store story challenge/action_steps as TEXT[] instead of flattened TEXT --- app/database/operations.py | 4 ++-- app/temporal/pii_and_abusive_activity.py | 9 +++---- app/temporal/story_rating_activity.py | 30 ++++++++++++++++++------ app/temporal/thematic_activity.py | 23 ++++++++---------- schema.sql | 4 ++-- 5 files changed, 42 insertions(+), 28 deletions(-) diff --git a/app/database/operations.py b/app/database/operations.py index 8f38074..6634294 100644 --- a/app/database/operations.py +++ b/app/database/operations.py @@ -301,8 +301,8 @@ async def insert_or_update_submission( "SELECT 1 FROM story_submissions WHERE submission_id = $1 AND tenant_code = $2", submission_id, tenant_code ) - challenges_joined = _normalize_string_list(data.get("challenges")) - action_steps_joined = _normalize_string_list(data.get("actionSteps")) + challenges_joined = _normalize_statement_list(data.get("challenges")) + action_steps_joined = _normalize_statement_list(data.get("actionSteps")) image_urls = _normalize_media_url_list(data.get("imageUrls")) pdf_urls, masked_pdf_urls = _normalize_pdf_urls(data.get("pdfUrls")) diff --git a/app/temporal/pii_and_abusive_activity.py b/app/temporal/pii_and_abusive_activity.py index ec25682..5c306a6 100644 --- a/app/temporal/pii_and_abusive_activity.py +++ b/app/temporal/pii_and_abusive_activity.py @@ -32,10 +32,11 @@ def _get_case_insensitive_key(d: dict, key: str) -> Any: def _parse_statement_list(raw_value: Any) -> Optional[List[str]]: """ - Returns raw_value if it's already a list — discussion columns like - challenges/solutions are stored as TEXT[] (per operations.py's - _normalize_statement_list), which asyncpg auto-decodes to a native Python list. - Returns None for scalar columns (a story's objective/challenge are plain text). + Returns raw_value if it's already a list — statement columns like + discussion's challenges/solutions and story's challenge/action_steps are + stored as TEXT[] (per operations.py's _normalize_statement_list), which + asyncpg auto-decodes to a native Python list. Returns None for scalar + columns (e.g. a story's objective, which is plain text). """ return raw_value if isinstance(raw_value, list) else None diff --git a/app/temporal/story_rating_activity.py b/app/temporal/story_rating_activity.py index 89850c8..89182f5 100644 --- a/app/temporal/story_rating_activity.py +++ b/app/temporal/story_rating_activity.py @@ -62,14 +62,30 @@ def _truncate_text(text: str, max_chars: int = settings.MAX_PDF_TEXT_CHARS) -> s return text[:max_chars] + "\n\n[... content truncated ...]" -def _build_fallback_text(challenge: Optional[str], action_steps: Optional[str], impact: Optional[str]) -> str: +def _stringify_field(value: Any) -> str: + """ + Renders a submission field as display text for the fallback prompt. challenge/ + action_steps are stored as TEXT[] (one element per discrete statement, see + operations.py's _normalize_statement_list) and come back from asyncpg as a + native Python list — joined here with newlines purely for human-readable + display, not for storage, so no round-trip ambiguity is introduced. + """ + if isinstance(value, list): + return "\n".join(str(v).strip() for v in value if v and str(v).strip()) + return str(value).strip() if value else "" + + +def _build_fallback_text(challenge: Any, action_steps: Any, impact: Optional[str]) -> str: parts = [] - if challenge and str(challenge).strip(): - parts.append(f"Challenges and Issues:\n{str(challenge).strip()}\n") - if action_steps and str(action_steps).strip(): - parts.append(f"Action Steps Taken:\n{str(action_steps).strip()}\n") - if impact and str(impact).strip(): - parts.append(f"Impact and Outcomes:\n{str(impact).strip()}\n") + challenge_text = _stringify_field(challenge) + if challenge_text: + parts.append(f"Challenges and Issues:\n{challenge_text}\n") + action_steps_text = _stringify_field(action_steps) + if action_steps_text: + parts.append(f"Action Steps Taken:\n{action_steps_text}\n") + impact_text = _stringify_field(impact) + if impact_text: + parts.append(f"Impact and Outcomes:\n{impact_text}\n") return "\n".join(parts) diff --git a/app/temporal/thematic_activity.py b/app/temporal/thematic_activity.py index 304ed31..d7915f8 100644 --- a/app/temporal/thematic_activity.py +++ b/app/temporal/thematic_activity.py @@ -713,20 +713,17 @@ async def thematic_classification_activity(params: Dict[str, Any]) -> Dict[str, # --- Step 1b: Build the list of statements to classify --- is_discussion = "discussion" in sub_type - if is_discussion: - # challenges/solutions are stored as TEXT[] (see operations.py's - # _normalize_statement_list) — asyncpg returns them as a native Python - # list already, one element per discrete statement. No delimiter/JSON - # parsing needed (and none would be safe, since a statement could - # legitimately contain any given delimiter character itself). - if isinstance(raw_value, list): - statements = [str(s).strip() for s in raw_value if s and str(s).strip()] - else: - # Defensive fallback in case of unexpected non-array data - fallback = str(raw_value).strip() - statements = [fallback] if fallback else [] + # Branch on the column's actual runtime shape, not submission type — + # TEXT[] columns (discussion's challenges/solutions, story's + # challenge/action_steps; see operations.py's _normalize_statement_list) + # come back from asyncpg as a native Python list, one element per + # discrete statement. No delimiter/JSON parsing needed (and none would + # be safe, since a statement could legitimately contain any given + # delimiter character itself). Scalar TEXT columns (e.g. a story's + # objective) are processed as a single unit. + if isinstance(raw_value, list): + statements = [str(s).strip() for s in raw_value if s and str(s).strip()] else: - # Story types are single-string columns — process as one unit raw_text = str(raw_value).strip() statements = [raw_text] if raw_text else [] diff --git a/schema.sql b/schema.sql index 3439256..ddad346 100644 --- a/schema.sql +++ b/schema.sql @@ -126,8 +126,8 @@ CREATE TABLE story_submissions ( tenant_code TEXT NOT NULL, title TEXT, objective TEXT, - challenge TEXT, - action_steps TEXT, + challenge TEXT[], -- one array element per discrete statement, same format as discussion_submissions.challenges (see operations.py's _normalize_statement_list) + action_steps TEXT[], -- same format as challenge impact TEXT, duration TEXT, blurb TEXT, From dccb06760e70b47e7e18b868de5a8d955bbfa520 Mon Sep 17 00:00:00 2001 From: Vivek M <125434153+Vivek-M-08@users.noreply.github.com> Date: Wed, 22 Jul 2026 16:49:51 +0530 Subject: [PATCH 2/3] Add Docker deployment: elevate-analytics image, containerized Kafka, split/single-container app modes Dockerfile + requirements-prod.txt build the app image (CPU-only torch, trimmed unused deps, embedding model baked in). docker-compose.yaml adds a KRaft Kafka broker and analytics-web/consumer/worker as three containers by default (COMPOSE_PROFILES=split), with run_all.sh + analytics-all as a single-container alternative (COMPOSE_PROFILES=single). Postgres stays host-local via host.docker.internal, same as temporal already did. --- .dockerignore | 10 ++++ Dockerfile | 36 ++++++++++++ docker-compose.yaml | 129 +++++++++++++++++++++++++++++++++++++++++- requirements-prod.txt | 15 +++++ run_all.sh | 16 ++++++ 5 files changed, 205 insertions(+), 1 deletion(-) create mode 100644 .dockerignore create mode 100644 Dockerfile create mode 100644 requirements-prod.txt create mode 100644 run_all.sh diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..a9c5ff4 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,10 @@ +thematic_analysis/ +.git +logs/ +downloads/ +outputs/ +.env +tests/ +__pycache__/ +.pytest_cache/ +.DS_Store diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..de07252 --- /dev/null +++ b/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"] diff --git a/docker-compose.yaml b/docker-compose.yaml index 728c75f..94a10a5 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -17,6 +17,15 @@ services: - TEMPORAL_VISIBILITY_DB=temporal_visibility extra_hosts: - "host.docker.internal:host-gateway" + healthcheck: + # The auto-setup image binds the frontend service to the container's own + # network IP, not loopback — so a /dev/tcp/127.0.0.1 probe never connects. + # Check /proc/net/tcp directly for a LISTEN (0A) socket on port 7233 (0x1C41). + test: ["CMD-SHELL", "awk '$2 ~ /:1C41$/ && $4==\"0A\" {f=1} END{exit !f}' /proc/net/tcp"] + interval: 5s + timeout: 5s + retries: 20 + start_period: 15s temporal-ui: image: temporalio/ui:2.34.0 @@ -27,4 +36,122 @@ services: - TEMPORAL_ADDRESS=temporal:7233 - TEMPORAL_UI_PORT=8080 depends_on: - - temporal \ No newline at end of file + - temporal + + # kafka: + # image: apache/kafka:3.9.0 + # container_name: kafka + # ports: + # - "29092:29092" # host-facing listener only — do NOT also publish 9092 + # environment: + # KAFKA_NODE_ID: 1 + # KAFKA_PROCESS_ROLES: broker,controller + # KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,PLAINTEXT_HOST://:29092 + # KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092 + # KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT + # KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER + # KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT + # KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 + # KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + # KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 + # CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk" + # volumes: + # - kafka_data:/var/lib/kafka/data + # healthcheck: + # test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server kafka:9092 || exit 1"] + # interval: 10s + # timeout: 5s + # retries: 10 + # start_period: 20s + + # # --- Split mode: one container per service (default — fault isolation, + # # independent scaling/restarts). Active when COMPOSE_PROFILES=split (see .env). + # analytics-web: + # profiles: ["split"] + # build: . + # image: elevate-analytics:latest + # env_file: .env + # environment: + # DATABASE_URL: postgresql://postgres:postgres@host.docker.internal:5432/analytics_db + # TEMPORAL_HOST: temporal:7233 + # KAFKA_BOOTSTRAP_SERVERS: kafka:9092 + # extra_hosts: + # - "host.docker.internal:host-gateway" + # depends_on: + # temporal: + # condition: service_healthy + # kafka: + # condition: service_healthy + # ports: + # - "8000:8000" + # volumes: + # - ./logs:/app/logs + # command: ["--mode", "web"] + + # analytics-consumer: + # profiles: ["split"] + # build: . + # image: elevate-analytics:latest + # env_file: .env + # environment: + # DATABASE_URL: postgresql://postgres:postgres@host.docker.internal:5432/analytics_db + # TEMPORAL_HOST: temporal:7233 + # KAFKA_BOOTSTRAP_SERVERS: kafka:9092 + # extra_hosts: + # - "host.docker.internal:host-gateway" + # depends_on: + # temporal: + # condition: service_healthy + # kafka: + # condition: service_healthy + # volumes: + # - ./logs:/app/logs + # command: ["--mode", "consumer"] + + # analytics-worker: + # profiles: ["split"] + # build: . + # image: elevate-analytics:latest + # env_file: .env + # environment: + # DATABASE_URL: postgresql://postgres:postgres@host.docker.internal:5432/analytics_db + # TEMPORAL_HOST: temporal:7233 + # KAFKA_BOOTSTRAP_SERVERS: kafka:9092 + # extra_hosts: + # - "host.docker.internal:host-gateway" + # depends_on: + # temporal: + # condition: service_healthy + # kafka: + # condition: service_healthy + # volumes: + # - ./logs:/app/logs + # command: ["--mode", "worker"] + + # --- Single mode: one container, all three services as separate processes + # (run_all.sh). Active when COMPOSE_PROFILES=single (see .env). Do not run + # this alongside the split services above — both bind host port 8000. + # analytics-all: + # profiles: ["single"] + # build: . + # image: elevate-analytics:latest + # env_file: .env + # environment: + # DATABASE_URL: postgresql://postgres:postgres@host.docker.internal:5432/analytics_db + # TEMPORAL_HOST: temporal:7233 + # KAFKA_BOOTSTRAP_SERVERS: kafka:9092 + # extra_hosts: + # - "host.docker.internal:host-gateway" + # depends_on: + # temporal: + # condition: service_healthy + # kafka: + # condition: service_healthy + # ports: + # - "8000:8000" + # volumes: + # - ./logs:/app/logs + # entrypoint: ["/app/run_all.sh"] + +volumes: + kafka_data: diff --git a/requirements-prod.txt b/requirements-prod.txt new file mode 100644 index 0000000..201ddf8 --- /dev/null +++ b/requirements-prod.txt @@ -0,0 +1,15 @@ +temporalio>=1.18.0 +confluent-kafka>=2.4.0 +asyncpg>=0.29.0 +pydantic-settings>=2.0.0 +deface>=1.5.0 +python-dotenv>=1.0.1 +fastapi>=0.112.0 +python-multipart>=0.0.9 +uvicorn>=0.30.0 +sentence-transformers==3.0.1 +scikit-learn +torch +numpy>=1.24.0 +google-cloud-storage +pypdf>=4.0.0 diff --git a/run_all.sh b/run_all.sh new file mode 100644 index 0000000..392844a --- /dev/null +++ b/run_all.sh @@ -0,0 +1,16 @@ +#!/bin/bash +# Runs web/consumer/worker as three separate processes inside one container +# (single-container alternative to the split analytics-web/consumer/worker +# services). Each still gets its own log pair via configure_logging(args.mode) +# since every process calls `python main.py --mode ` independently. +set -e + +python main.py --mode web & +python main.py --mode consumer & +python main.py --mode worker & + +# Exit as soon as ANY one of the three dies, propagating its exit code, so a +# crashed sub-process actually surfaces as a container failure instead of the +# container silently running in a degraded state forever. +wait -n +exit $? From ad723b6b718d268c0fb46adefb6b9a54087502e2 Mon Sep 17 00:00:00 2001 From: Vivek M <125434153+Vivek-M-08@users.noreply.github.com> Date: Wed, 22 Jul 2026 20:20:58 +0530 Subject: [PATCH 3/3] Add auth + CSV upload/process API, ported from analytics-service-1 [Rohan's codebase] Adopts the sibling project's layered app/api/ structure (routes/services/ validators/models) for a real Bearer-token-protected CSV upload pipeline (POST /v1/upload/, POST /v1/process/csv/{id}), replacing the old unauthenticated /api/submissions/trigger and mock /api/bulk/upload. Extracts the Kafka ingestion-schema validator into app/services/ingestion_validation.py so the CSV pipeline checks each row against STORY_KAFKA_SCHEMA/DISCUSSION_KAFKA_SCHEMA before publishing, not just the consumer after receiving. Stops auto-generating a session ID when missing from a CSV row so it's caught by that check instead of silently faked, and rejects bad-column CSVs before any GCS/DB write. Fixes made during the port: confluent_kafka.Producer instead of a second Kafka client library; blocking GCS/pandas calls wrapped in asyncio.to_thread inside Temporal activities; RecordNotPending -> 409 (was inconsistently 400). --- .env.example | 17 + app/api/bulk.py | 40 -- app/api/deps.py | 12 + app/api/exceptions.py | 111 ++++++ app/api/models/__init__.py | 0 app/api/models/uploads.py | 16 + app/api/response.py | 20 + app/api/router.py | 5 + app/api/routes.py | 58 --- app/api/routes/__init__.py | 0 app/api/routes/uploads.py | 61 +++ app/api/services/__init__.py | 0 app/api/services/uploads.py | 491 ++++++++++++++++++++++++ app/api/validators/__init__.py | 0 app/api/validators/uploads.py | 73 ++++ app/config.py | 69 ++++ app/database/operations.py | 171 +++++++++ app/kafka/consumer.py | 65 +--- app/services/gcp_storage.py | 32 ++ app/services/ingestion_validation.py | 68 ++++ app/temporal/csv_processing_activity.py | 295 ++++++++++++++ app/temporal/worker.py | 80 +++- app/temporal/workflows.py | 111 ++++++ main.py | 8 +- requirements-prod.txt | 1 + schema.sql | 37 ++ tests/TEST_CASES.csv | 94 +++++ 27 files changed, 1757 insertions(+), 178 deletions(-) delete mode 100644 app/api/bulk.py create mode 100644 app/api/deps.py create mode 100644 app/api/exceptions.py create mode 100644 app/api/models/__init__.py create mode 100644 app/api/models/uploads.py create mode 100644 app/api/response.py create mode 100644 app/api/router.py delete mode 100644 app/api/routes.py create mode 100644 app/api/routes/__init__.py create mode 100644 app/api/routes/uploads.py create mode 100644 app/api/services/__init__.py create mode 100644 app/api/services/uploads.py create mode 100644 app/api/validators/__init__.py create mode 100644 app/api/validators/uploads.py create mode 100644 app/services/ingestion_validation.py create mode 100644 app/temporal/csv_processing_activity.py diff --git a/.env.example b/.env.example index 23579e7..2f4c34c 100644 --- a/.env.example +++ b/.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 diff --git a/app/api/bulk.py b/app/api/bulk.py deleted file mode 100644 index 240c638..0000000 --- a/app/api/bulk.py +++ /dev/null @@ -1,40 +0,0 @@ -import logging -from fastapi import APIRouter, UploadFile, File, HTTPException -from pydantic import BaseModel - -logger = logging.getLogger("analytics_service.api.bulk") -router = APIRouter(prefix="/api/bulk", tags=["Bulk Uploads"]) - -class BulkUploadResponse(BaseModel): - status: str - message: str - processed_rows: int - csv_upload_id: str - -@router.post("/upload") -async def upload_csv_submissions( - tenant_code: str, - file: UploadFile = File(...) -) -> BulkUploadResponse: - """ - Skelton endpoint to upload a CSV sheet containing bulk submissions. - """ - logger.info(f"API: Received bulk CSV upload request from tenant {tenant_code} with file {file.filename}") - - if not file.filename.endswith(".csv"): - raise HTTPException(status_code=400, detail="Only CSV files are allowed.") - - # In a full implementation, this step would: - # 1. Parse the uploaded CSV using pandas or csv.DictReader. - # 2. Bulk insert records into submissions & type-specific tables with 'pending' status. - # 3. Create a csv_uploads log entry. - # 4. Trigger the batch Temporal schedule/workflow to process them. - - # Placeholder return - mock_csv_upload_id = "csv-uuid-placeholder-12345" - return BulkUploadResponse( - status="success", - message="CSV upload accepted and queued for batch execution.", - processed_rows=100, # Mock processed count - csv_upload_id=mock_csv_upload_id - ) diff --git a/app/api/deps.py b/app/api/deps.py new file mode 100644 index 0000000..f1c5eed --- /dev/null +++ b/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") diff --git a/app/api/exceptions.py b/app/api/exceptions.py new file mode 100644 index 0000000..4f829d6 --- /dev/null +++ b/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)}) diff --git a/app/api/models/__init__.py b/app/api/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/models/uploads.py b/app/api/models/uploads.py new file mode 100644 index 0000000..0bc6261 --- /dev/null +++ b/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 diff --git a/app/api/response.py b/app/api/response.py new file mode 100644 index 0000000..8e3ebf2 --- /dev/null +++ b/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) diff --git a/app/api/router.py b/app/api/router.py new file mode 100644 index 0000000..7108b50 --- /dev/null +++ b/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) diff --git a/app/api/routes.py b/app/api/routes.py deleted file mode 100644 index 3c887ef..0000000 --- a/app/api/routes.py +++ /dev/null @@ -1,58 +0,0 @@ -import logging -from fastapi import APIRouter, HTTPException -from pydantic import BaseModel -from temporalio.client import Client - -from app.config import settings -from app.temporal.workflows import ConfigDrivenProcessingWorkflow - -logger = logging.getLogger("analytics_service.api.routes") -router = APIRouter(prefix="/api/submissions", tags=["Submissions"]) - -class ManualTriggerRequest(BaseModel): - submission_id: str - tenant_code: str - submission_type: str - -@router.post("/trigger") -async def trigger_submission_manually(request: ManualTriggerRequest): - """ - Manually triggers real-time processing workflow for a submission. - """ - logger.info(f"API: Received manual trigger request for submission {request.submission_id}") - - # 1. Resolve steps config based on type - process_steps = settings.get_process_config(request.submission_type) - if not process_steps: - raise HTTPException( - status_code=400, - detail=f"No process configuration found for submission type: {request.submission_type}" - ) - - # 2. Connect to Temporal and trigger - try: - client = await Client.connect(settings.TEMPORAL_HOST) - workflow_id = f"manual-{request.submission_id}-{request.tenant_code}" - - handle = await client.start_workflow( - ConfigDrivenProcessingWorkflow.run, - { - "submission_id": request.submission_id, - "tenant_code": request.tenant_code, - "process_steps": process_steps - }, - id=workflow_id, - task_queue=settings.TEMPORAL_QUEUE - ) - return { - "status": "success", - "message": "Workflow started successfully", - "workflow_id": handle.id, - "run_id": handle.first_execution_run_id - } - except Exception as e: - logger.error(f"API: Manual trigger failed: {e}") - raise HTTPException( - status_code=500, - detail=f"Failed to start workflow: {str(e)}" - ) diff --git a/app/api/routes/__init__.py b/app/api/routes/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/routes/uploads.py b/app/api/routes/uploads.py new file mode 100644 index 0000000..d2604f1 --- /dev/null +++ b/app/api/routes/uploads.py @@ -0,0 +1,61 @@ +import logging +from fastapi import APIRouter, Depends, Form, UploadFile, File +from fastapi.security import HTTPAuthorizationCredentials + +from app.api.deps import verify_auth_token +from app.api.validators.uploads import ( + validate_report_type, + validate_extension, + validate_file_bytes, +) +from app.api.models.uploads import UploadResponse +from app.api.services import uploads as upload_service +from app.config import settings + +logger = logging.getLogger("analytics_service.api.routes.uploads") + +uploads_router = APIRouter(prefix="/v1", tags=["CSV Pipeline"]) + + +@uploads_router.post("/upload/", response_model=UploadResponse) +async def upload_report( + report_type: str = Form(...), + program_name: str = Form(...), + leader_category: str = Form(...), + tenant_code: str = Form(default="mitra"), + file: UploadFile = File(...), + _token: HTTPAuthorizationCredentials = Depends(verify_auth_token), +): + """ + Upload a CSV report file. + + The file is stored in GCS and a tracking record is created with + status='pending' (validation passed) or status='on_hold' (validation failed). + """ + # Pure request-shape checks + report_type = validate_report_type(report_type) + validate_extension(file.filename) + + file_bytes = await file.read(settings.MAX_CSV_UPLOAD_BYTES + 1) + validate_file_bytes(file_bytes) + + # Delegate to domain service + return await upload_service.handle_upload( + report_type=report_type, + program_name=program_name, + leader_category=leader_category, + tenant_code=tenant_code, + file_name=file.filename, + file_bytes=file_bytes, + ) + + +@uploads_router.post("/process/csv/{record_id}") +async def push_record( + record_id: int, + _token: HTTPAuthorizationCredentials = Depends(verify_auth_token), +): + """ + Manually trigger processing for a specific csv_upload record in pending status. + """ + return await upload_service.handle_push(record_id) diff --git a/app/api/services/__init__.py b/app/api/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/services/uploads.py b/app/api/services/uploads.py new file mode 100644 index 0000000..78eb44d --- /dev/null +++ b/app/api/services/uploads.py @@ -0,0 +1,491 @@ +from typing import Any, Dict, List, Optional, Union +import asyncio +import io +import json +import logging +from datetime import datetime +import pandas as pd +from temporalio.client import Client + +from app.config import settings +from app.api.validators.uploads import validate_columns +from app.services.gcp_storage import upload_csv +from app.database import operations +from app.api.exceptions import ( + DuplicateFile, + InvalidCsvColumns, + RecordNotFound, + RecordAlreadyProcessing, + RecordNotPending, +) + +logger = logging.getLogger("analytics_service.api.services.uploads") + + +# --------------------------------------------------------------------------- +# CSV Processing & Formatting Helpers +# --------------------------------------------------------------------------- + +def load_csv(csv_file: Union[io.BytesIO, bytes]) -> pd.DataFrame: + """Parse an in-memory CSV (BytesIO or raw bytes from GCS) into a DataFrame.""" + if isinstance(csv_file, bytes): + csv_file = io.BytesIO(csv_file) + else: + csv_file.seek(0) + return pd.read_csv(csv_file) + + +def split_csv(df: pd.DataFrame) -> List[pd.DataFrame]: + """ + Split a DataFrame into chunks for processing. + """ + return [df] + + +def get_csv_value(row_dict: Dict[str, Any], expected_cols: List[str], target_col: str, default: Any = None) -> Any: + """ + Finds target_col in expected_cols case-insensitively, + and returns the corresponding value from row_dict case-insensitively. + Falls back to alias matching for common CSV column name variations. + """ + _ALIASES = { + "district": ["matched_district"], + "organization": ["detected_organization"], + "transcript link": ["transcript_link"], + "image urls": ["image_urls"], + "pdf urls": ["pdf_urls"], + "session id": ["session"], + "user location": ["user_location"], + "report created at": ["created_at"], + "date of discussion": ["date_of_discussion"], + } + + target_lower = target_col.lower() + matched_col = None + for col in expected_cols: + if col.lower() == target_lower: + matched_col = col + break + + if not matched_col: + matched_col = target_col + + actual_lower = matched_col.lower() + for k, v in row_dict.items(): + if k.lower() == actual_lower: + return v + + aliases = _ALIASES.get(target_lower, []) + for alias in aliases: + for k, v in row_dict.items(): + if k.lower() == alias: + return v + + return default + + +def parse_csv_list(val) -> List[str]: + if pd.isna(val) or val is None or not str(val).strip(): + return [] + s = str(val).strip() + if (s.startswith("[") and s.endswith("]")) or \ + (s.startswith("(") and s.endswith(")")) or \ + (s.startswith("{") and s.endswith("}")): + s = s[1:-1].strip() + + if "|" in s: + raw_items = s.split("|") + else: + raw_items = s.split(",") + + cleaned = [] + for x in raw_items: + x_clean = x.strip().strip("'\"").strip() + if x_clean: + cleaned.append(x_clean) + return cleaned + + +def get_url_field(val): + urls = parse_csv_list(val) + if not urls: + return None + if len(urls) == 1: + return urls[0] + return urls + + +def clean_segment(s: str) -> str: + import re + s = s.strip() + pattern_num = r'^\s*\d+[\.\)]\s*' + s = re.sub(pattern_num, '', s).strip() + pattern_bullet = r'^\s*[\-\*•]\s*' + s = re.sub(pattern_bullet, '', s).strip() + return s + + +def parse_segments(val, delimiter="|") -> List[str]: + import re + if pd.isna(val) or val is None or not str(val).strip(): + return [] + s = str(val).strip() + if delimiter in s: + raw_segments = s.split(delimiter) + elif "\n" in s: + raw_segments = s.split("\n") + else: + numbered_pattern = r'(?:^|\s)\d+(?:\.(?!\d)|\))\s*\S' + has_numbering = len(re.findall(numbered_pattern, s)) >= 2 + if has_numbering: + raw_segments = re.split(r'(?:^|\s+)\d+(?:\.(?!\d)|\))\s*', s) + else: + raw_segments = [s] + + segments = [] + for x in raw_segments: + x_clean = clean_segment(x) + if x_clean: + segments.append(x_clean) + return segments + + +def format_datetime(val, with_ms=True) -> str: + if pd.isna(val) or val is None: + val = datetime.utcnow() + if isinstance(val, str): + try: + val = pd.to_datetime(val) + except Exception: + return val + if hasattr(val, "to_pydatetime"): + val = val.to_pydatetime() + if isinstance(val, datetime): + if with_ms: + return val.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" + else: + return val.strftime("%Y-%m-%dT%H:%M:%SZ") + return str(val) + + +def _is_row_complete(row_dict: Dict[str, Any], report_type: str) -> Any: + return True, [] + + +def row_to_json( + row: pd.Series, + report_type: str, + event_type: str = "create", + metadata: Optional[dict] = None, +) -> str: + row_dict = {k: (None if pd.isna(v) else v) for k, v in row.to_dict().items()} + + normalized_type = report_type.lower().strip() + raw_cols = settings.STORY_CSV_COLUMN if normalized_type == "story" else settings.DISCUSSION_CSV_COLUMN + try: + expected_cols = json.loads(raw_cols) + except Exception: + expected_cols = [] + + try: + submission_id = int(get_csv_value(row_dict, expected_cols, "id")) + except Exception: + submission_id = get_csv_value(row_dict, expected_cols, "id") + + # No fallback generation — a missing Session ID is left null so the + # pre-publish schema check (STORY_KAFKA_SCHEMA/DISCUSSION_KAFKA_SCHEMA + # both require sessionId) catches and reports it, instead of silently + # inventing an identifier for a row that never had one. + session_id = get_csv_value(row_dict, expected_cols, "Session ID") + + program_info = None + leader_info = None + tenant_code = "mitra" + + if metadata: + program_info = metadata.get("programInfo") + leader_info = metadata.get("LeaderCategoryInfo") + tenant_code = metadata.get("tenantCode", "mitra") + + designation = None + if normalized_type == "story": + designation = get_csv_value(row_dict, expected_cols, "Designation") + if not designation and leader_info and leader_info.get("name"): + designation = leader_info["name"].split("(")[0].strip() + + published_at_raw = get_csv_value(row_dict, expected_cols, "Report Created At") + event_published_at = format_datetime(published_at_raw, with_ms=True) + + user_name = get_csv_value(row_dict, expected_cols, "User name") + organization = get_csv_value(row_dict, expected_cols, "Organization") + district = get_csv_value(row_dict, expected_cols, "District") + + state = None + if normalized_type == "discussion": + state_raw = get_csv_value(row_dict, expected_cols, "User Location") + else: + state_raw = get_csv_value(row_dict, expected_cols, "Location") + + if state_raw: + state_str = str(state_raw) + if "," in state_str: + state_str = state_str.split(",")[-1].strip() + state = state_str.title() + + user_id = None + author_val = get_csv_value(row_dict, expected_cols, "Author") + if normalized_type == "discussion": + user_id = str(author_val) if author_val is not None else str(submission_id) + else: + user_id_val = get_csv_value(row_dict, expected_cols, "userId") + user_id = str(author_val) if author_val is not None else ( + str(user_id_val) if user_id_val is not None else str(submission_id) + ) + + if normalized_type == "discussion": + submission_date_raw = get_csv_value(row_dict, expected_cols, "Date of Discussion") + submission_date = format_datetime(submission_date_raw, with_ms=False) + else: + submission_date = format_datetime(published_at_raw, with_ms=False) + + pdf_col = "Pdf" if normalized_type == "story" else "PDF Urls" + original_pdf = get_url_field(get_csv_value(row_dict, expected_cols, pdf_col)) + pdf_urls = None + if original_pdf: + pdf_urls = {"original": original_pdf} + + tags = { + "state": state, + "district": district, + "organization": organization, + "programId": program_info.get("id") if program_info else None, + "programName": program_info.get("name") if program_info else None, + "leaderCategoryId": leader_info.get("id") if leader_info else None, + "leaderCategoryName": leader_info.get("name") if leader_info else None, + } + + if normalized_type == "discussion": + participants_data = [] + role_cols = settings.get_discussion_participants_map() + + total_role = None + for role, col_name in role_cols.items(): + if role.lower() == "participant count" or (col_name and col_name.lower() == "participant count"): + total_role = role + break + + total_count = None + if total_role: + col_name = role_cols[total_role] + if col_name: + val = get_csv_value(row_dict, expected_cols, col_name) + if val is not None: + try: + total_count = int(val) + except Exception: + pass + + if total_count is not None and total_count > 0: + participants_data.append({"role": total_role, "count": total_count}) + + for role, col_name in role_cols.items(): + if role == total_role or not col_name: + continue + val = get_csv_value(row_dict, expected_cols, col_name) + if val is not None: + try: + count = int(val) + if count > 0: + participants_data.append({"role": role, "count": count}) + except Exception: + pass + + data = { + "title": get_csv_value(row_dict, expected_cols, "Title"), + "userId": user_id, + "userName": user_name, + "designation": designation, + "submissionDate": submission_date, + "imageUrls": parse_csv_list(get_csv_value(row_dict, expected_cols, "Image Urls")), + "pdfUrls": pdf_urls, + "transcriptLink": get_csv_value(row_dict, expected_cols, "Transcript Link") or None, + "challenges": parse_segments(get_csv_value(row_dict, expected_cols, "Challenges")), + "solutions": parse_segments(get_csv_value(row_dict, expected_cols, "Solutions")), + "participantsData": participants_data, + "author": user_id, + "language": get_csv_value(row_dict, expected_cols, "Language") or "en", + } + else: # story + data = { + "title": get_csv_value(row_dict, expected_cols, "Title"), + "userId": user_id, + "userName": user_name, + "designation": designation, + "submissionDate": submission_date, + "imageUrls": parse_csv_list(get_csv_value(row_dict, expected_cols, "Images")), + "pdfUrls": pdf_urls, + "transcriptLink": get_csv_value(row_dict, expected_cols, "Transcript Link") or None, + "objective": get_csv_value(row_dict, expected_cols, "Objective"), + "challenges": parse_segments(get_csv_value(row_dict, expected_cols, "Challenges")), + "actionSteps": parse_segments(get_csv_value(row_dict, expected_cols, "Action Steps")), + "impact": get_csv_value(row_dict, expected_cols, "Impact"), + "duration": get_csv_value(row_dict, expected_cols, "Duration"), + "blurb": get_csv_value(row_dict, expected_cols, "Blurb"), + "content": get_csv_value(row_dict, expected_cols, "Content"), + } + + payload = { + "submissionId": submission_id, + "submissionType": report_type, + "sessionId": session_id, + "tenantCode": tenant_code, + "eventType": event_type, + "eventPublishedAt": event_published_at, + "tags": tags, + "data": data, + } + return json.dumps(payload, default=str) + + +def rows_to_json( + df: pd.DataFrame, + report_type: str, + event_type: str = "create", + metadata: Optional[dict] = None, +): + for _, row in df.iterrows(): + row_dict = {k: (None if pd.isna(v) else v) for k, v in row.to_dict().items()} + is_complete, missing_fields = _is_row_complete(row_dict, report_type) + if not is_complete: + logger.warning( + "Skipping CSV row due to missing required data: %s", + missing_fields, + ) + continue + yield row_to_json(row, report_type, event_type, metadata) + + +# --------------------------------------------------------------------------- +# Service Orchestration Logic +# --------------------------------------------------------------------------- + +async def handle_upload( + report_type: str, + program_name: str, + leader_category: str, + tenant_code: str, + file_name: str, + file_bytes: bytes, +) -> dict: + from app.temporal.workflows import CsvProcessingWorkflow + + normalized_type = report_type.lower().strip() + file_size = len(file_bytes) + + # Duplicate check + is_duplicate = await operations.check_duplicate_file( + program_name=program_name, + leader_category=leader_category, + report_type=normalized_type, + file_name=file_name, + file_size=file_size, + ) + if is_duplicate: + raise DuplicateFile("FILE ALREADY EXISTS") + + # Validate columns FIRST — reject before touching GCS or the DB, so a + # malformed CSV never leaves cloud-storage or tracking-table clutter behind. + try: + df = await asyncio.to_thread(pd.read_csv, io.BytesIO(file_bytes)) + except Exception as exc: + raise InvalidCsvColumns([f"Failed to parse CSV: {exc}"]) + + is_valid, errors = await asyncio.to_thread(validate_columns, df, normalized_type) + if not is_valid: + raise InvalidCsvColumns(errors) + + # Upload to GCS + try: + cloud_storage_path = await asyncio.to_thread(upload_csv, file_bytes, normalized_type, file_name) + except Exception as exc: + logger.error("GCS Upload failed: %s", exc) + raise RuntimeError(f"GCS Upload failed: {exc}. Please verify GCS settings.") + + meta_data = { + "original_filename": file_name, + "program_name": program_name, + "leader_category": leader_category, + "report_type": normalized_type, + "tenant_code": tenant_code, + } + + record_id = await operations.insert_upload_record( + report_type=normalized_type, + program_name=program_name, + leader_category=leader_category, + cloud_storage_path=cloud_storage_path, + file_name=file_name, + file_size=file_size, + meta_data=meta_data, + status="pending", + ) + + logger.info( + "CSV uploaded: id=%s, report_type=%s, status=pending, cloud_storage_path=%s", + record_id, normalized_type, cloud_storage_path, + ) + + # Trigger Temporal workflow in real-time mode + if settings.PROCESSING_MODE.lower().strip() == "real-time": + try: + temporal_client = await Client.connect(settings.TEMPORAL_HOST) + await temporal_client.start_workflow( + CsvProcessingWorkflow.run, + record_id, + id=f"csv-upload-{record_id}", + task_queue=settings.TEMPORAL_QUEUE, + ) + logger.info("Triggered real-time CsvProcessingWorkflow for upload ID %s", record_id) + except Exception as e: + logger.error("Failed to trigger real-time CsvProcessingWorkflow: %s", e) + await operations.update_status(record_id, "on_hold", {"error": f"Temporal trigger failed: {e}"}) + raise RuntimeError(f"Failed to start CSV processing workflow: {e}") + + return { + "message": "Successfully uploaded to cloud", + "id": record_id, + "status": "pending", + } + + +async def handle_push(record_id: int) -> dict: + from app.temporal.workflows import CsvProcessingWorkflow + + record = await operations.get_record(record_id) + if not record: + raise RecordNotFound("Record not found") + + status = record.get("status") + if status == "in_progress": + raise RecordAlreadyProcessing("Record is already being processed") + if status != "pending": + raise RecordNotPending("Only pending records can be processed") + + claim_status = await operations.try_claim_for_processing(record_id) + if claim_status is None: + raise RecordNotFound("Record not found") + if claim_status == "in_progress": + raise RecordAlreadyProcessing("Record is already being processed") + + try: + temporal_client = await Client.connect(settings.TEMPORAL_HOST) + await temporal_client.start_workflow( + CsvProcessingWorkflow.run, + record_id, + id=f"csv-upload-{record_id}", + task_queue=settings.TEMPORAL_QUEUE, + ) + return {"status": "success", "message": "CSV processing workflow started"} + except Exception as e: + await operations.update_status(record_id, "on_hold", {"error": str(e)}) + raise RuntimeError(f"Failed to start CSV processing workflow: {e}") diff --git a/app/api/validators/__init__.py b/app/api/validators/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/validators/uploads.py b/app/api/validators/uploads.py new file mode 100644 index 0000000..6ac623c --- /dev/null +++ b/app/api/validators/uploads.py @@ -0,0 +1,73 @@ +import json +import logging +from typing import List, Optional, Tuple +import pandas as pd +from app.api.exceptions import InvalidReportType, InvalidFileType, FileTooLarge, EmptyFile +from app.config import settings + +logger = logging.getLogger("analytics_service.api.validators.uploads") + + +def validate_report_type(report_type: str) -> str: + if not report_type: + raise InvalidReportType("Only 'story' or 'discussion' report types are accepted.") + normalized_type = report_type.lower().strip() + if normalized_type not in ("story", "discussion"): + raise InvalidReportType("Only 'story' or 'discussion' report types are accepted.") + return normalized_type + + +def validate_extension(filename: Optional[str]) -> None: + if not filename or not filename.lower().endswith(".csv"): + raise InvalidFileType("Only .csv files are accepted") + + +def validate_file_bytes(file_bytes: bytes) -> None: + if len(file_bytes) > settings.MAX_CSV_UPLOAD_BYTES: + raise FileTooLarge("Uploaded file is too large") + if not file_bytes: + raise EmptyFile("Uploaded file is empty") + + +def validate_columns(df: pd.DataFrame, report_type: str) -> Tuple[bool, List[str]]: + """ + Returns (is_valid, list_of_error_messages). + + Checks: + 1. All expected columns defined in settings for report_type are present (case-insensitive check). + """ + errors: List[str] = [] + + normalized_type = report_type.lower().strip() + if normalized_type == "story": + raw_cols = settings.STORY_CSV_COLUMN + elif normalized_type == "discussion": + raw_cols = settings.DISCUSSION_CSV_COLUMN + else: + return False, [f"No expected schema configured for report_type='{report_type}'"] + + try: + expected_cols = json.loads(raw_cols) + if not isinstance(expected_cols, list): + raise ValueError("Expected columns must be a JSON array") + except Exception as exc: + return False, [f"Failed to parse expected columns from settings for {report_type}: {exc}"] + + # Normalize both sides to lowercase for case-insensitive comparison + expected_cols_lower = {col.lower(): col for col in expected_cols} + actual_cols_lower = {col.lower(): col for col in df.columns} + + missing = set(expected_cols_lower.keys()) - set(actual_cols_lower.keys()) + if missing: + # Report using original expected column names for clarity + missing_originals = sorted(expected_cols_lower[m] for m in missing) + errors.append(f"Missing columns: {missing_originals}") + logger.warning(f"Validation failed for report_type='{report_type}'. Missing: {missing_originals}") + + extra = set(actual_cols_lower.keys()) - set(expected_cols_lower.keys()) + if extra: + extra_originals = sorted(actual_cols_lower[e] for e in extra) + errors.append(f"Extra/unexpected columns: {extra_originals}") + logger.warning(f"Validation failed for report_type='{report_type}'. Extra: {extra_originals}") + + return (len(errors) == 0), errors diff --git a/app/config.py b/app/config.py index 24f0210..9d6db17 100644 --- a/app/config.py +++ b/app/config.py @@ -34,6 +34,29 @@ class Settings(BaseSettings): TEMPORAL_HOST: str = Field(default="localhost:7233") TEMPORAL_QUEUE: str = Field(default="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, since this is an eagerly-evaluated singleton. + AUTH_TOKEN: str = Field(description="Bearer token for API authentication. Must be set via environment variable.") + + # CSV Upload / Processing Configuration + MAX_CSV_UPLOAD_BYTES: int = Field(default=10485760) # 10MB + CSV_BLOB_UPLOADS: str = Field(default="mitra_dashboard_api_output") + CSV_SCHEDULE_CRON_TIME: str = Field(default="40 15 * * *") + # Expected CSV column headers per report type (JSON arrays of column names, + # matched case-insensitively against the uploaded file's header row). + STORY_CSV_COLUMN: str = Field( + default='["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: str = Field( + default='["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"]' + ) + # Maps a discussion participant "role" to the CSV column name holding its + # count (JSON object). See get_discussion_participants_map(). + DISCUSSION_PARTICIPANTS_MAP: str = Field( + default='{"men": "Men", "women": "Women", "children": "Children", "teacher": "Teacher", "participant count": "Participant Count"}' + ) + # LLM / OpenRouter Configuration OPENROUTER_API_KEY: str = Field(default="") OPENROUTER_MODEL: str = Field(default="nvidia/nemotron-3-nano-omni-30b-a3b-reasoning:free") @@ -135,6 +158,30 @@ def validate_log_level(cls, v: str) -> str: ) return level + @field_validator("STORY_CSV_COLUMN", "DISCUSSION_CSV_COLUMN") + @classmethod + def validate_csv_column_json(cls, v: str, info) -> str: + try: + parsed = json.loads(v) + if not isinstance(parsed, list) or not all(isinstance(item, str) for item in parsed): + raise ValueError(f"{info.field_name} must be a JSON array of column-name strings.") + except (json.JSONDecodeError, TypeError) as e: + raise ValueError(f"Invalid JSON configuration for {info.field_name}: {e}") from e + return v + + @field_validator("DISCUSSION_PARTICIPANTS_MAP") + @classmethod + def validate_participants_map_json(cls, v: str) -> str: + if not v or not v.strip(): + return v + try: + parsed = json.loads(v) + if not isinstance(parsed, dict): + raise ValueError("DISCUSSION_PARTICIPANTS_MAP must be a JSON object mapping role names to CSV column names.") + except (json.JSONDecodeError, TypeError) as e: + raise ValueError(f"Invalid JSON configuration for DISCUSSION_PARTICIPANTS_MAP: {e}") from e + return v + @field_validator("STORY_KAFKA_SCHEMA", "DISCUSSION_KAFKA_SCHEMA") @classmethod def validate_kafka_ingestion_schema_json(cls, v: str, info) -> str: @@ -224,5 +271,27 @@ def get_kafka_ingestion_schema(self, submission_type: str) -> Dict[str, Any]: f"Failed to parse Kafka ingestion schema JSON for submission type {submission_type!r}: {e}" ) from e + def get_discussion_participants_map(self) -> Dict[str, str]: + """ + Dynamically returns the participant role-to-column mapping dictionary. + Falls back to empty dict if empty/invalid, or default if parsing fails. + """ + raw_map = self.DISCUSSION_PARTICIPANTS_MAP + if not raw_map or not str(raw_map).strip(): + return {} + try: + parsed = json.loads(raw_map) + if isinstance(parsed, dict): + return {str(k).strip(): str(v).strip() for k, v in parsed.items()} + return {} + except Exception: + return { + "men": "Men", + "women": "Women", + "children": "Children", + "teacher": "Teacher", + "participant count": "Participant Count" + } + # Singleton instance settings = Settings() diff --git a/app/database/operations.py b/app/database/operations.py index 6634294..607f8be 100644 --- a/app/database/operations.py +++ b/app/database/operations.py @@ -620,3 +620,174 @@ async def get_submission_type_and_payload(conn: asyncpg.Connection, submission_i return sub_type, dict(payload_row) +# CSV Upload Tracking (csv_uploads) +async def check_duplicate_file( + program_name: str, + leader_category: str, + report_type: str, + file_name: str, + file_size: int, +) -> bool: + """Return True if a matching file already exists in the tracker.""" + from app.database.db import db + if not db.pool: + await db.connect() + + async with db.pool.acquire() as conn: + exists = await conn.fetchval( + """ + SELECT 1 FROM csv_uploads + WHERE program_name = $1 + AND leader_category = $2 + AND report_type = $3 + AND file_name = $4 + AND file_size = $5 + LIMIT 1 + """, + program_name, + leader_category, + report_type, + file_name, + file_size, + ) + return exists is not None + + +async def insert_upload_record( + report_type: str, + program_name: str, + leader_category: str, + cloud_storage_path: str, + file_name: Optional[str] = None, + file_size: Optional[int] = None, + meta_data: Optional[Dict[str, Any]] = None, + status: str = "pending", +) -> int: + """Insert a new row with the given status. Returns the new row's id.""" + from app.database.db import db + if not db.pool: + await db.connect() + + async with db.pool.acquire() as conn: + try: + row = await conn.fetchrow( + """ + INSERT INTO csv_uploads + (report_type, program_name, leader_category, cloud_storage_path, + file_name, file_size, meta_data, status) + VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8) + RETURNING id + """, + report_type, + program_name, + leader_category, + cloud_storage_path, + file_name, + file_size, + json.dumps(meta_data or {}), + status, + ) + record_id = row["id"] + logger.info("Inserted csv_uploads record %s (status=%s)", record_id, status) + return record_id + except asyncpg.UniqueViolationError: + from app.api.exceptions import DuplicateFile + raise DuplicateFile("FILE ALREADY EXISTS") + + +async def get_record(record_id: int) -> Optional[dict]: + """Fetch a single tracker record by id.""" + from app.database.db import db + if not db.pool: + await db.connect() + + async with db.pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT * FROM csv_uploads WHERE id = $1", + record_id, + ) + return dict(row) if row else None + + +async def update_status( + record_id: int, + status: str, + meta_data: Optional[Dict[str, Any]] = None, +) -> None: + """ + Update status, optionally merging new keys into meta_data. + """ + from app.database.db import db + if not db.pool: + await db.connect() + + async with db.pool.acquire() as conn: + if meta_data is not None: + await conn.execute( + """ + UPDATE csv_uploads + SET status = $1, + meta_data = COALESCE(meta_data, '{}'::jsonb) || $2::jsonb + WHERE id = $3 + """, + status, + json.dumps(meta_data), + record_id, + ) + else: + await conn.execute( + "UPDATE csv_uploads SET status = $1 WHERE id = $2", + status, + record_id, + ) + + +async def list_by_status(status: str) -> list: + """List all tracker records with a given status.""" + from app.database.db import db + if not db.pool: + await db.connect() + + async with db.pool.acquire() as conn: + rows = await conn.fetch( + "SELECT * FROM csv_uploads WHERE status = $1 ORDER BY created_at", + status, + ) + return [dict(r) for r in rows] + + +async def try_claim_for_processing(record_id: int) -> Optional[str]: + """ + Atomically set status to 'in_progress' if record exists and its status is not 'in_progress'. + Returns: + - 'success' if successfully claimed/updated. + - 'in_progress' if it is already in progress. + - None if the record does not exist. + """ + from app.database.db import db + if not db.pool: + await db.connect() + + async with db.pool.acquire() as conn: + row = await conn.fetchrow( + """ + UPDATE csv_uploads + SET status = 'in_progress' + WHERE id = $1 AND status != 'in_progress' + RETURNING status + """, + record_id, + ) + if row: + logger.info("Atomically claimed record %s for processing", record_id) + return "success" + + # If update did not match any row, determine if it doesn't exist or is in_progress + exists = await conn.fetchval( + "SELECT 1 FROM csv_uploads WHERE id = $1 LIMIT 1", + record_id, + ) + if not exists: + return None + return "in_progress" + diff --git a/app/kafka/consumer.py b/app/kafka/consumer.py index 7234342..b7678ea 100644 --- a/app/kafka/consumer.py +++ b/app/kafka/consumer.py @@ -13,37 +13,12 @@ delete_submission, update_submission_status ) +from app.services.ingestion_validation import validate_ingestion_schema from app.temporal.workflows import ConfigDrivenProcessingWorkflow logger = logging.getLogger("analytics_service.kafka.consumer") -def _get_nested(obj: dict, dotted_path: str): - """Walks a dotted path (e.g. 'data.pdfUrls.original') through nested dicts. - Returns (value, found) — found is False if any segment is missing or not a dict.""" - current = obj - for segment in dotted_path.split("."): - if not isinstance(current, dict) or segment not in current: - return None, False - current = current[segment] - return current, True - - -def _is_empty(value) -> bool: - """True only for None, "", [], {} — explicitly NOT for 0 or False, which are - falsy-but-valid values (unlike a bare `not value` check).""" - if value is None: - return True - if isinstance(value, (str, list, dict, tuple)) and len(value) == 0: - return True - return False - - -def _emptiness_label(value) -> str: - """Distinguishes *why* a value tripped _is_empty(), for precise problem messages.""" - return "null" if value is None else "empty" - - def _payload_fingerprint(raw_payload: str) -> str: """A log-safe stand-in for a raw Kafka payload: a short hash plus byte length, enough to correlate a log line with the full message already preserved on the @@ -199,42 +174,6 @@ def _on_delivery(err, _msg): await asyncio.to_thread(_produce) logger.warning(f"Sent invalid message to DLQ topic '{self.dlq_topic}'. Reason: {reason}. Identifiers: {id_str}. Payload: {_payload_fingerprint(raw_payload)}") - def _validate_ingestion_schema(self, event: dict, submission_type: str, event_type: str) -> list: - """ - Validates a Kafka event against the configured required-fields schema for its - (submissionType, eventType) combination. Returns a list of problem descriptions; - an empty list means the event is valid. - """ - normalized_type = submission_type.lower().strip() if isinstance(submission_type, str) else "" - if event_type in ("create", "update") and "story" not in normalized_type and "discussion" not in normalized_type: - return [f"Unrecognized submissionType {submission_type!r}; no ingestion schema to validate against"] - - try: - schema = settings.get_kafka_ingestion_schema(submission_type) - except ValueError as e: - return [str(e)] - - event_schema = schema.get(event_type) - if event_schema is None: - return [f"No ingestion schema section defined for eventType {event_type!r}"] - - problems = [] - for path in event_schema.get("required", []): - value, found = _get_nested(event, path) - if not found: - problems.append(f"'{path}' is missing") - elif _is_empty(value): - problems.append(f"'{path}' is {_emptiness_label(value)}") - - if event_type == "update" and event_schema.get("newValuesNoEmpty"): - new_values = event.get("newValues") - if isinstance(new_values, dict): - for key, value in new_values.items(): - if _is_empty(value): - problems.append(f"'newValues.{key}' is {_emptiness_label(value)}") - - return problems - async def process_message(self, raw_payload: str) -> None: """ Processes a single deserialized Kafka message, executing DB changes and routing. @@ -264,7 +203,7 @@ async def process_message(self, raw_payload: str) -> None: event_type = event_type_raw.lower().strip() submission_type = event.get("submissionType") - problems = self._validate_ingestion_schema(event, submission_type, event_type) + problems = validate_ingestion_schema(event, submission_type, event_type) if problems: reason = f"Failed ingestion validation for submissionType {submission_type!r}, eventType {event_type!r}: {'; '.join(problems)}" logger.error(f"{reason}. Identifiers: {_format_identifiers(identifiers)}") diff --git a/app/services/gcp_storage.py b/app/services/gcp_storage.py index 1e1dd46..59bd42f 100644 --- a/app/services/gcp_storage.py +++ b/app/services/gcp_storage.py @@ -1,5 +1,6 @@ import json import logging +import uuid from pathlib import Path from google.cloud import storage from google.oauth2 import service_account @@ -62,3 +63,34 @@ def upload_to_gcp(local_file_path: str, blob_name: str) -> str: except Exception as e: logger.error(f"Failed to upload {local_file_path} to GCP bucket {settings.BUCKET_NAME}: {e}") raise + +def upload_csv(file_bytes: bytes, report_type: str, original_filename: str) -> str: + """Upload raw CSV bytes to the configured GCS bucket and return the object key.""" + if not settings.BUCKET_NAME: + raise ValueError("BUCKET_NAME is not configured in settings.") + + prefix = settings.CSV_BLOB_UPLOADS.strip("/") + object_key = f"{prefix}/{uuid.uuid4()}_{original_filename}" + + credentials = get_gcp_credentials() + client = storage.Client(credentials=credentials, project=settings.PROJECT_ID) + bucket = client.bucket(settings.BUCKET_NAME) + blob = bucket.blob(object_key) + blob.upload_from_string(file_bytes, content_type="text/csv") + + logger.info(f"Uploaded CSV to gs://{settings.BUCKET_NAME}/{object_key}") + return object_key + +def fetch_csv(bucket_path: str) -> bytes: + """Download CSV bytes from the configured GCS bucket.""" + if not settings.BUCKET_NAME: + raise ValueError("BUCKET_NAME is not configured in settings.") + + credentials = get_gcp_credentials() + client = storage.Client(credentials=credentials, project=settings.PROJECT_ID) + bucket = client.bucket(settings.BUCKET_NAME) + blob = bucket.blob(bucket_path) + data = blob.download_as_bytes() + + logger.info(f"Fetched CSV from gs://{settings.BUCKET_NAME}/{bucket_path} ({len(data)} bytes)") + return data diff --git a/app/services/ingestion_validation.py b/app/services/ingestion_validation.py new file mode 100644 index 0000000..25d2442 --- /dev/null +++ b/app/services/ingestion_validation.py @@ -0,0 +1,68 @@ +from app.config import settings + + +def _get_nested(obj: dict, dotted_path: str): + """Walks a dotted path (e.g. 'data.pdfUrls.original') through nested dicts. + Returns (value, found) — found is False if any segment is missing or not a dict.""" + current = obj + for segment in dotted_path.split("."): + if not isinstance(current, dict) or segment not in current: + return None, False + current = current[segment] + return current, True + + +def _is_empty(value) -> bool: + """True only for None, "", [], {} — explicitly NOT for 0 or False, which are + falsy-but-valid values (unlike a bare `not value` check).""" + if value is None: + return True + if isinstance(value, (str, list, dict, tuple)) and len(value) == 0: + return True + return False + + +def _emptiness_label(value) -> str: + """Distinguishes *why* a value tripped _is_empty(), for precise problem messages.""" + return "null" if value is None else "empty" + + +def validate_ingestion_schema(event: dict, submission_type: str, event_type: str) -> list: + """ + Validates a Kafka event against the configured required-fields schema for its + (submissionType, eventType) combination. Returns a list of problem descriptions; + an empty list means the event is valid. + + Shared between app/kafka/consumer.py (validates events arriving off the Kafka + topic) and app/temporal/csv_processing_activity.py (validates each CSV-derived + event against the same schema before it's ever published to Kafka). + """ + normalized_type = submission_type.lower().strip() if isinstance(submission_type, str) else "" + if event_type in ("create", "update") and "story" not in normalized_type and "discussion" not in normalized_type: + return [f"Unrecognized submissionType {submission_type!r}; no ingestion schema to validate against"] + + try: + schema = settings.get_kafka_ingestion_schema(submission_type) + except ValueError as e: + return [str(e)] + + event_schema = schema.get(event_type) + if event_schema is None: + return [f"No ingestion schema section defined for eventType {event_type!r}"] + + problems = [] + for path in event_schema.get("required", []): + value, found = _get_nested(event, path) + if not found: + problems.append(f"'{path}' is missing") + elif _is_empty(value): + problems.append(f"'{path}' is {_emptiness_label(value)}") + + if event_type == "update" and event_schema.get("newValuesNoEmpty"): + new_values = event.get("newValues") + if isinstance(new_values, dict): + for key, value in new_values.items(): + if _is_empty(value): + problems.append(f"'newValues.{key}' is {_emptiness_label(value)}") + + return problems diff --git a/app/temporal/csv_processing_activity.py b/app/temporal/csv_processing_activity.py new file mode 100644 index 0000000..e1956cd --- /dev/null +++ b/app/temporal/csv_processing_activity.py @@ -0,0 +1,295 @@ +import asyncio +import json +import logging +import uuid +from datetime import datetime +from typing import Any, Dict, List, Optional +from temporalio import activity +from confluent_kafka import Producer, KafkaException + +from app.config import settings +from app.database.db import db +from app.database import operations as csv_upload_repo +from app.api.services.uploads import load_csv, rows_to_json, split_csv +from app.api.validators.uploads import validate_columns +from app.services.gcp_storage import fetch_csv +from app.services.ingestion_validation import validate_ingestion_schema + +logger = logging.getLogger("analytics_service.temporal.csv_processing_activity") + +_producer: Optional[Producer] = None + + +def _get_producer() -> Producer: + global _producer + if _producer is None: + _producer = Producer({ + "bootstrap.servers": settings.KAFKA_BOOTSTRAP_SERVERS, + "acks": "all", + "enable.idempotence": True, + }) + return _producer + + +def _push_rows_sync(payloads: List[Any]) -> None: + """ + Runs in a worker thread (via asyncio.to_thread) — produce()/flush() are + blocking calls. Flushes once for the whole batch rather than per row (a + per-row flush forces a network round trip per row, far too slow for large + CSVs), mirroring app/kafka/consumer.py's DLQ producer pattern. + """ + producer = _get_producer() + delivery_error = {} + + def _on_delivery(err, _msg): + if err is not None: + delivery_error["error"] = err + + for payload, key in payloads: + producer.produce( + settings.KAFKA_TOPIC_INGESTION, + value=payload.encode("utf-8"), + key=key.encode("utf-8") if key else None, + callback=_on_delivery, + ) + producer.poll(0) + if "error" in delivery_error: + raise KafkaException(delivery_error["error"]) + + remaining = producer.flush(10) + if remaining > 0: + raise TimeoutError(f"Timed out waiting for Kafka delivery ({remaining} still in-flight)") + if "error" in delivery_error: + raise KafkaException(delivery_error["error"]) + + +@activity.defn +async def csv_fetch_and_validate_activity(record_id: int) -> bool: + """ + Temporal activity to fetch the CSV file from cloud storage and update status. + """ + record = await csv_upload_repo.get_record(record_id) + if not record: + raise ValueError(f"Record {record_id} not found in database.") + + cloud_storage_path = record["cloud_storage_path"] + + try: + # Fetch from GCS + parse — both blocking, offloaded from the event loop. + csv_file = await asyncio.to_thread(fetch_csv, cloud_storage_path) + df = await asyncio.to_thread(load_csv, csv_file) + except Exception as exc: + logger.exception("Failed to fetch/load CSV for record %s", record_id) + error_meta = { + "stage": "CSV Fetching", + "error": "Failed to fetch/load CSV from GCS", + "exception": str(exc), + "timestamp": datetime.utcnow().isoformat() + "Z" + } + await csv_upload_repo.update_status(record_id, "on_hold", error_meta) + return False + + is_valid, errors = await asyncio.to_thread(validate_columns, df, record["report_type"]) + if not is_valid: + logger.warning("Validation failed for record %s: %s", record_id, errors) + error_meta = { + "stage": "CSV Column Validation", + "error": "Invalid CSV schema", + "validation_errors": errors, + "timestamp": datetime.utcnow().isoformat() + "Z" + } + await csv_upload_repo.update_status(record_id, "on_hold", error_meta) + return False + + await csv_upload_repo.update_status(record_id, "in_progress") + return True + + +@activity.defn +async def csv_push_to_kafka_activity(record_id: int) -> Dict[str, Any]: + """ + Temporal activity to process the CSV file row by row and publish + individual messages to Kafka. Returns {"rows_pushed": int, + "schema_validation_errors": list} — rows failing pre-publish schema + validation are skipped (not published) and reported here instead. + """ + record = await csv_upload_repo.get_record(record_id) + if not record: + raise ValueError(f"Record {record_id} not found in database.") + + report_type = record["report_type"] + cloud_storage_path = record["cloud_storage_path"] + + # Load CSV — blocking I/O + parse, offloaded from the event loop. + csv_file = await asyncio.to_thread(fetch_csv, cloud_storage_path) + df = await asyncio.to_thread(load_csv, csv_file) + + is_valid, errors = await asyncio.to_thread(validate_columns, df, report_type) + if not is_valid: + logger.warning("Kafka push blocked for record %s due to invalid columns: %s", record_id, errors) + error_meta = { + "stage": "CSV Column Validation", + "error": "Invalid CSV schema - aborting Kafka push", + "validation_errors": errors, + "timestamp": datetime.utcnow().isoformat() + "Z" + } + await csv_upload_repo.update_status(record_id, "on_hold", error_meta) + raise ValueError("CSV validation failed for Kafka push") + + # Fetch programs / leader categories info from Postgres once for context mapping + program_info = None + leader_info = None + # Use tenant_code from the upload payload (stored in meta_data) as primary source, + # falling back to DB lookup and then to "mitra" as last resort. + record_meta = record.get("meta_data") or {} + if isinstance(record_meta, str): + try: + record_meta = json.loads(record_meta) + except json.JSONDecodeError: + record_meta = {} + + if not isinstance(record_meta, dict): + record_meta = {} + + tenant_code = record_meta.get("tenant_code") or "mitra" + + try: + async with db.pool.acquire() as conn: + leader_row = await conn.fetchrow( + "SELECT id, name, description, tenant_code FROM leader_category WHERE name = $1 LIMIT 1", + record.get("leader_category") + ) + if leader_row: + leader_info = { + "id": str(leader_row["id"]), + "name": leader_row["name"], + "description": leader_row["description"], + } + tenant_code = leader_row["tenant_code"] + + if leader_row: + program_row = await conn.fetchrow( + "SELECT id, name, description, tenant_code, leaders_id FROM programs WHERE name = $1 AND leaders_id = $2 LIMIT 1", + record.get("program_name"), leader_row["id"] + ) + else: + program_row = await conn.fetchrow( + "SELECT id, name, description, tenant_code, leaders_id FROM programs WHERE name = $1 LIMIT 1", + record.get("program_name") + ) + + if program_row: + program_info = { + "id": str(program_row["id"]), + "name": program_row["name"], + "description": program_row["description"], + } + tenant_code = program_row.get("tenant_code", tenant_code) + + if program_row and not leader_info: + leader_row_from_program = await conn.fetchrow( + "SELECT id, name, description, tenant_code FROM leader_category WHERE id = $1 LIMIT 1", + program_row["leaders_id"] + ) + if leader_row_from_program: + leader_info = { + "id": str(leader_row_from_program["id"]), + "name": leader_row_from_program["name"], + "description": leader_row_from_program["description"], + } + tenant_code = leader_row_from_program.get("tenant_code", tenant_code) + except Exception as db_exc: + logger.warning("Failed to query program/leader category metadata from DB: %s", db_exc) + + # Fallbacks if DB query returned nothing + if not leader_info: + leader_info = { + "id": str(uuid.uuid4()), + "name": record.get("leader_category") or "District Leader", + "description": f"Leader category: {record.get('leader_category') or 'District Leader'}", + } + if not program_info: + program_info = { + "id": str(uuid.uuid4()), + "name": record.get("program_name") or "My Program", + "description": f"Program: {record.get('program_name') or 'My Program'}", + } + + metadata = { + "programInfo": program_info, + "LeaderCategoryInfo": leader_info, + "tenantCode": tenant_code, + } + + chunks = split_csv(df) + payloads = [] + schema_errors = [] + row_number = 0 + + for chunk in chunks: + for payload_str in rows_to_json(chunk, report_type, metadata=metadata): + row_number += 1 + try: + payload_dict = json.loads(payload_str) + except json.JSONDecodeError as exc: + schema_errors.append({"row": row_number, "problems": [f"Failed to parse generated payload: {exc}"]}) + continue + + # Double-check the generated event against the exact same schema + # app/kafka/consumer.py enforces at ingestion — catches a row missing + # a required field (e.g. no Session ID, now that it's no longer + # auto-generated) here, before it's ever published, rather than + # relying on the consumer to silently DLQ it later. + problems = validate_ingestion_schema(payload_dict, report_type, "create") + if problems: + schema_errors.append({ + "row": row_number, + "submissionId": payload_dict.get("submissionId"), + "sessionId": payload_dict.get("sessionId"), + "problems": problems, + }) + continue + + payloads.append((payload_str, f"{record_id}-{len(payloads)}")) + + if schema_errors: + logger.warning( + "record %s: %d of %d row(s) failed pre-publish schema validation and were skipped: %s", + record_id, len(schema_errors), row_number, schema_errors, + ) + + if payloads: + try: + await asyncio.to_thread(_push_rows_sync, payloads) + except Exception as exc: + logger.exception("Kafka push failed for record %s", record_id) + error_meta = { + "stage": "Kafka Publishing", + "error": "Failed to publish record", + "exception": str(exc), + "timestamp": datetime.utcnow().isoformat() + "Z" + } + await csv_upload_repo.update_status(record_id, "on_hold", error_meta) + raise + + return {"rows_pushed": len(payloads), "schema_validation_errors": schema_errors} + + +@activity.defn +async def csv_update_status_activity(params: Dict[str, Any]) -> None: + """ + Temporal activity to update the overall processing status of a csv_upload in PostgreSQL. + """ + record_id = params["record_id"] + status = params["status"] + meta_data = params.get("meta_data") + await csv_upload_repo.update_status(record_id, status, meta_data) + + +@activity.defn +async def fetch_pending_csv_uploads_activity() -> List[int]: + """ + Temporal activity to fetch the IDs of all pending csv_upload records. + """ + records = await csv_upload_repo.list_by_status("pending") + return [r["id"] for r in records] diff --git a/app/temporal/worker.py b/app/temporal/worker.py index 5b44080..a398317 100644 --- a/app/temporal/worker.py +++ b/app/temporal/worker.py @@ -5,7 +5,12 @@ from app.config import settings from app.database.db import db -from app.temporal.workflows import ConfigDrivenProcessingWorkflow, BatchProcessingWorkflow +from app.temporal.workflows import ( + ConfigDrivenProcessingWorkflow, + BatchProcessingWorkflow, + CsvProcessingWorkflow, + CsvBatchProcessingWorkflow, +) from app.temporal.activities import ( update_status_activity, fetch_pending_submissions_activity @@ -14,6 +19,12 @@ from app.temporal.pii_and_abusive_activity import pii_and_abusive_language_detection_activity from app.temporal.thematic_activity import thematic_classification_activity from app.temporal.story_rating_activity import story_rating_activity +from app.temporal.csv_processing_activity import ( + csv_fetch_and_validate_activity, + csv_push_to_kafka_activity, + csv_update_status_activity, + fetch_pending_csv_uploads_activity, +) logger = logging.getLogger("analytics_service.temporal.worker") @@ -33,14 +44,23 @@ async def start_worker(): return # Define registered activities and workflows - workflows = [ConfigDrivenProcessingWorkflow, BatchProcessingWorkflow] + workflows = [ + ConfigDrivenProcessingWorkflow, + BatchProcessingWorkflow, + CsvProcessingWorkflow, + CsvBatchProcessingWorkflow, + ] activities = [ pii_and_abusive_language_detection_activity, thematic_classification_activity, deface_blur_activity, story_rating_activity, update_status_activity, - fetch_pending_submissions_activity + fetch_pending_submissions_activity, + csv_fetch_and_validate_activity, + csv_push_to_kafka_activity, + csv_update_status_activity, + fetch_pending_csv_uploads_activity, ] worker = Worker( @@ -50,17 +70,40 @@ async def start_worker(): activities=activities ) - # Register daily batch schedule if configured for batch mode + # Register daily batch schedules if configured for batch mode if settings.PROCESSING_MODE.lower().strip() == "batch": + from temporalio.client import ( + Schedule, + ScheduleActionStartWorkflow, + ScheduleSpec, + ScheduleAlreadyRunningError, + ) + + # 1. Register CSV batch processing schedule try: - from temporalio.client import ( - Schedule, - ScheduleActionStartWorkflow, - ScheduleSpec, - ScheduleAlreadyRunningError, + logger.info(f"Registering CSV batch schedule '{settings.CSV_SCHEDULE_CRON_TIME}' in Temporal...") + await client.create_schedule( + id="csv-batch-processing", + schedule=Schedule( + action=ScheduleActionStartWorkflow( + CsvBatchProcessingWorkflow.run, + id="csv-batch-processing-run", + task_queue=settings.TEMPORAL_QUEUE, + ), + spec=ScheduleSpec( + cron_expressions=[settings.CSV_SCHEDULE_CRON_TIME] + ), + ), ) + logger.info("CSV batch schedule successfully registered.") + except ScheduleAlreadyRunningError: + logger.info("CSV batch schedule already exists in Temporal. Skipping registration.") + except Exception as e: + logger.error(f"Failed to register CSV batch schedule in Temporal: {e}") - logger.info(f"Registering daily batch schedule '{settings.BATCH_SCHEDULE_CRON}' in Temporal...") + # 2. Register daily analysis batch processing schedule + try: + logger.info(f"Registering daily analysis batch schedule '{settings.BATCH_SCHEDULE_CRON}' in Temporal...") await client.create_schedule( id="daily-batch-processing", schedule=Schedule( @@ -75,11 +118,22 @@ async def start_worker(): ), ), ) - logger.info("Daily batch schedule successfully registered.") + logger.info("Daily analysis batch schedule successfully registered.") except ScheduleAlreadyRunningError: - logger.info("Daily batch schedule already exists in Temporal. Skipping registration.") + logger.info("Daily analysis batch schedule already exists in Temporal. Skipping registration.") except Exception as e: - logger.error(f"Failed to register daily batch schedule in Temporal: {e}") + logger.error(f"Failed to register daily analysis batch schedule in Temporal: {e}") + else: + # Real-time mode: clean up any leftover batch schedules from Temporal Server + # (prevents a schedule left behind from a prior batch-mode config from + # silently retrying forever with outdated arguments). + for sched_id in ("csv-batch-processing", "daily-batch-processing"): + try: + handle = client.get_schedule_handle(sched_id) + await handle.delete() + logger.info("Deleted stale batch schedule '%s' (PROCESSING_MODE=real-time).", sched_id) + except Exception: + pass # schedule doesn't exist — nothing to clean up logger.info(f"🚀 Temporal Worker started. Listening on task queue '{settings.TEMPORAL_QUEUE}'...") try: diff --git a/app/temporal/workflows.py b/app/temporal/workflows.py index 141e3a5..1514a03 100644 --- a/app/temporal/workflows.py +++ b/app/temporal/workflows.py @@ -13,6 +13,12 @@ from app.temporal.pii_and_abusive_activity import pii_and_abusive_language_detection_activity from app.temporal.thematic_activity import thematic_classification_activity from app.temporal.story_rating_activity import story_rating_activity + from app.temporal.csv_processing_activity import ( + csv_fetch_and_validate_activity, + csv_push_to_kafka_activity, + csv_update_status_activity, + fetch_pending_csv_uploads_activity + ) @workflow.defn class ConfigDrivenProcessingWorkflow: @@ -303,3 +309,108 @@ async def run(self, batch_size: int, carry_over: Optional[Dict[str, Any]] = None "failed_count": total_failed, "chunks": chunk_index, } + + +@workflow.defn +class CsvProcessingWorkflow: + @workflow.run + async def run(self, record_id: int) -> Dict[str, Any]: + """ + Orchestrates processing for a single csv_upload record. + 1. Fetch CSV and validate columns. + 2. Publish each row to Kafka. + 3. Mark as success. + """ + retry_policy = RetryPolicy( + maximum_attempts=3, + initial_interval=timedelta(seconds=2), + backoff_coefficient=2.0 + ) + + # 1. Fetch and validate columns + is_valid = await workflow.execute_activity( + csv_fetch_and_validate_activity, + record_id, + start_to_close_timeout=timedelta(minutes=5), + retry_policy=retry_policy + ) + + if not is_valid: + return {"status": "on_hold", "reason": "Validation or fetching failed"} + + # 2. Push rows to Kafka + # Note: If any error happens during push, the activity itself catches it, + # sets status to 'on_hold' with error info, and raises an exception. + push_result = await workflow.execute_activity( + csv_push_to_kafka_activity, + record_id, + start_to_close_timeout=timedelta(minutes=30), + retry_policy=RetryPolicy(maximum_attempts=1) # No auto-retries for Kafka pushes to prevent duplicate writes + ) + rows_pushed = push_result["rows_pushed"] + schema_validation_errors = push_result.get("schema_validation_errors") or [] + + # 3. Update status to 'success'. Rows that failed pre-publish schema + # validation (e.g. a row missing a required field) are recorded here + # rather than silently dropped, even though the upload as a whole succeeded. + final_meta = {"rows_pushed": rows_pushed, "processed_at": workflow.now().isoformat()} + if schema_validation_errors: + final_meta["schema_validation_errors"] = schema_validation_errors + + await workflow.execute_activity( + csv_update_status_activity, + { + "record_id": record_id, + "status": "success", + "meta_data": final_meta + }, + start_to_close_timeout=timedelta(seconds=10), + retry_policy=retry_policy + ) + + return {"status": "success", "rows_pushed": rows_pushed, "schema_validation_errors": schema_validation_errors} + + +@workflow.defn +class CsvBatchProcessingWorkflow: + @workflow.run + async def run(self) -> Dict[str, Any]: + """ + Runs batch execution for all pending CSV uploads. + Retrieves pending records and executes child workflows in parallel. + """ + retry_policy = RetryPolicy( + maximum_attempts=2, + initial_interval=timedelta(seconds=2) + ) + + # Fetch pending csv_upload IDs + pending_ids: List[int] = await workflow.execute_activity( + fetch_pending_csv_uploads_activity, + start_to_close_timeout=timedelta(minutes=2), + retry_policy=retry_policy + ) + + if not pending_ids: + return {"processed_count": 0, "message": "No pending CSV uploads found."} + + # Fan-out child workflows to process each CSV in parallel + child_tasks = [] + for pid in pending_ids: + child_tasks.append( + workflow.execute_child_workflow( + CsvProcessingWorkflow.run, + pid, + id=f"csv-batch-child-{pid}" + ) + ) + + results = await asyncio.gather(*child_tasks, return_exceptions=True) + success_count = sum(1 for r in results if not isinstance(r, Exception)) + failed_count = len(results) - success_count + + return { + "processed_count": len(pending_ids), + "success_count": success_count, + "failed_count": failed_count + } diff --git a/main.py b/main.py index 2aab7a3..8e7f1c9 100644 --- a/main.py +++ b/main.py @@ -6,8 +6,8 @@ import uvicorn from fastapi import FastAPI -from app.api.bulk import router as bulk_router -from app.api.routes import router as submissions_router +from app.api.router import api_router +from app.api.exceptions import register_exception_handlers from app.kafka.consumer import IngestionConsumer from app.logging_config import configure_logging from app.temporal.worker import start_worker @@ -26,8 +26,8 @@ def run_web(): version="1.0.0", ) - app.include_router(submissions_router) - app.include_router(bulk_router) + register_exception_handlers(app) + app.include_router(api_router) @app.get("/health") def health_check(): diff --git a/requirements-prod.txt b/requirements-prod.txt index 201ddf8..22954ef 100644 --- a/requirements-prod.txt +++ b/requirements-prod.txt @@ -11,5 +11,6 @@ sentence-transformers==3.0.1 scikit-learn torch numpy>=1.24.0 +pandas>=2.0.0 google-cloud-storage pypdf>=4.0.0 diff --git a/schema.sql b/schema.sql index ddad346..704eafd 100644 --- a/schema.sql +++ b/schema.sql @@ -257,6 +257,43 @@ CREATE TABLE submission_metrics ( REFERENCES submissions(submission_id, tenant_code) ON DELETE CASCADE ); +-- ========================================================================= +-- 11. CSV UPLOAD TRACKING +-- ========================================================================= + +CREATE TABLE csv_uploads ( + id SERIAL PRIMARY KEY, + report_type VARCHAR(100) NOT NULL, + program_name VARCHAR(255), + leader_category VARCHAR(255), + file_name VARCHAR(500), + file_size BIGINT, + cloud_storage_path TEXT NOT NULL, + meta_data JSONB DEFAULT '{}'::jsonb, + status VARCHAR(20) NOT NULL DEFAULT 'pending', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT uq_csv_uploads UNIQUE (program_name, leader_category, report_type, file_name, file_size) +); + +CREATE INDEX idx_csv_uploads_status + ON csv_uploads (status); + +-- Keep updated_at fresh automatically +CREATE OR REPLACE FUNCTION set_updated_at() +RETURNS TRIGGER AS $$ +BEGIN + NEW.updated_at = now(); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_csv_uploads_updated_at ON csv_uploads; +CREATE TRIGGER trg_csv_uploads_updated_at + BEFORE UPDATE ON csv_uploads + FOR EACH ROW + EXECUTE FUNCTION set_updated_at(); + -- ========================================================================= -- INDEXES FOR HIGH-PERFORMANCE ANALYTICS -- ========================================================================= diff --git a/tests/TEST_CASES.csv b/tests/TEST_CASES.csv index 3487551..1dd30d3 100644 --- a/tests/TEST_CASES.csv +++ b/tests/TEST_CASES.csv @@ -243,3 +243,97 @@ LLM-002,LLM & Cost Tracking,split_llm_usage separates prompt/completion tokens f LLM-003,LLM & Cost Tracking,Token/cost fallback estimate used only when no usage was ever obtained,Edge,Medium,An LLM call that fails before returning any response,"Trigger a failure prior to receiving a response, then log to llm_logs","Falls back to a word-count estimate for prompt/completion tokens only in this case; real usage is used whenever available, even on a later (e.g. parse) failure",Verified,"1) Force an LLM call to fail at the network level (e.g. invalid API key) for one submission 2) DB: SELECT prompt_tokens, completion_tokens, meta_data FROM llm_logs WHERE submission_id='' AND status='failed'; 3) meta_data should be NULL/empty (no real usage was ever returned) and token counts should be rough word-count estimates, not zero" +UPLOAD-001,CSV Upload & Process API,Missing Authorization header rejected on upload,Security,Critical,Web API running (--mode web or analytics-web container),POST /v1/upload/ with a valid multipart form body but NO Authorization header,"401/403 rejection before any validation, GCS upload, or DB write occurs",Verified,"curl -s -o /dev/null -w ""%{http_code}\n"" -X POST http://localhost:8000/v1/upload/ +-> expect 403 (FastAPI's HTTPBearer default response for a missing Authorization header)" +UPLOAD-002,CSV Upload & Process API,Invalid Bearer token rejected on upload,Security,Critical,Web API running,POST /v1/upload/ with Authorization: Bearer ,401 Unauthorized: Invalid token (secrets.compare_digest check in app/api/deps.py fails),Verified,"curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer wrong-token"" -w ""\n%{http_code}\n"" +-> expect {""detail"":""Unauthorized: Invalid token""} and HTTP 401" +UPLOAD-003,CSV Upload & Process API,Valid story CSV uploads successfully,Positive,Critical,Web API running; GCS credentials configured; a valid AUTH_TOKEN,"POST /v1/upload/ with report_type=story, valid program_name/leader_category/tenant_code, and a CSV matching STORY_CSV_COLUMN headers exactly",200 {status: pending}; a csv_uploads row is created (status=pending); the raw CSV is uploaded to gs:////_,Verified,"1) AUTH_TOKEN=$(grep ""^AUTH_TOKEN="" .env | cut -d= -f2) +curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer $AUTH_TOKEN"" -F ""report_type=story"" -F ""program_name=P"" -F ""leader_category=L"" -F ""tenant_code=mitra"" -F ""file=@tests/csv_uploads/valid_story.csv;type=text/csv"" -w ""\n%{http_code}\n"" +2) DB: SELECT id, status, cloud_storage_path FROM csv_uploads ORDER BY id DESC LIMIT 1; -> status='pending', cloud_storage_path populated +3) Confirm the object exists in GCS at that cloud_storage_path" +UPLOAD-004,CSV Upload & Process API,Valid discussion CSV uploads successfully,Positive,Critical,Web API running; GCS credentials configured,POST /v1/upload/ with report_type=discussion and a CSV matching DISCUSSION_CSV_COLUMN headers exactly,200 {status: pending}; csv_uploads row created with report_type='discussion',Verified,"curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer $AUTH_TOKEN"" -F ""report_type=discussion"" -F ""program_name=P"" -F ""leader_category=L"" -F ""tenant_code=mitra"" -F ""file=@tests/csv_uploads/valid_discussion.csv;type=text/csv"" -w ""\n%{http_code}\n"" +DB: SELECT id, report_type, status FROM csv_uploads ORDER BY id DESC LIMIT 1; -> report_type='discussion', status='pending'" +UPLOAD-005,CSV Upload & Process API,Invalid report_type rejected,Negative,High,Web API running,POST /v1/upload/ with report_type=survey (not story/discussion),"400 ""Only 'story' or 'discussion' report types are accepted.""; no GCS upload or DB row created",Verified,"curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer $AUTH_TOKEN"" -F ""report_type=survey"" -F ""program_name=P"" -F ""leader_category=L"" -F ""tenant_code=mitra"" -F ""file=@tests/csv_uploads/valid_story.csv;type=text/csv"" -w ""\n%{http_code}\n"" +-> expect HTTP 400 with that detail message" +UPLOAD-006,CSV Upload & Process API,Non-.csv file extension rejected,Negative,High,Web API running,POST /v1/upload/ with a file whose name does not end in .csv (e.g. .txt),"400 ""Only .csv files are accepted""",Verified,"curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer $AUTH_TOKEN"" -F ""report_type=story"" -F ""program_name=P"" -F ""leader_category=L"" -F ""tenant_code=mitra"" -F ""file=@tests/csv_uploads/not_a_csv.txt;type=text/plain"" -w ""\n%{http_code}\n"" +-> expect HTTP 400 ""Only .csv files are accepted""" +UPLOAD-007,CSV Upload & Process API,Empty (0-byte) file rejected,Negative,High,Web API running,POST /v1/upload/ with a genuinely empty (0-byte) .csv file,"400 ""Uploaded file is empty""",Verified,"curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer $AUTH_TOKEN"" -F ""report_type=story"" -F ""program_name=P"" -F ""leader_category=L"" -F ""tenant_code=mitra"" -F ""file=@tests/csv_uploads/empty.csv;type=text/csv"" -w ""\n%{http_code}\n"" +-> expect HTTP 400 ""Uploaded file is empty""" +UPLOAD-008,CSV Upload & Process API,File exceeding MAX_CSV_UPLOAD_BYTES rejected,Negative,Medium,Web API running; MAX_CSV_UPLOAD_BYTES default 10485760 (10MB),POST /v1/upload/ with a CSV file larger than MAX_CSV_UPLOAD_BYTES,"413 ""Uploaded file is too large""; the route reads only MAX_CSV_UPLOAD_BYTES+1 bytes so the whole file is never buffered",Not Tested,"1) python3 -c ""open('/tmp/big.csv','wb').write(b'id,Title\n' + b'1,x\n'*3000000)"" (produce >10MB) +2) curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer $AUTH_TOKEN"" -F ""report_type=story"" -F ""program_name=P"" -F ""leader_category=L"" -F ""tenant_code=mitra"" -F ""file=@/tmp/big.csv;type=text/csv"" -w ""\n%{http_code}\n"" +-> expect HTTP 413" +UPLOAD-009,CSV Upload & Process API,CSV missing required columns rejected with no GCS/DB side effects,Negative,Critical,Web API running,POST /v1/upload/ with a CSV whose header row is missing one or more columns from STORY_CSV_COLUMN (e.g. no 'Session ID'),"400 with detail 'CSV column mismatch...' and errors listing the missing columns; crucially, NO GCS object is created and NO csv_uploads row is inserted (validated before any side effect)",Verified,"1) DB: SELECT count(*) FROM csv_uploads; (note baseline) +2) curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer $AUTH_TOKEN"" -F ""report_type=story"" -F ""program_name=P"" -F ""leader_category=L"" -F ""tenant_code=mitra"" -F ""file=@tests/csv_uploads/missing_columns.csv;type=text/csv"" -w ""\n%{http_code}\n"" +-> expect HTTP 400, errors=[""Missing columns: ['Session ID']""] +3) DB: re-run the count -> UNCHANGED from step 1 +4) grep app/web log for 'Uploaded CSV to gs://' -> no new line for this request" +UPLOAD-010,CSV Upload & Process API,CSV with extra/unexpected columns rejected with no GCS/DB side effects,Negative,Critical,Web API running,POST /v1/upload/ with a CSV containing all expected columns PLUS unexpected extra ones,"400 with errors listing the extra/unexpected columns; no GCS upload, no DB row",Verified,"curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer $AUTH_TOKEN"" -F ""report_type=story"" -F ""program_name=P"" -F ""leader_category=L"" -F ""tenant_code=mitra"" -F ""file=@tests/csv_uploads/extra_columns.csv;type=text/csv"" -w ""\n%{http_code}\n"" +-> expect HTTP 400, errors=[""Extra/unexpected columns: ['pri_member_info', 'school_representative_info', 'session___i']""] +DB: confirm csv_uploads row count unchanged" +UPLOAD-011,CSV Upload & Process API,Malformed/unparseable CSV content rejected,Negative,High,Web API running,POST /v1/upload/ with a .csv file whose content is corrupt (e.g. unterminated quoted field) so pandas fails to parse it,"400 CSV column mismatch (pd.read_csv failure is caught and reported as InvalidCsvColumns, not a 500)",Verified,"curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer $AUTH_TOKEN"" -F ""report_type=story"" -F ""program_name=P"" -F ""leader_category=L"" -F ""tenant_code=mitra"" -F ""file=@tests/csv_uploads/malformed.csv;type=text/csv"" -w ""\n%{http_code}\n"" +-> expect HTTP 400 (not 500); errors mention missing/extra columns from the failed parse" +UPLOAD-012,CSV Upload & Process API,Duplicate file upload rejected,Negative,High,A prior successful upload exists with the same program_name/leader_category/report_type/file_name/file_size,Upload the exact same CSV file with the exact same program_name/leader_category/tenant_code a second time,"400 ""FILE ALREADY EXISTS"" (DuplicateFile, backed by the uq_csv_uploads unique constraint)",Verified,"1) Upload tests/csv_uploads/valid_story.csv with program_name=P once (succeeds) +2) Upload the exact same file with the exact same program_name/leader_category/tenant_code again +-> expect HTTP 400 {""detail"":""FILE ALREADY EXISTS""} +3) DB: SELECT count(*) FROM csv_uploads WHERE file_name='valid_story.csv'; -> exactly 1" +UPLOAD-013,CSV Upload & Process API,tenant_code Form field defaults to 'mitra' when omitted,Edge,Low,Web API running,POST /v1/upload/ omitting the tenant_code form field entirely,"Request succeeds; the resulting csv_uploads.meta_data.tenant_code is 'mitra' (the Form(default=""mitra"") value)",Not Tested,"curl -s -X POST http://localhost:8000/v1/upload/ -H ""Authorization: Bearer $AUTH_TOKEN"" -F ""report_type=story"" -F ""program_name=P"" -F ""leader_category=L"" -F ""file=@tests/csv_uploads/valid_story.csv;type=text/csv"" -w ""\n%{http_code}\n"" (no tenant_code field) +DB: SELECT meta_data->>'tenant_code' FROM csv_uploads ORDER BY id DESC LIMIT 1; -> expect 'mitra'" +UPLOAD-014,CSV Upload & Process API,real-time mode triggers CsvProcessingWorkflow immediately on valid upload,Positive,Critical,PROCESSING_MODE=real-time; Temporal + worker running,Upload a valid CSV,CsvProcessingWorkflow (id=csv-upload-) starts immediately; record eventually reaches status='success' or 'on_hold' without any manual /v1/process/csv call,Verified,"1) Confirm PROCESSING_MODE=real-time +2) Upload tests/csv_uploads/valid_story.csv +3) Temporal UI (localhost:8233): search workflow id 'csv-upload-' -> started within seconds +4) DB: SELECT status FROM csv_uploads WHERE id=; -> progresses from 'pending' to 'in_progress' to a terminal status without calling /v1/process/csv" +UPLOAD-015,CSV Upload & Process API,"batch mode leaves a valid upload pending, no workflow started",Positive,High,PROCESSING_MODE=batch,Upload a valid CSV while PROCESSING_MODE=batch,csv_uploads row created with status='pending'; no CsvProcessingWorkflow is started at upload time,Verified,"1) Start the web process with PROCESSING_MODE=batch +2) Upload tests/csv_uploads/valid_story.csv +3) DB: SELECT status FROM csv_uploads WHERE id=; -> remains 'pending' (does not change on its own) +4) Temporal UI: confirm no 'csv-upload-' workflow was started" +UPLOAD-016,CSV Upload & Process API,GCS upload failure prevents any csv_uploads row from being created,Edge,Medium,GCS temporarily unreachable/misconfigured (e.g. invalid BUCKET_NAME),Upload a column-valid CSV while GCS is unreachable,"500/RuntimeError 'GCS Upload failed...'; NO csv_uploads row is created, since the DB insert only happens after a successful GCS upload",Not Tested,"1) Temporarily set BUCKET_NAME to a nonexistent bucket in a test .env, restart web +2) Upload tests/csv_uploads/valid_story.csv +3) Expect a 500 error mentioning 'GCS Upload failed' +4) DB: SELECT count(*) FROM csv_uploads; -> unchanged from before the request +5) Restore BUCKET_NAME afterward" +UPLOAD-017,CSV Upload & Process API,Temporal unreachable during real-time trigger marks record on_hold,Edge,High,PROCESSING_MODE=real-time; Temporal server stopped/unreachable,Upload a valid CSV while the Temporal server is down,"GCS upload and DB row ARE created (status starts 'pending'), but the workflow-start attempt fails; record is updated to status='on_hold' with the error captured in meta_data, and the request raises a 500",Not Tested,"1) Stop the Temporal server +2) Upload tests/csv_uploads/valid_story.csv with PROCESSING_MODE=real-time +3) DB: SELECT status, meta_data->>'error' FROM csv_uploads ORDER BY id DESC LIMIT 1; -> status='on_hold', error mentions 'Temporal trigger failed' +4) Restart Temporal afterward" +UPLOAD-018,CSV Upload & Process API,Manually processing a pending record starts CsvProcessingWorkflow,Positive,Critical,A csv_uploads row exists with status='pending' (e.g. uploaded under batch mode),POST /v1/process/csv/{record_id} for that pending record,"200 {status: success, message: 'CSV processing workflow started'}; record status flips to 'in_progress' (atomically claimed) then eventually to a terminal status",Verified,"curl -s -X POST http://localhost:8000/v1/process/csv/ -H ""Authorization: Bearer $AUTH_TOKEN"" -w ""\n%{http_code}\n"" +-> expect 200; DB: SELECT status FROM csv_uploads WHERE id=; -> 'in_progress' shortly after, then terminal once a worker processes it" +UPLOAD-019,CSV Upload & Process API,Processing a nonexistent record_id returns 404,Negative,Medium,Web API running,POST /v1/process/csv/999999 (an id that does not exist),"404 ""Record not found""",Verified,"curl -s -X POST http://localhost:8000/v1/process/csv/999999 -H ""Authorization: Bearer $AUTH_TOKEN"" -w ""\n%{http_code}\n"" +-> expect HTTP 404 {""detail"":""Record not found""}" +UPLOAD-020,CSV Upload & Process API,Processing an already in_progress record returns 409,Edge,High,A csv_uploads row currently has status='in_progress',POST /v1/process/csv/{record_id} for that in_progress record,"409 ""Record is already being processed"" (RecordAlreadyProcessing)",Verified,"1) POST /v1/process/csv/ once on a pending record (flips to in_progress) +2) Immediately POST /v1/process/csv/ again +-> expect HTTP 409 {""detail"":""Record is already being processed""}" +UPLOAD-021,CSV Upload & Process API,Reprocessing a terminal-status record returns 409,Negative,High,A csv_uploads row has a terminal status (success or on_hold),POST /v1/process/csv/{record_id} for a record whose status is 'success' (or 'on_hold'),"409 ""Only pending records can be processed"" (RecordNotPending — fixed from the ported sibling's inconsistent 400 to match RecordAlreadyProcessing's 409, since both are 'conflicts with current status')",Verified,"1) Let a record reach status='success' (upload + let it process fully) +2) POST /v1/process/csv/ again +-> expect HTTP 409 {""detail"":""Only pending records can be processed""}" +UPLOAD-022,CSV Upload & Process API,Process endpoint requires the same Bearer auth as upload,Security,Critical,Web API running,POST /v1/process/csv/{id} with no Authorization header,403 rejection before any record lookup or claim attempt,Verified,"curl -s -o /dev/null -w ""%{http_code}\n"" -X POST http://localhost:8000/v1/process/csv/1 +-> expect 403" +UPLOAD-023,CSV Upload & Process API,Concurrent process calls on the same pending record are race-safe,Edge,High,A csv_uploads row exists with status='pending',Fire two POST /v1/process/csv/{record_id} requests for the same pending record at (as close to) the same time,"Exactly one request succeeds and starts the workflow; the other gets 409 RecordAlreadyProcessing — try_claim_for_processing's UPDATE ... WHERE status != 'in_progress' RETURNING status is an atomic compare-and-swap, so no double-processing is possible regardless of timing",Not Tested,"1) Create a pending record +2) Fire two curl POST /v1/process/csv/ calls in parallel (e.g. via `&` backgrounding in the same shell, both against the same id) +3) Confirm exactly one response is 200 and the other is 409; DB: only one workflow id csv-upload- was ever started (check Temporal UI for a single execution, not two)" +UPLOAD-024,CSV Upload & Process API,CSV row missing Session ID is skipped pre-publish and recorded in meta_data,Negative,Critical,A csv_uploads record whose source CSV has a blank Session ID cell for one row,"Upload + process tests/csv_uploads/missing_session_id_value.csv (valid columns present, but the Session ID value is blank)",The row is NOT auto-assigned a generated session id (removed behavior); validate_ingestion_schema flags 'sessionId' is null against STORY_KAFKA_SCHEMA; the row is skipped (not published to Kafka) and the problem is recorded in csv_uploads.meta_data.schema_validation_errors; rows_pushed reflects only the rows that DID pass,Verified,"1) Upload tests/csv_uploads/missing_session_id_value.csv, then POST /v1/process/csv/ +2) DB: SELECT meta_data FROM csv_uploads WHERE id=; +-> meta_data.schema_validation_errors contains an entry with ""'sessionId' is null"" and sessionId: null; meta_data.rows_pushed does not count this row +3) Confirm no Kafka message was published for this row (e.g. no matching submissionId reaches the submissions table)" +UPLOAD-025,CSV Upload & Process API,CSV row missing other required schema fields is skipped pre-publish and recorded,Negative,High,"A CSV row that is missing/blank on required STORY_KAFKA_SCHEMA fields other than sessionId (e.g. Transcript Link, Blurb, Content)",Upload + process a story CSV row with those fields left blank,"Each missing/empty required field is listed in that row's schema_validation_errors entry (e.g. ""'data.transcriptLink' is null"", ""'data.blurb' is null""); the row is skipped, not published",Verified,"Upload+process a story CSV with Session ID populated but Transcript Link/Blurb/Content left blank +DB: SELECT meta_data->'schema_validation_errors' FROM csv_uploads WHERE id=; -> lists each missing field explicitly" +UPLOAD-026,CSV Upload & Process API,A fully-populated story CSV row still fails schema check on data.pdfUrls.masked (known/accepted gap),Edge,Medium,"A story CSV row with EVERY column populated, including Pdf/Transcript Link/Blurb/Content/Session ID",Upload + process tests/csv_uploads/valid_story.csv (fully populated),"Row STILL fails pre-publish validation with exactly one problem: ""'data.pdfUrls.masked' is missing"" — because the CSV-to-payload mapping only ever populates pdfUrls.original (masking happens downstream, after ingestion, per schema.sql's masked_pdf_urls/pii_masked_at columns). This is a known, currently-accepted gap in STORY_KAFKA_SCHEMA (confirmed with the team; schema intentionally left as-is for now) rather than a bug in the CSV pipeline itself.",Verified,"Upload+process tests/csv_uploads/valid_story.csv (or any fully-populated story row) +DB: SELECT meta_data->'schema_validation_errors' FROM csv_uploads WHERE id=; -> exactly one problem, ""'data.pdfUrls.masked' is missing""; rows_pushed=0" +UPLOAD-027,CSV Upload & Process API,Kafka broker unreachable during row push marks record on_hold and raises,Edge,Critical,A record has passed pre-publish schema validation for at least one row; Kafka broker stopped/unreachable,"Process a record whose row(s) would pass schema validation, while Kafka is down",confluent_kafka.Producer.flush()/produce() fails; csv_upload_repo.update_status sets status='on_hold' with stage='Kafka Publishing' in meta_data; the activity re-raises (Temporal will not silently report success),Not Tested,"1) Stop the Kafka broker +2) Process a record with at least one schema-valid row (would require adjusting STORY_KAFKA_SCHEMA or using a discussion CSV without the pdfUrls.masked gap) +3) DB: SELECT status, meta_data->>'stage' FROM csv_uploads WHERE id=; -> status='on_hold', stage='Kafka Publishing' +4) Temporal UI: confirm the csv_push_to_kafka_activity attempt shows a failure, not a false success +5) Restart Kafka afterward" +UPLOAD-028,CSV Upload & Process API,Missing program/leader-category DB match falls back to a generated UUID and generic name,Edge,Medium,program_name/leader_category values on the upload do not match any existing programs/leader_category row,Upload + process a CSV with a program_name/leader_category that has never been seen before (no matching DB row),"csv_processing_activity.py's DB lookup finds no match; leader_info/program_info fall back to {""id"": str(uuid.uuid4()), ""name"": , ...} rather than failing — the row is still processed (assuming it otherwise passes schema validation) with a freshly generated UUID standing in for the real id",Verified,"Code-level confirmation (app/temporal/csv_processing_activity.py lines ~201-213): when leader_row/program_row from the DB query is None, leader_info/program_info are built with id=str(uuid.uuid4()). To observe directly, use a program_name/leader_category guaranteed not to exist in the programs/leader_category tables and inspect the constructed payload (e.g. via a temporary debug log) to confirm tags.programId/tags.leaderCategoryId are freshly generated UUIDs, not matching any row in programs/leader_category." +UPLOAD-029,CSV Upload & Process API,CsvBatchProcessingWorkflow fans out all pending records as child workflows,Positive,High,PROCESSING_MODE=batch; multiple csv_uploads rows with status='pending',Manually start CsvBatchProcessingWorkflow (or wait for the csv-batch-processing schedule to fire),fetch_pending_csv_uploads_activity returns all pending ids; one CsvProcessingWorkflow child (id=csv-batch-child-) is started per pending record; results are aggregated into processed_count/success_count/failed_count,Not Tested,"1) Ensure 2+ csv_uploads rows have status='pending' +2) Start CsvBatchProcessingWorkflow manually via a Temporal client, or trigger the csv-batch-processing schedule +3) Temporal UI: confirm one csv-batch-child- child workflow per pending record +4) DB: confirm all previously-pending rows have moved to a terminal status" +UPLOAD-030,CSV Upload & Process API,CsvBatchProcessingWorkflow with zero pending records returns processed_count=0,Edge,Low,No csv_uploads rows with status='pending',Trigger CsvBatchProcessingWorkflow,"Returns {processed_count: 0, message: 'No pending CSV uploads found.'} without starting any child workflow",Not Tested,"1) DB: confirm SELECT count(*) FROM csv_uploads WHERE status='pending'; = 0 +2) Trigger CsvBatchProcessingWorkflow +3) Temporal UI: confirm the workflow result is {""processed_count"": 0, ""message"": ""No pending CSV uploads found.""}" +UPLOAD-031,CSV Upload & Process API,csv-batch-processing and daily-batch-processing schedules register when PROCESSING_MODE=batch,Positive,Medium,PROCESSING_MODE=batch,Start the Temporal worker with PROCESSING_MODE=batch,Both the 'csv-batch-processing' (cron CSV_SCHEDULE_CRON_TIME) and 'daily-batch-processing' (cron BATCH_SCHEDULE_CRON) schedules register successfully in Temporal on worker startup,Verified,"1) PROCESSING_MODE=batch python main.py --mode worker +2) Worker log should show 'CSV batch schedule successfully registered.' and 'Daily analysis batch schedule successfully registered.' +3) Temporal CLI/UI: confirm both schedule ids exist and show the correct cron expressions" +UPLOAD-032,CSV Upload & Process API,Stale batch schedules are deleted when switching back to real-time,Positive,Medium,Both batch schedules from UPLOAD-031 are currently registered,Restart the Temporal worker with PROCESSING_MODE=real-time,Both 'csv-batch-processing' and 'daily-batch-processing' schedules are deleted on startup — prevents a schedule left behind from a prior batch-mode config from silently retrying forever with outdated arguments,Verified,"1) With both schedules registered (see UPLOAD-031), restart: python main.py --mode worker (PROCESSING_MODE=real-time, the .env default) +2) Worker log should show 'Deleted stale batch schedule 'csv-batch-processing' (PROCESSING_MODE=real-time).' and the same for 'daily-batch-processing' +3) Temporal CLI/UI: confirm neither schedule id exists anymore"