feat: download campaign report

api/routes/campaign.py

@@ -1,12 +1,16 @@
+import csv
+import io
 import json
 from datetime import datetime
 from typing import List, Optional
 from zoneinfo import ZoneInfo
 
 from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field, field_validator, model_validator
 
 from api.constants import (
+    BACKEND_API_ENDPOINT,
     DEFAULT_CAMPAIGN_RETRY_CONFIG,
     DEFAULT_ORG_CONCURRENCY_LIMIT,
 )
@@ -18,6 +22,7 @@ from api.services.campaign.runner import campaign_runner_service
 from api.services.campaign.source_sync_factory import get_sync_service
 from api.services.quota_service import check_dograh_quota
 from api.services.storage import storage_fs
+from api.utils.transcript import generate_transcript_text
 
 router = APIRouter(prefix="/campaign")
 
@@ -662,3 +667,76 @@ async def get_campaign_source_download_url(
         raise HTTPException(
             status_code=500, detail=f"Failed to generate download URL: {str(e)}"
         )
+
+
+def _transcript_from_logs(logs: dict | None) -> str:
+    """Extract transcript text from workflow run logs JSON."""
+    if not logs:
+        return ""
+    events = logs.get("realtime_feedback_events", [])
+    return generate_transcript_text(events).strip()
+
+
+@router.get("/{campaign_id}/report")
+async def download_campaign_report(
+    campaign_id: int,
+    user: UserModel = Depends(get_user),
+) -> StreamingResponse:
+    """Download a CSV report of completed campaign runs."""
+    campaign = await db_client.get_campaign(campaign_id, user.selected_organization_id)
+    if not campaign:
+        raise HTTPException(status_code=404, detail="Campaign not found")
+
+    runs = await db_client.get_completed_runs_for_report(campaign_id)
+
+    output = io.StringIO()
+    writer = csv.writer(output)
+    writer.writerow(
+        [
+            "Run ID",
+            "Created At",
+            "Customer Name",
+            "Phone Number",
+            "Call Tags",
+            "Call Duration (s)",
+            "Transcript",
+            "Recording URL",
+        ]
+    )
+
+    for run in runs:
+        initial = run.initial_context or {}
+        gathered = run.gathered_context or {}
+        cost = run.cost_info or {}
+
+        recording_url = ""
+        if run.public_access_token:
+            recording_url = (
+                f"{BACKEND_API_ENDPOINT}/api/v1/public/download/workflow"
+                f"/{run.public_access_token}/recording"
+            )
+
+        call_tags = gathered.get("call_tags", [])
+        if isinstance(call_tags, list):
+            call_tags = ", ".join(str(t) for t in call_tags)
+
+        writer.writerow(
+            [
+                run.id,
+                run.created_at.isoformat() if run.created_at else "",
+                initial.get("first_name", ""),
+                initial.get("phone_number", ""),
+                call_tags,
+                cost.get("call_duration_seconds", ""),
+                _transcript_from_logs(run.logs),
+                recording_url,
+            ]
+        )
+
+    output.seek(0)
+    filename = f"campaign_{campaign_id}_report.csv"
+    return StreamingResponse(
+        output,
+        media_type="text/csv",
+        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+    )

api/routes/user.py

@@ -3,7 +3,7 @@ from typing import List, Literal, Optional, TypedDict, Union
 
 from fastapi import APIRouter, Depends, HTTPException, Query
 from loguru import logger
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 
 from api.db import db_client
 from api.db.models import (
@@ -15,7 +15,7 @@ from api.services.configuration.check_validity import (
     UserConfigurationValidator,
 )
 from api.services.configuration.defaults import DEFAULT_SERVICE_PROVIDERS
-from api.services.configuration.masking import mask_user_config
+from api.services.configuration.masking import check_for_masked_keys, mask_user_config
 from api.services.configuration.merge import merge_user_configurations
 from api.services.configuration.registry import REGISTRY, ServiceType
 from api.services.mps_service_key_client import mps_service_key_client
@@ -113,7 +113,15 @@ async def update_user_configurations(
     incoming_dict.pop("organization_pricing", None)
 
     # Merge via helper
-    user_configurations = merge_user_configurations(existing_config, incoming_dict)
+    try:
+        user_configurations = merge_user_configurations(existing_config, incoming_dict)
+    except ValidationError as e:
+        raise HTTPException(status_code=422, detail=str(e))
+
+    try:
+        check_for_masked_keys(user_configurations)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
 
     try:
         validator = UserConfigurationValidator()