135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
try:
|
|
from qcloud_cos import CosConfig, CosS3Client
|
|
except ModuleNotFoundError: # pragma: no cover - exercised in runtime env
|
|
CosConfig = None
|
|
CosS3Client = None
|
|
|
|
from app.settings import settings
|
|
|
|
|
|
_CONTENT_TYPE_TO_EXT = {
|
|
"image/jpeg": "jpg",
|
|
"image/png": "png",
|
|
"image/webp": "webp",
|
|
}
|
|
_EXTENSION_ALIASES = {
|
|
".jpg": "jpg",
|
|
".jpeg": "jpg",
|
|
".png": "png",
|
|
".webp": "webp",
|
|
}
|
|
_EXT_TO_CONTENT_TYPE = {
|
|
"jpg": "image/jpeg",
|
|
"png": "image/png",
|
|
"webp": "image/webp",
|
|
}
|
|
|
|
|
|
class AvatarStorageError(Exception):
|
|
def __init__(self, message: str, status_code: int = 400) -> None:
|
|
super().__init__(message)
|
|
self.status_code = status_code
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StoredAvatar:
|
|
file_key: str
|
|
|
|
|
|
class AvatarStorageService:
|
|
def __init__(self) -> None:
|
|
self._client = None
|
|
|
|
def upload_avatar(
|
|
self,
|
|
*,
|
|
user_id: int,
|
|
filename: str | None,
|
|
content_type: str | None,
|
|
content: bytes,
|
|
) -> StoredAvatar:
|
|
self._assert_ready()
|
|
normalized_ext = self._normalize_extension(filename=filename, content_type=content_type)
|
|
if not content:
|
|
raise AvatarStorageError("avatar file is empty")
|
|
if len(content) > settings.cos_avatar_max_bytes:
|
|
raise AvatarStorageError("avatar file too large", status_code=413)
|
|
|
|
key = self._build_key(user_id=user_id, extension=normalized_ext)
|
|
self._get_client().put_object(
|
|
Bucket=settings.cos_bucket_ava,
|
|
Body=content,
|
|
Key=key,
|
|
ContentType=_EXT_TO_CONTENT_TYPE[normalized_ext],
|
|
EnableMD5=False,
|
|
)
|
|
return StoredAvatar(file_key=key)
|
|
|
|
def get_avatar_url(self, file_key: str) -> str:
|
|
self._assert_ready()
|
|
if not file_key:
|
|
raise AvatarStorageError("avatar file key is required", status_code=500)
|
|
return self._get_client().get_presigned_url(
|
|
Bucket=settings.cos_bucket_ava,
|
|
Key=file_key,
|
|
Method="GET",
|
|
Expired=settings.cos_avatar_url_expire_seconds,
|
|
)
|
|
|
|
def delete_avatar(self, file_key: str) -> None:
|
|
self._assert_ready()
|
|
if not file_key:
|
|
return
|
|
self._get_client().delete_object(Bucket=settings.cos_bucket_ava, Key=file_key)
|
|
|
|
def _assert_ready(self) -> None:
|
|
if CosConfig is None or CosS3Client is None:
|
|
raise AvatarStorageError("COS SDK is not installed", status_code=500)
|
|
|
|
required_pairs = {
|
|
"COS_SECRET_ID": settings.cos_secret_id,
|
|
"COS_SECRET_KEY": settings.cos_secret_key,
|
|
"COS_REGION": settings.cos_region,
|
|
"COS_BUCKET_AVA": settings.cos_bucket_ava,
|
|
}
|
|
missing = [key for key, value in required_pairs.items() if not value]
|
|
if missing:
|
|
raise AvatarStorageError(
|
|
f"missing COS avatar config: {', '.join(missing)}",
|
|
status_code=500,
|
|
)
|
|
|
|
def _get_client(self):
|
|
if self._client is None:
|
|
config = CosConfig(
|
|
Region=settings.cos_region,
|
|
SecretId=settings.cos_secret_id,
|
|
SecretKey=settings.cos_secret_key,
|
|
Scheme="https",
|
|
)
|
|
self._client = CosS3Client(config)
|
|
return self._client
|
|
|
|
def _normalize_extension(self, *, filename: str | None, content_type: str | None) -> str:
|
|
if content_type in _CONTENT_TYPE_TO_EXT:
|
|
return _CONTENT_TYPE_TO_EXT[content_type]
|
|
|
|
suffix = Path(filename or "").suffix.lower()
|
|
if suffix in _EXTENSION_ALIASES:
|
|
return _EXTENSION_ALIASES[suffix]
|
|
|
|
raise AvatarStorageError("unsupported avatar file type", status_code=415)
|
|
|
|
def _build_key(self, *, user_id: int, extension: str) -> str:
|
|
prefix = settings.cos_avatar_prefix.strip("/") or "avatars"
|
|
now = datetime.now(UTC)
|
|
return (
|
|
f"{prefix}/{user_id}/{now.strftime('%Y/%m/%d')}/"
|
|
f"{uuid4().hex}.{extension}"
|
|
)
|