xiaoxia
3ae15eb4fc
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
511 lines
20 KiB
Python
Executable File
511 lines
20 KiB
Python
Executable File
"""统一存储服务 — API 和 Worker 共用的唯一存储入口。
|
||
|
||
实现 StoragePort 端口接口，整合原来分散在各处的存储能力：
|
||
- API端 SharedStorageService 的全部能力（上传/下载/签名URL/直传POST）
|
||
- Worker端 oss_helpers 的高级能力（分片上传/超时保护/HTTP下载/Asset路径解析）
|
||
|
||
所有服务都通过这个统一入口与存储交互，消除重复实现。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import datetime as dt
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import logging
|
||
import os
|
||
import threading
|
||
from pathlib import Path
|
||
from typing import Optional, Union
|
||
from urllib.parse import unquote, urlparse
|
||
|
||
import requests
|
||
|
||
try:
|
||
import oss2
|
||
except ImportError: # pragma: no cover
|
||
oss2 = None
|
||
|
||
from packages.config import get_shared_settings
|
||
from packages.ports.storage_port import StoragePort
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── OSS 高级配置（从 oss_helpers 合并）─────────────────────────────────
|
||
OSS_CONNECT_TIMEOUT = 10 # 连接超时（秒）
|
||
OSS_UPLOAD_TOTAL_TIMEOUT = 300 # 单文件上传总超时（秒）
|
||
OSS_MULTIPART_THRESHOLD = 100 * 1024 * 1024 # 分片上传阈值：100MB
|
||
OSS_PART_SIZE = 8 * 1024 * 1024 # 分片大小：8MB
|
||
OSS_MULTIPART_NUM_THREADS = 3 # 分片上传并发数
|
||
OSS_HTTP_DOWNLOAD_TIMEOUT = 300 # HTTP下载超时（秒）
|
||
|
||
|
||
class SharedStorageService(StoragePort):
|
||
"""统一存储服务 — 实现 StoragePort，API 和 Worker 共用。
|
||
|
||
整合了原 SharedStorageService + oss_helpers 的全部能力。
|
||
"""
|
||
|
||
def __init__(self):
|
||
settings = get_shared_settings()
|
||
self.bucket_name = settings.oss_bucket_name
|
||
self.endpoint = settings.oss_endpoint
|
||
self.public_url = f"https://{settings.oss_bucket_name}.{settings.oss_endpoint}"
|
||
self.local_url_prefix = os.getenv("GENERATED_FILES_URL_PREFIX", "/generated-files")
|
||
self.bucket = None
|
||
|
||
self.access_key_id = settings.oss_access_key_id
|
||
self.access_key_secret = settings.oss_access_key_secret
|
||
|
||
has_key_id = bool(self.access_key_id)
|
||
has_key_secret = bool(self.access_key_secret)
|
||
|
||
if has_key_id and has_key_secret:
|
||
if oss2 is not None:
|
||
try:
|
||
# endpoint 不带 scheme 时补 https:// 前缀
|
||
bucket_endpoint = self.endpoint
|
||
if not bucket_endpoint.startswith(("http://", "https://")):
|
||
bucket_endpoint = f"https://{bucket_endpoint}"
|
||
auth = oss2.Auth(self.access_key_id, self.access_key_secret)
|
||
self.bucket = oss2.Bucket(
|
||
auth,
|
||
bucket_endpoint,
|
||
self.bucket_name,
|
||
connect_timeout=OSS_CONNECT_TIMEOUT,
|
||
)
|
||
logger.info(
|
||
"OSS initialized: endpoint=%s bucket=%s",
|
||
self.endpoint,
|
||
self.bucket_name,
|
||
)
|
||
except Exception as error:
|
||
logger.error("Failed to initialize OSS bucket client: %s", error)
|
||
else:
|
||
logger.error("oss2 SDK is not installed — OSS operations will fail")
|
||
else:
|
||
missing = []
|
||
if not has_key_id:
|
||
missing.append("OSS_ACCESS_KEY_ID")
|
||
if not has_key_secret:
|
||
missing.append("OSS_ACCESS_KEY_SECRET")
|
||
logger.error("OSS credentials not configured — missing: %s", ", ".join(missing))
|
||
|
||
# ── 诊断 ───────────────────────────────────────────────────────────
|
||
|
||
def diagnose(self) -> None:
|
||
"""输出存储配置诊断日志。"""
|
||
key_id_display = (
|
||
f"{self.access_key_id[:4]}...{self.access_key_id[-4:]}" if len(self.access_key_id) > 8 else "(empty)"
|
||
)
|
||
logger.info(
|
||
"[OSS诊断] endpoint=%s bucket_name=%s access_key_id=%s",
|
||
self.endpoint,
|
||
self.bucket_name,
|
||
key_id_display,
|
||
)
|
||
if self.bucket is None:
|
||
logger.error(
|
||
"[OSS诊断] ❌ bucket=None — 预签名URL不可用！"
|
||
"原因: OSS_ACCESS_KEY_ID/OSS_ACCESS_KEY_SECRET 未配置或 oss2 未安装。"
|
||
)
|
||
else:
|
||
logger.info("[OSS诊断] ✅ bucket 已配置，预签名URL可用")
|
||
|
||
# ── 工具方法 ───────────────────────────────────────────────────────
|
||
|
||
def _is_local_generated_url(self, storage_key_or_url: str) -> bool:
|
||
parsed = urlparse(storage_key_or_url)
|
||
path = parsed.path if parsed.scheme else storage_key_or_url
|
||
return path.startswith(f"{self.local_url_prefix}/")
|
||
|
||
def _normalize_storage_key(self, storage_key_or_url: str) -> str:
|
||
"""从 URL 提取存储键，并做 URL 解码。
|
||
|
||
防止 URL 编码的字符（空格=%20、中文=%XX）导致签名不匹配。
|
||
"""
|
||
if storage_key_or_url.startswith("http://") or storage_key_or_url.startswith("https://"):
|
||
parsed = urlparse(storage_key_or_url)
|
||
return unquote(parsed.path.lstrip("/"))
|
||
return storage_key_or_url.lstrip("/")
|
||
|
||
def normalize_storage_key(self, storage_key_or_url: str) -> str:
|
||
"""从 URL 提取存储键（公开方法）。"""
|
||
return self._normalize_storage_key(storage_key_or_url)
|
||
|
||
# ── 上传 ───────────────────────────────────────────────────────────
|
||
|
||
def upload_file(
|
||
self,
|
||
file_or_path: Union[str, Path, object],
|
||
storage_key: str,
|
||
content_type: str = "application/octet-stream",
|
||
) -> str:
|
||
"""上传文件到存储，返回公开 URL（简单上传，API端原有行为）。
|
||
|
||
- 路径字符串 → bucket.put_object_from_file
|
||
- 类文件对象 → bucket.put_object
|
||
- bucket未配置 → 抛 RuntimeError
|
||
"""
|
||
if self.bucket is None:
|
||
raise RuntimeError("OSS storage is not configured")
|
||
|
||
try:
|
||
if isinstance(file_or_path, (str, Path)):
|
||
self.bucket.put_object_from_file(storage_key, str(file_or_path), headers={"Content-Type": content_type})
|
||
else:
|
||
file_or_path.seek(0) # type: ignore[attr-defined]
|
||
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}") from e
|
||
|
||
def upload_file_smart(
|
||
self,
|
||
local_path: Union[str, Path],
|
||
storage_key: str,
|
||
) -> Optional[str]:
|
||
"""智能上传：大文件自动分片+超时保护（从 oss_helpers 合并）。
|
||
|
||
- 大文件（>100MB）走分片上传，3 线程并发
|
||
- 总超时 300s，防止网络异常时挂死
|
||
- 成功返回 URL，失败返回 None（不抛异常）
|
||
|
||
Worker端 oss_helpers.upload_to_oss 的统一入口。
|
||
"""
|
||
local_path = Path(local_path)
|
||
if not local_path.exists():
|
||
logger.error("上传文件不存在: %s", local_path)
|
||
return None
|
||
if self.bucket is None:
|
||
logger.error("OSS未配置，无法上传: %s", storage_key[:80])
|
||
return None
|
||
|
||
result: dict = {"url": None, "error": None, "file_size": 0}
|
||
done = threading.Event()
|
||
|
||
def _do_upload():
|
||
try:
|
||
try:
|
||
file_size = local_path.stat().st_size
|
||
result["file_size"] = file_size
|
||
use_multipart = file_size >= OSS_MULTIPART_THRESHOLD
|
||
except OSError:
|
||
use_multipart = False
|
||
file_size = 0
|
||
|
||
if use_multipart:
|
||
logger.info(
|
||
"大文件分片上传: storage_key=%s, size=%.1fMB, part_size=%dMB, threads=%d",
|
||
storage_key[:80],
|
||
file_size / 1024 / 1024,
|
||
OSS_PART_SIZE // 1024 // 1024,
|
||
OSS_MULTIPART_NUM_THREADS,
|
||
)
|
||
oss2.resumable_upload(
|
||
self.bucket,
|
||
storage_key,
|
||
str(local_path),
|
||
multipart_threshold=OSS_MULTIPART_THRESHOLD,
|
||
part_size=OSS_PART_SIZE,
|
||
num_threads=OSS_MULTIPART_NUM_THREADS,
|
||
)
|
||
else:
|
||
self.bucket.put_object_from_file(storage_key, str(local_path))
|
||
|
||
result["url"] = f"{self.public_url}/{storage_key}"
|
||
except Exception as e:
|
||
result["error"] = e
|
||
logger.exception("上传 OSS 失败: %s", storage_key)
|
||
finally:
|
||
done.set()
|
||
|
||
upload_thread = threading.Thread(target=_do_upload, daemon=True)
|
||
upload_thread.start()
|
||
finished = done.wait(timeout=OSS_UPLOAD_TOTAL_TIMEOUT)
|
||
|
||
if not finished:
|
||
logger.error(
|
||
"OSS 上传超时（%.0fs），强制中止: storage_key=%s, size=%.1fMB",
|
||
OSS_UPLOAD_TOTAL_TIMEOUT,
|
||
storage_key[:80],
|
||
result["file_size"] / 1024 / 1024 if result["file_size"] else 0,
|
||
)
|
||
return None
|
||
|
||
if result["error"]:
|
||
return None
|
||
|
||
return result["url"]
|
||
|
||
# ── 下载 ───────────────────────────────────────────────────────────
|
||
|
||
def download_file(self, storage_key: str, local_path: Union[str, Path]) -> None:
|
||
"""从 OSS 下载文件（简单下载，API端原有行为）。
|
||
|
||
bucket未配置 → 抛 RuntimeError
|
||
"""
|
||
if self.bucket is None:
|
||
raise RuntimeError("OSS storage is not configured")
|
||
|
||
local_path = Path(local_path)
|
||
os.makedirs(local_path.parent, exist_ok=True)
|
||
try:
|
||
self.bucket.get_object_to_file(self._normalize_storage_key(storage_key), str(local_path))
|
||
except Exception as e:
|
||
raise Exception(f"Failed to download file from OSS: {e}") from e
|
||
|
||
def download_asset(self, asset_storage_key: str, local_path: Union[str, Path]) -> bool:
|
||
"""下载素材（从 oss_helpers 合并）。
|
||
|
||
自动识别输入类型：
|
||
- 完整 URL → 走 HTTP 下载（支持预签名URL）
|
||
- 存储键 → 走 oss2 SDK 下载
|
||
|
||
成功返回 True，失败返回 False（不抛异常）。
|
||
"""
|
||
local_path = Path(local_path)
|
||
os.makedirs(local_path.parent, exist_ok=True)
|
||
|
||
# 完整URL走HTTP下载（兼容预签名URL）
|
||
if asset_storage_key.startswith(("http://", "https://")):
|
||
return self._download_via_http(asset_storage_key, local_path)
|
||
|
||
# OSS存储键走SDK
|
||
if self.bucket is None:
|
||
logger.error("OSS not configured, cannot download: %s", asset_storage_key[:80])
|
||
return False
|
||
try:
|
||
self.bucket.get_object_to_file(self._normalize_storage_key(asset_storage_key), str(local_path))
|
||
return local_path.exists() and local_path.stat().st_size > 0
|
||
except Exception:
|
||
logger.exception("下载素材失败: %s", asset_storage_key)
|
||
return False
|
||
|
||
def _download_via_http(self, url: str, local_path: Path) -> bool:
|
||
"""通过 HTTP 下载文件（支持预签名 URL）。
|
||
|
||
流式下载避免大文件内存溢出。
|
||
"""
|
||
try:
|
||
resp = requests.get(url, stream=True, timeout=OSS_HTTP_DOWNLOAD_TIMEOUT)
|
||
resp.raise_for_status()
|
||
with open(local_path, "wb") as f:
|
||
for chunk in resp.iter_content(chunk_size=8 * 1024 * 1024):
|
||
if chunk:
|
||
f.write(chunk)
|
||
return local_path.exists() and local_path.stat().st_size > 0
|
||
except Exception:
|
||
logger.exception("HTTP下载素材失败: %s", url[:100])
|
||
return False
|
||
|
||
# ── URL 生成 ──────────────────────────────────────────────────────
|
||
|
||
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。
|
||
|
||
bucket未配置时降级为公开URL；本地产物URL直接返回。
|
||
"""
|
||
if self.bucket is None:
|
||
if self._is_local_generated_url(storage_key_or_url):
|
||
return storage_key_or_url
|
||
logger.warning(
|
||
"get_download_url: OSS bucket not configured, returning raw URL. key=%s",
|
||
storage_key_or_url[:200],
|
||
)
|
||
return self.get_url(self.normalize_storage_key(storage_key_or_url))
|
||
|
||
storage_key = self.normalize_storage_key(storage_key_or_url)
|
||
try:
|
||
signed = self.bucket.sign_url("GET", storage_key, expires_seconds)
|
||
logger.info(
|
||
"get_download_url: signed URL generated. key=%s url_prefix=%s",
|
||
storage_key[:80],
|
||
signed[:60],
|
||
)
|
||
return signed
|
||
except Exception:
|
||
logger.exception(
|
||
"get_download_url: sign_url failed, falling back to raw URL. key=%s",
|
||
storage_key[:200],
|
||
)
|
||
return self.get_url(storage_key)
|
||
|
||
# ── 浏览器直传 POST ────────────────────────────────────────────────
|
||
|
||
def create_direct_upload_post(
|
||
self,
|
||
storage_key: str,
|
||
content_type: str,
|
||
max_size_bytes: int,
|
||
expires_seconds: int,
|
||
) -> dict[str, object]:
|
||
"""创建浏览器直传 POST 表单。"""
|
||
if not self.access_key_id or not self.access_key_secret:
|
||
raise RuntimeError("OSS storage is not configured")
|
||
normalized_key = self.normalize_storage_key(storage_key)
|
||
if not normalized_key.startswith("uploads/"):
|
||
raise ValueError("direct upload key must be under uploads/")
|
||
|
||
expiration = (dt.datetime.now(dt.timezone.utc) + dt.timedelta(seconds=expires_seconds)).strftime(
|
||
"%Y-%m-%dT%H:%M:%S.000Z"
|
||
)
|
||
policy = {
|
||
"expiration": expiration,
|
||
"conditions": [
|
||
{"bucket": self.bucket_name},
|
||
{"key": normalized_key},
|
||
["content-length-range", 1, max_size_bytes],
|
||
[
|
||
"starts-with",
|
||
"$Content-Type",
|
||
content_type.split("/", 1)[0] + "/" if "/" in content_type else "",
|
||
],
|
||
],
|
||
}
|
||
encoded_policy = base64.b64encode(json.dumps(policy, separators=(",", ":")).encode("utf-8")).decode("ascii")
|
||
signature = base64.b64encode(
|
||
hmac.new(
|
||
self.access_key_secret.encode("utf-8"),
|
||
encoded_policy.encode("utf-8"),
|
||
hashlib.sha1,
|
||
).digest()
|
||
).decode("ascii")
|
||
|
||
return {
|
||
"url": self.public_url,
|
||
"method": "POST",
|
||
"storage_key": normalized_key,
|
||
"expires_at": expiration,
|
||
"fields": {
|
||
"key": normalized_key,
|
||
"OSSAccessKeyId": self.access_key_id,
|
||
"policy": encoded_policy,
|
||
"Signature": signature,
|
||
"success_action_status": "201",
|
||
"Content-Type": content_type,
|
||
},
|
||
}
|
||
|
||
# ── 文件操作 ───────────────────────────────────────────────────────
|
||
|
||
def delete_file(self, storage_key: str) -> None:
|
||
"""删除文件（不抛异常）。"""
|
||
if self.bucket is None:
|
||
return
|
||
try:
|
||
self.bucket.delete_object(storage_key)
|
||
except Exception as error:
|
||
logger.warning("Failed to delete file from OSS", extra={"storage_key": storage_key, "error": str(error)})
|
||
|
||
def file_exists(self, storage_key: str) -> bool:
|
||
"""检查文件是否存在。"""
|
||
if self.bucket is None:
|
||
return False
|
||
return self.bucket.object_exists(storage_key)
|
||
|
||
# ── Asset 路径解析（Worker 用）────────────────────────────────────
|
||
|
||
def resolve_asset_path(self, asset_id: str, work_dir: Union[str, Path]) -> Optional[Path]:
|
||
"""从 asset_id 解析到本地文件路径。
|
||
|
||
策略（按优先级）：
|
||
1. 本地绝对路径（在允许目录内）→ 直接返回
|
||
2. work_dir 缓存命中 → 返回缓存路径
|
||
3. 从OSS下载到缓存 → 返回下载路径
|
||
4. 全部失败 → None
|
||
|
||
从 oss_helpers.resolve_asset_path 合并而来。
|
||
"""
|
||
# 延迟导入，避免循环依赖
|
||
from video_processing.path_security import ( # type: ignore[import-not-found]
|
||
PathSecurityError,
|
||
get_allowed_local_dirs,
|
||
is_in_allowed_dirs,
|
||
sanitize_filename,
|
||
)
|
||
|
||
if not asset_id or not isinstance(asset_id, str):
|
||
return None
|
||
|
||
work_dir = Path(work_dir)
|
||
os.makedirs(work_dir, exist_ok=True)
|
||
|
||
# 空字节检测
|
||
if "\x00" in asset_id:
|
||
logger.warning("asset_id 包含空字节，拒绝: %s", asset_id[:50])
|
||
return None
|
||
|
||
# 1. 本地绝对路径 — 必须在允许的目录内
|
||
if asset_id.startswith("/") and os.path.exists(asset_id):
|
||
try:
|
||
resolved = Path(asset_id).resolve()
|
||
if is_in_allowed_dirs(resolved, get_allowed_local_dirs()):
|
||
return resolved
|
||
else:
|
||
logger.warning(
|
||
"本地素材路径不在允许目录内，拒绝: %s (allowed=%s)",
|
||
asset_id[:80],
|
||
get_allowed_local_dirs(),
|
||
)
|
||
return None
|
||
except (OSError, PathSecurityError):
|
||
return None
|
||
|
||
# 2. 缓存命中（SHA256 hash 防路径遍历）
|
||
cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
|
||
safe_name = sanitize_filename(cache_hash)
|
||
cached_path = work_dir / f"{safe_name}.mp4"
|
||
if cached_path.exists() and cached_path.stat().st_size > 0:
|
||
return cached_path
|
||
|
||
# 3. 从 OSS 下载（先标准化 key，防路径遍历注入）
|
||
safe_key = self.normalize_storage_key(asset_id)
|
||
if ".." in safe_key or safe_key.startswith("/"):
|
||
logger.warning("asset_id 包含路径遍历模式，拒绝下载: %s", asset_id[:80])
|
||
return None
|
||
|
||
if self.download_asset(safe_key, cached_path):
|
||
return cached_path
|
||
|
||
return None
|
||
|
||
def resolve_asset_ids_to_paths(
|
||
self,
|
||
asset_ids: list[str],
|
||
work_dir: Union[str, Path],
|
||
) -> dict[str, Path]:
|
||
"""批量解析 asset_id → 本地路径。"""
|
||
result: dict[str, Path] = {}
|
||
for aid in asset_ids:
|
||
local_path = self.resolve_asset_path(aid, work_dir)
|
||
if local_path:
|
||
result[aid] = local_path
|
||
return result
|
||
|
||
|
||
# ── 单例管理 ────────────────────────────────────────────────────────────
|
||
|
||
_storage_service: Optional[SharedStorageService] = None
|
||
|
||
|
||
def get_shared_storage_service() -> SharedStorageService:
|
||
"""获取统一存储服务单例。"""
|
||
global _storage_service
|
||
if _storage_service is None:
|
||
_storage_service = SharedStorageService()
|
||
_storage_service.diagnose()
|
||
return _storage_service
|
||
|
||
|
||
# 向后兼容别名
|
||
def get_storage_service() -> SharedStorageService:
|
||
"""向后兼容：返回统一存储服务。"""
|
||
return get_shared_storage_service()
|