audio upload extension with gdrive credentials

backends/advanced/pyproject.toml

@@ -25,6 +25,9 @@ dependencies = [
     "redis>=5.0.0",
     "rq>=1.16.0",
     "soundfile>=0.12.1",
+    "google-api-python-client>=2.0.0",
+    "google-auth-oauthlib>=1.0.0",
+    "google-auth-httplib2>=0.2.0",
     "websockets>=12.0",
 ]
 

backends/advanced/src/advanced_omi_backend/app_config.py

@@ -85,6 +85,9 @@ def __init__(self):
         # Memory service configuration
         self.memory_service_supports_threshold = self.memory_provider == "friend_lite"
 
+        self.gdrive_credentials_path = "data/gdrive_service_account.json"
+        self.gdrive_scopes = ["https://www.googleapis.com/auth/drive.readonly"]
+
 
 # Global configuration instance
 app_config = AppConfig()

backends/advanced/src/advanced_omi_backend/clients/gdrive_audio_client.py

@@ -0,0 +1,29 @@
+import os 
+from google.oauth2.service_account import Credentials
+from googleapiclient.discovery import build
+from advanced_omi_backend.app_config import get_app_config
+
+_drive_client_cache = None
+
+def get_google_drive_client():
+    """Singleton Google Drive client."""
+    global _drive_client_cache
+
+    if _drive_client_cache:
+        return _drive_client_cache
+
+    config = get_app_config()
+
+    if not os.path.exists(config.gdrive_credentials_path):
+        raise FileNotFoundError(
+            f"Missing Google Drive credentials at {config.gdrive_credentials_path}"
+        )
+
+    creds = Credentials.from_service_account_file(
+        config.gdrive_credentials_path,
+        scopes=config.gdrive_scopes
+    )
+
+    _drive_client_cache = build("drive", "v3", credentials=creds)
+
+    return _drive_client_cache

backends/advanced/src/advanced_omi_backend/controllers/audio_controller.py

@@ -81,6 +81,8 @@ async def upload_and_process_audio_files(
                 # Read file content
                 content = await file.read()
 
+                gdrive_file_id = getattr(file, "gdrive_file_id", None)
+
                 # Generate audio UUID and timestamp
                 audio_uuid = str(uuid.uuid4())
                 timestamp = int(time.time() * 1000)
@@ -103,7 +105,8 @@ async def upload_and_process_audio_files(
                         user_email=user.email,
                         timestamp=timestamp,
                         chunk_dir=chunk_dir,
-                        validate=True  # Validate WAV format, convert stereo→mono
+                        validate=True,  # Validate WAV format, convert stereo→mono
+                        gdrive_file_id=gdrive_file_id
                     )
                 except AudioValidationError as e:
                     processed_files.append({
@@ -133,7 +136,7 @@ async def upload_and_process_audio_files(
                 # Use the relative path returned by write_audio_file (already includes folder prefix if applicable)
                 conversation.audio_path = relative_audio_path
                 await conversation.insert()
-                conversation_id = conversation.conversation_id  # Get the auto-generated ID
+                conversation_id = conversation.conversation_id  # Get the auto-gener    ated ID
 
                 audio_logger.info(f"📝 Created conversation {conversation_id} for uploaded file")
 

backends/advanced/src/advanced_omi_backend/models/audio_file.py

@@ -51,11 +51,15 @@ class AudioFile(Document):
         description="Speech detection results"
     )
 
+    gdrive_file_id: Optional[str] = Field(default=None, description="Google Drive file id")
+
+
     class Settings:
         name = "audio_chunks"
         indexes = [
             "audio_uuid",
             "client_id",
             "user_id",
-            "timestamp"
+            "timestamp", 
+            "gdrive_file_id"
         ]

backends/advanced/src/advanced_omi_backend/routers/modules/audio_routes.py

@@ -11,10 +11,29 @@
 from advanced_omi_backend.auth import current_superuser, current_active_user_optional, get_user_from_token_param
 from advanced_omi_backend.controllers import audio_controller
 from advanced_omi_backend.models.user import User
+from advanced_omi_backend.app_config import get_audio_chunk_dir
+from advanced_omi_backend.utils.gdrive_audio_utils import download_audio_files_from_drive, AudioValidationError
 
 router = APIRouter(prefix="/audio", tags=["audio"])
 
 
+@router.post("/upload_audio_from_gdrive")
+async def upload_audio_from_drive_folder(
+    gdrive_folder_id: str = Query(..., description="Google Drive Folder ID containing audio files (e.g., the string after /folders/ in the URL)"),
+    current_user: User = Depends(current_superuser),
+    device_name: str = Query(default="upload"),
+    auto_generate_client: bool = Query(default=True),
+):
+    try: 
+        files = await download_audio_files_from_drive(gdrive_folder_id)
+    except AudioValidationError as e: 
+        raise HTTPException(status_code=400, detail=str(e))
+
+    return await audio_controller.upload_and_process_audio_files(
+        current_user, files, device_name, auto_generate_client
+    )
+
+
 @router.get("/get_audio/{conversation_id}")
 async def get_conversation_audio(
     conversation_id: str,

backends/advanced/src/advanced_omi_backend/utils/audio_utils.py

@@ -107,7 +107,8 @@ async def write_audio_file(
     user_email: str,
     timestamp: int,
     chunk_dir: Optional[Path] = None,
-    validate: bool = True
+    validate: bool = True,
+    gdrive_file_id: Optional[str] = None,
 ) -> tuple[str, str, float]:
     """
     Validate, write audio data to WAV file, and create AudioSession database entry.
@@ -203,7 +204,8 @@ async def write_audio_file(
         user_id=user_id,
         user_email=user_email,
         has_speech=False,  # Will be updated by transcription
-        speech_analysis={}
+        speech_analysis={}, 
+        gdrive_file_id=gdrive_file_id
     )
     await audio_file.insert()
 

backends/advanced/src/advanced_omi_backend/utils/gdrive_audio_utils.py

@@ -0,0 +1,122 @@
+import io
+import tempfile
+from typing import List
+import logging
+from starlette.datastructures import UploadFile as StarletteUploadFile
+from googleapiclient.http import MediaIoBaseDownload
+from advanced_omi_backend.clients.gdrive_audio_client import get_google_drive_client
+from advanced_omi_backend.models.audio_file import AudioFile
+from advanced_omi_backend.utils.audio_utils import AudioValidationError
+
+
+logger = logging.getLogger(__name__)
+audio_logger = logging.getLogger("audio_processing")
+
+AUDIO_EXTENSIONS = (".wav", ".mp3", ".flac", ".ogg", ".m4a")
+FOLDER_MIMETYPE = "application/vnd.google-apps.folder"
+
+
+
+async def download_and_wrap_drive_file(service, file_item):
+    file_id = file_item["id"]
+    name = file_item["name"]
+
+    request = service.files().get_media(fileId=file_id)
+
+    fh = io.BytesIO()
+    downloader = MediaIoBaseDownload(fh, request)
+
+    done = False
+    while not done:
+        _status, done = downloader.next_chunk()
+
+    content = fh.getvalue()
+
+    if not content:
+        raise AudioValidationError(f"Downloaded Google Drive file '{name}' was empty")
+
+    tmp_file = tempfile.SpooledTemporaryFile(max_size=10*1024*1024)  # 10 MB
+    tmp_file.write(content)
+    tmp_file.seek(0)
+    upload_file = StarletteUploadFile(filename=name, file=tmp_file)
+
+    original_close = upload_file.close
+
+    def wrapped_close():
+        try:
+            original_close()
+        finally:
+            # SpooledTemporaryFile auto-cleans when closed; no unlink needed
+            pass
+
+    upload_file.close = wrapped_close
+
+    return upload_file
+
+# -------------------------------------------------------------
+# LIST + DOWNLOAD FILES IN FOLDER (OAUTH)
+# -------------------------------------------------------------
+async def download_audio_files_from_drive(folder_id: str) -> List[StarletteUploadFile]:
+    if not folder_id:
+        raise AudioValidationError("Google Drive folder ID is required.")
+
+    service = get_google_drive_client()
+
+    try:
+        escaped_folder_id = folder_id.replace("\\", "\\\\").replace("'", "\\'")
+        query = f"'{escaped_folder_id}' in parents and trashed = false"
+
+        response = service.files().list(
+            q=query,
+            fields="files(id, name, mimeType)",
+            includeItemsFromAllDrives=False,
+            supportsAllDrives=False,
+        ).execute()
+
+        all_files = response.get("files", [])
+
+        audio_files_metadata = [
+            f for f in all_files
+            if f["name"].lower().endswith(AUDIO_EXTENSIONS)
+        ]
+
+        if not audio_files_metadata:
+            raise AudioValidationError("No audio files found in folder.")
+
+        wrapped_files = []
+        skipped_count = 0
+
+        for item in audio_files_metadata:
+            file_id = item["id"] # Get the Google Drive File ID
+
+            #  Check if the file is already processed
+            if await is_drive_file_already_processed(file_id):
+                audio_logger.info(f"Skipping already processed file: {item['name']}") # Use your logger
+                skipped_count += 1
+                continue
+
+            # synchronous call now (but make the parent function async)
+            wrapped_file = await download_and_wrap_drive_file(service, item)
+            #  Attach the file_id to the UploadFile object for later use
+            wrapped_file.gdrive_file_id = file_id
+            wrapped_files.append(wrapped_file)
+
+        if not wrapped_files and skipped_count > 0:
+            raise AudioValidationError(f"All {skipped_count} files in the folder have already been processed.")
+
+        return wrapped_files
+
+    except Exception as e:
+        if isinstance(e, AudioValidationError):
+            raise
+        raise AudioValidationError(f"Google Drive API Error: {e}") from e
+
+
+async def is_drive_file_already_processed(gdrive_file_id: str) -> bool:
+    """Check if an AudioFile document already exists for the given GDrive File ID."""
+    if not gdrive_file_id:
+        return False
+    existing_file = await AudioFile.find_one(
+        AudioFile.gdrive_file_id == gdrive_file_id
+    )
+    return existing_file is not None