52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
"""阿里云 OSS 图片上传工具。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from urllib.parse import unquote, urlparse
|
|
from uuid import uuid4
|
|
|
|
import oss2
|
|
|
|
from app.config import settings
|
|
from app.core.logger import log
|
|
from app.tool.image_download import download_image
|
|
|
|
_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)
|
|
|
|
|
|
def _get_file_name(image_url: str) -> str:
|
|
"""从图片 URL 提取文件名，无扩展名时默认使用 PNG。"""
|
|
path = unquote(urlparse(image_url).path)
|
|
file_name = os.path.basename(path)
|
|
if not file_name or not os.path.splitext(file_name)[1]:
|
|
return "image.png"
|
|
return file_name
|
|
|
|
|
|
def upload(data: bytes, file_name: str) -> str:
|
|
"""上传图片字节到固定目录，返回 OSS URL。"""
|
|
key = f"{settings.oss_upload_dir}/{uuid4().hex[:18]}{os.path.splitext(file_name)[1]}"
|
|
_bucket.put_object(key, data)
|
|
url = f"{settings.oss_domain}/{key}"
|
|
log.info("OSS 上传成功 [{}] -> {}", file_name, url)
|
|
return url
|
|
|
|
|
|
def upload_image_url(image_url: str) -> str | None:
|
|
"""下载图片 URL 并上传到 OSS，成功时返回 OSS URL。"""
|
|
if not image_url or not image_url.strip():
|
|
return None
|
|
|
|
image_url = image_url.strip()
|
|
image = download_image(image_url)
|
|
if image is None:
|
|
return None
|
|
|
|
try:
|
|
return upload(image, _get_file_name(image_url))
|
|
except Exception as exc:
|
|
log.warning("图片上传 OSS 失败 [{}]: {}", image_url, exc)
|
|
return None
|