admin
620e95ff91
- 恢复演示出厂设置后直接解析演视LC视频序列并生成可打开帧序列 - 保持演视DICOM序列按文件名自然顺序恢复并生成帧 - 增加 MinIO 浏览器访问端点配置，修复 Docker 部署中封面和帧图预签名地址使用容器内主机名的问题 - 更新管理员恢复测试覆盖视频和 DICOM 帧数量 - 更新 README 和前后端契约/设计/测试文档中的演示恢复说明
143 lines
3.8 KiB
Python
143 lines
3.8 KiB
Python
"""MinIO client wrapper for object storage operations."""
|
|
|
|
import io
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from minio import Minio
|
|
from minio.error import S3Error
|
|
|
|
from config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BUCKET_NAME = "seg-media"
|
|
|
|
_minio_client: Optional[Minio] = None
|
|
_minio_public_client: Optional[Minio] = None
|
|
|
|
|
|
def get_minio_client() -> Minio:
|
|
"""Return a singleton MinIO client instance."""
|
|
global _minio_client
|
|
if _minio_client is None:
|
|
_minio_client = Minio(
|
|
settings.minio_endpoint,
|
|
access_key=settings.minio_access_key,
|
|
secret_key=settings.minio_secret_key,
|
|
secure=settings.minio_secure,
|
|
)
|
|
return _minio_client
|
|
|
|
|
|
def get_minio_public_client() -> Minio:
|
|
"""Return a MinIO client configured for browser-facing presigned URLs."""
|
|
global _minio_public_client
|
|
if _minio_public_client is None:
|
|
endpoint = settings.minio_public_endpoint or settings.minio_endpoint
|
|
_minio_public_client = Minio(
|
|
endpoint,
|
|
access_key=settings.minio_access_key,
|
|
secret_key=settings.minio_secret_key,
|
|
secure=settings.minio_secure,
|
|
)
|
|
return _minio_public_client
|
|
|
|
|
|
def ensure_bucket_exists() -> None:
|
|
"""Create the bucket if it does not already exist."""
|
|
client = get_minio_client()
|
|
try:
|
|
if not client.bucket_exists(BUCKET_NAME):
|
|
client.make_bucket(BUCKET_NAME)
|
|
logger.info("Created MinIO bucket: %s", BUCKET_NAME)
|
|
else:
|
|
logger.info("MinIO bucket %s already exists", BUCKET_NAME)
|
|
except S3Error as exc:
|
|
logger.error("MinIO bucket check/creation failed: %s", exc)
|
|
raise
|
|
|
|
|
|
def upload_file(
|
|
object_name: str,
|
|
data: bytes,
|
|
content_type: str = "application/octet-stream",
|
|
length: int = -1,
|
|
) -> str:
|
|
"""Upload bytes to MinIO and return the object name.
|
|
|
|
Args:
|
|
object_name: Destination path inside the bucket.
|
|
data: Raw bytes or a file-like object.
|
|
content_type: MIME type of the object.
|
|
length: Object size; -1 for unknown (uses chunked upload).
|
|
|
|
Returns:
|
|
The object name (same as input).
|
|
"""
|
|
client = get_minio_client()
|
|
if isinstance(data, bytes):
|
|
data = io.BytesIO(data)
|
|
length = len(data.getvalue())
|
|
|
|
try:
|
|
client.put_object(
|
|
BUCKET_NAME,
|
|
object_name,
|
|
data,
|
|
length=length,
|
|
content_type=content_type,
|
|
)
|
|
logger.info("Uploaded to MinIO: %s", object_name)
|
|
return object_name
|
|
except S3Error as exc:
|
|
logger.error("MinIO upload failed: %s", exc)
|
|
raise
|
|
|
|
|
|
from datetime import timedelta
|
|
|
|
def get_presigned_url(
|
|
object_name: str,
|
|
expires: int = 3600,
|
|
method: str = "GET",
|
|
) -> str:
|
|
"""Generate a presigned URL for an object.
|
|
|
|
Args:
|
|
object_name: Path inside the bucket.
|
|
expires: Expiration time in seconds (default 1 hour).
|
|
method: HTTP method (GET or PUT).
|
|
|
|
Returns:
|
|
Presigned URL string.
|
|
"""
|
|
client = get_minio_public_client()
|
|
try:
|
|
url = client.get_presigned_url(method, BUCKET_NAME, object_name, expires=timedelta(seconds=expires))
|
|
return url
|
|
except S3Error as exc:
|
|
logger.error("MinIO presigned URL failed: %s", exc)
|
|
raise
|
|
|
|
|
|
def download_file(object_name: str) -> bytes:
|
|
"""Download an object from MinIO and return its bytes.
|
|
|
|
Args:
|
|
object_name: Path inside the bucket.
|
|
|
|
Returns:
|
|
Raw bytes of the object.
|
|
"""
|
|
client = get_minio_client()
|
|
try:
|
|
response = client.get_object(BUCKET_NAME, object_name)
|
|
data = response.read()
|
|
response.close()
|
|
response.release_conn()
|
|
return data
|
|
except S3Error as exc:
|
|
logger.error("MinIO download failed: %s", exc)
|
|
raise
|