yml2213
df8c63b829
协作台账：新表 huya_external_orders 替代 Excel 多人协同表（设计见
docs/设计-虎牙外部CK协作台账.md），一条 CK 一行，所有人共享、改动留痕。
- 前端 pages/HuyaCollabPage.tsx + 菜单「虎牙 → 外部单协作」/huya/collab：
内联编辑（带乐观锁 version）、图片缩略图、详情抽屉（字段表单 + 两个图片槽位 +
操作时间线）、粘贴导入弹窗（带 TSV 预览）、批量/单条验证 CK、页头统计条、筛选
- 两套状态机：业务 pending→bound→recharged→scripted→done（旁路 void）、
凭据 unverified/valid/invalid/disabled；唯一写入口在 services/huya_collab_service.py，
字段级审计 + 整数 version 乐观锁（不用 updated_at：DATETIME 只到秒）
- CK 只读探针 check_user_bind_game_account（真机判据：响应非 None 且 status == 200），
成功即覆盖写绑定快照 bind_game_account/bind_game_role/bind_game_channel（迁移 0056），
失败只改凭据状态、不动业务状态；人工只能停用/恢复（POST /{id}/ck/status）
- 贴/换/重建 CK、TSV 批量导入（SAVEPOINT 单行隔离，部分失败不整体失败）、
权限 huya:collab:view/edit/manage（删除只给运营）
- 孤儿图清理脚本 scripts/cleanup_orphan_file_assets.py（默认 --dry-run）
图片上传（协作台账截图用）：OSS/本地双后端 + 缩略图变体 + /api/files 代理回读，
权限 file:upload / file:view。
验证：pytest 880 passed / 6 skipped；ruff format 全绿、ruff check 仅剩既有告警；
前端 tsc -b / eslint / vite build 通过；接口端到端冒烟 29 项全绿（TestClient，探针用假客户端）；
探针真机冒烟 3 例（有效 CK=True、别人 uid=False、裸 cookie=False）。
435 lines
15 KiB
Python
435 lines
15 KiB
Python
"""对象存储适配层：本地目录后端 + S3 兼容对象存储（阿里云 OSS / MinIO）。
|
||
|
||
约定对齐两个参考实现（order_site 的 ``apps/backend/src/services/file-storage/*``、
|
||
hfb_sys 的 ``backend/internal/modules/file/storage.go``）：
|
||
|
||
- **服务端代理读写**：对外只暴露 ``/api/files/object?key=...``，不下发 OSS 直链，
|
||
也不发签名链接 —— 签名链接会被业务方原样存进库里，过期后到处裂图；
|
||
- **内网 endpoint**：ECS 与桶同地域时用 ``oss-<region>-internal.aliyuncs.com``，
|
||
免公网流量费且延迟低；
|
||
- **寻址风格**：阿里云 OSS 必须 virtual-host style，自建 MinIO 走 path-style；
|
||
- **S3 校验头**：OSS 的 S3 兼容层不认 botocore 默认的 CRC32 请求校验头，需要关掉。
|
||
|
||
boto3 只在真正走 OSS 时才导入：本地开发/测试用 ``local`` 后端，没装 boto3 也能跑。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import io
|
||
import os
|
||
import threading
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
from typing import Any, BinaryIO, Protocol
|
||
from urllib.parse import quote, urlsplit
|
||
|
||
from loguru import logger
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
||
DEFAULT_LOCAL_ROOT = PROJECT_ROOT / "data" / "file_storage"
|
||
|
||
MAX_UPLOAD_SIZE_MB_DEFAULT = 10
|
||
|
||
|
||
class FileStorageError(RuntimeError):
|
||
"""对象存储不可用（配置缺失、依赖缺失或远端拒绝）。"""
|
||
|
||
|
||
class StorageUnavailableError(FileStorageError):
|
||
"""存储后端没有配置好或连不上，属于「暂时不可用」而非「文件不存在」。"""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class StorageSettings:
|
||
"""从环境变量读到的存储配置。"""
|
||
|
||
mode: str
|
||
endpoint: str
|
||
bucket: str
|
||
access_key_id: str
|
||
secret_access_key: str
|
||
region: str
|
||
local_root: Path
|
||
max_upload_size_mb: int
|
||
|
||
@property
|
||
def is_remote(self) -> bool:
|
||
return self.mode == "oss"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class StoredObject:
|
||
"""读取到的对象：调用方负责关闭 ``reader``。"""
|
||
|
||
reader: BinaryIO
|
||
content_type: str
|
||
size: int
|
||
etag: str = ""
|
||
last_modified: datetime | None = None
|
||
|
||
|
||
class ObjectStorage(Protocol):
|
||
"""存储后端接口。缺 key 一律抛 ``FileNotFoundError``。"""
|
||
|
||
def put_object(
|
||
self,
|
||
*,
|
||
key: str,
|
||
data: bytes,
|
||
content_type: str,
|
||
metadata: dict[str, str] | None = None,
|
||
) -> None: ...
|
||
|
||
def get_object(self, key: str) -> StoredObject: ...
|
||
|
||
def delete_object(self, key: str) -> None: ...
|
||
|
||
def ensure_ready(self) -> None: ...
|
||
|
||
|
||
def _env_int(name: str, default: int) -> int:
|
||
"""读取正整数环境变量，非法值回退到默认值。"""
|
||
try:
|
||
value = int(os.getenv(name, str(default)))
|
||
except ValueError:
|
||
return default
|
||
return value if value > 0 else default
|
||
|
||
|
||
def load_storage_settings() -> StorageSettings:
|
||
"""读取 ``OSS_*`` 环境变量。未知 mode 一律回退到 local。
|
||
|
||
变量名沿用仓库既有约定（compose 早已透传、生产 ``.env`` 里也已写入）：
|
||
``OSS_ENDPOINT`` 用内网端点，``OSS_BUCKET`` 是本项目自己的桶。
|
||
"""
|
||
raw_mode = (os.getenv("OSS_MODE", "") or "").strip().lower()
|
||
mode = raw_mode if raw_mode in ("local", "oss") else "local"
|
||
endpoint = (os.getenv("OSS_ENDPOINT", "") or "").strip()
|
||
local_root = (os.getenv("OSS_LOCAL_ROOT", "") or "").strip()
|
||
region = (os.getenv("OSS_REGION", "") or "").strip() or _derive_region(endpoint)
|
||
return StorageSettings(
|
||
mode=mode,
|
||
endpoint=endpoint,
|
||
bucket=(os.getenv("OSS_BUCKET", "") or "").strip(),
|
||
access_key_id=(os.getenv("OSS_ACCESS_KEY_ID", "") or "").strip(),
|
||
secret_access_key=(os.getenv("OSS_ACCESS_KEY_SECRET", "") or "").strip(),
|
||
region=region,
|
||
local_root=Path(local_root) if local_root else DEFAULT_LOCAL_ROOT,
|
||
max_upload_size_mb=_env_int(
|
||
"OSS_MAX_UPLOAD_SIZE_MB", MAX_UPLOAD_SIZE_MB_DEFAULT
|
||
),
|
||
)
|
||
|
||
|
||
def _derive_region(endpoint: str) -> str:
|
||
"""从 endpoint 反推 region。
|
||
|
||
``oss-cn-hangzhou-internal.aliyuncs.com`` → ``oss-cn-hangzhou``。
|
||
生产 ``.env`` 只要求配 endpoint/bucket/AK 三项，但 sigv4 需要 region，
|
||
所以没显式配 ``OSS_REGION`` 时按端点推导，省一个必填项、也避免填错。
|
||
"""
|
||
if not endpoint:
|
||
return ""
|
||
candidate = endpoint if "//" in endpoint else f"//{endpoint}"
|
||
host = urlsplit(candidate).hostname or ""
|
||
label = host.split(".")[0]
|
||
if label.startswith("oss-") and label.endswith("-internal"):
|
||
return label[: -len("-internal")]
|
||
return label if label.startswith("oss-") else ""
|
||
|
||
|
||
def max_upload_size_bytes(settings: StorageSettings | None = None) -> int:
|
||
"""上传大小上限（字节）。"""
|
||
resolved = settings or load_storage_settings()
|
||
return max(1, resolved.max_upload_size_mb) * 1024 * 1024
|
||
|
||
|
||
def normalize_object_key(raw_key: Any) -> str:
|
||
"""归一化对象 key 并挡住路径穿越。
|
||
|
||
与 order_site 的 ``normalizeObjectKey`` 同一套判据：去掉前导斜杠，
|
||
出现 ``..`` 或反斜杠直接拒绝。
|
||
"""
|
||
key = str(raw_key or "").strip().lstrip("/")
|
||
if not key or ".." in key or "\\" in key:
|
||
raise ValueError("文件 key 不正确")
|
||
return key
|
||
|
||
|
||
class LocalObjectStorage:
|
||
"""本地目录后端：仅用于开发与测试，生产走 OSS。
|
||
|
||
内容类型按扩展名推断（本地模式下原图与变体的扩展名都由上传流程按 MIME 生成，
|
||
因此推断是可靠的）；etag 用内容 md5，便于回读接口做 304。
|
||
"""
|
||
|
||
def __init__(self, root: Path):
|
||
self._root = Path(root)
|
||
|
||
@property
|
||
def root(self) -> Path:
|
||
return self._root
|
||
|
||
def _path_for(self, key: str) -> Path:
|
||
return self._root / normalize_object_key(key)
|
||
|
||
def ensure_ready(self) -> None:
|
||
self._root.mkdir(parents=True, exist_ok=True)
|
||
|
||
def put_object(
|
||
self,
|
||
*,
|
||
key: str,
|
||
data: bytes,
|
||
content_type: str,
|
||
metadata: dict[str, str] | None = None,
|
||
) -> None:
|
||
del content_type, metadata # 本地后端不保存对象元数据
|
||
path = self._path_for(key)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_bytes(data)
|
||
|
||
def get_object(self, key: str) -> StoredObject:
|
||
path = self._path_for(key)
|
||
try:
|
||
data = path.read_bytes()
|
||
except FileNotFoundError as exc:
|
||
raise FileNotFoundError(key) from exc
|
||
return StoredObject(
|
||
reader=io.BytesIO(data),
|
||
content_type=_guess_content_type(path.name),
|
||
size=len(data),
|
||
etag=f'"{hashlib.md5(data).hexdigest()}"',
|
||
last_modified=datetime.fromtimestamp(path.stat().st_mtime, UTC),
|
||
)
|
||
|
||
def delete_object(self, key: str) -> None:
|
||
path = self._path_for(key)
|
||
try:
|
||
path.unlink()
|
||
except FileNotFoundError:
|
||
return
|
||
|
||
|
||
class S3ObjectStorage:
|
||
"""S3 兼容后端（阿里云 OSS / MinIO）。boto3 延迟导入。"""
|
||
|
||
def __init__(self, settings: StorageSettings):
|
||
self._settings = settings
|
||
self._client: Any = None
|
||
self._bucket_ready = False
|
||
self._lock = threading.Lock()
|
||
|
||
@property
|
||
def is_aliyun(self) -> bool:
|
||
return "aliyuncs.com" in self._settings.endpoint.lower()
|
||
|
||
def _build_client(self) -> Any:
|
||
settings = self._settings
|
||
if not (settings.endpoint and settings.bucket):
|
||
raise StorageUnavailableError(
|
||
"对象存储未配置：需要同时设置 OSS_ENDPOINT 与 OSS_BUCKET"
|
||
)
|
||
try:
|
||
import boto3
|
||
except ImportError as exc: # pragma: no cover - 取决于部署环境
|
||
raise StorageUnavailableError(
|
||
"缺少 boto3 依赖，无法使用 OSS 存储（执行 uv sync 安装后重启服务）"
|
||
) from exc
|
||
|
||
from botocore.config import Config as BotoConfig
|
||
|
||
addons: dict[str, Any] = {}
|
||
try:
|
||
# 新版 botocore 才支持这两个开关；旧版直接省略。
|
||
addons = {
|
||
"request_checksum_calculation": "when_required",
|
||
"response_checksum_validation": "when_required",
|
||
}
|
||
config = BotoConfig(
|
||
signature_version="s3v4",
|
||
s3={"addressing_style": "virtual" if self.is_aliyun else "path"},
|
||
retries={"max_attempts": 3, "mode": "standard"},
|
||
connect_timeout=_env_int("OSS_CONNECT_TIMEOUT", 5),
|
||
read_timeout=_env_int("OSS_READ_TIMEOUT", 30),
|
||
max_pool_connections=_env_int("OSS_MAX_POOL_CONNECTIONS", 20),
|
||
**addons,
|
||
)
|
||
except TypeError: # pragma: no cover - 仅旧版 botocore 走到
|
||
config = BotoConfig(
|
||
signature_version="s3v4",
|
||
s3={"addressing_style": "virtual" if self.is_aliyun else "path"},
|
||
)
|
||
|
||
return boto3.client(
|
||
"s3",
|
||
endpoint_url=_normalize_endpoint_url(settings.endpoint, self.is_aliyun),
|
||
aws_access_key_id=settings.access_key_id or None,
|
||
aws_secret_access_key=settings.secret_access_key or None,
|
||
region_name=settings.region or None,
|
||
config=config,
|
||
)
|
||
|
||
def _get_client(self) -> Any:
|
||
if self._client is None:
|
||
with self._lock:
|
||
if self._client is None:
|
||
self._client = self._build_client()
|
||
return self._client
|
||
|
||
def ensure_ready(self) -> None:
|
||
"""确保桶可用。
|
||
|
||
阿里云 OSS 的桶由控制台预先创建 —— HeadBucket 在 RAM 权限收紧时会 403
|
||
甚至挂起，所以直接跳过；自建 MinIO 才做「不存在就建」。
|
||
"""
|
||
if self._bucket_ready or self.is_aliyun:
|
||
self._bucket_ready = True
|
||
return
|
||
client = self._get_client()
|
||
with self._lock:
|
||
if self._bucket_ready:
|
||
return
|
||
try:
|
||
client.head_bucket(Bucket=self._settings.bucket)
|
||
except Exception as exc: # 兼容层错误码不规范，只能按形状兜底
|
||
if not _is_not_found(exc):
|
||
raise
|
||
client.create_bucket(Bucket=self._settings.bucket)
|
||
self._bucket_ready = True
|
||
|
||
def put_object(
|
||
self,
|
||
*,
|
||
key: str,
|
||
data: bytes,
|
||
content_type: str,
|
||
metadata: dict[str, str] | None = None,
|
||
) -> None:
|
||
self.ensure_ready()
|
||
try:
|
||
self._get_client().put_object(
|
||
Bucket=self._settings.bucket,
|
||
Key=normalize_object_key(key),
|
||
Body=data,
|
||
ContentType=content_type or "application/octet-stream",
|
||
**({"Metadata": _sanitize_metadata(metadata)} if metadata else {}),
|
||
)
|
||
except Exception as exc: # 边界统一转成存储异常
|
||
raise StorageUnavailableError(f"对象写入失败: {exc}") from exc
|
||
|
||
def get_object(self, key: str) -> StoredObject:
|
||
self.ensure_ready()
|
||
try:
|
||
output = self._get_client().get_object(
|
||
Bucket=self._settings.bucket, Key=normalize_object_key(key)
|
||
)
|
||
except Exception as exc: # 边界统一转成存储异常
|
||
if _is_not_found(exc):
|
||
raise FileNotFoundError(key) from exc
|
||
raise StorageUnavailableError(f"对象读取失败: {exc}") from exc
|
||
|
||
body = output.get("Body")
|
||
if body is None:
|
||
raise FileNotFoundError(key)
|
||
return StoredObject(
|
||
reader=body,
|
||
content_type=str(output.get("ContentType") or "application/octet-stream"),
|
||
size=int(output.get("ContentLength") or 0),
|
||
etag=str(output.get("ETag") or ""),
|
||
last_modified=output.get("LastModified"),
|
||
)
|
||
|
||
def delete_object(self, key: str) -> None:
|
||
self.ensure_ready()
|
||
try:
|
||
self._get_client().delete_object(
|
||
Bucket=self._settings.bucket, Key=normalize_object_key(key)
|
||
)
|
||
except Exception as exc: # noqa: BLE001 删除失败不该打断业务异常链
|
||
logger.warning(f"对象删除失败 key={key}: {exc}")
|
||
|
||
|
||
_storage: ObjectStorage | None = None
|
||
_storage_lock = threading.Lock()
|
||
|
||
|
||
def _build_storage(settings: StorageSettings) -> ObjectStorage:
|
||
"""按配置构造后端：oss → S3 兼容，其余 → 本地目录。"""
|
||
if settings.is_remote:
|
||
return S3ObjectStorage(settings)
|
||
return LocalObjectStorage(settings.local_root)
|
||
|
||
|
||
def get_object_storage(settings: StorageSettings | None = None) -> ObjectStorage:
|
||
"""返回当前配置对应的存储后端；显式传 settings 时不走单例（测试用）。"""
|
||
global _storage
|
||
if settings is not None:
|
||
return _build_storage(settings)
|
||
if _storage is None:
|
||
with _storage_lock:
|
||
if _storage is None:
|
||
_storage = _build_storage(load_storage_settings())
|
||
return _storage
|
||
|
||
|
||
def reset_object_storage_cache() -> None:
|
||
"""清空单例 —— 测试里切换 ``STORAGE_*`` 环境变量后调用。"""
|
||
global _storage
|
||
with _storage_lock:
|
||
_storage = None
|
||
|
||
|
||
def _guess_content_type(filename: str) -> str:
|
||
lowered = filename.lower()
|
||
if lowered.endswith((".jpg", ".jpeg")):
|
||
return "image/jpeg"
|
||
if lowered.endswith(".png"):
|
||
return "image/png"
|
||
if lowered.endswith(".webp"):
|
||
return "image/webp"
|
||
if lowered.endswith(".pdf"):
|
||
return "application/pdf"
|
||
return "application/octet-stream"
|
||
|
||
|
||
def _normalize_endpoint_url(endpoint: str, is_aliyun: bool) -> str:
|
||
"""补全 scheme：阿里云走 https，自建 MinIO 走 http。"""
|
||
candidate = endpoint.strip()
|
||
if not candidate:
|
||
raise StorageUnavailableError("对象存储 endpoint 未配置")
|
||
if not candidate.startswith(("http://", "https://")):
|
||
candidate = f"{'https' if is_aliyun else 'http'}://{candidate}"
|
||
parsed = urlsplit(candidate)
|
||
if not parsed.hostname:
|
||
raise StorageUnavailableError(f"对象存储 endpoint 不合法: {endpoint}")
|
||
return candidate.rstrip("/")
|
||
|
||
|
||
def _sanitize_metadata(metadata: dict[str, str] | None) -> dict[str, str]:
|
||
"""元数据会作为 HTTP header 下发，值统一转义为 ASCII（中文文件名否则会 400）。"""
|
||
if not metadata:
|
||
return {}
|
||
sanitized: dict[str, str] = {}
|
||
for raw_key, raw_value in metadata.items():
|
||
key = str(raw_key or "").strip()
|
||
if not key:
|
||
continue
|
||
sanitized[key] = quote(str(raw_value or ""), safe="")
|
||
return sanitized
|
||
|
||
|
||
def _is_not_found(error: Exception) -> bool:
|
||
"""把各兼容层形状不一的不存在错误统一识别出来。"""
|
||
code = getattr(error, "response", None)
|
||
if isinstance(code, dict):
|
||
error_code = str((code.get("Error") or {}).get("Code") or "")
|
||
status = (code.get("ResponseMetadata") or {}).get("HTTPStatusCode")
|
||
if error_code in ("404", "NoSuchKey", "NoSuchBucket", "NotFound"):
|
||
return True
|
||
if status == 404:
|
||
return True
|
||
return getattr(error, "name", "") in ("NotFound", "NoSuchKey")
|