小鱼开发
588c2236e6
- 背景图配置改为从七牛云 CDN 远程拉取，支持不发版新增图片 - 删除本地 public/bg/ 图片副本，减小包体积 - 封面制作：'换一组'按钮移至标签右侧、优化按钮样式、删除冗余提示 - 封面制作/视频合成右侧预览区分别添加'封面预览'/'视频预览'标题 - 素材匹配服务支持远程 fetch materials.json + reload API
145 lines
4.3 KiB
Python
145 lines
4.3 KiB
Python
"""
|
||
空镜素材服务
|
||
============
|
||
|
||
优先从远程 CDN 加载素材配置，失败时 fallback 到本地 JSON 文件。
|
||
支持手动 reload，新增素材只需更新远程配置文件即可生效。
|
||
duration 从文件名 `_{N}s_` 中自动解析。
|
||
"""
|
||
|
||
import json
|
||
import random
|
||
import re
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
from app.config import get_settings
|
||
|
||
# 正则：从文件名中提取时长，如 plumbing_10s_a23f8fcb.mp4 → 10
|
||
_DURATION_RE = re.compile(r"_(\d+)s_")
|
||
|
||
# 远程配置 URL（七牛云 CDN）
|
||
REMOTE_CONFIG_URL = "https://media.liche.cn/meijiaka-zy/materials.json"
|
||
|
||
# 配置缓存（启动时加载，运行时 reload 更新）
|
||
_keywords: dict[str, str] = {}
|
||
_materials: dict[str, list[dict]] = {}
|
||
|
||
|
||
def _get_config_path() -> Path:
|
||
"""获取本地配置文件绝对路径"""
|
||
return Path(__file__).resolve().parent.parent.parent / "config" / "materials.json"
|
||
|
||
|
||
def _parse_duration(url: str) -> float:
|
||
"""从 URL 文件名中解析时长（秒）"""
|
||
filename = url.split("/")[-1]
|
||
match = _DURATION_RE.search(filename)
|
||
if not match:
|
||
raise ValueError(f"无法从文件名解析时长: {filename}")
|
||
return float(match.group(1))
|
||
|
||
|
||
def _apply_config(data: dict) -> None:
|
||
"""将原始配置数据解析到内存缓存"""
|
||
global _keywords, _materials
|
||
|
||
_keywords = data.get("keywords", {})
|
||
raw_materials = data.get("materials", {})
|
||
|
||
_materials = {}
|
||
for slug, entries in raw_materials.items():
|
||
parsed = []
|
||
for entry in entries:
|
||
url = entry["url"]
|
||
duration = _parse_duration(url)
|
||
parsed.append({"url": url, "duration": duration})
|
||
_materials[slug] = parsed
|
||
|
||
|
||
def load_config() -> None:
|
||
"""加载素材配置：优先远程 CDN，fallback 本地文件"""
|
||
global _keywords, _materials
|
||
|
||
# 1. 尝试从远程 CDN 加载
|
||
try:
|
||
with httpx.Client(timeout=10.0) as client:
|
||
response = client.get(REMOTE_CONFIG_URL)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
_apply_config(data)
|
||
return
|
||
except Exception as e:
|
||
# 远程加载失败，记录日志后继续尝试本地
|
||
import logging
|
||
|
||
logging.getLogger(__name__).warning(f"远程素材配置加载失败: {e}，尝试本地 fallback")
|
||
|
||
# 2. Fallback 到本地文件
|
||
config_path = _get_config_path()
|
||
if not config_path.exists():
|
||
raise FileNotFoundError(f"素材配置文件不存在: {config_path}")
|
||
|
||
with open(config_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
|
||
_apply_config(data)
|
||
|
||
|
||
def reload_config() -> bool:
|
||
"""手动重新加载配置（用于更新素材后即时生效）"""
|
||
try:
|
||
load_config()
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def match_material(
|
||
scene: str,
|
||
required_duration: float,
|
||
exclude_urls: list[str] | None = None,
|
||
strict: bool = False,
|
||
) -> dict | None:
|
||
"""
|
||
根据场景描述和所需时长匹配空镜素材
|
||
|
||
策略：
|
||
1. 严格匹配分类（scene 必须完全匹配 keywords 中的关键词）
|
||
2. 过滤掉时长小于 required_duration 的素材
|
||
3. 从剩余素材中排除已使用的，随机选取
|
||
|
||
Args:
|
||
scene: 分镜场景描述
|
||
required_duration: 所需时长（秒）
|
||
exclude_urls: 已使用的素材 URL 列表（避免重复）
|
||
|
||
Returns:
|
||
匹配到的素材 {url, duration}，无匹配返回 None
|
||
"""
|
||
exclude_urls = exclude_urls or []
|
||
for keyword, slug in _keywords.items():
|
||
if keyword == scene:
|
||
all_materials = _materials.get(slug, [])
|
||
if not all_materials:
|
||
return None
|
||
|
||
# 1. 过滤掉时长小于 required_duration 的素材
|
||
matching = [m for m in all_materials if m["duration"] >= required_duration]
|
||
if not matching:
|
||
return None
|
||
|
||
# 2. 排除已使用的，从中随机选
|
||
unused = [m for m in matching if m["url"] not in exclude_urls]
|
||
if unused:
|
||
return random.choice(unused)
|
||
|
||
# 3. 严格模式下不允许返回已排除的素材
|
||
if strict:
|
||
return None
|
||
|
||
# 4. 非严格模式：全部用完则允许重复
|
||
return random.choice(matching)
|
||
|
||
return None
|