diff --git a/src/backend/core/recording/event/notification.py b/src/backend/core/recording/event/notification.py index 46bd1dc4..19ab31cd 100644 --- a/src/backend/core/recording/event/notification.py +++ b/src/backend/core/recording/event/notification.py @@ -58,6 +58,9 @@ class NotificationService: email_success = self._notify_user_by_email(recording) return email_success and summary_success + if True: # TODO: add condition (in recording or global setting) + self._notify_push_file_service(recording) + logger.error( "Unknown recording mode %s for recording %s", recording.mode, @@ -222,6 +225,31 @@ class NotificationService: f"Unknown summary service version: {settings.SUMMARY_SERVICE_VERSION}" ) + @staticmethod + def _notify_push_file_service(recording: models.Recording): + """Notify push to file storage service.""" + + payload = {""} + + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {settings.SUMMARY_SERVICE_API_TOKEN}", + } + + try: + response = requests.post( + "=http://app-summary-dev:8000/api/v2/async-jobs/push-recording/", # TODO, replace by: settings.PUSH_RECORDING_SERVICE_ENDPOINT, + json=payload, + headers=headers, + timeout=30, + ) + response.raise_for_status() + except requests.RequestException as exc: + logger.exception("Push failed" # TODO: better messagea + ) + return False + + @staticmethod def _notify_summary_service_v1(recording: models.Recording): """Notify summary service about a new recording.""" @@ -386,7 +414,7 @@ class NotificationService: try: response = requests.post( - settings.SUMMARY_SERVICE_ENDPOINT, + settings.SUMMARY_SERVICE_ENDPOINT, # process_audio_transcribe_v2_task json=payload, headers=headers, timeout=30, diff --git a/src/summary/summary/core/celery_worker.py b/src/summary/summary/core/celery_worker.py index fb65f71b..8f1c9da5 100644 --- a/src/summary/summary/core/celery_worker.py +++ b/src/summary/summary/core/celery_worker.py @@ -5,8 +5,9 @@ import json import time from datetime import datetime, timezone +from pathlib import Path from typing import Any -from urllib.parse import urljoin +from urllib.parse import urljoin, urlparse import requests import sentry_sdk @@ -22,6 +23,7 @@ from summary.core.file_service import ( CorruptedAudioFile, FileService, FileServiceException, + SizedStream, TranscribeError, ) from summary.core.llm_service import LLMException, LLMObservability, LLMService @@ -564,6 +566,104 @@ def process_audio_transcribe_v2_task( return success_payload.model_dump() + +@celery.task( + bind=True, + autoretry_for=[exceptions.HTTPError], + max_retries=settings.celery_max_retries, + queue=settings.push_recording, +) +def push_recording( + self, + payload: dict, +): + """ + Create a new file in the main workspace. + + The recording is streamed straight from object storage to Drive's presigned + URL: it is never fully downloaded to the worker's disk or memory. + + Mostly taken from: https://github.com/suitenumerique/drive/blob/main/docs/resource_server.md# + """ + + # Get the access token from the session + access_token = payload['oidc_access_token'] + cloud_storage_url = payload['cloud_storage_url'] + filename = Path(urlparse(cloud_storage_url).path).name + + # Get the main workspace + response = requests.get( + f"{settings.DRIVE_API}/items/", + headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}, + ) + response.raise_for_status() + data = response.json() + items = data['results'] + main_workspace = None + for item in items: + if item['main_workspace']: + main_workspace = item + break + + if not main_workspace: + raise Exception(status=404, data={"error": "No main workspace found"}) + + # Create a new file in the main workspace + response = requests.post( + f"{settings.DRIVE_API}/items/{main_workspace['id']}/children/", + json={ + "type": "file", + "filename": filename, + }, + headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}, + ) + response.raise_for_status() + item = response.json() + policy = item['policy'] + + # Upload file content using the presigned URL, streaming the recording + # straight from object storage: bytes are relayed chunk by chunk and the + # file is not fully held in memory. + with requests.get( + cloud_storage_url, + stream=True, + timeout=(10, 300), + ) as download_response: + download_response.raise_for_status() + + content_length = download_response.headers.get("Content-Length") + if content_length is None: + raise FileServiceException( + "Object storage did not return the recording size, " + "cannot stream it to Drive." + ) + + upload_response = requests.put( + policy, + data=SizedStream(download_response.raw, int(content_length)), + headers={ + "Content-Type": download_response.headers.get( + "Content-Type", "application/octet-stream" + ), + "Content-Length": content_length, + "x-amz-acl": "private", + }, + timeout=(10, 300), + ) + upload_response.raise_for_status() + + # Tell the Drive API that the upload is ended + # drive_api = settings.DRIVE_API + drive_api = "" + response = requests.post( + f"{drive_api}/items/{item['id']}/upload-ended/", + json={}, + headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}, + ) + response.raise_for_status() + + return data + @signals.task_prerun.connect(sender=process_audio_transcribe_v2_task) def task_started_transcript(task_id=None, task=None, args=None, **kwargs): """Signal handler called before task execution begins.""" diff --git a/src/summary/summary/core/config.py b/src/summary/summary/core/config.py index 4afb377e..e4cd27bd 100644 --- a/src/summary/summary/core/config.py +++ b/src/summary/summary/core/config.py @@ -72,10 +72,11 @@ class Settings(BaseSettings): celery_result_backend: str = "redis://redis/0" celery_max_retries: int = 1 - # v2 tasks + # tasks transcribe_queue_v2: str = "transcribe-queue-v2" summarize_queue_v2: str = "summarize-queue-v2" call_webhook_queue_v2: str = "call-webhook-queue-v2" + push_recording: str = "push-recording" # Minio settings aws_storage_bucket_name: str diff --git a/src/summary/summary/core/file_service.py b/src/summary/summary/core/file_service.py index 3c10f26d..c634b737 100644 --- a/src/summary/summary/core/file_service.py +++ b/src/summary/summary/core/file_service.py @@ -266,6 +266,33 @@ class FileServiceException(Exception): pass +class SizedStream: + """Read-only byte stream of a known size, suitable as a `requests` body. + + `requests` falls back to a chunked transfer encoding when it cannot guess + the body size upfront, which presigned S3 uploads reject. Advertising the + size through `__len__` makes it send a plain `Content-Length` instead, + while the underlying stream is still consumed chunk by chunk. + """ + + def __init__(self, stream, length: int): + """Wrap `stream`, whose full content is `length` bytes long.""" + self._stream = stream + self._length = length + + def __len__(self) -> int: + """Return the total size of the stream, in bytes.""" + return self._length + + def __iter__(self): + """Iterate over the stream, required for `requests` to stream the body.""" + return iter(self._stream) + + def read(self, amt=None) -> bytes: + """Read up to `amt` bytes from the stream.""" + return self._stream.read(amt) + + class FileService: """Service for downloading and preparing files from MinIO storage."""