53 lines
2.0 KiB
Python
53 lines
2.0 KiB
Python
import oss2
|
||
import httpx
|
||
import os
|
||
import io
|
||
from fastapi import HTTPException
|
||
|
||
# 加载配置
|
||
ACCESS_KEY_ID = os.getenv("S3_ACCESS_KEY_ID")
|
||
ACCESS_KEY_SECRET = os.getenv("S3_SECRET_ACCESS_KEY")
|
||
ENDPOINT = os.getenv("S3_ENDPOINT", "https://oss-ap-northeast-1.aliyuncs.com")
|
||
BUCKET_NAME = os.getenv("S3_STORAGE_BUCKET_NAME", "character-roleplay-dev-asset")
|
||
|
||
# 初始化阿里云 Auth 和 Bucket
|
||
auth = oss2.Auth(ACCESS_KEY_ID, ACCESS_KEY_SECRET)
|
||
bucket = oss2.Bucket(auth, ENDPOINT, BUCKET_NAME)
|
||
|
||
async def upload_google_avatar_to_s3(google_avatar_url: str, user_id: str) -> str:
|
||
"""
|
||
下载 Google 头像并上传到阿里云 OSS
|
||
"""
|
||
try:
|
||
# 1. 使用 httpx 下载图片
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
response = await client.get(google_avatar_url)
|
||
if response.status_code != 200:
|
||
return google_avatar_url
|
||
|
||
image_data = response.content
|
||
# 获取内容类型（如 image/jpeg）
|
||
content_type = response.headers.get("Content-Type", "image/jpeg")
|
||
|
||
# 构造 OSS 中的存储路径
|
||
object_name = f"avatars/{user_id}/avatar.jpg"
|
||
|
||
headers = {"Content-Type": content_type}
|
||
|
||
# oss2.Bucket.put_object 是同步阻塞调用
|
||
# 在高并发下建议使用 run_in_executor，初期可以直接这样写
|
||
bucket.put_object(object_name, image_data, headers=headers)
|
||
|
||
# 4. 生成公开访问的 URL
|
||
# 阿里云 OSS 标准 URL 格式: https://bucket-name.endpoint/object-name
|
||
# 例如: https://character-roleplay-dev-asset.oss-ap-northeast-1.aliyuncs.com/avatars/xxx/avatar.jpg
|
||
host = ENDPOINT.replace("https://", "")
|
||
oss_url = f"https://{BUCKET_NAME}.{host}/{object_name}"
|
||
|
||
return oss_url
|
||
|
||
except Exception as e:
|
||
# 打印错误日志，生产环境建议使用 logger
|
||
print(f"OSS Upload Error: {str(e)}")
|
||
# 降级处理：上传失败则返回原始 Google URL，确保用户能注册成功
|
||
return google_avatar_url |