diff --git a/src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl b/src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl index 8d8a51d2..6bb6fdf5 100644 --- a/src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl +++ b/src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl @@ -6,7 +6,7 @@ _summaryEnvVars: &summaryEnvVars AWS_S3_ACCESS_KEY_ID: meet AWS_S3_SECRET_ACCESS_KEY: password AWS_S3_SECURE_ACCESS: False - AUTHORIZED_TENANTS: '[{"id": "dictaphone", "api_key": "dictaphone_token", "webhook_url": "http://dictaphone-backend.dictaphone.svc.cluster.local/api/v1.0/files/ai-webhook/", "webhook_api_key": "token_summary"}]' + AUTHORIZED_TENANTS: '[{"id": "dictaphone", "api_key": "dictaphone_token", "webhook_url": "http://dictaphone-backend.dictaphone.svc.cluster.local/api/v1.0/ai-jobs/webhook/", "webhook_api_key": "token_summary"}]' WHISPERX_API_KEY: secretKeyRef: name: secret-dev diff --git a/src/summary/summary/api/route/tasks_v2.py b/src/summary/summary/api/route/tasks_v2.py index f53b5e9a..e8f8cc78 100644 --- a/src/summary/summary/api/route/tasks_v2.py +++ b/src/summary/summary/api/route/tasks_v2.py @@ -1,15 +1,24 @@ """API routes related to application tasks (V2 / tenant friendly).""" from celery.result import AsyncResult -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Request from summary.core.celery_worker import ( + celery, process_audio_transcribe_v2_task, summarize_v2_task, ) from summary.core.config import AuthorizedTenant from summary.core.models import SummarizeTaskV2Request, TranscribeTaskV2Request from summary.core.security import verify_tenant_api_key_v2 +from summary.core.shared_models import ( + SummarizeWebhookFailurePayload, + SummarizeWebhookPendingPayload, + SummarizeWebhookSuccessPayload, + TranscribeWebhookFailurePayload, + TranscribeWebhookPendingPayload, + TranscribeWebhookSuccessPayload, +) router_tasks_v2 = APIRouter() @@ -24,7 +33,7 @@ async def create_transcribe_task_v2( args=[{**request.model_dump(), "tenant_id": request_tenant.id}] ) - return {"job_id": task.id, "message": "Transcribe job created"} + return TranscribeWebhookPendingPayload(job_id=task.id).model_dump() @router_tasks_v2.post("/async-jobs/summarize") @@ -37,25 +46,64 @@ async def create_summarize_task_v2( args=[{**request.model_dump(), "tenant_id": request_tenant.id}] ) - return {"job_id": task.id, "message": "Summarize job created"} + return SummarizeWebhookPendingPayload(job_id=task.id).model_dump() @router_tasks_v2.get("/async-jobs/transcribe/{job_id}") -@router_tasks_v2.get("/async-jobs/summarize/{job_id}") -async def get_task_status( +async def get_transcribe_job_status( job_id: str, + request: Request, request_tenant: AuthorizedTenant = Depends(verify_tenant_api_key_v2), ): - """Check task status by ID.""" - task = AsyncResult(job_id) - try: - if ( - isinstance(task.args, (list, tuple)) - and len(task.args) > 0 - and isinstance(task.args[0], dict) - and task.args[0].get("tenant_id") == request_tenant.id - ): - return {"job_id": job_id, "status": task.status} - except (TypeError, KeyError): - pass - raise HTTPException(status_code=404, detail="Not found") + """Check transcription task status by ID.""" + # We have to look directly in Redis to check if the task exists + redis_client = celery.backend.client + key = f"celery-task-meta-{job_id}" + if not redis_client.exists(key): + raise HTTPException(status_code=404, detail="Not found") + + task = AsyncResult(job_id, app=celery) + task_tenant_id = task.args[0]["tenant_id"] + if task_tenant_id != request_tenant.id: + raise HTTPException(status_code=403, detail="Forbidden") + + if task.status == "SUCCESS": + result = task.result + return TranscribeWebhookSuccessPayload.model_validate(result).model_dump() + + if task.status == "FAILURE": + return TranscribeWebhookFailurePayload( + job_id=job_id, error_code="unknown_error" + ).model_dump() + + return TranscribeWebhookPendingPayload(job_id=job_id).model_dump() + + +@router_tasks_v2.get("/async-jobs/summarize/{job_id}") +async def get_summarize_job_status( + job_id: str, + request: Request, + request_tenant: AuthorizedTenant = Depends(verify_tenant_api_key_v2), +): + """Check summarize task status by ID.""" + # We have to look directly in Redis to check if the task exists + redis_client = celery.backend.client + key = f"celery-task-meta-{job_id}" + if not redis_client.exists(key): + raise HTTPException(status_code=404, detail="Not found") + + task = AsyncResult(job_id, app=celery) + task_tenant_id = task.args[0]["tenant_id"] + if task_tenant_id != request_tenant.id: + raise HTTPException(status_code=403, detail="Forbidden") + + if task.status == "SUCCESS": + result = task.result + return SummarizeWebhookSuccessPayload.model_validate(result).model_dump() + + if task.status == "FAILURE": + return SummarizeWebhookFailurePayload( + job_id=job_id, error_code="unknown_error" + ).model_dump() + + return SummarizeWebhookPendingPayload(job_id=job_id).model_dump() diff --git a/src/summary/summary/core/celery_worker.py b/src/summary/summary/core/celery_worker.py index 7a88aae3..7c0b8e4d 100644 --- a/src/summary/summary/core/celery_worker.py +++ b/src/summary/summary/core/celery_worker.py @@ -58,6 +58,9 @@ celery = Celery( broker=settings.celery_broker_url, backend=settings.celery_result_backend, broker_connection_retry_on_startup=True, + # To store the tasks args too in results and make the + # V2 API work + result_extended=True, ) celery.config_from_object("summary.core.celery_config") @@ -465,15 +468,14 @@ def process_audio_transcribe_v2_task( job_id=job_id, ) - call_webhook_v2_task.apply_async( - args=[ - TranscribeWebhookSuccessPayload( - job_id=job_id, - transcription_data_url=file_service.get_transcript_signed_url(job_id), - ).model_dump(), - payload.tenant_id, - ] + success_payload = TranscribeWebhookSuccessPayload( + job_id=job_id, + transcription_data_url=file_service.get_transcript_signed_url(job_id), ) + call_webhook_v2_task.apply_async( + args=[success_payload.model_dump(), payload.tenant_id] + ) + return success_payload.model_dump() @signals.task_failure.connect(sender=process_audio_transcribe_v2_task) @@ -531,14 +533,17 @@ def summarize_v2_task( transcript=payload.content, session_id=self.request.id, ) - call_webhook_v2_task.apply_async( - args=[ - SummarizeWebhookSuccessPayload( - job_id=self.request.id, summary=summary - ).model_dump(), - payload.tenant_id, - ] + job_id = self.request.id + file_service.store_summary(summary=summary, job_id=job_id) + + success_payload = SummarizeWebhookSuccessPayload( + job_id=job_id, + summary_data_url=file_service.get_summary_signed_url(job_id), ) + call_webhook_v2_task.apply_async( + args=[success_payload.model_dump(), payload.tenant_id] + ) + return success_payload.model_dump() @signals.task_failure.connect(sender=summarize_v2_task) diff --git a/src/summary/summary/core/config.py b/src/summary/summary/core/config.py index 5b0096af..74753ab3 100644 --- a/src/summary/summary/core/config.py +++ b/src/summary/summary/core/config.py @@ -78,6 +78,7 @@ class Settings(BaseSettings): aws_s3_secret_access_key: SecretStr aws_s3_secure_access: bool = True aws_transcript_path: str = "transcripts" + aws_summary_path: str = "summaries" # AI-related settings whisperx_api_key: SecretStr diff --git a/src/summary/summary/core/file_service.py b/src/summary/summary/core/file_service.py index 2f945c71..45636fe0 100644 --- a/src/summary/summary/core/file_service.py +++ b/src/summary/summary/core/file_service.py @@ -293,5 +293,27 @@ class FileService: transcript_path = f"{settings.aws_transcript_path}/{job_id}.json" logger.debug("Transcript path: %s", transcript_path) return self._minio_client.presigned_get_object( - self._bucket_name, transcript_path, expires=timedelta(hours=1) + self._bucket_name, transcript_path, expires=timedelta(hours=24) + ) + + def store_summary(self, *, summary: str, job_id: str) -> None: + """Store summary in MinIO.""" + logger.info("Storing summary for job id %s", job_id) + summary_path = f"{settings.aws_summary_path}/{job_id}.txt" + logger.debug("Summary path: %s", summary_path) + data = summary.encode() + self._minio_client.put_object( + self._bucket_name, + summary_path, + io.BytesIO(data), + length=len(data), + ) + logger.info("Summary stored successfully for job id %s", job_id) + + def get_summary_signed_url(self, job_id: str) -> str: + """Get signed URL for summary file.""" + summary_path = f"{settings.aws_summary_path}/{job_id}.txt" + logger.debug("Summary path: %s", summary_path) + return self._minio_client.presigned_get_object( + self._bucket_name, summary_path, expires=timedelta(hours=24) ) diff --git a/src/summary/summary/core/shared_models.py b/src/summary/summary/core/shared_models.py index f213da2c..340a9959 100644 --- a/src/summary/summary/core/shared_models.py +++ b/src/summary/summary/core/shared_models.py @@ -65,6 +65,13 @@ class TranscribeWebhookSuccessPayload(BaseWebhook): ) +class TranscribeWebhookPendingPayload(BaseWebhook): + """Payload for a pending transcription webhook-like response.""" + + type: Literal["transcript"] = Field(default="transcript") + status: Literal["pending"] = Field(default="pending") + + class TranscribeWebhookFailurePayload(BaseWebhook): """Payload for a failed transcription webhook.""" @@ -76,7 +83,11 @@ class TranscribeWebhookFailurePayload(BaseWebhook): TranscribeWebhookPayloads = Annotated[ - Union[TranscribeWebhookSuccessPayload, TranscribeWebhookFailurePayload], + Union[ + TranscribeWebhookSuccessPayload, + TranscribeWebhookPendingPayload, + TranscribeWebhookFailurePayload, + ], Field(discriminator="status"), ] @@ -86,7 +97,16 @@ class SummarizeWebhookSuccessPayload(BaseWebhook): type: Literal["summary"] = Field(default="summary") status: Literal["success"] = Field(default="success") - summary: str = Field(title="Summary", description="The summary of the text.") + summary_data_url: str = Field( + title="Summary", description="URL to the raw summary data." + ) + + +class SummarizeWebhookPendingPayload(BaseWebhook): + """Payload for a pending summarization webhook-like response.""" + + type: Literal["summary"] = Field(default="summary") + status: Literal["pending"] = Field(default="pending") class SummarizeWebhookFailurePayload(BaseWebhook): @@ -100,7 +120,11 @@ class SummarizeWebhookFailurePayload(BaseWebhook): SummarizeWebhookPayloads = Annotated[ - Union[SummarizeWebhookSuccessPayload, SummarizeWebhookFailurePayload], + Union[ + SummarizeWebhookSuccessPayload, + SummarizeWebhookPendingPayload, + SummarizeWebhookFailurePayload, + ], Field(discriminator="status"), ] @@ -114,8 +138,10 @@ webhook_payload_adapter = TypeAdapter(WebhookPayloads) __all__ = [ "TranscribeWebhookSuccessPayload", + "TranscribeWebhookPendingPayload", "TranscribeWebhookFailurePayload", "SummarizeWebhookSuccessPayload", + "SummarizeWebhookPendingPayload", "SummarizeWebhookFailurePayload", "TranscribeWebhookPayloads", "SummarizeWebhookPayloads", diff --git a/src/summary/tests/api/test_api_tasks_v2.py b/src/summary/tests/api/test_api_tasks_v2.py index ffde24ad..0fac599e 100644 --- a/src/summary/tests/api/test_api_tasks_v2.py +++ b/src/summary/tests/api/test_api_tasks_v2.py @@ -1,6 +1,6 @@ """Integration tests for the V2 task API endpoints.""" -from unittest.mock import MagicMock, patch +from unittest.mock import ANY, MagicMock, patch class TestTasksV2: @@ -26,7 +26,8 @@ class TestTasksV2: assert response.status_code == 200 assert response.json() == { "job_id": "transcribe-task-id-abc", - "message": "Transcribe job created", + "type": "transcript", + "status": "pending", } args = mock_apply_async.call_args.kwargs["args"] @@ -58,7 +59,8 @@ class TestTasksV2: assert response.status_code == 200 assert response.json() == { "job_id": "summarize-task-id-abc", - "message": "Summarize job created", + "type": "summary", + "status": "pending", } args = mock_apply_async.call_args.kwargs["args"] @@ -70,11 +72,13 @@ class TestTasksV2: } ] + @patch("summary.api.route.tasks_v2.celery") @patch("summary.api.route.tasks_v2.AsyncResult") def test_get_transcribe_task_status_returns_status_for_same_tenant( - self, mock_async_result, client + self, mock_async_result, mock_celery, client ): """GET /async-jobs/transcribe/{id} returns status when tenant matches.""" + mock_celery.backend.client.exists.return_value = True mock_async_result.return_value = MagicMock( status="PENDING", args=[{"tenant_id": "test-tenant"}], @@ -86,18 +90,28 @@ class TestTasksV2: ) assert response.status_code == 200 - assert response.json() == {"job_id": "task-id-abc", "status": "PENDING"} + assert response.json() == { + "job_id": "task-id-abc", + "type": "transcript", + "status": "pending", + } - mock_async_result.assert_called_once_with("task-id-abc") + mock_async_result.assert_called_once_with("task-id-abc", app=ANY) + @patch("summary.api.route.tasks_v2.celery") @patch("summary.api.route.tasks_v2.AsyncResult") def test_get_summarize_task_status_returns_status_for_same_tenant( - self, mock_async_result, client + self, mock_async_result, mock_celery, client ): """GET /async-jobs/summarize/{id} returns status when tenant matches.""" + mock_celery.backend.client.exists.return_value = True mock_async_result.return_value = MagicMock( status="SUCCESS", args=[{"tenant_id": "test-tenant"}], + result={ + "job_id": "task-id-abc", + "summary_data_url": "https://example.com/summary.json", + }, ) response = client.get( @@ -106,16 +120,22 @@ class TestTasksV2: ) assert response.status_code == 200 - assert response.json() == {"job_id": "task-id-abc", "status": "SUCCESS"} + assert response.json() == { + "job_id": "task-id-abc", + "type": "summary", + "status": "success", + "summary_data_url": "https://example.com/summary.json", + } - mock_async_result.assert_called_once_with("task-id-abc") + mock_async_result.assert_called_once_with("task-id-abc", app=ANY) + @patch("summary.api.route.tasks_v2.celery") @patch("summary.api.route.tasks_v2.AsyncResult") - def test_get_task_status_returns_404_when_args_are_missing( - self, mock_async_result, client + def test_get_task_status_returns_404_when_job_does_not_exist( + self, mock_async_result, mock_celery, client ): - """GET /async-jobs/.../{id} returns 404 when task args are invalid.""" - mock_async_result.return_value = MagicMock(status="SUCCESS", args=None) + """GET /async-jobs/.../{id} returns 404 when task key is not in Redis.""" + mock_celery.backend.client.exists.return_value = False response = client.get( "/api/v2/async-jobs/transcribe/task-id-abc", @@ -124,3 +144,24 @@ class TestTasksV2: assert response.status_code == 404 assert response.json() == {"detail": "Not found"} + mock_async_result.assert_not_called() + + @patch("summary.api.route.tasks_v2.celery") + @patch("summary.api.route.tasks_v2.AsyncResult") + def test_get_task_status_returns_403_for_other_tenant( + self, mock_async_result, mock_celery, client + ): + """GET /async-jobs/.../{id} returns 403 when task belongs to another tenant.""" + mock_celery.backend.client.exists.return_value = True + mock_async_result.return_value = MagicMock( + status="PENDING", + args=[{"tenant_id": "another-tenant"}], + ) + + response = client.get( + "/api/v2/async-jobs/summarize/task-id-abc", + headers={"Authorization": "Bearer test-api-token"}, + ) + + assert response.status_code == 403 + assert response.json() == {"detail": "Forbidden"}