Xiaoxia AI
153 lines
4.5 KiB
Python
153 lines
4.5 KiB
Python
"""阿里云 OSS 存储服务"""
|
||
from datetime import timedelta
|
||
from typing import BinaryIO, Optional
|
||
from urllib.parse import urlparse
|
||
import os
|
||
|
||
import oss2
|
||
from app.config import get_settings
|
||
|
||
|
||
class OSSStorageService:
|
||
"""阿里云 OSS 存储服务"""
|
||
|
||
def __init__(self):
|
||
settings = get_settings()
|
||
auth = oss2.Auth(
|
||
settings.OSS_ACCESS_KEY_ID,
|
||
settings.OSS_ACCESS_KEY_SECRET
|
||
)
|
||
self.bucket = oss2.Bucket(
|
||
auth,
|
||
settings.OSS_ENDPOINT,
|
||
settings.OSS_BUCKET_NAME
|
||
)
|
||
self.bucket_name = settings.OSS_BUCKET_NAME
|
||
self.public_url = f"https://{settings.OSS_BUCKET_NAME}.{settings.OSS_ENDPOINT}"
|
||
|
||
def upload_file(
|
||
self,
|
||
file_or_path,
|
||
storage_key: str,
|
||
content_type: str = "application/octet-stream",
|
||
) -> str:
|
||
"""
|
||
上传文件到 OSS
|
||
|
||
Args:
|
||
file_or_path: 文件对象或本地文件路径
|
||
storage_key: 存储键（文件路径）
|
||
content_type: 内容类型
|
||
|
||
Returns:
|
||
文件公网 URL
|
||
"""
|
||
try:
|
||
# 如果是字符串路径，从本地文件上传
|
||
if isinstance(file_or_path, str):
|
||
self.bucket.put_object_from_file(
|
||
storage_key,
|
||
file_or_path,
|
||
headers={'Content-Type': content_type}
|
||
)
|
||
else:
|
||
# 文件对象
|
||
file_or_path.seek(0)
|
||
self.bucket.put_object(
|
||
storage_key,
|
||
file_or_path,
|
||
headers={'Content-Type': content_type}
|
||
)
|
||
|
||
return f"{self.public_url}/{storage_key}"
|
||
except Exception as e:
|
||
raise Exception(f"Failed to upload file to OSS: {e}")
|
||
|
||
def get_url(self, storage_key: str) -> str:
|
||
"""获取文件公网 URL"""
|
||
return f"{self.public_url}/{storage_key}"
|
||
|
||
def get_download_url(self, storage_key_or_url: str, expires_seconds: int = 3600) -> str:
|
||
"""
|
||
获取文件下载签名 URL（用于私有文件）
|
||
|
||
Args:
|
||
storage_key_or_url: 存储键或完整 URL
|
||
expires_seconds: 过期时间（秒）
|
||
|
||
Returns:
|
||
签名 URL
|
||
"""
|
||
storage_key = self._normalize_storage_key(storage_key_or_url)
|
||
try:
|
||
return self.bucket.sign_url('GET', storage_key, expires_seconds)
|
||
except Exception:
|
||
return self.get_url(storage_key)
|
||
|
||
def _normalize_storage_key(self, storage_key_or_url: str) -> str:
|
||
"""从 URL 中提取存储键"""
|
||
if storage_key_or_url.startswith("http://") or storage_key_or_url.startswith("https://"):
|
||
parsed = urlparse(storage_key_or_url)
|
||
# 移除开头的 /
|
||
return parsed.path.lstrip("/")
|
||
return storage_key_or_url.lstrip("/")
|
||
|
||
def download_file(self, storage_key: str, local_path: str):
|
||
"""
|
||
从 OSS 下载文件到本地
|
||
|
||
Args:
|
||
storage_key: 存储键
|
||
local_path: 本地文件路径
|
||
"""
|
||
try:
|
||
os.makedirs(os.path.dirname(local_path), exist_ok=True)
|
||
self.bucket.get_object_to_file(storage_key, local_path)
|
||
except Exception as e:
|
||
raise Exception(f"Failed to download file from OSS: {e}")
|
||
|
||
def delete_file(self, storage_key: str):
|
||
"""
|
||
删除 OSS 文件
|
||
|
||
Args:
|
||
storage_key: 存储键
|
||
"""
|
||
try:
|
||
self.bucket.delete_object(storage_key)
|
||
except Exception as e:
|
||
print(f"Error deleting file from OSS: {e}")
|
||
|
||
def file_exists(self, storage_key: str) -> bool:
|
||
"""
|
||
检查文件是否存在
|
||
|
||
Args:
|
||
storage_key: 存储键
|
||
|
||
Returns:
|
||
是否存在
|
||
"""
|
||
return self.bucket.object_exists(storage_key)
|
||
|
||
|
||
# 向后兼容的服务类名
|
||
MinIOService = OSSStorageService
|
||
|
||
|
||
_storage_service = None
|
||
|
||
|
||
def get_storage_service() -> OSSStorageService:
|
||
"""获取存储服务实例（全局单例）"""
|
||
global _storage_service
|
||
if _storage_service is None:
|
||
_storage_service = OSSStorageService()
|
||
return _storage_service
|
||
|
||
|
||
# 向后兼容的别名
|
||
def get_minio_service() -> OSSStorageService:
|
||
"""向后兼容：返回 OSS 服务（已替换 MinIO）"""
|
||
return get_storage_service()
|