fix bug 5330, add piii version 2 , remove temporal for csv uploads

.env.example

@@ -71,7 +71,6 @@ 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"}
@@ -92,7 +91,7 @@ MAX_PDF_TEXT_CHARS=40000
 # missing/null/empty a "required" path is rejected and routed to the DLQ topic
 # instead of being ingested. See app/kafka/consumer.py's _validate_ingestion_schema.
 STORY_KAFKA_SCHEMA={"create": {"required": ["submissionId", "submissionType", "sessionId", "tenantCode", "eventType", "eventPublishedAt", "tags.state", "tags.district", "tags.organization", "tags.programId", "tags.programName", "tags.leaderCategoryId", "tags.leaderCategoryName", "data.title", "data.designation", "data.submissionDate", "data.pdfUrls.original", "data.pdfUrls.masked", "data.transcriptLink", "data.challenges", "data.objective", "data.actionSteps", "data.impact", "data.duration", "data.blurb", "data.content"], "optional": ["data.imageUrls"]}, "update": {"required": ["submissionId", "submissionType", "sessionId", "tenantCode", "eventType", "eventPublishedAt"], "newValuesNoEmpty": true}, "delete": {"required": ["submissionId", "submissionType", "sessionId", "tenantCode", "eventType", "eventPublishedAt"]}}
-DISCUSSION_KAFKA_SCHEMA={"create": {"required": ["submissionId", "submissionType", "sessionId", "tenantCode", "eventType", "eventPublishedAt", "tags.state", "tags.district", "tags.organization", "tags.programId", "tags.programName", "tags.leaderCategoryId", "tags.leaderCategoryName", "data.title", "data.designation", "data.submissionDate", "data.pdfUrls.original", "data.pdfUrls.masked", "data.transcriptLink", "data.challenges", "data.solutions", "data.participantsData"], "optional": ["data.author", "data.language", "data.imageUrls"]}, "update": {"required": ["submissionId", "submissionType", "sessionId", "tenantCode", "eventType", "eventPublishedAt"], "newValuesNoEmpty": true}, "delete": {"required": ["submissionId", "submissionType", "sessionId", "tenantCode", "eventType", "eventPublishedAt"]}}
+DISCUSSION_KAFKA_SCHEMA={"create": {"required": ["submissionId", "submissionType", "sessionId", "tenantCode", "eventType", "eventPublishedAt", "tags.state", "tags.district", "tags.organization", "tags.programId", "tags.programName", "tags.leaderCategoryId", "tags.leaderCategoryName", "data.title", "data.designation", "data.submissionDate", "data.discussionDate", "data.pdfUrls.original", "data.pdfUrls.masked", "data.transcriptLink", "data.challenges", "data.solutions", "data.participantsData"], "optional": ["data.author", "data.language", "data.imageUrls"]}, "update": {"required": ["submissionId", "submissionType", "sessionId", "tenantCode", "eventType", "eventPublishedAt"], "newValuesNoEmpty": true}, "delete": {"required": ["submissionId", "submissionType", "sessionId", "tenantCode", "eventType", "eventPublishedAt"]}}
 
 # Scoped Ingestion Pipeline Configurations (JSON strings)
 # Each step optionally accepts "llm_model" / "max_tokens" / "llm_timeout_seconds" to override

app/api/routes/uploads.py

@@ -1,5 +1,5 @@
 import logging
-from fastapi import APIRouter, Depends, Form, UploadFile, File
+from fastapi import APIRouter, Depends, Form, UploadFile, File, BackgroundTasks
 from fastapi.security import HTTPAuthorizationCredentials
 
 from app.api.deps import verify_auth_token
@@ -19,6 +19,7 @@
 
 @uploads_router.post("/upload/", response_model=UploadResponse)
 async def upload_report(
+    background_tasks: BackgroundTasks,
     report_type: str = Form(...),
     program_name: str = Form(...),
     leader_category: str = Form(...),
@@ -30,7 +31,8 @@ async def upload_report(
     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).
+    status='pending'. Processing (column validation, Kafka publishing)
+    runs in the background.
     """
     # Pure request-shape checks
     report_type = validate_report_type(report_type)
@@ -47,15 +49,18 @@ async def upload_report(
         tenant_code=tenant_code,
         file_name=file.filename,
         file_bytes=file_bytes,
+        background_tasks=background_tasks,
     )
 
 
 @uploads_router.post("/process/csv/{record_id}")
 async def push_record(
     record_id: int,
+    background_tasks: BackgroundTasks,
     _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)
+    return await upload_service.handle_push(record_id, background_tasks)
+