43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
"""阿里云 OSS 上传：往固定目录上传，返回下载地址"""
|
||
|
||
import asyncio
|
||
import base64
|
||
import os
|
||
from uuid import uuid4
|
||
|
||
import oss2
|
||
|
||
from app.config import settings
|
||
from app.core.logger import log
|
||
|
||
_auth = oss2.Auth(settings.oss_access_key_id, settings.oss_access_key_secret)
|
||
_bucket = oss2.Bucket(_auth, f"https://{settings.oss_endpoint}", settings.oss_bucket)
|
||
|
||
|
||
async def upload(data: bytes, file_name: str) -> str:
|
||
"""上传字节内容到固定目录，返回下载地址"""
|
||
key = f"{settings.oss_upload_dir}/{uuid4().hex[:18]}{os.path.splitext(file_name)[1]}"
|
||
await asyncio.to_thread(_bucket.put_object, key, data)
|
||
url = f"{settings.oss_domain}/{key}"
|
||
log.info("OSS 上传成功 [{}] -> {}", file_name, url)
|
||
return url
|
||
|
||
|
||
async def upload_base64(b64: str, file_name: str = "image.png") -> str | None:
|
||
"""上传 base64 图片到固定目录，返回下载地址；空值或解码失败返回 None
|
||
|
||
:param b64: base64 字符串，兼容 data URI 前缀（data:image/png;base64,xxx）
|
||
:param file_name: 用于取后缀，默认 png
|
||
"""
|
||
if not b64 or not b64.strip():
|
||
return None
|
||
# 去掉 data URI 前缀
|
||
if "," in b64 and b64.strip().startswith("data:"):
|
||
b64 = b64.split(",", 1)[1]
|
||
try:
|
||
data = base64.b64decode(b64)
|
||
except Exception as e:
|
||
log.warning("base64 解码失败: {}", e)
|
||
return None
|
||
return await upload(data, file_name)
|