File size: 1,920 Bytes
e562a59 |
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 |
import hashlib
from enum import Enum
import boto3
from botocore.client import BaseClient
# S3 HANDLING ######################################################################################
def get_md5(fpath):
with open(fpath, "rb") as f:
file_hash = hashlib.md5()
while chunk := f.read(8192):
file_hash.update(chunk)
return file_hash.hexdigest()
def upload_file_to_bucket(s3_client, file_obj, bucket, s3key):
"""Upload a file to an S3 bucket
:param file_obj: File to upload
:param bucket: Bucket to upload to
:param s3key: s3key
:param object_name: S3 object name. If not specified then file_name is used
:return: True if file was uploaded, else False
"""
# Upload the file
return s3_client.upload_fileobj(
file_obj, bucket, s3key,
ExtraArgs={"ACL": "public-read", "ContentType": "Content-Type: audio/mpeg"}
)
def s3_auth(aws_access_key_id, aws_secret_access_key, region_name) -> BaseClient:
s3 = boto3.client(
service_name='s3',
aws_access_key_id=aws_access_key_id,
aws_secret_access_key=aws_secret_access_key,
region_name=region_name
)
return s3
def get_list_of_buckets(s3: BaseClient):
response = s3.list_buckets()
buckets = {}
for buckets in response['Buckets']:
buckets[response['Name']] = response['Name']
BucketName = Enum('BucketName', buckets)
return BucketName
if __name__ == '__main__':
import os
AWS_ACCESS_KEY_ID = os.environ['AWS_ACCESS_KEY_ID']
AWS_SECRET_ACCESS_KEY = os.environ['AWS_SECRET_ACCESS_KEY']
S3_BUCKET = "synthia-research"
S3_FOLDER = "huggingface_spaces_demo"
AWS_REGION = "eu-west-3"
s3 = s3_auth(AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION)
print(s3.list_buckets())
s3key = f'{S3_FOLDER}/015.WAV'
#print(upload_file_to_bucket(s3, file_obj, S3_BUCKET, s3key))
|