Repository navigation
audio upload extension with gdrive credentials #175
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 7 commits
44beeac
d5b9518
9392989
5b5ea64
5abd99d
3d00bac
b036185
cff1a4c
6534288
1ff28cb
8e5a6b2
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -11,10 +11,26 @@ | |
| 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 | ||
|
|
||
| router = APIRouter(prefix="/audio", tags=["audio"]) | ||
|
|
||
|
|
||
| @router.post("/upload_audio_from_url") | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this router name suggests I can use any url, but looks like it's just Gdrive? I'm not super familiar with how gdrive works though |
||
| async def upload_audio_from_drive_folder( | ||
| drive_folder_id: str = Query(...,alias="url", 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), | ||
| ): | ||
| files = await download_audio_files_from_drive(drive_folder_id) | ||
|
|
||
| return await audio_controller.upload_and_process_audio_files( | ||
| current_user, files, device_name, auto_generate_client | ||
| ) | ||
|
coderabbitai[bot] marked this conversation as resolved.
Outdated
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
|
|
||
| @router.get("/get_audio/{conversation_id}") | ||
| async def get_conversation_audio( | ||
| conversation_id: str, | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,123 @@ | ||
| 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.app_config import get_app_config | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
| audio_logger = logging.getLogger("audio_processing") | ||
|
|
||
| AUDIO_EXTENSIONS = (".wav", ".mp3", ".flac", ".ogg", ".m4a") | ||
| FOLDER_MIMETYPE = "application/vnd.google-apps.folder" | ||
|
|
||
|
|
||
| class AudioValidationError(Exception): | ||
| pass | ||
|
|
||
|
coderabbitai[bot] marked this conversation as resolved.
Outdated
|
||
|
|
||
| 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 | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
| # ------------------------------------------------------------- | ||
| # 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_app_config().get_gdrive_service() | ||
|
|
||
| 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 | ||
| setattr(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 | ||
| from advanced_omi_backend.models.audio_file import AudioFile | ||
| existing_file = await AudioFile.find_one( | ||
| AudioFile.gdrive_file_id == gdrive_file_id | ||
| ) | ||
| return existing_file is not None | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This probably shouldn't be in app_config. can have a gdrive service in services or have a google_client in clients. Then we initialise it at the point of use