UPDATE: code has been refactore for UUID for diffrent audio upload sources

Author: 01PrathamS

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

@@ -40,6 +40,7 @@ async def upload_and_process_audio_files(
     device_name: str = "upload",
     auto_generate_client: bool = True,
     folder: str = None,
+    source: str = "upload"
 ) -> dict:
     """
     Upload audio files and process them directly.
@@ -81,10 +82,12 @@ 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())
+                if source == "gdrive":
+                    audio_uuid = getattr(file, "audio_uuid", None)
+                else: 
+                    audio_uuid = str(uuid.uuid4())
                 timestamp = int(time.time() * 1000)
 
                 # Determine output directory (with optional subfolder)
@@ -100,13 +103,13 @@ async def upload_and_process_audio_files(
                     relative_audio_path, file_path, duration = await write_audio_file(
                         raw_audio_data=content,
                         audio_uuid=audio_uuid,
+                        source=source,
                         client_id=client_id,
                         user_id=user.user_id,
                         user_email=user.email,
                         timestamp=timestamp,
                         chunk_dir=chunk_dir,
                         validate=True,  # Validate WAV format, convert stereo→mono
-                        gdrive_file_id=gdrive_file_id
                     )
                 except AudioValidationError as e:
                     processed_files.append({

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

@@ -29,6 +29,10 @@ class AudioFile(Document):
 
     # Core identifiers
     audio_uuid: Indexed(str, unique=True) = Field(description="Unique audio identifier")
+    source: Indexed(str) = Field(
+        default="upload",
+        description="Source of the audio (upload, gdrive, etc.)"
+    )
     audio_path: str = Field(description="Path to raw audio file")
     client_id: Indexed(str) = Field(description="Client device identifier")
     timestamp: Indexed(int) = Field(description="Unix timestamp in milliseconds")
@@ -51,7 +55,6 @@ class AudioFile(Document):
         description="Speech detection results"
     )
 
-    gdrive_file_id: Optional[str] = Field(default=None, description="Google Drive file id")
 
 
     class Settings:
@@ -61,5 +64,4 @@ class Settings:
             "client_id",
             "user_id",
             "timestamp", 
-            "gdrive_file_id"
         ]

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

@@ -30,7 +30,7 @@ async def upload_audio_from_drive_folder(
         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
+        current_user, files, device_name, auto_generate_client, source="gdrive"
     )
 
 

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

@@ -102,13 +102,13 @@ async def validate_and_prepare_audio(
 async def write_audio_file(
     raw_audio_data: bytes,
     audio_uuid: str,
+    source: str,
     client_id: str,
     user_id: str,
     user_email: str,
     timestamp: int,
     chunk_dir: Optional[Path] = None,
     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.
@@ -198,14 +198,14 @@ async def write_audio_file(
     # Create AudioFile database entry using Beanie model
     audio_file = AudioFile(
         audio_uuid=audio_uuid,
+        source=source,
         audio_path=wav_filename,
         client_id=client_id,
         timestamp=timestamp,
         user_id=user_id,
         user_email=user_email,
         has_speech=False,  # Will be updated by transcription
         speech_analysis={}, 
-        gdrive_file_id=gdrive_file_id
     )
     await audio_file.insert()
 

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

@@ -0,0 +1,150 @@
+import io
+import tempfile
+from typing import List
+import logging
+from starlette.datastructures import UploadFile as StarletteUploadFile
+import requests
+from advanced_omi_backend.app_config import get_app_config
+
+logger = logging.getLogger(__name__)
+audio_logger = logging.getLogger("audio_processing")
+
+AUDIO_EXTENSIONS = (".wav", ".mp3", ".flac", ".ogg", ".m4a")
+
+class AudioValidationError(Exception):
+    pass
+
+
+async def download_and_wrap_dropbox_file(file_metadata: dict):
+    access_token = get_app_config().dropbox_access_token    
+    if not access_token:
+        raise AudioValidationError("Dropbox access token is missing.")
+
+    file_path = file_metadata["path_lower"]
+    name = file_metadata["name"]
+
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "Dropbox-API-Arg": f'{{"path": "{file_path}"}}'
+    }
+
+    response = requests.post(
+        "https://content.dropboxapi.com/2/files/download",
+        headers=headers
+    )
+
+    if response.status_code != 200:
+        raise AudioValidationError(
+            f"Failed to download Dropbox file {name}: {response.text}"
+        )
+
+    content = response.content
+    if not content:
+        raise AudioValidationError(f"Downloaded Dropbox file '{name}' was empty")
+
+    tmp_file = tempfile.SpooledTemporaryFile(max_size=10 * 1024 * 1024)
+    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:
+            pass
+
+    upload_file.close = wrapped_close
+    return upload_file
+
+
+async def download_audio_files_from_dropbox(folder_path: str) -> List[StarletteUploadFile]:
+    if not folder_path:
+        raise AudioValidationError("Dropbox folder path is required.")
+
+    access_token = get_app_config().dropbox_access_token
+    if not access_token:
+        raise AudioValidationError("Dropbox access token is missing.")
+
+    try:
+        # -----------------------------------------------
+        # Step 1: List files in folder
+        # -----------------------------------------------
+        list_headers = {
+            "Authorization": f"Bearer {access_token}",
+            "Content-Type": "application/json"
+        }
+
+        list_body = {
+            "path": folder_path,
+            "recursive": False
+        }
+
+        list_res = requests.post(
+            "https://api.dropboxapi.com/2/files/list_folder",
+            headers=list_headers,
+            json=list_body
+        )
+
+        if list_res.status_code != 200:
+            raise AudioValidationError(f"Dropbox API list_folder error: {list_res.text}")
+
+        items = list_res.json().get("entries", [])
+
+        # Filter audio files
+        audio_files_metadata = [
+            f for f in items
+            if f[".tag"] == "file" and f["name"].lower().endswith(AUDIO_EXTENSIONS)
+        ]
+
+        if not audio_files_metadata:
+            raise AudioValidationError("No audio files found in folder.")
+
+        wrapped_files = []
+        skipped_count = 0
+
+        # -----------------------------------------------
+        # Step 2: Download each audio file
+        # -----------------------------------------------
+        for item in audio_files_metadata:
+            dropbox_id = item["id"]  # Dropbox file ID
+
+            # Check if already processed
+            if await is_dropbox_file_already_processed(dropbox_id):
+                audio_logger.info(f"Skipping already processed file: {item['name']}")
+                skipped_count += 1
+                continue
+
+            wrapped_file = await download_and_wrap_dropbox_file(access_token, item)
+
+            # Attach Dropbox file ID
+            setattr(wrapped_file, "dropbox_file_id", dropbox_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"Dropbox API Error: {e}") from e
+
+
+async def is_dropbox_file_already_processed(dropbox_file_id: str) -> bool:
+    if not dropbox_file_id:
+        return False
+
+    from advanced_omi_backend.models.audio_file import AudioFile
+
+    existing_file = await AudioFile.find_one(
+        AudioFile.dropbox_file_id == dropbox_file_id
+    )
+
+    return existing_file is not None

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

@@ -90,15 +90,20 @@ async def download_audio_files_from_drive(folder_id: str) -> List[StarletteUploa
             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
+            existing = await AudioFile.find_one({
+                "audio_uuid": file_id,
+                "source": "gdrive"
+            })
+
+            if existing:
+                audio_logger.info(f"Skipping already processed file: {item['name']}")
                 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_file.audio_uuid = file_id
             wrapped_files.append(wrapped_file)
 
         if not wrapped_files and skipped_count > 0:
@@ -112,11 +117,3 @@ async def download_audio_files_from_drive(folder_id: str) -> List[StarletteUploa
         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