64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
import uuid
|
|
import boto3
|
|
from botocore.client import Config
|
|
from botocore.exceptions import ClientError
|
|
from app.core.config import settings
|
|
|
|
|
|
def _client():
|
|
return boto3.client(
|
|
"s3",
|
|
endpoint_url=settings.S3_ENDPOINT_URL,
|
|
aws_access_key_id=settings.S3_ACCESS_KEY,
|
|
aws_secret_access_key=settings.S3_SECRET_KEY,
|
|
region_name=settings.S3_REGION,
|
|
config=Config(signature_version="s3v4"),
|
|
)
|
|
|
|
|
|
def ensure_bucket() -> None:
|
|
s3 = _client()
|
|
try:
|
|
s3.head_bucket(Bucket=settings.S3_BUCKET)
|
|
except ClientError:
|
|
s3.create_bucket(Bucket=settings.S3_BUCKET)
|
|
# Make objects publicly readable so the frontend can <img src=...> directly.
|
|
policy = (
|
|
'{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":["*"]},'
|
|
'"Action":["s3:GetObject"],"Resource":["arn:aws:s3:::'
|
|
+ settings.S3_BUCKET
|
|
+ '/*"]}]}'
|
|
)
|
|
try:
|
|
s3.put_bucket_policy(Bucket=settings.S3_BUCKET, Policy=policy)
|
|
except ClientError:
|
|
pass
|
|
|
|
|
|
def upload_image(file_bytes: bytes, filename: str, content_type: str) -> tuple[str, str]:
|
|
"""Upload bytes to S3, return (object_key, public_url)."""
|
|
ext = ""
|
|
if "." in filename:
|
|
ext = "." + filename.rsplit(".", 1)[-1].lower()
|
|
key = f"characters/{uuid.uuid4().hex}{ext}"
|
|
s3 = _client()
|
|
s3.put_object(
|
|
Bucket=settings.S3_BUCKET,
|
|
Key=key,
|
|
Body=file_bytes,
|
|
ContentType=content_type or "application/octet-stream",
|
|
)
|
|
public_url = f"{settings.S3_PUBLIC_URL.rstrip('/')}/{settings.S3_BUCKET}/{key}"
|
|
return key, public_url
|
|
|
|
|
|
def delete_object_by_url(url: str) -> None:
|
|
prefix = f"{settings.S3_PUBLIC_URL.rstrip('/')}/{settings.S3_BUCKET}/"
|
|
if not url.startswith(prefix):
|
|
return
|
|
key = url[len(prefix):]
|
|
try:
|
|
_client().delete_object(Bucket=settings.S3_BUCKET, Key=key)
|
|
except ClientError:
|
|
pass
|