Spaces:
Running
Running
File size: 6,346 Bytes
e8d4b95 2a0bdb6 e8d4b95 8350d4d e8d4b95 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 |
# 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}")
|