You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
82 lines
3.1 KiB
82 lines
3.1 KiB
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi.responses import FileResponse, RedirectResponse
|
|
|
|
from .config import settings
|
|
|
|
|
|
class StorageService:
|
|
"""Store files locally for development or in Aliyun OSS in production."""
|
|
|
|
def __init__(self) -> None:
|
|
self.enabled = settings.oss_enabled
|
|
self.bucket = None
|
|
self.domain = settings.oss_domain.rstrip("/") if settings.oss_domain else ""
|
|
self.prefix = settings.oss_base_prefix.strip("/")
|
|
if not self.enabled:
|
|
return
|
|
if not all((settings.oss_endpoint, settings.oss_access_key_id,
|
|
settings.oss_access_key_secret, settings.oss_bucket_name)):
|
|
raise RuntimeError(
|
|
"OSS_ENABLED=true 时必须配置 OSS_ENDPOINT、OSS_ACCESS_KEY_ID、"
|
|
"OSS_ACCESS_KEY_SECRET 和 OSS_BUCKET_NAME"
|
|
)
|
|
try:
|
|
import oss2
|
|
except ImportError as exc: # pragma: no cover
|
|
raise RuntimeError("OSS_ENABLED=true 时请先安装 oss2 依赖") from exc
|
|
endpoint = settings.oss_endpoint.replace("http://", "https://")
|
|
self.bucket = oss2.Bucket(
|
|
oss2.Auth(settings.oss_access_key_id, settings.oss_access_key_secret),
|
|
endpoint,
|
|
settings.oss_bucket_name,
|
|
)
|
|
|
|
def make_key(self, relative_key: str) -> str:
|
|
relative_key = relative_key.lstrip("/")
|
|
return f"{self.prefix}/{relative_key}" if self.prefix else relative_key
|
|
|
|
def finalize(self, local_path: Path, relative_key: str) -> str:
|
|
if not self.enabled:
|
|
return str(local_path)
|
|
key = self.make_key(relative_key)
|
|
try:
|
|
self.bucket.put_object_from_file(key, str(local_path))
|
|
finally:
|
|
local_path.unlink(missing_ok=True)
|
|
return f"oss://{key}"
|
|
|
|
def remove(self, storage_path: str) -> None:
|
|
if storage_path.startswith("oss://"):
|
|
if self.enabled:
|
|
self.bucket.delete_object(storage_path[6:])
|
|
return
|
|
Path(storage_path).unlink(missing_ok=True)
|
|
|
|
def response(self, storage_path: str, mime_type: Optional[str], filename: Optional[str]):
|
|
if storage_path.startswith("oss://"):
|
|
if not self.enabled:
|
|
raise FileNotFoundError(storage_path)
|
|
url = self.bucket.sign_url("GET", storage_path[6:], 3600)
|
|
return RedirectResponse(url)
|
|
path = Path(storage_path)
|
|
if not path.is_file():
|
|
raise FileNotFoundError(storage_path)
|
|
return FileResponse(path, media_type=mime_type, filename=filename)
|
|
|
|
def access_url(self, storage_path: str, fallback_url: str) -> str:
|
|
"""Return a browser-ready URL without exposing private OSS credentials.
|
|
|
|
Production files live in a private OSS bucket, so public profile media
|
|
uses a one-hour signed URL. Local development keeps using the API route
|
|
supplied by the caller.
|
|
"""
|
|
if storage_path.startswith("oss://") and self.enabled:
|
|
return self.bucket.sign_url("GET", storage_path[6:], 3600)
|
|
return fallback_url
|
|
|
|
|
|
storage = StorageService()
|
|
|