76 lines
2.2 KiB
Python
76 lines
2.2 KiB
Python
import uuid
|
|
from datetime import datetime
|
|
from io import BytesIO
|
|
|
|
from qcloud_cos import CosConfig, CosS3Client
|
|
|
|
from app.config import settings
|
|
|
|
config = CosConfig(
|
|
Region=settings.cos_region,
|
|
SecretId=settings.cos_secret_id,
|
|
SecretKey=settings.cos_secret_key,
|
|
)
|
|
client = CosS3Client(config)
|
|
|
|
ALLOWED_IMAGE_TYPES = {
|
|
"image/jpeg": ".jpg",
|
|
"image/png": ".png",
|
|
"image/gif": ".gif",
|
|
"image/webp": ".webp",
|
|
}
|
|
|
|
|
|
def upload_bytes(key: str, data: bytes, content_type: str) -> str:
|
|
"""Upload raw bytes to COS, return the public URL."""
|
|
client.put_object(
|
|
Bucket=settings.cos_bucket,
|
|
Key=key,
|
|
Body=BytesIO(data),
|
|
ContentType=content_type,
|
|
)
|
|
return f"{settings.cos_base_url}/{key}"
|
|
|
|
|
|
def upload_image(prefix: str, filename: str, data: bytes, content_type: str) -> str:
|
|
"""
|
|
Upload an image to COS under the given prefix.
|
|
Returns the public URL.
|
|
"""
|
|
if content_type not in ALLOWED_IMAGE_TYPES:
|
|
raise ValueError(f"Unsupported image type: {content_type}")
|
|
|
|
ext = ALLOWED_IMAGE_TYPES[content_type]
|
|
date_path = datetime.now().strftime("%Y/%m/%d")
|
|
unique_name = uuid.uuid4().hex[:12]
|
|
key = f"{prefix}/{date_path}/{unique_name}{ext}"
|
|
|
|
return upload_bytes(key, data, content_type)
|
|
|
|
|
|
FILE_TYPE_EXTENSIONS = {
|
|
"application/pdf": ".pdf",
|
|
"application/msword": ".doc",
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
|
|
"application/vnd.ms-excel": ".xls",
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
|
|
"application/vnd.ms-powerpoint": ".ppt",
|
|
"application/vnd.openxmlformats-officedocument.presentationml.presentation": ".pptx",
|
|
"application/zip": ".zip",
|
|
"text/plain": ".txt",
|
|
"image/jpeg": ".jpg",
|
|
"image/png": ".png",
|
|
"image/gif": ".gif",
|
|
"image/webp": ".webp",
|
|
}
|
|
|
|
|
|
def upload_file(prefix: str, filename: str, data: bytes, content_type: str) -> str:
|
|
"""Upload any allowed file type to COS. Returns the public URL."""
|
|
ext = FILE_TYPE_EXTENSIONS.get(content_type, "")
|
|
date_path = datetime.now().strftime("%Y/%m/%d")
|
|
unique_name = uuid.uuid4().hex[:12]
|
|
key = f"{prefix}/{date_path}/{unique_name}{ext}"
|
|
|
|
return upload_bytes(key, data, content_type)
|