gsfdc/server/app/services/storage.py

import json
import os
import re
import uuid
from datetime import datetime, timedelta, timezone
from io import BytesIO
from urllib.parse import quote

import urllib3
from fastapi import HTTPException, status
from minio import Minio
from minio.error import S3Error

from app.core.config import get_settings

_EXT = re.compile(r"^\.[a-z0-9]{1,8}$")
_MISSING = {"NoSuchKey", "NoSuchBucket", "NoSuchObject", "NotFound"}
_EXISTS = {"BucketAlreadyOwnedByYou", "BucketAlreadyExists"}
_NO_POLICY = {"NoSuchBucketPolicy", "NoSuchBucket"}


def _unavailable() -> None:
    raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="对象存储不可用")


def _unconfigured() -> None:
    raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="对象存储未配置")


def _endpoint() -> tuple[str, bool]:
    settings = get_settings()
    endpoint = settings.minio_endpoint.strip()
    secure = bool(settings.minio_secure)
    if endpoint.startswith("https://"):
        endpoint = endpoint[len("https://") :]
        secure = True
    elif endpoint.startswith("http://"):
        endpoint = endpoint[len("http://") :]
        secure = False
    return endpoint.rstrip("/"), secure


def _client() -> Minio:
    settings = get_settings()
    endpoint, secure = _endpoint()
    if not endpoint or not settings.minio_access_key or not settings.minio_secret_key:
        _unconfigured()
    http_client = urllib3.PoolManager(
        timeout=urllib3.Timeout(connect=3.0, read=60.0),
        retries=False,
    )
    return Minio(
        endpoint,
        access_key=settings.minio_access_key,
        secret_key=settings.minio_secret_key,
        secure=secure,
        http_client=http_client,
    )


def bucket_name(kind: str) -> str:
    settings = get_settings()
    if kind == "public":
        name = settings.minio_bucket_public.strip()
    elif kind == "private":
        name = settings.minio_bucket_private.strip()
    else:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="bucket 只能是 public 或 private")
    if not name:
        _unconfigured()
    return name


def _public_policy(bucket: str) -> str:
    return json.dumps(
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "PublicRead",
                    "Effect": "Allow",
                    "Principal": {"AWS": ["*"]},
                    "Action": ["s3:GetObject"],
                    "Resource": [f"arn:aws:s3:::{bucket}/*"],
                }
            ],
        }
    )


def _ensure_bucket(client: Minio, name: str) -> None:
    if client.bucket_exists(name):
        return
    try:
        client.make_bucket(name)
    except S3Error as exc:
        if exc.code not in _EXISTS:
            raise


def ensure_buckets() -> None:
    client = _client()
    public = bucket_name("public")
    private = bucket_name("private")
    _ensure_bucket(client, public)
    _ensure_bucket(client, private)
    client.set_bucket_policy(public, _public_policy(public))
    try:
        client.delete_bucket_policy(private)
    except S3Error as exc:
        if exc.code not in _NO_POLICY:
            raise


def safe_filename(filename: str | None) -> str:
    name = (filename or "").replace("\\", "/").split("/")[-1].replace("\x00", "").strip()
    if not name or name in {".", ".."}:
        return "file"
    return name[:255]


def make_object_key(filename: str) -> str:
    suffix = os.path.splitext(safe_filename(filename))[1].lower()
    ext = suffix if _EXT.fullmatch(suffix) else ""
    day = datetime.now(timezone.utc).strftime("%Y/%m/%d")
    return f"{day}/{uuid.uuid4().hex}{ext}"


def public_object_url(object_key: str) -> str:
    base = get_settings().minio_base_url.strip().rstrip("/")
    if not base:
        _unconfigured()
    parts = [quote(part, safe="") for part in object_key.split("/") if part]
    return f"{base}/{'/'.join(parts)}"


def private_object_url(object_key: str, expires_seconds: int = 60) -> tuple[str, datetime]:
    if expires_seconds < 1 or expires_seconds > 604800:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="链接有效期无效")
    expires = timedelta(seconds=expires_seconds)
    try:
        url = _client().presigned_get_object(bucket_name("private"), object_key, expires=expires)
    except HTTPException:
        raise
    except Exception:
        _unavailable()
    return url, datetime.now(timezone.utc) + expires


def upload_object(kind: str, object_key: str, data: bytes, content_type: str) -> None:
    try:
        _client().put_object(
            bucket_name(kind),
            object_key,
            BytesIO(data),
            length=len(data),
            content_type=content_type or "application/octet-stream",
        )
    except HTTPException:
        raise
    except Exception:
        _unavailable()


def remove_stored_object(kind: str, object_key: str) -> None:
    try:
        _client().remove_object(bucket_name(kind), object_key)
    except S3Error as exc:
        if exc.code in _MISSING:
            return
        _unavailable()
    except HTTPException:
        raise
    except Exception:
        _unavailable()