File size: 2,527 Bytes
534df99 | 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 | import boto3
from botocore.client import Config
import uuid
from datetime import datetime
import config
class StorageService:
def __init__(self):
self.bucket = config.DO_SPACES_BUCKET
self.region = config.DO_SPACES_REGION
# Endpoint for boto3 client (region-based)
self.endpoint_url = f"https://{self.region}.digitaloceanspaces.com"
# Public URL base (bucket-based subdomain)
self.public_url_base = f"https://{self.bucket}.{self.region}.digitaloceanspaces.com"
self.client = boto3.client(
's3',
region_name=self.region,
endpoint_url=self.endpoint_url,
aws_access_key_id=config.DO_SPACES_KEY,
aws_secret_access_key=config.DO_SPACES_SECRET,
config=Config(signature_version='s3v4')
)
def _generate_filename(self, user_id: str, category_id: str, extension: str) -> str:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
unique_id = str(uuid.uuid4())[:8]
return f"{user_id}_{category_id}_{timestamp}_{unique_id}.{extension}"
def _get_public_url(self, file_path: str) -> str:
"""Generate correct public URL for Digital Ocean Spaces"""
# Format: https://{bucket}.{region}.digitaloceanspaces.com/{path}
return f"{self.public_url_base}/{file_path}"
def upload_source_image(self, file_content: bytes, user_id: str, category_id: str, content_type: str) -> str:
extension = content_type.split('/')[-1]
if extension == 'jpeg':
extension = 'jpg'
filename = self._generate_filename(user_id, category_id, extension)
file_path = f"{config.SOURCE_PATH}/{filename}"
self.client.put_object(
Bucket=self.bucket,
Key=file_path,
Body=file_content,
ContentType=content_type,
ACL='public-read'
)
return self._get_public_url(file_path)
def upload_result_video(self, video_content: bytes, user_id: str, category_id: str) -> str:
filename = self._generate_filename(user_id, category_id, "mp4")
file_path = f"{config.RESULTS_PATH}/{filename}"
self.client.put_object(
Bucket=self.bucket,
Key=file_path,
Body=video_content,
ContentType='video/mp4',
ACL='public-read'
)
return self._get_public_url(file_path)
storage_service = StorageService()
|