Spaces:
Running
Running
| # from googleapiclient.discovery import build | |
| # from googleapiclient.http import MediaFileUpload, MediaIoBaseDownload | |
| # from google.oauth2 import service_account | |
| # import os | |
| # import datetime | |
| # import io | |
| # import json | |
| # SCOPES = ['https://www.googleapis.com/auth/drive.file'] | |
| # class GoogleDriveAudioManager: | |
| # def __init__(self): | |
| # # Load service account JSON from environment variable | |
| # credentials_dict = json.loads(os.environ['GOOGLE_CREDS_JSON']) | |
| # self.creds = service_account.Credentials.from_service_account_info( | |
| # credentials_dict, scopes=SCOPES | |
| # ) | |
| # self.service = build('drive', 'v3', credentials=self.creds) | |
| # def upload_audio(self, file_path: str = "output.wav", filename: str = "output.wav"): | |
| # file_metadata = { | |
| # 'name': filename, | |
| # 'description': f"Uploaded on {datetime.datetime.utcnow()}", | |
| # 'mimeType': 'audio/wav' | |
| # } | |
| # media = MediaFileUpload(file_path, mimetype='audio/wav') | |
| # file = self.service.files().create( | |
| # body=file_metadata, media_body=media, fields='id, name').execute() | |
| # print(f"✅ Uploaded '{filename}' to Google Drive with file ID: {file.get('id')}") | |
| # return file.get('id') | |
| # def download_audio(self, file_id: str, save_path: str = "downloaded_output.wav"): | |
| # request = self.service.files().get_media(fileId=file_id) | |
| # fh = io.FileIO(save_path, 'wb') | |
| # downloader = MediaIoBaseDownload(fh, request) | |
| # done = False | |
| # while not done: | |
| # status, done = downloader.next_chunk() | |
| # print(f"⬇️ Download {int(status.progress() * 100)}% complete.") | |
| # fh.close() | |
| # print(f"✅ Audio downloaded to {save_path}") | |
| # manager = GoogleDriveAudioManager() | |
| # # Upload | |
| # file_id = manager.upload_audio("output.wav") | |
| # # Download | |
| # manager.download_audio(file_id, "downloaded_output.wav") | |
| from dotenv import load_dotenv | |
| from googleapiclient.discovery import build | |
| from googleapiclient.http import MediaFileUpload, MediaIoBaseDownload | |
| from google.oauth2 import service_account | |
| import os | |
| import datetime | |
| import io | |
| import json | |
| SCOPES = ['https://www.googleapis.com/auth/drive.file'] | |
| load_dotenv() | |
| class GoogleDriveAudioManager: | |
| def __init__(self): | |
| # Load service account JSON from environment variable | |
| credentials_dict = { | |
| "type": "service_account", | |
| "project_id": "winged-moon-458905-s2", | |
| "private_key_id": "0f63ac56d16ca67a7b0359128c6704b0d1ad346b", | |
| "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQDM59sgtCZ4swKo\naWyDu7Lpme64x9oc2sn8xDhloujhb3fLAry6T6YG8TaFir3VWwLXc2PRlWrf6SUm\nIvyKfIhUtfLJmWFRtEYnNUHeK+lJwCgShF+UhMqW73rUYYz26VyxA3we/K9GRTlx\ngzwufLtW95TS8k40sXxUQyMVc9KE3xeqVEkYGyDWz6jgPzgQwAq2/Z5lqigb1Utv\n2QG0pGaPFwqYQQTVoKpWlt1i2cRZYdQzGkGrRtF+18uvm64DMFu6ShkMnmVJ7Kk5\nF5O4Uzl6LC01UAY7A88siHa6l5zBmPZz2OwJ2ieEDRUQJ/YIQKjMHNgCGa3FYxlJ\nqU+s8EQjAgMBAAECggEAXyr+mOFjDuMhhGO2jTaTwbCHchYMZSNu9fGSug3geLfj\nXQwv6J6On6HPr1WBlsL2/mcvAMwTf9ySrvRH8/AkB6gcE+sjiqWegZ5YMd6WMvVH\nB1ao59uHFnQyycpnBNTdAq/6KZA1vf7vz5tiy7qN7jX+Tr1QemPGuYIXHOMW7zOz\nFuWdefWqGF9yKZWg9q0XeJu9nueJEC9zdIWBW+Bu72rcv9pRhrrgdENU1Pv1tCqt\n0Bp0GAhkuHhkw17IBy+oHvI83KNwfQPLJ//f+3McdZRYRAwVlfZiry3QYaaYZzIM\nYCJo36lK+/D2uHMsWEKxNzx7UUw/36lHQ0GoDV/FuQKBgQD452gvUhnuLM0FO3Ns\nOdMtLcakJP7Dsi5cs27GkObIs2plvcqrcqgyVQG1iQ4xZ/yOAYAyfHEg/06ZbC6A\n7+hC0u957dvBpSyVvAc5/m+r4Iv9U5XuwF9F5QgC+m083vIqhGOMrS5cqXfCS0Nj\nr016LpLAME9zKs+gtwMQOUkweQKBgQDSv1VhAkj4yI1l34+x8yRFmKEV4BdvzYAQ\nDdUKN6D7FZ6bCExlF+35qfkptxh8TAB4tEntdAGUyXzUwi73FteFf3pyuQI81LvE\nwIk6qNpSz06IIPhen0y41dsyF0fZLX8BpIYLypxFEDvSmo5IXa0fp2UOsR1vj4cY\ngTcfchtKewKBgCs0f/E23ylovW3bE/4FS5LcD08O3Oo7U7DjtKbjICxWeHGQsE1y\naZ/TAajxUXViB4Te3dHlp/2CU7JuBOR8wGRZGRihITRL56FWMiru75N2NtaSW8RF\nwEPxraJWtf+K855NvLn+lb2f7Rnc1qXHGDvNOcwClsf29nN/URjP2NSBAoGAHSj0\nO1QBnMrDJdFlJI4r+qU/xLMZCxM4IQCI9bz0vUkkhPcE+wfwmZb3LtpyvOAXQJxR\njlV60od+ikHEFdRHReV5n94RfTFCOdsaIc+zRboBd7x4IT+RchHFHo0yF86/wA5v\nrbn/HV89p8pS3Zh4l+Sliv7ev8zwOGfIpMU5GXkCgYAfqLNwTXI9vfUBJw0uhufH\n7zdMPGTf8Jkjpq4gIkXG/7xZYxqKtFG6QItbFHN5E4zqJAEn5zvBXaYpno1GplRs\n+aFlae+IJS7i9WJICHYTvskvMtoNqIgAAcLxlHG2fcKr5QtA/G2zdmUdkYmK8ESA\n/s+J3gnPPKo/wggV4TfIkw==\n-----END PRIVATE KEY-----\n", | |
| "client_email": "capstone-braille@winged-moon-458905-s2.iam.gserviceaccount.com", | |
| "client_id": "109756671550003136073", | |
| "auth_uri": "https://accounts.google.com/o/oauth2/auth", | |
| "token_uri": "https://oauth2.googleapis.com/token", | |
| "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs", | |
| "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/capstone-braille%40winged-moon-458905-s2.iam.gserviceaccount.com", | |
| "universe_domain": "googleapis.com" | |
| } | |
| # Use the service account directly instead of OAuth flow | |
| self.creds = service_account.Credentials.from_service_account_info( | |
| credentials_dict, scopes=SCOPES | |
| ) | |
| # Build the drive service | |
| self.service = build('drive', 'v3', credentials=self.creds) | |
| def upload_audio(self, file_path: str = "output.wav", filename: str = "output.wav"): | |
| file_metadata = { | |
| 'name': filename, | |
| 'description': f"Uploaded on {datetime.datetime.utcnow()}", | |
| 'mimeType': 'audio/wav' | |
| } | |
| media = MediaFileUpload(file_path, mimetype='audio/wav') | |
| file = self.service.files().create( | |
| body=file_metadata, media_body=media, fields='id, name').execute() | |
| print(f"✅ Uploaded '{filename}' to Google Drive with file ID: {file.get('id')}") | |
| return file.get('id') | |
| print("lol") | |
| def download_audio(self, file_id: str, save_path: str = "downloaded_output.wav"): | |
| request = self.service.files().get_media(fileId=file_id) | |
| fh = io.FileIO(save_path, 'wb') | |
| downloader = MediaIoBaseDownload(fh, request) | |
| done = False | |
| while not done: | |
| status, done = downloader.next_chunk() | |
| print(f"⬇️ Download {int(status.progress() * 100)}% complete.") | |
| fh.close() | |
| print(f"✅ Audio downloaded to {save_path}") | |