封装OSS 过程

app/core/oss.py

@@ -0,0 +1,51 @@
+"""阿里云 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