wip

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,

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."""

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

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."""