fix: call_imports

README.md

@@ -429,6 +429,7 @@ gcs:
   bucket_name: "your-gcs-bucket-name"
   project_id: "your-gcp-project-id"
   credentials_path: null  # Optional; falls back to GOOGLE_APPLICATION_CREDENTIALS / ADC
+  signing_service_account_email: null  # Optional override for IAM signBlob (GKE Workload Identity)
   prefix: "audio/"  # Same object key layout as S3
 
 # CORS Settings

app/api/v1/routes/call_import_evaluations.py

@@ -1737,6 +1737,12 @@ async def export_call_import_evaluation_csv(
                     continue  # would clobber a real column
                 custom_export.append((name, csv_header))
 
+    if (
+        call_import.source_format == "audio"
+        and "conversation_id" not in standard_export_headers
+    ):
+        standard_export_headers.insert(0, "conversation_id")
+
     # Build the metric columns: each parent (if any) gets a value column
     # and (when capture_rationale=true) a "<Parent> - LLM Rationale"
     # column. The per-child boolean columns are intentionally suppressed
@@ -1824,6 +1830,8 @@ def _project_rows() -> Iterator[Dict[str, str]]:
             )
             for header in standard_export_headers:
                 value = raw.get(header)
+                if value is None and header == "conversation_id":
+                    value = source_row.conversation_id
                 row_out[header] = "" if value is None else str(value)
             for export_header, csv_header in custom_export:
                 value = raw.get(csv_header)

app/api/v1/routes/call_import_schemas.py

@@ -7,9 +7,10 @@
 parameters are mapped to source columns.
 
 Every schema MUST contain exactly one parameter with
-``type='conversation_id'``. At most one parameter each may use
-``type='recording_url'``, ``type='recording_date'``, or
-``type='transcript'``. Only ``conversation_id`` is forced to
+``type='conversation_id'`` and exactly one with
+``type='recording_url'``. At most one parameter each may use
+``type='recording_date'`` or ``type='transcript'``.
+``conversation_id`` and ``recording_url`` are forced to
 ``is_required=True``. The invariant is enforced here on create + update
 because it spans the parent (`call_import_schemas`) and the children
 (`call_import_schema_parameters`) which are written in the same
@@ -77,7 +78,11 @@ def _materialize_parameters(
     rows: List[CallImportSchemaParameter] = []
     for idx, param in enumerate(payload_params):
         is_required = (
-            param.type == CallImportParameterType.CONVERSATION_ID
+            param.type
+            in (
+                CallImportParameterType.CONVERSATION_ID,
+                CallImportParameterType.RECORDING_URL,
+            )
             or bool(param.is_required)
         )
         rows.append(

app/api/v1/routes/call_imports.py

@@ -378,6 +378,20 @@ def _coerce_parameter_value(
     return cell
 
 
+def _parameter_is_required(param: CallImportSchemaParameter) -> bool:
+    """Return whether a schema parameter must be mapped on every upload."""
+    if param.is_required:
+        return True
+    try:
+        param_type = CallImportParameterType(param.type)
+    except ValueError:
+        return False
+    return param_type in (
+        CallImportParameterType.CONVERSATION_ID,
+        CallImportParameterType.RECORDING_URL,
+    )
+
+
 def _apply_schema_mapping(
     fieldnames: List[str],
     rows_iter: Iterable[Dict[str, str]],
@@ -439,7 +453,7 @@ def _apply_schema_mapping(
             if mapped_header
             else None
         )
-        if param.is_required and canonical is None:
+        if _parameter_is_required(param) and canonical is None:
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail=(
@@ -543,7 +557,7 @@ def _apply_schema_mapping(
                 row_idx=idx,
                 param_name=param.name,
             )
-            if param.is_required and coerced is None:
+            if _parameter_is_required(param) and coerced is None:
                 raise HTTPException(
                     status_code=status.HTTP_400_BAD_REQUEST,
                     detail=(
@@ -2248,6 +2262,37 @@ async def list_call_import_datasets(
     return [row[0] for row in rows if row[0]]
 
 
+@router.get(
+    "/diarisation-prompt-default",
+    response_model=CallImportDiarisationPromptDefaultResponse,
+    operation_id="getCallImportDiarisationPromptDefault",
+)
+async def get_call_import_diarisation_prompt_default(
+    api_key: str = Depends(get_api_key),
+    organization_id: UUID = Depends(get_organization_id),
+) -> CallImportDiarisationPromptDefaultResponse:
+    """Return the canonical LLM diariser prompt.
+
+    The Transcribe / Run Evaluation modals call this on open so they
+    can pre-fill the prompt textarea. Returning the constant from the
+    backend (rather than hard-coding it in the frontend) keeps the
+    fallback used by the worker and the placeholder shown in the UI
+    in lock-step — operators always see the *actual* default they'd
+    get if they leave the field blank.
+
+    Registered before ``GET /{call_import_id}`` so the static path is
+    not mistaken for a UUID import id (which would 422).
+    """
+    del api_key, organization_id
+    from app.workers.tasks.helpers.llm_diarisation import (
+        DEFAULT_DIARIZATION_PROMPT,
+    )
+
+    return CallImportDiarisationPromptDefaultResponse(
+        prompt=DEFAULT_DIARIZATION_PROMPT
+    )
+
+
 @router.patch(
     "/{call_import_id}",
     response_model=CallImportResponse,
@@ -2983,34 +3028,6 @@ async def bulk_delete_call_import_rows(
 # ---------------------------------------------------------------------------
 
 
-@router.get(
-    "/diarisation-prompt-default",
-    response_model=CallImportDiarisationPromptDefaultResponse,
-    operation_id="getCallImportDiarisationPromptDefault",
-)
-async def get_call_import_diarisation_prompt_default(
-    api_key: str = Depends(get_api_key),
-    organization_id: UUID = Depends(get_organization_id),
-) -> CallImportDiarisationPromptDefaultResponse:
-    """Return the canonical LLM diariser prompt.
-
-    The Transcribe / Run Evaluation modals call this on open so they
-    can pre-fill the prompt textarea. Returning the constant from the
-    backend (rather than hard-coding it in the frontend) keeps the
-    fallback used by the worker and the placeholder shown in the UI
-    in lock-step — operators always see the *actual* default they'd
-    get if they leave the field blank.
-    """
-    del api_key, organization_id
-    from app.workers.tasks.helpers.llm_diarisation import (
-        DEFAULT_DIARIZATION_PROMPT,
-    )
-
-    return CallImportDiarisationPromptDefaultResponse(
-        prompt=DEFAULT_DIARIZATION_PROMPT
-    )
-
-
 def _select_rows_for_transcription(
     db: Session,
     call_import: CallImport,

app/config.py

@@ -56,6 +56,7 @@ class Settings(BaseSettings):
     GCS_BUCKET_NAME: Optional[str] = None
     GCS_PROJECT_ID: Optional[str] = None
     GCS_CREDENTIALS_PATH: Optional[str] = None
+    GCS_SIGNING_SERVICE_ACCOUNT_EMAIL: Optional[str] = None
     GCS_PREFIX: str = "audio/"
 
     # Azure Blob Storage Configuration
@@ -458,6 +459,10 @@ def load_config_from_file(config_path: str) -> None:
             settings.GCS_PROJECT_ID = gcs_config["project_id"]
         if "credentials_path" in gcs_config:
             settings.GCS_CREDENTIALS_PATH = gcs_config["credentials_path"]
+        if "signing_service_account_email" in gcs_config:
+            settings.GCS_SIGNING_SERVICE_ACCOUNT_EMAIL = gcs_config[
+                "signing_service_account_email"
+            ]
         if "prefix" in gcs_config:
             settings.GCS_PREFIX = gcs_config["prefix"]
 

app/models/schemas.py

@@ -2209,9 +2209,10 @@ class CallImportSchemaParameterBase(BaseModel):
             "Parameter type. One of conversation_id / recording_url / "
             "recording_date / transcript / text / number / boolean / "
             "datetime / url. Exactly one parameter of type "
-            "'conversation_id' must be present; at most one each of "
-            "'recording_url', 'recording_date', and 'transcript'. Only "
-            "'conversation_id' is forced required."
+            "'conversation_id' and exactly one of type 'recording_url' "
+            "must be present; at most one each of 'recording_date' and "
+            "'transcript'. Both conversation_id and recording_url are "
+            "forced required."
         ),
     )
     description: Optional[str] = Field(
@@ -2223,8 +2224,9 @@ class CallImportSchemaParameterBase(BaseModel):
         default=False,
         description=(
             "When True, the parameter must be mapped to a CSV column on "
-            "every upload. The ``conversation_id`` parameter is always "
-            "required and is force-set to True by the server."
+            "every upload. The ``conversation_id`` and ``recording_url`` "
+            "parameters are always required and are force-set to True by "
+            "the server."
         ),
     )
 
@@ -2279,6 +2281,11 @@ def _validate_schema_parameters(
             "Schema must contain exactly one parameter of type "
             "'conversation_id'."
         )
+    if rec_url_count != 1:
+        raise ValueError(
+            "Schema must contain exactly one parameter of type "
+            "'recording_url'."
+        )
     if recording_date_count > 1:
         raise ValueError(
             "Schema may contain at most one parameter of type "

app/services/storage/gcs_service.py

@@ -69,6 +69,16 @@ def _resolve_credentials_path() -> Optional[str]:
     return str(resolved) if resolved.exists() else str(path)
 
 
+_GCS_SIGNING_UNAVAILABLE_MSG = (
+    "GCS signed URLs require signing credentials. Provide a service account JSON "
+    "(gcs.credentials_path or GOOGLE_APPLICATION_CREDENTIALS), or configure "
+    "GKE Workload Identity with the IAM Credentials API enabled and grant "
+    "roles/iam.serviceAccountTokenCreator to the workload service account on itself. "
+    "Optional: set gcs.signing_service_account_email when the service account email "
+    "is not detected from ADC."
+)
+
+
 class GcsService:
     """Service for managing GCS file storage."""
 
@@ -129,6 +139,35 @@ def _get_signing_credentials(self):
 
         return None
 
+    def _get_iam_signing_params(self) -> Optional[Tuple[str, str]]:
+        """Return (service_account_email, access_token) for IAM signBlob URL signing."""
+        if self.gcs_client is None:
+            return None
+
+        creds = getattr(self.gcs_client, "_credentials", None)
+        if creds is None:
+            return None
+
+        sa_email = settings.GCS_SIGNING_SERVICE_ACCOUNT_EMAIL or getattr(
+            creds, "service_account_email", None
+        ) or getattr(creds, "signer_email", None)
+        if not sa_email:
+            return None
+
+        try:
+            from google.auth.transport import requests as auth_requests
+
+            auth_request = auth_requests.Request()
+            if not creds.valid:
+                creds.refresh(auth_request)
+            token = creds.token
+            if token:
+                return sa_email, token
+        except Exception:
+            return None
+
+        return None
+
     def _ensure_initialized(self):
         """Lazily initialize GCS client if not already initialized."""
         if self.gcs_client is not None and self.bucket is not None:
@@ -524,20 +563,28 @@ def generate_presigned_url_by_key(self, key: str, expiration: int = 3600) -> str
         _, NotFound, GoogleCloudError = _get_gcs_exception_types()
 
         credentials = self._get_signing_credentials()
-        if credentials is None:
-            raise StorageError(
-                "GCS signed URLs require a service account JSON with a private key. "
-                "Set gcs.credentials_path or GOOGLE_APPLICATION_CREDENTIALS."
-            )
+        iam_params = None if credentials is not None else self._get_iam_signing_params()
+        if credentials is None and iam_params is None:
+            raise StorageError(_GCS_SIGNING_UNAVAILABLE_MSG)
 
         try:
             blob = self.bucket.blob(key)
-            url = blob.generate_signed_url(
-                version="v4",
-                expiration=timedelta(seconds=expiration),
-                method="GET",
-                credentials=credentials,
-            )
+            if credentials is not None:
+                url = blob.generate_signed_url(
+                    version="v4",
+                    expiration=timedelta(seconds=expiration),
+                    method="GET",
+                    credentials=credentials,
+                )
+            else:
+                sa_email, access_token = iam_params
+                url = blob.generate_signed_url(
+                    version="v4",
+                    expiration=timedelta(seconds=expiration),
+                    method="GET",
+                    service_account_email=sa_email,
+                    access_token=access_token,
+                )
             return url
         except GoogleCloudError as e:
             raise StorageError(f"Failed to generate signed URL: {str(e)}")