Compare commits

...

17 Commits

Author SHA1 Message Date
xiaoxia 2023ddc8cf fix: 恢复 LayerConfig.tsx 主入口,re-export 子目录保持路径兼容
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 6s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m8s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m2s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 1m45s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 44s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 35s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 1m49s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 51s
AI Code Review / AI Code Review (pull_request) Successful in 1m28s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 48s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m46s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 15m15s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 25s
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 20s
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1220h39m40s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 1221h13m58s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1221h14m0s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1221h13m59s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1222h58m2s
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 1222h58m21s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1222h58m4s
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 1222h58m23s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 1222h58m25s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1224h15m59s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1224h16m3s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1221h46m45s
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 1223h31m4s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1224h48m46s
2026-07-27 12:41:35 +08:00
xiaoxia 7ea0a1397e Merge branch 'develop' into refactor/layer-config-split
CI/CD Pipeline / Check if frontend-only change (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Lint (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build API Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Web Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Staging API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
AI Code Review / AI Code Review (pull_request) Has been cancelled
Preview Deploy / Deploy Preview Environment (pull_request) Has been cancelled
PR Automation / Auto Approve on CI Green (pull_request) Successful in 1m3s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 58s
2026-07-27 12:34:17 +08:00
CI Bot 7936be3339 style: auto-format with black + isort + prettier
CI/CD Pipeline / Frontend Lint (push) Successful in 57s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 2m10s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m18s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 2m29s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m18s
CI/CD Pipeline / Validate - Code Quality (push) Failing after 3m28s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 5m16s
CI/CD Pipeline / Integration Tests (push) Successful in 3m51s
CI/CD Pipeline / Unit Tests (push) Successful in 9m12s
CI/CD Pipeline / Build Staging API Image (push) Successful in 14m13s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m45s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 40s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 1m51s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m3s
CI/CD Pipeline / Canary Release to Production (push) Failing after 1227h50m10s
CI/CD Pipeline / Production Browser E2E (push) Failing after 1228h7m30s
CI/CD Pipeline / Build Production API Image (push) Failing after 1228h9m44s
CI/CD Pipeline / Deploy Production (push) Failing after 1228h8m31s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 1228h10m40s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 1228h9m40s
CI/CD Pipeline / PR Build Web Image (push) Failing after 1228h10m41s
CI/CD Pipeline / PR Build API Image (push) Failing after 1228h10m41s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 1228h10m42s
CI/CD Pipeline / Build Production Web Image (push) Failing after 1228h42m27s
2026-07-27 03:08:54 +00:00
xiaoxia d2c067e9bd test(wave121): 抽离xfade_builder领域模型 + 77单测
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m55s
CI/CD Pipeline / Validate - Code Quality (push) Failing after 4m21s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 2m35s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 2m25s
CI/CD Pipeline / Unit Tests (push) Successful in 9m9s
CI/CD Pipeline / Integration Tests (push) Successful in 3m0s
CI/CD Pipeline / Frontend Lint (push) Successful in 2m44s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 2m13s
CI/CD Pipeline / Build Staging API Image (push) Successful in 12m53s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m34s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 5m7s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 54s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 2m23s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m44s
CI/CD Pipeline / Canary Release to Production (push) Failing after 1228h3m34s
CI/CD Pipeline / Production Browser E2E (push) Failing after 1228h20m56s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 1228h21m2s
CI/CD Pipeline / Build Production Web Image (push) Failing after 1228h21m3s
CI/CD Pipeline / Build Production API Image (push) Failing after 1228h21m4s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 1228h23m39s
CI/CD Pipeline / PR Build Web Image (push) Failing after 1228h23m39s
CI/CD Pipeline / PR Build API Image (push) Failing after 1228h23m39s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 1228h23m40s
CI/CD Pipeline / Deploy Production (push) Failing after 1228h53m43s
2026-07-27 11:04:17 +08:00
CI Bot 925b365d6a style: auto-format with black + isort + prettier
CI/CD Pipeline / Frontend Lint (push) Successful in 57s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m22s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m23s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m56s
CI/CD Pipeline / Validate - Code Quality (push) Failing after 2m56s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m21s
CI/CD Pipeline / Integration Tests (push) Successful in 2m3s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 5m22s
CI/CD Pipeline / Unit Tests (push) Successful in 7m2s
CI/CD Pipeline / Build Staging API Image (push) Successful in 14m26s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m42s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 37s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 3m8s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m7s
CI/CD Pipeline / Canary Release to Production (push) Failing after 1228h35m18s
CI/CD Pipeline / Production Browser E2E (push) Failing after 1228h50m27s
CI/CD Pipeline / Deploy Production (push) Failing after 1228h52m51s
CI/CD Pipeline / PR Build Web Image (push) Failing after 1228h55m41s
CI/CD Pipeline / PR Build API Image (push) Failing after 1228h55m43s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 1228h54m17s
CI/CD Pipeline / Build Production Web Image (push) Failing after 1228h54m19s
CI/CD Pipeline / Build Production API Image (push) Failing after 1228h54m21s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 1228h55m45s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 1229h28m24s
2026-07-27 02:19:20 +00:00
xiaoxia 7c7f33fbd6 refactor(subscription): 拆分 UpgradeSubscription 页面,抽离 Hook + UI 组件 (#1020)
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m58s
CI/CD Pipeline / Validate - Code Quality (push) Failing after 1m33s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 44s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m1s
CI/CD Pipeline / Unit Tests (push) Successful in 6m19s
CI/CD Pipeline / Integration Tests (push) Successful in 2m17s
CI/CD Pipeline / Frontend Lint (push) Failing after 40s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m35s
CI/CD Pipeline / Build Staging API Image (push) Successful in 13m4s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m15s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 5m13s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 58s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 2m22s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m55s
CI/CD Pipeline / Canary Release to Production (push) Failing after 1228h47m52s
CI/CD Pipeline / Production Browser E2E (push) Failing after 1229h6m28s
CI/CD Pipeline / Build Production API Image (push) Failing after 1229h8m23s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 1229h8m52s
CI/CD Pipeline / PR Build API Image (push) Failing after 1229h8m54s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 1229h10m29s
CI/CD Pipeline / Deploy Production (push) Failing after 1229h6m35s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 1229h8m19s
CI/CD Pipeline / Build Production Web Image (push) Failing after 1229h41m6s
CI/CD Pipeline / PR Build Web Image (push) Failing after 1229h41m38s
2026-07-27 10:09:26 +08:00
xiaoxia b891eeab43 refactor(templates): 拆分 MyTemplates 页面,抽离数据 Hook + 卡片组件 (#1019) 2026-07-27 10:09:26 +08:00
xiaoxia dfed224b7d test(wave120): 抽离ass_subtitle_builder领域模型 + 68单测 (#1016) 2026-07-27 10:09:14 +08:00
xiaoxia 9a849d319e test(wave119): 抽离noise_reduction_config领域模型 + 39单测 (#1014) 2026-07-27 10:09:14 +08:00
xiaoxia 07bbe7ee01 test(wave118): 抽离chroma_key_config领域模型 + 45单测 (#1013) 2026-07-27 10:09:14 +08:00
CI Bot af7088a549 style: auto-format with black + isort + prettier
CI/CD Pipeline / Canary Release to Production (push) Failing after 1229h6m26s
CI/CD Pipeline / Production Browser E2E (push) Failing after 1229h11m58s
CI/CD Pipeline / Deploy Production (push) Failing after 1229h14m11s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 1229h31m3s
CI/CD Pipeline / PR Build Web Image (push) Failing after 1229h31m35s
CI/CD Pipeline / Build Production Web Image (push) Failing after 1229h31m3s
CI/CD Pipeline / PR Build API Image (push) Failing after 1229h31m37s
CI/CD Pipeline / Build Production API Image (push) Failing after 1229h31m5s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 1229h32m47s
CI/CD Pipeline / Validate - Code Quality (push) Failing after 1m40s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m1s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 54s
CI/CD Pipeline / Unit Tests (push) Successful in 5m23s
CI/CD Pipeline / Integration Tests (push) Successful in 2m17s
CI/CD Pipeline / Frontend Lint (push) Successful in 47s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 52s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 1230h4m18s
CI/CD Pipeline / Build Staging API Image (push) Successful in 17m13s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m41s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 7m42s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m35s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 1m50s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m1s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 51s
2026-07-27 01:19:35 +00:00
xiaoxia 9f61f76019 refactor(editing-planner): 拆分 LayerConfig 组件,按职责拆为 3 个子面板
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 28s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m2s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m0s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 40s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 1m47s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 48s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 1m12s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m19s
Preview Deploy / Deploy Preview Environment (pull_request) Failing after 53s
AI Code Review / AI Code Review (pull_request) Successful in 1m42s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 2m41s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 8m5s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 42s
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1229h13m40s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 1229h25m42s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1229h25m46s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1229h25m58s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1229h41m22s
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 1229h44m48s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1229h41m24s
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 1229h44m50s
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 1229h44m52s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 1229h44m54s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1230h1m35s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1230h1m36s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1230h1m38s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1229h58m28s
主文件 286→55 行(-81%),拆分为:
- PipPreview: 图层迷你预览
- LayerPositionSize: 素材类型/URL/位置/尺寸/圆角/透明度
- LayerTimingAnimation: 时间/入场动画/滑入方向
- index: 组合入口

导入路径保持向后兼容
2026-07-27 09:17:16 +08:00
xiaoxia 7d75ee6586 refactor(editing-planner): 拆分 useEditPlanClips Hook,按职责拆为子模块 (#1007)
CI/CD Pipeline / Validate - Code Quality (push) Failing after 3m6s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m6s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m1s
CI/CD Pipeline / Unit Tests (push) Successful in 5m42s
CI/CD Pipeline / Integration Tests (push) Successful in 1m58s
CI/CD Pipeline / Frontend Lint (push) Successful in 45s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m8s
CI/CD Pipeline / Build Staging API Image (push) Successful in 20m24s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m43s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 7m31s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m57s
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Failing after 1229h26m0s
CI/CD Pipeline / Deploy Production (push) Failing after 1229h44m58s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 1230h3m29s
CI/CD Pipeline / Build Production Web Image (push) Failing after 1230h3m31s
CI/CD Pipeline / Build Production API Image (push) Failing after 1230h3m33s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 1230h11m0s
CI/CD Pipeline / PR Build API Image (push) Failing after 1230h11m4s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 1230h11m47s
CI/CD Pipeline / PR Build Web Image (push) Failing after 1230h43m46s
2026-07-27 08:47:23 +08:00
xiaoxia 83bf454daf refactor(generate): 拆分 useStep4Title Hook,按职责拆为 AI生成/历史/模板 (#1004) 2026-07-27 08:47:23 +08:00
xiaoxia ea04bb8525 test(wave116): 抽离trim_config领域模型 + 64单测 (#1005) 2026-07-27 08:47:23 +08:00
xiaoxia d537dd2ed0 test(wave115): 抽离watermark_config领域模型 + 76单测 (#1001)
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
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
2026-07-27 08:44:53 +08:00
CI Bot 18590e22e5 style: auto-format with black + isort + prettier
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
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
2026-07-27 00:03:46 +00:00
42 changed files with 5373 additions and 2213 deletions
+4 -6
View File
@@ -16,12 +16,10 @@ import json
import logging
from typing import Any, Dict, List, Optional
from packages.domain.ai_parsing import (
generate_titles_fallback as _generate_titles_fallback_base,
keyword_match_fallback as _semantic_match_fallback_base,
parse_semantic_match_response as _parse_semantic_match_base,
parse_titles_from_response as _parse_titles_from_response,
)
from packages.domain.ai_parsing import generate_titles_fallback as _generate_titles_fallback_base
from packages.domain.ai_parsing import keyword_match_fallback as _semantic_match_fallback_base
from packages.domain.ai_parsing import parse_semantic_match_response as _parse_semantic_match_base
from packages.domain.ai_parsing import parse_titles_from_response as _parse_titles_from_response
from packages.shared.ai_client import get_doubao_client
logger = logging.getLogger(__name__)
@@ -90,8 +90,11 @@ const EditingPlanner: React.FC = () => {
}
/* ── 配音素材 ── */
const { voiceMaterials, loading: voiceMaterialsLoading, refetch: refetchVoiceMaterials } =
useVoiceMaterials()
const {
voiceMaterials,
loading: voiceMaterialsLoading,
refetch: refetchVoiceMaterials,
} = useVoiceMaterials()
/* ── 派生计算 ── */
const totalDuration = clips.reduce((sum, c) => sum + c.duration, 0)
+3 -284
View File
@@ -1,286 +1,5 @@
/**
* 混剪单图层配置区
* LayerConfig 入口(向后兼容)
* 实际实现位于 ./layer-config/ 目录
*/
import React from "react"
import type {
PipLayer,
PipAnimType,
PipSlideDirection,
PipGridPosition,
} from "@/pages/editing-planner/types"
import {
GRID_POSITIONS,
ANIM_OPTIONS,
SLIDE_DIR_OPTIONS,
LAYER_COLORS,
} from "@/pages/editing-planner/constants/pipConfig"
interface LayerConfigProps {
layer: PipLayer | null
layers: PipLayer[]
totalDuration: number
onUpdate: (id: string, partial: Partial<PipLayer>) => void
onGridClick: (pos: PipGridPosition) => void
onWidthChange: (val: number) => void
onHeightChange: (val: number) => void
}
const LayerConfig: React.FC<LayerConfigProps> = ({
layer,
layers,
totalDuration,
onUpdate,
onGridClick,
onWidthChange,
onHeightChange,
}) => {
if (!layer) {
return (
<div className="pip-config-area">
<div className="pip-config-empty">选择或添加图层以配置</div>
</div>
)
}
return (
<div className="pip-config-area">
{/* ── 迷你预览 ── */}
<div className="pip-preview-box">
{layers.map((l, idx) => (
<div
key={l.id}
className={`pip-preview-layer${layer.id === l.id ? " selected" : ""}`}
style={{
left: `${l.x}%`,
top: `${l.y}%`,
width: `${l.width}%`,
height: `${l.height}%`,
background: LAYER_COLORS[idx % LAYER_COLORS.length],
opacity: l.opacity / 100,
borderRadius: `${l.border_radius}%`,
}}
>
<span className="pip-preview-label">{l.name}</span>
</div>
))}
</div>
{/* ── 素材类型 ── */}
<div className="pip-field">
<label className="pip-field-label">素材类型</label>
<div className="pip-type-btns">
<button
className={`pip-type-btn${layer.material_type === "image" ? " active" : ""}`}
onClick={() => onUpdate(layer.id, { material_type: "image" })}
>
🖼️ 图片
</button>
<button
className={`pip-type-btn${layer.material_type === "video" ? " active" : ""}`}
onClick={() => onUpdate(layer.id, { material_type: "video" })}
>
🎬 视频
</button>
</div>
</div>
{/* ── 素材 URL ── */}
<div className="pip-field">
<label className="pip-field-label">
{layer.material_type === "image" ? "图片" : "视频"} URL
</label>
<input
className="pip-input"
type="text"
placeholder={
layer.material_type === "image"
? "https://example.com/image.png"
: "https://example.com/video.mp4"
}
value={layer.material_url}
onChange={(e) => onUpdate(layer.id, { material_url: e.target.value })}
/>
</div>
{/* ── 位置:九宫格 + 坐标 ── */}
<div className="pip-field">
<label className="pip-field-label">位置</label>
<div style={{ display: "flex", gap: 16, alignItems: "flex-start" }}>
<div className="pip-grid">
{GRID_POSITIONS.map((pos) => (
<button
key={pos}
className={`pip-grid-btn${layer.grid_position === pos ? " active" : ""}`}
onClick={() => onGridClick(pos)}
>
<span className="pip-grid-dot" />
</button>
))}
</div>
<div className="pip-field-row" style={{ flex: 1 }}>
<div>
<label className="pip-field-label">X (%)</label>
<input
className="pip-number"
type="number"
min={0}
max={100}
value={layer.x}
onChange={(e) => onUpdate(layer.id, { x: Number(e.target.value) })}
/>
</div>
<div>
<label className="pip-field-label">Y (%)</label>
<input
className="pip-number"
type="number"
min={0}
max={100}
value={layer.y}
onChange={(e) => onUpdate(layer.id, { y: Number(e.target.value) })}
/>
</div>
</div>
</div>
</div>
{/* ── 尺寸 ── */}
<div className="pip-field">
<label className="pip-field-label">尺寸</label>
<div className="pip-slider-row">
<span style={{ fontSize: 12, color: "#999", width: 20 }}>宽</span>
<input
className="pip-slider"
type="range"
min={10}
max={80}
value={layer.width}
onChange={(e) => onWidthChange(Number(e.target.value))}
/>
<span className="pip-slider-value">{layer.width}%</span>
</div>
<div className="pip-slider-row" style={{ marginTop: 6 }}>
<span style={{ fontSize: 12, color: "#999", width: 20 }}>高</span>
<input
className="pip-slider"
type="range"
min={10}
max={80}
value={layer.height}
onChange={(e) => onHeightChange(Number(e.target.value))}
/>
<span className="pip-slider-value">{layer.height}%</span>
</div>
<div
className="pip-lock-row"
style={{ marginTop: 6 }}
onClick={() => onUpdate(layer.id, { aspect_lock: !layer.aspect_lock })}
>
<span className="pip-lock-icon">{layer.aspect_lock ? "🔒" : "🔓"}</span>
<span>{layer.aspect_lock ? "已锁定比例" : "锁定宽高比"}</span>
</div>
</div>
{/* ── 圆角 ── */}
<div className="pip-field">
<label className="pip-field-label">圆角</label>
<div className="pip-slider-row">
<input
className="pip-slider"
type="range"
min={0}
max={50}
value={layer.border_radius}
onChange={(e) => onUpdate(layer.id, { border_radius: Number(e.target.value) })}
/>
<span className="pip-slider-value">{layer.border_radius}%</span>
</div>
</div>
{/* ── 透明度 ── */}
<div className="pip-field">
<label className="pip-field-label">透明度</label>
<div className="pip-slider-row">
<input
className="pip-slider"
type="range"
min={0}
max={100}
value={layer.opacity}
onChange={(e) => onUpdate(layer.id, { opacity: Number(e.target.value) })}
/>
<span className="pip-slider-value">{layer.opacity}%</span>
</div>
</div>
{/* ── 时间 ── */}
<div className="pip-field">
<label className="pip-field-label">时间</label>
<div className="pip-field-row">
<div>
<label className="pip-field-label">开始 (s)</label>
<input
className="pip-number"
type="number"
min={0}
max={totalDuration || 999}
step={0.1}
value={layer.start_time}
onChange={(e) => onUpdate(layer.id, { start_time: Number(e.target.value) })}
/>
</div>
<div>
<label className="pip-field-label">持续 (s)</label>
<input
className="pip-number"
type="number"
min={0.1}
max={totalDuration || 999}
step={0.1}
value={layer.duration}
onChange={(e) => onUpdate(layer.id, { duration: Number(e.target.value) })}
/>
</div>
</div>
</div>
{/* ── 入场动画 ── */}
<div className="pip-field">
<label className="pip-field-label">入场动画</label>
<select
className="pip-select"
value={layer.animation}
onChange={(e) => onUpdate(layer.id, { animation: e.target.value as PipAnimType })}
>
{ANIM_OPTIONS.map((opt) => (
<option key={opt.value} value={opt.value}>
{opt.label}
</option>
))}
</select>
</div>
{/* 滑入方向(仅 slide_in 时显示) */}
{layer.animation === "slide_in" && (
<div className="pip-field">
<label className="pip-field-label">滑入方向</label>
<select
className="pip-select"
value={layer.slide_direction}
onChange={(e) =>
onUpdate(layer.id, { slide_direction: e.target.value as PipSlideDirection })
}
>
{SLIDE_DIR_OPTIONS.map((opt) => (
<option key={opt.value} value={opt.value}>
{opt.label}
</option>
))}
</select>
</div>
)}
</div>
)
}
export default LayerConfig
export { default } from "./layer-config"
@@ -0,0 +1,172 @@
import React from "react"
import type { PipLayer, PipGridPosition } from "@/pages/editing-planner/types"
import { GRID_POSITIONS } from "@/pages/editing-planner/constants/pipConfig"
interface LayerPositionSizeProps {
layer: PipLayer
onUpdate: (id: string, partial: Partial<PipLayer>) => void
onGridClick: (pos: PipGridPosition) => void
onWidthChange: (val: number) => void
onHeightChange: (val: number) => void
}
/**
* 图层位置与尺寸配置面板
*/
export const LayerPositionSize: React.FC<LayerPositionSizeProps> = ({
layer,
onUpdate,
onGridClick,
onWidthChange,
onHeightChange,
}) => (
<>
{/* 素材类型 */}
<div className="pip-field">
<label className="pip-field-label">素材类型</label>
<div className="pip-type-btns">
<button
className={`pip-type-btn${layer.material_type === "image" ? " active" : ""}`}
onClick={() => onUpdate(layer.id, { material_type: "image" })}
>
🖼️ 图片
</button>
<button
className={`pip-type-btn${layer.material_type === "video" ? " active" : ""}`}
onClick={() => onUpdate(layer.id, { material_type: "video" })}
>
🎬 视频
</button>
</div>
</div>
{/* 素材 URL */}
<div className="pip-field">
<label className="pip-field-label">
{layer.material_type === "image" ? "图片" : "视频"} URL
</label>
<input
className="pip-input"
type="text"
placeholder={
layer.material_type === "image"
? "https://example.com/image.png"
: "https://example.com/video.mp4"
}
value={layer.material_url}
onChange={(e) => onUpdate(layer.id, { material_url: e.target.value })}
/>
</div>
{/* 位置:九宫格 + 坐标 */}
<div className="pip-field">
<label className="pip-field-label">位置</label>
<div style={{ display: "flex", gap: 16, alignItems: "flex-start" }}>
<div className="pip-grid">
{GRID_POSITIONS.map((pos) => (
<button
key={pos}
className={`pip-grid-btn${layer.grid_position === pos ? " active" : ""}`}
onClick={() => onGridClick(pos)}
>
<span className="pip-grid-dot" />
</button>
))}
</div>
<div className="pip-field-row" style={{ flex: 1 }}>
<div>
<label className="pip-field-label">X (%)</label>
<input
className="pip-number"
type="number"
min={0}
max={100}
value={layer.x}
onChange={(e) => onUpdate(layer.id, { x: Number(e.target.value) })}
/>
</div>
<div>
<label className="pip-field-label">Y (%)</label>
<input
className="pip-number"
type="number"
min={0}
max={100}
value={layer.y}
onChange={(e) => onUpdate(layer.id, { y: Number(e.target.value) })}
/>
</div>
</div>
</div>
</div>
{/* 尺寸 */}
<div className="pip-field">
<label className="pip-field-label">尺寸</label>
<div className="pip-slider-row">
<span style={{ fontSize: 12, color: "#999", width: 20 }}>宽</span>
<input
className="pip-slider"
type="range"
min={10}
max={80}
value={layer.width}
onChange={(e) => onWidthChange(Number(e.target.value))}
/>
<span className="pip-slider-value">{layer.width}%</span>
</div>
<div className="pip-slider-row" style={{ marginTop: 6 }}>
<span style={{ fontSize: 12, color: "#999", width: 20 }}>高</span>
<input
className="pip-slider"
type="range"
min={10}
max={80}
value={layer.height}
onChange={(e) => onHeightChange(Number(e.target.value))}
/>
<span className="pip-slider-value">{layer.height}%</span>
</div>
<div
className="pip-lock-row"
style={{ marginTop: 6 }}
onClick={() => onUpdate(layer.id, { aspect_lock: !layer.aspect_lock })}
>
<span className="pip-lock-icon">{layer.aspect_lock ? "🔒" : "🔓"}</span>
<span>{layer.aspect_lock ? "已锁定比例" : "锁定宽高比"}</span>
</div>
</div>
{/* 圆角 */}
<div className="pip-field">
<label className="pip-field-label">圆角</label>
<div className="pip-slider-row">
<input
className="pip-slider"
type="range"
min={0}
max={50}
value={layer.border_radius}
onChange={(e) => onUpdate(layer.id, { border_radius: Number(e.target.value) })}
/>
<span className="pip-slider-value">{layer.border_radius}%</span>
</div>
</div>
{/* 透明度 */}
<div className="pip-field">
<label className="pip-field-label">透明度</label>
<div className="pip-slider-row">
<input
className="pip-slider"
type="range"
min={0}
max={100}
value={layer.opacity}
onChange={(e) => onUpdate(layer.id, { opacity: Number(e.target.value) })}
/>
<span className="pip-slider-value">{layer.opacity}%</span>
</div>
</div>
</>
)
@@ -0,0 +1,87 @@
import React from "react"
import type { PipLayer, PipAnimType, PipSlideDirection } from "@/pages/editing-planner/types"
import { ANIM_OPTIONS, SLIDE_DIR_OPTIONS } from "@/pages/editing-planner/constants/pipConfig"
interface LayerTimingAnimationProps {
layer: PipLayer
totalDuration: number
onUpdate: (id: string, partial: Partial<PipLayer>) => void
}
/**
* 图层时间与动画配置面板
*/
export const LayerTimingAnimation: React.FC<LayerTimingAnimationProps> = ({
layer,
totalDuration,
onUpdate,
}) => (
<>
{/* 时间 */}
<div className="pip-field">
<label className="pip-field-label">时间</label>
<div className="pip-field-row">
<div>
<label className="pip-field-label">开始 (s)</label>
<input
className="pip-number"
type="number"
min={0}
max={totalDuration || 999}
step={0.1}
value={layer.start_time}
onChange={(e) => onUpdate(layer.id, { start_time: Number(e.target.value) })}
/>
</div>
<div>
<label className="pip-field-label">持续 (s)</label>
<input
className="pip-number"
type="number"
min={0.1}
max={totalDuration || 999}
step={0.1}
value={layer.duration}
onChange={(e) => onUpdate(layer.id, { duration: Number(e.target.value) })}
/>
</div>
</div>
</div>
{/* 入场动画 */}
<div className="pip-field">
<label className="pip-field-label">入场动画</label>
<select
className="pip-select"
value={layer.animation}
onChange={(e) => onUpdate(layer.id, { animation: e.target.value as PipAnimType })}
>
{ANIM_OPTIONS.map((opt) => (
<option key={opt.value} value={opt.value}>
{opt.label}
</option>
))}
</select>
</div>
{/* 滑入方向(仅 slide_in 时显示) */}
{layer.animation === "slide_in" && (
<div className="pip-field">
<label className="pip-field-label">滑入方向</label>
<select
className="pip-select"
value={layer.slide_direction}
onChange={(e) =>
onUpdate(layer.id, { slide_direction: e.target.value as PipSlideDirection })
}
>
{SLIDE_DIR_OPTIONS.map((opt) => (
<option key={opt.value} value={opt.value}>
{opt.label}
</option>
))}
</select>
</div>
)}
</>
)
@@ -0,0 +1,33 @@
import React from "react"
import type { PipLayer } from "@/pages/editing-planner/types"
import { LAYER_COLORS } from "@/pages/editing-planner/constants/pipConfig"
interface PipPreviewProps {
layers: PipLayer[]
selectedId: string
}
/**
* PIP 图层迷你预览组件
*/
export const PipPreview: React.FC<PipPreviewProps> = ({ layers, selectedId }) => (
<div className="pip-preview-box">
{layers.map((l, idx) => (
<div
key={l.id}
className={`pip-preview-layer${selectedId === l.id ? " selected" : ""}`}
style={{
left: `${l.x}%`,
top: `${l.y}%`,
width: `${l.width}%`,
height: `${l.height}%`,
background: LAYER_COLORS[idx % LAYER_COLORS.length],
opacity: l.opacity / 100,
borderRadius: `${l.border_radius}%`,
}}
>
<span className="pip-preview-label">{l.name}</span>
</div>
))}
</div>
)
@@ -0,0 +1,57 @@
/**
* 混剪单图层配置区
*/
import React from "react"
import type { PipLayer, PipGridPosition } from "@/pages/editing-planner/types"
import { PipPreview } from "./PipPreview"
import { LayerPositionSize } from "./LayerPositionSize"
import { LayerTimingAnimation } from "./LayerTimingAnimation"
interface LayerConfigProps {
layer: PipLayer | null
layers: PipLayer[]
totalDuration: number
onUpdate: (id: string, partial: Partial<PipLayer>) => void
onGridClick: (pos: PipGridPosition) => void
onWidthChange: (val: number) => void
onHeightChange: (val: number) => void
}
const LayerConfig: React.FC<LayerConfigProps> = ({
layer,
layers,
totalDuration,
onUpdate,
onGridClick,
onWidthChange,
onHeightChange,
}) => {
if (!layer) {
return (
<div className="pip-config-area">
<div className="pip-config-empty">选择或添加图层以配置</div>
</div>
)
}
return (
<div className="pip-config-area">
{/* 迷你预览 */}
<PipPreview layers={layers} selectedId={layer.id} />
{/* 位置与尺寸 */}
<LayerPositionSize
layer={layer}
onUpdate={onUpdate}
onGridClick={onGridClick}
onWidthChange={onWidthChange}
onHeightChange={onHeightChange}
/>
{/* 时间与动画 */}
<LayerTimingAnimation layer={layer} totalDuration={totalDuration} onUpdate={onUpdate} />
</div>
)
}
export default LayerConfig
@@ -0,0 +1,83 @@
import { useUndoRedo } from "../useUndoRedo"
import type { EditPlanClip } from "@/api/template-editor"
import { useEditPlanClipList } from "./useEditPlanClipList"
import { useEditPlanClipMutations } from "./useEditPlanClipMutations"
/**
* 模板片段管理 Hook
* 对接后端 PR#389 片段 CRUD API
*
* 功能:
* - 加载/刷新片段列表
* - 单个增删改查
* - 批量删除
* - 拖拽重排序
* - 从素材批量导入
* - 乐观更新 + 撤销重做
*/
export function useEditPlanClips(planId: string | undefined) {
// 列表数据 + 选中状态
const {
clips,
clipsTotal,
clipsLoading,
refetchClips,
selectedClipId,
setSelectedClipId,
selectedClip,
} = useEditPlanClipList(planId)
// CRUD 操作
const mutations = useEditPlanClipMutations({
planId,
selectedClipId,
setSelectedClipId,
clipsLength: clips.length,
})
// 本地撤销重做(供拖拽等即时操作使用)
const {
state: localClips,
set: setLocalClips,
undo,
redo,
canUndo,
canRedo,
reset: resetLocalClips,
} = useUndoRedo<EditPlanClip[]>([])
return {
// 数据
clips,
clipsTotal,
clipsLoading,
selectedClipId,
selectedClip,
// 选中
setSelectedClipId,
// 操作
addClip: mutations.addClip,
updateClip: mutations.updateClip,
removeClip: mutations.removeClip,
batchRemoveClips: mutations.batchRemoveClips,
reorderClips: mutations.reorderClips,
importFromAssets: mutations.importFromAssets,
refetchClips,
// 状态
isCreating: mutations.isCreating,
isUpdating: mutations.isUpdating,
isDeleting: mutations.isDeleting,
isReordering: mutations.isReordering,
isImporting: mutations.isImporting,
// 本地撤销重做
localClips,
setLocalClips,
undo,
redo,
canUndo,
canRedo,
resetLocalClips,
}
}
export default useEditPlanClips
@@ -0,0 +1,49 @@
import { useState } from "react"
import { useQuery } from "@tanstack/react-query"
import type { EditPlanClip } from "@/api/template-editor"
import { getEditPlanClips } from "@/api/template-editor"
const QUERY_KEY = "editPlanClips"
interface UseEditPlanClipListResult {
clips: EditPlanClip[]
clipsTotal: number
clipsLoading: boolean
refetchClips: () => void
selectedClipId: string | null
setSelectedClipId: (id: string | null) => void
selectedClip: EditPlanClip | null
}
/**
* 编辑计划片段列表 Hook
* 封装片段列表查询、选中状态
*/
export function useEditPlanClipList(planId: string | undefined): UseEditPlanClipListResult {
const {
data: clipListData,
isLoading: clipsLoading,
refetch: refetchClips,
} = useQuery({
queryKey: [QUERY_KEY, planId],
queryFn: () => getEditPlanClips(planId!, { limit: 500 }),
enabled: !!planId,
staleTime: 30_000,
})
const clips: EditPlanClip[] = clipListData?.items ?? []
const clipsTotal = clipListData?.total ?? 0
const [selectedClipId, setSelectedClipId] = useState<string | null>(null)
const selectedClip = clips.find((c) => c.id === selectedClipId) ?? null
return {
clips,
clipsTotal,
clipsLoading,
refetchClips,
selectedClipId,
setSelectedClipId,
selectedClip,
}
}
@@ -1,26 +1,12 @@
/**
* 模板片段管理 Hook
* 对接后端 PR#389 片段 CRUD API,替代原来的 config.segments 模式
*
* 功能:
* - 加载/刷新片段列表
* - 单个增删改查
* - 批量删除
* - 拖拽重排序
* - 从素材批量导入
* - 乐观更新 + 撤销重做
*/
import { useCallback, useState } from "react"
import { useCallback } from "react"
import { message } from "antd"
import { useQuery, useQueryClient, useMutation } from "@tanstack/react-query"
import { useQueryClient, useMutation } from "@tanstack/react-query"
import type {
EditPlanClip,
CreateEditPlanClipRequest,
UpdateEditPlanClipRequest,
ClipReorderItem,
} from "@/api/template-editor"
import {
getEditPlanClips,
createEditPlanClip,
updateEditPlanClip,
deleteEditPlanClip,
@@ -28,51 +14,37 @@ import {
batchDeleteEditPlanClips,
createClipsFromAssets,
} from "@/api/template-editor"
import { useUndoRedo } from "./useUndoRedo"
const QUERY_KEY = "editPlanClips"
export function useEditPlanClips(planId: string | undefined) {
interface UseEditPlanClipMutationsOptions {
planId: string | undefined
selectedClipId: string | null
setSelectedClipId: (id: string | null) => void
clipsLength: number
}
/**
* 编辑计划片段 CRUD Hook
* 封装创建、更新、删除、批量删除、重排序、素材导入等操作
*/
export function useEditPlanClipMutations({
planId,
selectedClipId,
setSelectedClipId,
clipsLength,
}: UseEditPlanClipMutationsOptions) {
const queryClient = useQueryClient()
/* ── 片段列表查询 ── */
const {
data: clipListData,
isLoading: clipsLoading,
refetch: refetchClips,
} = useQuery({
queryKey: [QUERY_KEY, planId],
queryFn: () => getEditPlanClips(planId!, { limit: 500 }),
enabled: !!planId,
staleTime: 30_000,
})
const clips: EditPlanClip[] = clipListData?.items ?? []
const clipsTotal = clipListData?.total ?? 0
/* ── 选中片段 ── */
const [selectedClipId, setSelectedClipId] = useState<string | null>(null)
const selectedClip = clips.find((c) => c.id === selectedClipId) ?? null
/* ── 本地撤销/重做(用于拖拽等即时操作的回退) ── */
const {
state: localClips,
set: setLocalClips,
undo,
redo,
canUndo,
canRedo,
reset: resetLocalClips,
} = useUndoRedo<EditPlanClip[]>([])
// 当服务端数据变化时同步本地
// 注意:实际使用时以服务端为准,本地仅用于拖拽等临时操作
const invalidate = () => {
queryClient.invalidateQueries({ queryKey: [QUERY_KEY, planId] })
}
/* ── 创建片段 ── */
const createMutation = useMutation({
mutationFn: (data: CreateEditPlanClipRequest) => createEditPlanClip(planId!, data),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: [QUERY_KEY, planId] })
invalidate()
message.success("片段已添加")
},
onError: () => {
@@ -83,10 +55,10 @@ export function useEditPlanClips(planId: string | undefined) {
const addClip = useCallback(
(data: Omit<CreateEditPlanClipRequest, "order"> & { order?: number }) => {
if (!planId) return
const order = data.order ?? clips.length
const order = data.order ?? clipsLength
createMutation.mutate({ ...data, order })
},
[planId, clips.length, createMutation],
[planId, clipsLength, createMutation],
)
/* ── 更新片段 ── */
@@ -94,7 +66,7 @@ export function useEditPlanClips(planId: string | undefined) {
mutationFn: ({ clipId, data }: { clipId: string; data: UpdateEditPlanClipRequest }) =>
updateEditPlanClip(planId!, clipId, data),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: [QUERY_KEY, planId] })
invalidate()
},
onError: () => {
message.error("更新片段失败")
@@ -113,7 +85,7 @@ export function useEditPlanClips(planId: string | undefined) {
const deleteMutation = useMutation({
mutationFn: (clipId: string) => deleteEditPlanClip(planId!, clipId),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: [QUERY_KEY, planId] })
invalidate()
message.success("片段已删除")
},
onError: () => {
@@ -129,14 +101,14 @@ export function useEditPlanClips(planId: string | undefined) {
}
deleteMutation.mutate(clipId)
},
[planId, selectedClipId, deleteMutation],
[planId, selectedClipId, setSelectedClipId, deleteMutation],
)
/* ── 批量删除 ── */
const batchDeleteMutation = useMutation({
mutationFn: (clipIds: string[]) => batchDeleteEditPlanClips(planId!, clipIds),
onSuccess: (res) => {
queryClient.invalidateQueries({ queryKey: [QUERY_KEY, planId] })
invalidate()
message.success(`已删除 ${res.deleted_count} 个片段`)
},
onError: () => {
@@ -152,19 +124,18 @@ export function useEditPlanClips(planId: string | undefined) {
}
batchDeleteMutation.mutate(clipIds)
},
[planId, selectedClipId, batchDeleteMutation],
[planId, selectedClipId, setSelectedClipId, batchDeleteMutation],
)
/* ── 重排序(拖拽结束后一次性提交) ── */
/* ── 重排序 ── */
const reorderMutation = useMutation({
mutationFn: (items: ClipReorderItem[]) => reorderEditPlanClips(planId!, items),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: [QUERY_KEY, planId] })
invalidate()
},
onError: () => {
message.error("排序失败")
// 失败后刷新回服务端状态
queryClient.invalidateQueries({ queryKey: [QUERY_KEY, planId] })
invalidate()
},
})
@@ -180,7 +151,7 @@ export function useEditPlanClips(planId: string | undefined) {
const importFromAssetsMutation = useMutation({
mutationFn: (assetIds: string[]) => createClipsFromAssets(planId!, assetIds),
onSuccess: (res) => {
queryClient.invalidateQueries({ queryKey: [QUERY_KEY, planId] })
invalidate()
message.success(`已导入 ${res.created_count} 个素材片段`)
},
onError: () => {
@@ -197,37 +168,16 @@ export function useEditPlanClips(planId: string | undefined) {
)
return {
// 数据
clips,
clipsTotal,
clipsLoading,
selectedClipId,
selectedClip,
// 选中
setSelectedClipId,
// 操作
addClip,
updateClip,
removeClip,
batchRemoveClips,
reorderClips,
importFromAssets,
refetchClips,
// 状态
isCreating: createMutation.isPending,
isUpdating: updateMutation.isPending,
isDeleting: deleteMutation.isPending,
isReordering: reorderMutation.isPending,
isImporting: importFromAssetsMutation.isPending,
// 本地撤销重做(供拖拽等场景使用)
localClips,
setLocalClips,
undo,
redo,
canUndo,
canRedo,
resetLocalClips,
}
}
export default useEditPlanClips
@@ -4,21 +4,13 @@
*/
/* 转场 */
export {
type TransitionType,
type TransitionConfig,
DEFAULT_TRANSITION,
} from "./transition"
export { type TransitionType, type TransitionConfig, DEFAULT_TRANSITION } from "./transition"
/* 调速 */
export { type SpeedConfig, DEFAULT_SPEED } from "./speed"
/* TTS 配音 */
export {
type TtsMode,
type TtsConfig,
DEFAULT_TTS_CONFIG,
} from "./tts"
export { type TtsMode, type TtsConfig, DEFAULT_TTS_CONFIG } from "./tts"
/* 裁剪 */
export { type TrimConfig } from "./trim"
@@ -80,11 +72,7 @@ export {
} from "./sticker"
/* 封面 */
export {
type CoverMode,
type CoverConfig,
DEFAULT_COVER_CONFIG,
} from "./cover"
export { type CoverMode, type CoverConfig, DEFAULT_COVER_CONFIG } from "./cover"
/* 片段数据 */
export { type ClipType, type ClipData } from "./clip"
@@ -93,7 +81,4 @@ export { type ClipType, type ClipData } from "./clip"
export { type TitleSettings } from "./title"
/* 字幕样式 */
export {
type SubtitleStyleConfig,
DEFAULT_SUBTITLE_STYLE,
} from "./subtitle"
export { type SubtitleStyleConfig, DEFAULT_SUBTITLE_STYLE } from "./subtitle"
@@ -1,251 +0,0 @@
/**
* Step 4 标题设置 Hook
* 封装 AI 标题生成、标题样式设置等逻辑
*/
import { useState, useCallback, useMemo } from "react"
import { message } from "antd"
import { useQuery } from "@tanstack/react-query"
import { getTitles } from "@/api/titles"
import { TITLE_PRESETS, AI_TITLE_TEMPLATES } from "../constants"
import type { TitleSettings } from "../types"
interface UseStep4TitleProps {
titleSettings: TitleSettings
onTitleSettingsChange: (settings: TitleSettings) => void
}
interface AiTitleItem {
title: string
highlight: string
style: "catchy" | "emotional" | "informative"
}
export function useStep4Title({ titleSettings, onTitleSettingsChange }: UseStep4TitleProps) {
/* ── 标题库 API ── */
const { data: userTitles = [] } = useQuery({
queryKey: ["titles"],
queryFn: () => getTitles(),
staleTime: 30_000,
})
/* ── AI 标题生成状态 ── */
const [aiTitleInput, setAiTitleInput] = useState("")
const [aiTitleGenerating, setAiTitleGenerating] = useState(false)
const [aiTitleResults, setAiTitleResults] = useState<AiTitleItem[]>([])
const [hasGeneratedTitles, setHasGeneratedTitles] = useState(false)
/* ── 辅助函数 ── */
const extractTopic = (text: string): string => {
const keywords = text
.replace(/[,。!?、,.!?]/g, " ")
.split(/\s+/)
.filter(Boolean)
if (keywords.length === 0) return "这个话题"
// 取前3个关键词组合
return keywords.slice(0, 3).join("")
}
const getActivePreset = (settings: TitleSettings): string | null => {
for (const p of TITLE_PRESETS) {
if (
settings.size === p.style.size &&
settings.color === p.style.color &&
settings.bold === p.style.bold &&
settings.italic === p.style.italic &&
settings.stroke === p.style.stroke &&
settings.shadow === p.style.shadow
) {
return p.key
}
}
return null
}
const activePreset = useMemo(() => getActivePreset(titleSettings), [titleSettings])
/* ── AI 标题生成 ── */
const handleGenerateAiTitles = useCallback(async () => {
if (!aiTitleInput.trim()) {
message.warning("请先输入视频描述或关键词")
return
}
setAiTitleGenerating(true)
setHasGeneratedTitles(true)
// 模拟 AI 生成延迟
await new Promise((resolve) => setTimeout(resolve, 1200))
const topic = extractTopic(aiTitleInput)
const results: AiTitleItem[] = []
const styles: Array<"catchy" | "emotional" | "informative"> = [
"catchy",
"emotional",
"informative",
]
styles.forEach((style) => {
const templates = AI_TITLE_TEMPLATES[style]
// 每种风格随机选2个
const shuffled = [...templates].sort(() => Math.random() - 0.5).slice(0, 2)
shuffled.forEach((tpl) => {
const title = tpl.replace(/\{topic\}/g, topic)
const highlights = {
catchy: "吸睛标题",
emotional: "情感共鸣",
informative: "知识干货",
}
results.push({
title,
highlight: highlights[style],
style,
})
})
})
// 打乱顺序
results.sort(() => Math.random() - 0.5)
setAiTitleResults(results)
setAiTitleGenerating(false)
}, [aiTitleInput])
const handleSelectAiTitle = useCallback(
(title: string) => {
onTitleSettingsChange({ ...titleSettings, title, aiAutoSelect: false })
message.success("已选用此标题")
},
[titleSettings, onTitleSettingsChange],
)
const handleRefreshAiTitles = useCallback(async () => {
if (!aiTitleInput.trim()) return
setAiTitleGenerating(true)
await new Promise((resolve) => setTimeout(resolve, 800))
// 重新生成一批
const topic = extractTopic(aiTitleInput)
const results: AiTitleItem[] = []
const styles: Array<"catchy" | "emotional" | "informative"> = [
"catchy",
"emotional",
"informative",
]
const highlights = { catchy: "吸睛标题", emotional: "情感共鸣", informative: "知识干货" }
styles.forEach((style) => {
const templates = AI_TITLE_TEMPLATES[style]
const shuffled = [...templates].sort(() => Math.random() - 0.5).slice(0, 2)
shuffled.forEach((tpl) => {
results.push({
title: tpl.replace(/\{topic\}/g, topic),
highlight: highlights[style],
style,
})
})
})
results.sort(() => Math.random() - 0.5)
setAiTitleResults(results)
setAiTitleGenerating(false)
}, [aiTitleInput])
/* ── 标题设置更新 ── */
const updateTitle = useCallback(
(title: string) => {
onTitleSettingsChange({ ...titleSettings, title })
},
[titleSettings, onTitleSettingsChange],
)
const toggleAiAutoSelect = useCallback(() => {
onTitleSettingsChange({ ...titleSettings, aiAutoSelect: !titleSettings.aiAutoSelect })
}, [titleSettings, onTitleSettingsChange])
const updatePosition = useCallback(
(position: string) => {
onTitleSettingsChange({ ...titleSettings, position })
},
[titleSettings, onTitleSettingsChange],
)
const updateFont = useCallback(
(font: string) => {
onTitleSettingsChange({ ...titleSettings, font })
},
[titleSettings, onTitleSettingsChange],
)
const updateSize = useCallback(
(size: number) => {
onTitleSettingsChange({ ...titleSettings, size })
},
[titleSettings, onTitleSettingsChange],
)
const updateColor = useCallback(
(color: string) => {
onTitleSettingsChange({ ...titleSettings, color })
},
[titleSettings, onTitleSettingsChange],
)
const toggleBold = useCallback(() => {
onTitleSettingsChange({ ...titleSettings, bold: !titleSettings.bold })
}, [titleSettings, onTitleSettingsChange])
const toggleItalic = useCallback(() => {
onTitleSettingsChange({ ...titleSettings, italic: !titleSettings.italic })
}, [titleSettings, onTitleSettingsChange])
const toggleStroke = useCallback(() => {
onTitleSettingsChange({ ...titleSettings, stroke: !titleSettings.stroke })
}, [titleSettings, onTitleSettingsChange])
const toggleShadow = useCallback(() => {
onTitleSettingsChange({ ...titleSettings, shadow: !titleSettings.shadow })
}, [titleSettings, onTitleSettingsChange])
const applyPreset = useCallback(
(presetKey: string) => {
const preset = TITLE_PRESETS.find((p) => p.key === presetKey)
if (!preset) return
onTitleSettingsChange({
...titleSettings,
size: preset.style.size,
color: preset.style.color,
bold: preset.style.bold,
italic: preset.style.italic,
stroke: preset.style.stroke,
shadow: preset.style.shadow,
})
},
[titleSettings, onTitleSettingsChange],
)
return {
// 数据
userTitles,
titleSettings,
aiTitleInput,
setAiTitleInput,
aiTitleGenerating,
aiTitleResults,
hasGeneratedTitles,
activePreset,
titlePresets: TITLE_PRESETS,
// AI 标题操作
handleGenerateAiTitles,
handleSelectAiTitle,
handleRefreshAiTitles,
// 标题设置操作
updateTitle,
toggleAiAutoSelect,
updatePosition,
updateFont,
updateSize,
updateColor,
toggleBold,
toggleItalic,
toggleStroke,
toggleShadow,
applyPreset,
}
}
export default useStep4Title
@@ -0,0 +1,61 @@
import { useQuery } from "@tanstack/react-query"
import { getTitles } from "@/api/titles"
import type { TitleSettings } from "../../types"
import { useAiTitleGenerator } from "./useAiTitleGenerator"
import { useTitleStyleUpdaters } from "./useTitleStyleUpdaters"
interface UseStep4TitleProps {
titleSettings: TitleSettings
onTitleSettingsChange: (settings: TitleSettings) => void
}
/**
* Step 4 标题设置 Hook
* 封装 AI 标题生成、标题样式设置等逻辑
*/
export function useStep4Title({ titleSettings, onTitleSettingsChange }: UseStep4TitleProps) {
// 标题库数据
const { data: userTitles = [] } = useQuery({
queryKey: ["titles"],
queryFn: () => getTitles(),
staleTime: 30_000,
})
// AI 标题生成
const aiGenerator = useAiTitleGenerator({ titleSettings, onTitleSettingsChange })
// 样式更新
const styleUpdaters = useTitleStyleUpdaters({ titleSettings, onTitleSettingsChange })
return {
// 数据
userTitles,
titleSettings,
// AI 标题状态
aiTitleInput: aiGenerator.aiTitleInput,
setAiTitleInput: aiGenerator.setAiTitleInput,
aiTitleGenerating: aiGenerator.aiTitleGenerating,
aiTitleResults: aiGenerator.aiTitleResults,
hasGeneratedTitles: aiGenerator.hasGeneratedTitles,
activePreset: styleUpdaters.activePreset,
titlePresets: styleUpdaters.titlePresets,
// AI 标题操作
handleGenerateAiTitles: aiGenerator.handleGenerateAiTitles,
handleSelectAiTitle: aiGenerator.handleSelectAiTitle,
handleRefreshAiTitles: aiGenerator.handleRefreshAiTitles,
// 标题设置操作
updateTitle: styleUpdaters.updateTitle,
toggleAiAutoSelect: styleUpdaters.toggleAiAutoSelect,
updatePosition: styleUpdaters.updatePosition,
updateFont: styleUpdaters.updateFont,
updateSize: styleUpdaters.updateSize,
updateColor: styleUpdaters.updateColor,
toggleBold: styleUpdaters.toggleBold,
toggleItalic: styleUpdaters.toggleItalic,
toggleStroke: styleUpdaters.toggleStroke,
toggleShadow: styleUpdaters.toggleShadow,
applyPreset: styleUpdaters.applyPreset,
}
}
export default useStep4Title
@@ -0,0 +1,102 @@
import { useState, useCallback } from "react"
import { message } from "antd"
import { AI_TITLE_TEMPLATES } from "../../constants"
import type { TitleSettings } from "../../types"
export interface AiTitleItem {
title: string
highlight: string
style: "catchy" | "emotional" | "informative"
}
interface UseAiTitleGeneratorOptions {
titleSettings: TitleSettings
onTitleSettingsChange: (settings: TitleSettings) => void
}
/**
* AI 标题生成 Hook
* 封装 AI 标题生成、刷新、选择等逻辑
*/
export function useAiTitleGenerator({
titleSettings,
onTitleSettingsChange,
}: UseAiTitleGeneratorOptions) {
const [aiTitleInput, setAiTitleInput] = useState("")
const [aiTitleGenerating, setAiTitleGenerating] = useState(false)
const [aiTitleResults, setAiTitleResults] = useState<AiTitleItem[]>([])
const [hasGeneratedTitles, setHasGeneratedTitles] = useState(false)
const extractTopic = (text: string): string => {
const keywords = text
.replace(/[,。!?、,.!?]/g, " ")
.split(/\s+/)
.filter(Boolean)
if (keywords.length === 0) return "这个话题"
return keywords.slice(0, 3).join("")
}
const generateTitlesFromTopic = (topic: string): AiTitleItem[] => {
const results: AiTitleItem[] = []
const styles: Array<"catchy" | "emotional" | "informative"> = [
"catchy",
"emotional",
"informative",
]
const highlights = { catchy: "吸睛标题", emotional: "情感共鸣", informative: "知识干货" }
styles.forEach((style) => {
const templates = AI_TITLE_TEMPLATES[style]
const shuffled = [...templates].sort(() => Math.random() - 0.5).slice(0, 2)
shuffled.forEach((tpl) => {
results.push({
title: tpl.replace(/\{topic\}/g, topic),
highlight: highlights[style],
style,
})
})
})
results.sort(() => Math.random() - 0.5)
return results
}
const handleGenerateAiTitles = useCallback(async () => {
if (!aiTitleInput.trim()) {
message.warning("请先输入视频描述或关键词")
return
}
setAiTitleGenerating(true)
setHasGeneratedTitles(true)
await new Promise((resolve) => setTimeout(resolve, 1200))
const topic = extractTopic(aiTitleInput)
setAiTitleResults(generateTitlesFromTopic(topic))
setAiTitleGenerating(false)
}, [aiTitleInput])
const handleSelectAiTitle = useCallback(
(title: string) => {
onTitleSettingsChange({ ...titleSettings, title, aiAutoSelect: false })
message.success("已选用此标题")
},
[titleSettings, onTitleSettingsChange],
)
const handleRefreshAiTitles = useCallback(async () => {
if (!aiTitleInput.trim()) return
setAiTitleGenerating(true)
await new Promise((resolve) => setTimeout(resolve, 800))
const topic = extractTopic(aiTitleInput)
setAiTitleResults(generateTitlesFromTopic(topic))
setAiTitleGenerating(false)
}, [aiTitleInput])
return {
aiTitleInput,
setAiTitleInput,
aiTitleGenerating,
aiTitleResults,
hasGeneratedTitles,
handleGenerateAiTitles,
handleSelectAiTitle,
handleRefreshAiTitles,
}
}
@@ -0,0 +1,123 @@
import { useCallback, useMemo } from "react"
import { TITLE_PRESETS } from "../../constants"
import type { TitleSettings } from "../../types"
interface UseTitleStyleUpdatersOptions {
titleSettings: TitleSettings
onTitleSettingsChange: (settings: TitleSettings) => void
}
/**
* 标题样式更新 Hook
* 封装标题文字、位置、字体、样式等所有设置更新函数
*/
export function useTitleStyleUpdaters({
titleSettings,
onTitleSettingsChange,
}: UseTitleStyleUpdatersOptions) {
const getActivePreset = (settings: TitleSettings): string | null => {
for (const p of TITLE_PRESETS) {
if (
settings.size === p.style.size &&
settings.color === p.style.color &&
settings.bold === p.style.bold &&
settings.italic === p.style.italic &&
settings.stroke === p.style.stroke &&
settings.shadow === p.style.shadow
) {
return p.key
}
}
return null
}
const activePreset = useMemo(() => getActivePreset(titleSettings), [titleSettings])
const updateTitle = useCallback(
(title: string) => {
onTitleSettingsChange({ ...titleSettings, title })
},
[titleSettings, onTitleSettingsChange],
)
const toggleAiAutoSelect = useCallback(() => {
onTitleSettingsChange({ ...titleSettings, aiAutoSelect: !titleSettings.aiAutoSelect })
}, [titleSettings, onTitleSettingsChange])
const updatePosition = useCallback(
(position: string) => {
onTitleSettingsChange({ ...titleSettings, position })
},
[titleSettings, onTitleSettingsChange],
)
const updateFont = useCallback(
(font: string) => {
onTitleSettingsChange({ ...titleSettings, font })
},
[titleSettings, onTitleSettingsChange],
)
const updateSize = useCallback(
(size: number) => {
onTitleSettingsChange({ ...titleSettings, size })
},
[titleSettings, onTitleSettingsChange],
)
const updateColor = useCallback(
(color: string) => {
onTitleSettingsChange({ ...titleSettings, color })
},
[titleSettings, onTitleSettingsChange],
)
const toggleBold = useCallback(() => {
onTitleSettingsChange({ ...titleSettings, bold: !titleSettings.bold })
}, [titleSettings, onTitleSettingsChange])
const toggleItalic = useCallback(() => {
onTitleSettingsChange({ ...titleSettings, italic: !titleSettings.italic })
}, [titleSettings, onTitleSettingsChange])
const toggleStroke = useCallback(() => {
onTitleSettingsChange({ ...titleSettings, stroke: !titleSettings.stroke })
}, [titleSettings, onTitleSettingsChange])
const toggleShadow = useCallback(() => {
onTitleSettingsChange({ ...titleSettings, shadow: !titleSettings.shadow })
}, [titleSettings, onTitleSettingsChange])
const applyPreset = useCallback(
(presetKey: string) => {
const preset = TITLE_PRESETS.find((p) => p.key === presetKey)
if (!preset) return
onTitleSettingsChange({
...titleSettings,
size: preset.style.size,
color: preset.style.color,
bold: preset.style.bold,
italic: preset.style.italic,
stroke: preset.style.stroke,
shadow: preset.style.shadow,
})
},
[titleSettings, onTitleSettingsChange],
)
return {
activePreset,
titlePresets: TITLE_PRESETS,
updateTitle,
toggleAiAutoSelect,
updatePosition,
updateFont,
updateSize,
updateColor,
toggleBold,
toggleItalic,
toggleStroke,
toggleShadow,
applyPreset,
}
}
+24 -167
View File
@@ -3,126 +3,39 @@
* 卡片视图展示用户已保存的剪辑模板
* 支持搜索、分类筛选、编辑/复制/删除/使用模板生成
*/
import React, { useState } from "react"
import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"
import {
Typography,
Card,
Input,
Select,
Tag,
Button,
Space,
Empty,
Spin,
Tooltip,
message,
Popconfirm,
Row,
Col,
} from "antd"
import {
SearchOutlined,
EditOutlined,
CopyOutlined,
DeleteOutlined,
VideoCameraOutlined,
AppstoreOutlined,
PlusOutlined,
} from "@ant-design/icons"
import React from "react"
import { Typography, Input, Select, Button, Empty, Spin, Row, Col } from "antd"
import { SearchOutlined, AppstoreOutlined, PlusOutlined } from "@ant-design/icons"
import { useNavigate } from "react-router-dom"
import {
getEditingTemplates,
getTemplateCategories,
deleteEditingTemplate,
createEditingTemplate,
MODE_LABELS,
MODE_COLORS,
type EditingTemplate,
type TemplateMode,
} from "@/api/editing-planner"
import type { EditingTemplate } from "@/api/editing-planner"
import { useMyTemplates } from "./hooks/useMyTemplates"
import { TemplateCard } from "./components/TemplateCard"
import "./MyTemplates.css"
const { Title, Text } = Typography
const MyTemplates: React.FC = () => {
const navigate = useNavigate()
const queryClient = useQueryClient()
const {
searchText,
setSearchText,
filterCategory,
setFilterCategory,
templates,
categories,
isLoading,
handleCopy,
handleDelete,
} = useMyTemplates()
const [searchText, setSearchText] = useState("")
const [filterCategory, setFilterCategory] = useState("")
/* ── 数据查询 ── */
const { data: templates = [], isLoading } = useQuery({
queryKey: ["editing-templates", filterCategory, searchText],
queryFn: () =>
getEditingTemplates({
category: filterCategory || undefined,
tag: searchText || undefined,
}),
})
const { data: categories = [] } = useQuery({
queryKey: ["template-categories"],
queryFn: getTemplateCategories,
})
/* ── Mutations ── */
const deleteMutation = useMutation({
mutationFn: deleteEditingTemplate,
onSuccess: () => {
message.success("模板已删除")
queryClient.invalidateQueries({ queryKey: ["editing-templates"] })
},
onError: (err: unknown) => {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("删除失败")
},
})
const copyMutation = useMutation({
mutationFn: (tpl: EditingTemplate) =>
createEditingTemplate({
name: `${tpl.name}(副本)`,
mode: tpl.mode,
category: tpl.category,
tags: tpl.tags,
title_config: tpl.title_config,
subtitle_config: tpl.subtitle_config,
bgm_config: tpl.bgm_config,
estimated_duration:
tpl.estimated_duration ??
Math.round(
tpl.segments.reduce((s, seg) => s + (seg.duration_min + seg.duration_max) / 2, 0),
),
segments: tpl.segments.map(({ id: _id, ...rest }) => rest),
}),
onSuccess: () => {
message.success("模板已复制")
queryClient.invalidateQueries({ queryKey: ["editing-templates"] })
},
onError: (err: unknown) => {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("复制失败")
},
})
/* ── 操作 ── */
const handleEdit = (tpl: EditingTemplate) => {
navigate(`/editing-planner?template=${tpl.id}`)
}
const handleGenerate = (tpl: EditingTemplate) => {
// 跳转到智能剪辑页面,统一从智能剪辑出片
navigate(`/generate?templateId=${tpl.id}`)
}
const handleCopy = (tpl: EditingTemplate) => {
copyMutation.mutate(tpl)
}
const handleDelete = (id: string) => {
deleteMutation.mutate(id)
}
return (
<div className="mt-page">
{/* 页面头部 */}
@@ -179,69 +92,13 @@ const MyTemplates: React.FC = () => {
<Row gutter={[16, 16]}>
{templates.map((tpl) => (
<Col key={tpl.id} xs={24} sm={12} md={8} lg={6}>
<Card
className="mt-card"
hoverable
actions={[
<Tooltip title="编辑" key="edit">
<EditOutlined onClick={() => handleEdit(tpl)} />
</Tooltip>,
<Tooltip title="复制" key="copy">
<CopyOutlined onClick={() => handleCopy(tpl)} />
</Tooltip>,
<Tooltip title="使用模板生成" key="generate">
<VideoCameraOutlined onClick={() => handleGenerate(tpl)} />
</Tooltip>,
<Popconfirm
key="delete"
title="确定删除此模板?"
onConfirm={() => handleDelete(tpl.id)}
okText="删除"
cancelText="取消"
>
<Tooltip title="删除">
<DeleteOutlined style={{ color: "#ff4d4f" }} />
</Tooltip>
</Popconfirm>,
]}
>
<div className="mt-card-head">
<Text strong ellipsis style={{ fontSize: 15 }}>
{tpl.name}
</Text>
<Tag color={MODE_COLORS[tpl.mode as TemplateMode] || "default"}>
{MODE_LABELS[tpl.mode as TemplateMode] || tpl.mode}
</Tag>
<Tag color="green">用户自制</Tag>
</div>
<div className="mt-card-meta">
<Text type="secondary" style={{ fontSize: 12 }}>
{tpl.segments.length} 片段 · 预估 ~{tpl.estimated_duration}s
</Text>
{tpl.category && (
<Tag style={{ fontSize: 11, marginTop: 4 }}>{tpl.category}</Tag>
)}
</div>
{tpl.tags.length > 0 && (
<div className="mt-card-tags">
{tpl.tags.map((tag) => (
<Tag key={tag} style={{ fontSize: 11 }}>
{tag}
</Tag>
))}
</div>
)}
<div className="mt-card-config">
<Space size={4} wrap>
{tpl.title_config.ai_auto_select && <Tag color="cyan">AI标题</Tag>}
{tpl.subtitle_config.enabled && <Tag color="geekblue">字幕</Tag>}
{tpl.bgm_config.enabled && <Tag color="pink">BGM</Tag>}
</Space>
</div>
</Card>
<TemplateCard
tpl={tpl}
onEdit={handleEdit}
onCopy={handleCopy}
onGenerate={handleGenerate}
onDelete={handleDelete}
/>
</Col>
))}
</Row>
@@ -0,0 +1,92 @@
import React from "react"
import { Card, Tag, Tooltip, Popconfirm, Space, Typography } from "antd"
import { EditOutlined, CopyOutlined, DeleteOutlined, VideoCameraOutlined } from "@ant-design/icons"
import {
MODE_LABELS,
MODE_COLORS,
type EditingTemplate,
type TemplateMode,
} from "@/api/editing-planner"
const { Text } = Typography
interface TemplateCardProps {
tpl: EditingTemplate
onEdit: (tpl: EditingTemplate) => void
onCopy: (tpl: EditingTemplate) => void
onGenerate: (tpl: EditingTemplate) => void
onDelete: (id: string) => void
}
/**
* 单个模板卡片组件
*/
export const TemplateCard: React.FC<TemplateCardProps> = ({
tpl,
onEdit,
onCopy,
onGenerate,
onDelete,
}) => (
<Card
className="mt-card"
hoverable
actions={[
<Tooltip title="编辑" key="edit">
<EditOutlined onClick={() => onEdit(tpl)} />
</Tooltip>,
<Tooltip title="复制" key="copy">
<CopyOutlined onClick={() => onCopy(tpl)} />
</Tooltip>,
<Tooltip title="使用模板生成" key="generate">
<VideoCameraOutlined onClick={() => onGenerate(tpl)} />
</Tooltip>,
<Popconfirm
key="delete"
title="确定删除此模板?"
onConfirm={() => onDelete(tpl.id)}
okText="删除"
cancelText="取消"
>
<Tooltip title="删除">
<DeleteOutlined style={{ color: "#ff4d4f" }} />
</Tooltip>
</Popconfirm>,
]}
>
<div className="mt-card-head">
<Text strong ellipsis style={{ fontSize: 15 }}>
{tpl.name}
</Text>
<Tag color={MODE_COLORS[tpl.mode as TemplateMode] || "default"}>
{MODE_LABELS[tpl.mode as TemplateMode] || tpl.mode}
</Tag>
<Tag color="green">用户自制</Tag>
</div>
<div className="mt-card-meta">
<Text type="secondary" style={{ fontSize: 12 }}>
{tpl.segments.length} 片段 · 预估 ~{tpl.estimated_duration}s
</Text>
{tpl.category && <Tag style={{ fontSize: 11, marginTop: 4 }}>{tpl.category}</Tag>}
</div>
{tpl.tags.length > 0 && (
<div className="mt-card-tags">
{tpl.tags.map((tag) => (
<Tag key={tag} style={{ fontSize: 11 }}>
{tag}
</Tag>
))}
</div>
)}
<div className="mt-card-config">
<Space size={4} wrap>
{tpl.title_config.ai_auto_select && <Tag color="cyan">AI标题</Tag>}
{tpl.subtitle_config.enabled && <Tag color="geekblue">字幕</Tag>}
{tpl.bgm_config.enabled && <Tag color="pink">BGM</Tag>}
</Space>
</div>
</Card>
)
@@ -0,0 +1,100 @@
import { useState } from "react"
import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"
import { message } from "antd"
import {
getEditingTemplates,
getTemplateCategories,
deleteEditingTemplate,
createEditingTemplate,
type EditingTemplate,
} from "@/api/editing-planner"
/**
* 我的模板数据 Hook
* 封装模板列表查询、筛选、删除、复制等数据操作
*/
export function useMyTemplates() {
const queryClient = useQueryClient()
const [searchText, setSearchText] = useState("")
const [filterCategory, setFilterCategory] = useState("")
/* 模板列表 */
const { data: templates = [], isLoading } = useQuery({
queryKey: ["editing-templates", filterCategory, searchText],
queryFn: () =>
getEditingTemplates({
category: filterCategory || undefined,
tag: searchText || undefined,
}),
})
/* 分类列表 */
const { data: categories = [] } = useQuery({
queryKey: ["template-categories"],
queryFn: getTemplateCategories,
})
/* 删除 mutation */
const deleteMutation = useMutation({
mutationFn: deleteEditingTemplate,
onSuccess: () => {
message.success("模板已删除")
queryClient.invalidateQueries({ queryKey: ["editing-templates"] })
},
onError: (err: unknown) => {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("删除失败")
},
})
/* 复制 mutation */
const copyMutation = useMutation({
mutationFn: (tpl: EditingTemplate) =>
createEditingTemplate({
name: `${tpl.name}(副本)`,
mode: tpl.mode,
category: tpl.category,
tags: tpl.tags,
title_config: tpl.title_config,
subtitle_config: tpl.subtitle_config,
bgm_config: tpl.bgm_config,
estimated_duration:
tpl.estimated_duration ??
Math.round(
tpl.segments.reduce((s, seg) => s + (seg.duration_min + seg.duration_max) / 2, 0),
),
segments: tpl.segments.map(({ id: _id, ...rest }) => rest),
}),
onSuccess: () => {
message.success("模板已复制")
queryClient.invalidateQueries({ queryKey: ["editing-templates"] })
},
onError: (err: unknown) => {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("复制失败")
},
})
const handleCopy = (tpl: EditingTemplate) => {
copyMutation.mutate(tpl)
}
const handleDelete = (id: string) => {
deleteMutation.mutate(id)
}
return {
// 状态
searchText,
setSearchText,
filterCategory,
setFilterCategory,
// 数据
templates,
categories,
isLoading,
// 操作
handleCopy,
handleDelete,
isDeleting: deleteMutation.isPending,
isCopying: copyMutation.isPending,
}
}
@@ -78,11 +78,7 @@ export const ProductEmptyState: React.FC<ProductEmptyStateProps> = ({
<VideoCameraOutlined />
</div>
<p>暂无成片,去智能剪辑吧</p>
<Button
buttonType="primary"
buttonSize="sm"
onClick={() => message.info("跳转到生成页面")}
>
<Button buttonType="primary" buttonSize="sm" onClick={() => message.info("跳转到生成页面")}>
去生成
</Button>
</div>
@@ -2,151 +2,59 @@
* 升级/降级/续费页面
* P1-3: antd Button/Modal/Radio/Spin → 自定义 UI 组件
*/
import React, { useState, useEffect } from "react"
import { message } from "antd"
import { Button, Modal } from "@/components/ui"
import React from "react"
import { Modal } from "@/components/ui"
import { useNavigate } from "react-router-dom"
import {
getCurrentSubscription,
changePlan,
toggleAutoRenew,
cancelSubscription,
} from "@/api/subscription"
import type { SubscriptionInfo, PlanType, BillingCycle } from "@/api/subscription"
import type { PlanType } from "@/api/subscription"
import PageHead from "@/components/layout/PageHead"
import { Button } from "@/components/ui"
import { PLANS_META, getPlanName, getPlanPrice } from "./constants"
import { BillingCycleSwitch, Spinner } from "./components/SubscriptionUI"
import { useSubscription } from "./hooks/useSubscription"
import "./UpgradeSubscription.css"
const PLANS_META: Record<string, { name: string; price: number; yearlyPrice: number }> = {
free: { name: "体验版", price: 0, yearlyPrice: 0 },
standard: { name: "标准版", price: 99, yearlyPrice: 990 },
pro: { name: "专业版", price: 299, yearlyPrice: 2990 },
enterprise: { name: "企业版", price: 0, yearlyPrice: 0 },
}
/** 自定义计费周期切换组件 */
const BillingCycleSwitch: React.FC<{
value: BillingCycle
onChange: (cycle: BillingCycle) => void
monthlyPrice: number
yearlyPrice: number
}> = ({ value, onChange, monthlyPrice, yearlyPrice }) => (
<div className="xx-billing-cycle-switch">
<button
type="button"
className={`xx-billing-cycle-btn ${value === "monthly" ? "active" : ""}`}
onClick={() => onChange("monthly")}
>
¥{monthlyPrice}/月
</button>
<button
type="button"
className={`xx-billing-cycle-btn ${value === "yearly" ? "active" : ""}`}
onClick={() => onChange("yearly")}
>
¥{yearlyPrice}/年
{yearlyPrice > 0 && monthlyPrice > 0 && (
<span className="xx-save">省 ¥{monthlyPrice * 12 - yearlyPrice}</span>
)}
</button>
</div>
)
/** 自定义 Spinner 组件 */
const Spinner: React.FC<{ size?: "small" | "large" }> = ({ size = "large" }) => (
<div className={`xx-spinner xx-spinner--${size}`}>
<div className="xx-spinner-dot" />
<div className="xx-spinner-dot" />
<div className="xx-spinner-dot" />
</div>
)
const UpgradeSubscription: React.FC = () => {
const navigate = useNavigate()
const [subscription, setSubscription] = useState<SubscriptionInfo | null>(null)
const [loading, setLoading] = useState(true)
const [submitting, setSubmitting] = useState(false)
const [selectedPlan, setSelectedPlan] = useState<PlanType>("standard")
const [billingCycle, setBillingCycle] = useState<BillingCycle>("monthly")
const {
subscription,
loading,
submitting,
selectedPlan,
billingCycle,
setSelectedPlan,
setBillingCycle,
executeChangePlan,
handleToggleAutoRenew,
handleCancel,
} = useSubscription()
useEffect(() => {
loadSubscription()
}, [])
const loadSubscription = async () => {
try {
const data = await getCurrentSubscription()
setSubscription(data)
setSelectedPlan(data.plan_id)
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("获取订阅信息失败")
} finally {
setLoading(false)
}
}
const handleUpgrade = async () => {
const handleUpgradeClick = () => {
if (!subscription) return
if (selectedPlan === subscription.plan_id && billingCycle === subscription.billing_cycle) {
message.info("当前已是该套餐")
return
}
const plan = PLANS_META[selectedPlan]
const price = billingCycle === "yearly" ? plan.yearlyPrice : plan.price
const price = getPlanPrice(selectedPlan, billingCycle)
Modal.confirm({
title: "确认变更套餐",
content: `即将变更为「${plan.name}」(${billingCycle === "monthly" ? "月付" : "年付"}),${price > 0 ? `费用 ¥${price}${billingCycle === "monthly" ? "/月" : "/年"}` : "免费"}。变更立即生效。`,
okText: "确认变更",
cancelText: "取消",
onOk: async () => {
try {
setSubmitting(true)
const res = await changePlan({
target_plan_id: selectedPlan,
billing_cycle: billingCycle,
})
if (res.success) {
message.success(res.message)
setSubscription(res.new_subscription ?? null)
} else {
message.error(res.message)
}
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("套餐变更失败,请重试")
} finally {
setSubmitting(false)
}
},
onOk: executeChangePlan,
})
}
const handleToggleAutoRenew = async (enabled: boolean) => {
try {
const res = await toggleAutoRenew(enabled)
message.success(res.message)
if (subscription) {
setSubscription({ ...subscription, auto_renew: enabled })
}
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("操作失败")
}
}
const handleCancel = () => {
const handleCancelClick = () => {
Modal.confirm({
title: "确认取消订阅",
content: "取消后,当前周期结束前仍可正常使用,到期后降级为体验版。",
okText: "确认取消",
cancelText: "再想想",
onOk: async () => {
try {
const res = await cancelSubscription()
message.success(res.message)
navigate("/app/subscription")
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("取消失败")
}
const ok = await handleCancel()
if (ok) navigate("/app/subscription")
},
})
}
@@ -163,10 +71,7 @@ const UpgradeSubscription: React.FC = () => {
return (
<div className="xx-upgrade-page">
<PageHead
title="变更订阅方案"
description={`当前套餐:${PLANS_META[currentPlan]?.name ?? "体验版"}`}
/>
<PageHead title="变更订阅方案" description={`当前套餐:${getPlanName(currentPlan)}`} />
<div className="xx-upgrade-plans">
{(["standard", "pro", "enterprise"] as PlanType[]).map((planId) => {
@@ -198,7 +103,7 @@ const UpgradeSubscription: React.FC = () => {
buttonType="primary"
buttonSize="lg"
disabled={submitting || selectedPlan === currentPlan}
onClick={handleUpgrade}
onClick={handleUpgradeClick}
>
{submitting ? "处理中..." : "确认变更"}
</Button>
@@ -220,7 +125,7 @@ const UpgradeSubscription: React.FC = () => {
<Button
buttonType="danger"
buttonSize="sm"
onClick={handleCancel}
onClick={handleCancelClick}
className="xx-cancel-btn"
>
取消订阅
@@ -0,0 +1,50 @@
import React from "react"
import type { BillingCycle } from "@/api/subscription"
interface BillingCycleSwitchProps {
value: BillingCycle
onChange: (cycle: BillingCycle) => void
monthlyPrice: number
yearlyPrice: number
}
/** 自定义计费周期切换组件 */
export const BillingCycleSwitch: React.FC<BillingCycleSwitchProps> = ({
value,
onChange,
monthlyPrice,
yearlyPrice,
}) => (
<div className="xx-billing-cycle-switch">
<button
type="button"
className={`xx-billing-cycle-btn ${value === "monthly" ? "active" : ""}`}
onClick={() => onChange("monthly")}
>
¥{monthlyPrice}/月
</button>
<button
type="button"
className={`xx-billing-cycle-btn ${value === "yearly" ? "active" : ""}`}
onClick={() => onChange("yearly")}
>
¥{yearlyPrice}/年
{yearlyPrice > 0 && monthlyPrice > 0 && (
<span className="xx-save">省 ¥{monthlyPrice * 12 - yearlyPrice}</span>
)}
</button>
</div>
)
interface SpinnerProps {
size?: "small" | "large"
}
/** 自定义 Spinner 组件 */
export const Spinner: React.FC<SpinnerProps> = ({ size = "large" }) => (
<div className={`xx-spinner xx-spinner--${size}`}>
<div className="xx-spinner-dot" />
<div className="xx-spinner-dot" />
<div className="xx-spinner-dot" />
</div>
)
@@ -0,0 +1,16 @@
import type { PlanType, BillingCycle } from "@/api/subscription"
export const PLANS_META: Record<string, { name: string; price: number; yearlyPrice: number }> = {
free: { name: "体验版", price: 0, yearlyPrice: 0 },
standard: { name: "标准版", price: 99, yearlyPrice: 990 },
pro: { name: "专业版", price: 299, yearlyPrice: 2990 },
enterprise: { name: "企业版", price: 0, yearlyPrice: 0 },
}
export const getPlanName = (planId: PlanType | string) => PLANS_META[planId]?.name ?? "体验版"
export const getPlanPrice = (planId: PlanType | string, cycle: BillingCycle) => {
const plan = PLANS_META[planId]
if (!plan) return 0
return cycle === "yearly" ? plan.yearlyPrice : plan.price
}
@@ -0,0 +1,111 @@
import { useState, useEffect, useCallback } from "react"
import { message } from "antd"
import {
getCurrentSubscription,
changePlan,
toggleAutoRenew,
cancelSubscription,
type SubscriptionInfo,
type PlanType,
type BillingCycle,
} from "@/api/subscription"
/**
* 订阅管理 Hook
* 封装订阅信息查询、套餐变更、自动续费切换、取消订阅等逻辑
*/
export function useSubscription() {
const [subscription, setSubscription] = useState<SubscriptionInfo | null>(null)
const [loading, setLoading] = useState(true)
const [submitting, setSubmitting] = useState(false)
const [selectedPlan, setSelectedPlan] = useState<PlanType>("standard")
const [billingCycle, setBillingCycle] = useState<BillingCycle>("monthly")
const loadSubscription = useCallback(async () => {
try {
const data = await getCurrentSubscription()
setSubscription(data)
setSelectedPlan(data.plan_id)
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("获取订阅信息失败")
} finally {
setLoading(false)
}
}, [])
useEffect(() => {
loadSubscription()
}, [loadSubscription])
const handleUpgrade = useCallback(async () => {
if (!subscription) return
if (selectedPlan === subscription.plan_id && billingCycle === subscription.billing_cycle) {
message.info("当前已是该套餐")
return
}
// 由调用方决定是否弹确认框
}, [subscription, selectedPlan, billingCycle])
const executeChangePlan = useCallback(async () => {
try {
setSubmitting(true)
const res = await changePlan({
target_plan_id: selectedPlan,
billing_cycle: billingCycle,
})
if (res.success) {
message.success(res.message)
setSubscription(res.new_subscription ?? null)
} else {
message.error(res.message)
}
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("套餐变更失败,请重试")
} finally {
setSubmitting(false)
}
}, [selectedPlan, billingCycle])
const handleToggleAutoRenew = useCallback(
async (enabled: boolean) => {
try {
const res = await toggleAutoRenew(enabled)
message.success(res.message)
if (subscription) {
setSubscription({ ...subscription, auto_renew: enabled })
}
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("操作失败")
}
},
[subscription],
)
const handleCancel = useCallback(async () => {
try {
const res = await cancelSubscription()
message.success(res.message)
return true
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("取消失败")
return false
}
}, [])
return {
// 状态
subscription,
loading,
submitting,
selectedPlan,
billingCycle,
setSelectedPlan,
setBillingCycle,
// 操作
loadSubscription,
handleUpgrade,
executeChangePlan,
handleToggleAutoRenew,
handleCancel,
}
}
+21 -221
View File
@@ -2,129 +2,33 @@
支持将指定颜色(默认绿色)变为透明,可用于虚拟背景、画中画背景替换等场景。
使用方式:
config = ChromaKeyConfig(key_color="#00FF00", similarity=0.3, blend=0.1)
engine = ChromaKeyEngine(config)
filter_str = engine.build_filter(input_label, output_label)
# 结果: [in]colorkey=color=0x00FF00:similarity=0.3:blend=0.1[out]
降级策略:
- 参数越界自动钳制
- 素材格式不支持时跳过(调用方捕获异常)
注:核心领域模型已抽离到 packages/domain/chroma_key_config.py,
本模块保留薄包装层,确保向后兼容。
"""
from __future__ import annotations
import logging
import re
from dataclasses import dataclass
from typing import Optional
from packages.domain.chroma_key_config import (
CHROMA_KEY_PRESETS,
ChromaKeyConfig,
apply_chroma_key_if_needed,
)
from packages.domain.chroma_key_config import ( # noqa: F401 — 向后兼容
build_chromakey_filter as _build_chromakey_filter_base,
)
from packages.domain.chroma_key_config import build_colorkey_filter as _build_colorkey_filter_base
from packages.domain.chroma_key_config import normalize_color as _normalize_color_base
logger = logging.getLogger(__name__)
# ── 配置模型 ──────────────────────────────────────────────────────────────────
@dataclass
class ChromaKeyConfig:
"""绿幕抠像配置。
Attributes:
enabled: 是否启用抠像
key_color: 要抠除的颜色,支持 hex 格式(如 "#00FF00")或颜色名
similarity: 颜色相似度阈值 0.01~1.0,值越大抠除范围越大
blend: 边缘平滑/混合度 0.0~1.0,值越大边缘越柔和
spill_suppress: 溢色抑制 0.0~1.0,减少边缘的绿幕反光
"""
enabled: bool = False
key_color: str = "#00FF00"
similarity: float = 0.3
blend: float = 0.1
spill_suppress: float = 0.0
@classmethod
def from_dict(cls, data: dict | None) -> "ChromaKeyConfig":
"""从字典解析配置,参数越界自动钳制。"""
if not data or not data.get("enabled", False):
return cls(enabled=False)
key_color = str(data.get("key_color", "#00FF00")).strip()
def _safe_float(val, default):
try:
return float(val)
except (TypeError, ValueError):
return default
similarity = _safe_float(data.get("similarity", 0.3), 0.3)
blend = _safe_float(data.get("blend", 0.1), 0.1)
spill_suppress = _safe_float(data.get("spill_suppress", 0.0), 0.0)
# 钳制到合法范围
similarity = max(0.01, min(1.0, similarity))
blend = max(0.0, min(1.0, blend))
spill_suppress = max(0.0, min(1.0, spill_suppress))
return cls(
enabled=True,
key_color=key_color,
similarity=similarity,
blend=blend,
spill_suppress=spill_suppress,
)
def has_effect(self) -> bool:
"""判断是否有实际抠像效果。"""
return self.enabled and self.similarity > 0
# ── 预设配置 ──────────────────────────────────────────────────────────────────
# 常见绿幕/蓝幕预设
CHROMA_KEY_PRESETS = {
"green_screen": {
"key_color": "#00FF00",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.5,
},
"blue_screen": {
"key_color": "#0000FF",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.5,
},
"red_screen": {
"key_color": "#FF0000",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.0,
},
"precise_green": {
"key_color": "#00FF00",
"similarity": 0.2,
"blend": 0.05,
"spill_suppress": 0.3,
},
"soft_green": {
"key_color": "#00FF00",
"similarity": 0.45,
"blend": 0.2,
"spill_suppress": 0.5,
},
}
# ── 引擎实现 ──────────────────────────────────────────────────────────────────
class ChromaKeyEngine:
"""绿幕抠像引擎。
"""绿幕抠像引擎.
基于 FFmpeg colorkey 滤镜实现,将指定颜色变为透明。
适用于绿幕/蓝幕视频的背景去除,配合画中画或 overlay 实现虚拟背景。
薄包装层,实际逻辑委托给 packages.domain.chroma_key_config。
"""
def __init__(self, config: ChromaKeyConfig):
@@ -132,117 +36,13 @@ class ChromaKeyEngine:
@staticmethod
def _normalize_color(color_str: str) -> str:
"""将颜色字符串转为 FFmpeg colorkey 接受的格式。
支持:
- "#RRGGBB" / "#RRGGBBAA" → 0xRRGGBB
- "0xRRGGBB" → 直接使用
- 颜色名(green/blue/red/black/white 等)→ 直接透传
"""
color = color_str.strip()
# hex 格式
hex_match = re.match(r"^#?([0-9a-fA-F]{6})([0-9a-fA-F]{2})?$", color)
if hex_match:
return f"0x{hex_match.group(1).upper()}"
# 已经是 0x 格式
if color.lower().startswith("0x"):
return color.upper()
# 颜色名直接透传(FFmpeg 支持常见颜色名)
return color
"""将颜色字符串转为 FFmpeg colorkey 接受的格式."""
return _normalize_color_base(color_str)
def build_filter(self, input_label: str, output_label: str) -> str:
"""构建 colorkey 滤镜字符串。
Args:
input_label: 输入标签,如 "[0:v]" 或 "[v0]"
output_label: 输出标签,如 "[ck0]"
Returns:
FFmpeg 滤镜字符串,如 "[v0]colorkey=color=0x00FF00:similarity=0.3:blend=0.1[ck0]"
Raises:
ValueError: 配置无效时抛出(调用方应捕获并降级)
"""
if not self.config.has_effect():
# 无效果,直接直通
return f"{input_label}copy{output_label}"
color = self._normalize_color(self.config.key_color)
similarity = self.config.similarity
blend = self.config.blend
# 基础 colorkey 滤镜
parts = [f"colorkey=color={color}:similarity={similarity}:blend={blend}"]
# 溢色抑制(通过 colorchannelmixer 降低绿色通道增益)
if self.config.spill_suppress > 0:
# 降低绿通道增益,减少绿幕反光溢出
spill = self.config.spill_suppress
# 绿通道增益 = 1 - spill_factor
g_gain = max(0.3, 1.0 - spill * 0.7)
# 同时稍微提升红和蓝来补偿色偏
r_gain = 1.0 + spill * 0.15
b_gain = 1.0 + spill * 0.15
parts.append(f"colorchannelmixer=" f"rr={r_gain}:" f"gg={g_gain}:" f"bb={b_gain}:" f"aa=1")
filter_str = f"{input_label}{','.join(parts)}{output_label}"
return filter_str
"""构建 colorkey 滤镜字符串."""
return _build_colorkey_filter_base(self.config, input_label, output_label)
def build_filter_chromakey(self, input_label: str, output_label: str) -> str:
"""使用 chromakey 滤镜(更高级的版本,支持更多参数)。
注意:并非所有 FFmpeg 版本都支持 chromakey 滤镜,
优先使用 colorkey(兼容性更好)。
Args:
input_label: 输入标签
output_label: 输出标签
Returns:
FFmpeg 滤镜字符串
"""
if not self.config.has_effect():
return f"{input_label}copy{output_label}"
color = self._normalize_color(self.config.key_color)
similarity = self.config.similarity
blend = self.config.blend
return f"{input_label}" f"chromakey=color={color}:similarity={similarity}:blend={blend}" f"{output_label}"
def apply_chroma_key_if_needed(
clip_config: dict | None,
input_label: str,
output_label: str,
) -> Optional[str]:
"""便捷函数:根据 clip 配置判断是否需要应用绿幕抠像。
Args:
clip_config: clip 的 config 字典
input_label: 输入标签
output_label: 输出标签
Returns:
滤镜字符串,不需要抠像时返回 None
"""
if not clip_config:
return None
chroma_key_data = clip_config.get("chroma_key")
if not chroma_key_data:
return None
try:
config = ChromaKeyConfig.from_dict(chroma_key_data)
if not config.has_effect():
return None
engine = ChromaKeyEngine(config)
return engine.build_filter(input_label, output_label)
except Exception as e:
logger.warning("[chroma-key] 应用抠像失败,跳过: %s", e)
return None
"""使用 chromakey 滤镜(更高级的版本,支持更多参数)."""
return _build_chromakey_filter_base(self.config, input_label, output_label)
+38 -144
View File
@@ -20,6 +20,17 @@ from shared.ffmpeg_utils import ( # noqa: F401
run_ffmpeg,
)
# xfade 转场纯逻辑已抽离到 domain 层,这里 re-export 保持向后兼容
from packages.domain.xfade_builder import DEFAULT_TRANSITION_DURATION as _default_transition_duration_base # noqa: F401
from packages.domain.xfade_builder import (
SUPPORTED_TRANSITIONS,
XFADE_TRANSITION_MAP,
XFade_TRANSITION_NAMES,
)
from packages.domain.xfade_builder import build_xfade_filter_chain as _build_xfade_filter_chain_base
from packages.domain.xfade_builder import chain_filters as _chain_filters_base
from packages.domain.xfade_builder import resolve_xfade_transition as _resolve_xfade_transition_base
logger = logging.getLogger(__name__)
# ── 常量(Worker 层业务相关) ────────────────────────────────────────────────
@@ -28,43 +39,34 @@ DEFAULT_OUTPUT_WIDTH = 1280
DEFAULT_OUTPUT_HEIGHT = 720
DEFAULT_FPS = 25
# xfade 转场映射:transition_effect 名称 → FFmpeg xfade transition 名称
# 键同时支持 TransitionEffect 枚举值和字符串名称(向后兼容)
# "cut" 为特殊值:硬切,不使用 xfade(由调用方特殊处理)
XFADE_TRANSITION_MAP: dict[str, str] = {
# 基础
"fade": "fade",
"dissolve": "dissolve",
"crossfade": "dissolve",
"crossdissolve": "dissolve",
# 滑入系列
"slideleft": "slideleft",
"slide_left": "slideleft",
"slideright": "slideright",
"slide_right": "slideright",
"slideup": "slideup",
"slide_up": "slideup",
"slidedown": "slidedown",
"slide_down": "slidedown",
"slide": "slideleft", # 默认向左滑
# 缩放
"zoom": "zoomin",
"zoomin": "zoomin",
"zoomout": "zoomout",
# 擦除系列
"wipe": "wipeleft", # 默认向左擦
"wipeleft": "wipeleft",
"wiperight": "wiperight",
"wipeup": "wipeup",
"wipedown": "wipedown",
# 特殊效果
"circlecrop": "circlecrop",
"circle": "circlecrop",
"rectcrop": "rectcrop",
"rect": "rectcrop",
}
# 向后兼容:DEFAULT_TRANSITION_DURATION 从 domain 层导出
DEFAULT_TRANSITION_DURATION = _default_transition_duration_base
DEFAULT_TRANSITION_DURATION = 0.5
# 向后兼容:薄包装函数
def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
return _chain_filters_base(filters, output_label, input_label=input_label)
def resolve_xfade_transition(transition_name: Any) -> str:
return _resolve_xfade_transition_base(transition_name)
def build_xfade_filter_chain(
clip_durations: list[float],
clip_video_labels: list[str],
transitions: list[str],
*,
transition_duration: float = DEFAULT_TRANSITION_DURATION,
output_label: str = "outv",
) -> tuple[str, float]:
return _build_xfade_filter_chain_base(
clip_durations,
clip_video_labels,
transitions,
transition_duration=transition_duration,
output_label=output_label,
)
# ── FFprobe 探测 ──────────────────────────────────────────────────────────────
@@ -304,111 +306,3 @@ def normalize_video(
]
run_ffmpeg(command)
return {"width": width, "height": height, "path": output_path}
# ── xfade / concat 滤镜构建 ──────────────────────────────────────────────────
def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
"""将滤镜列表串联为 FFmpeg 滤镜字符串。
例:chain_filters(["scale=1280:720", "fps=25"], "v0")
→ "[0:v]scale=1280:720,fps=25[v0]"
"""
filter_body = ",".join(filters)
return f"[{input_label}]{filter_body}[{output_label}]"
def resolve_xfade_transition(transition_name: str) -> str:
"""将转场效果名称映射为 FFmpeg xfade transition 名称。
支持 TransitionEffect 枚举值和字符串名称,未知值回退到 "fade"。
"""
# 兼容 TransitionEffect 枚举(有 .value 属性)
if hasattr(transition_name, "value"):
transition_name = transition_name.value
return XFADE_TRANSITION_MAP.get(transition_name, "fade")
def build_xfade_filter_chain(
clip_durations: list[float],
clip_video_labels: list[str],
transitions: list[str],
*,
transition_duration: float = DEFAULT_TRANSITION_DURATION,
output_label: str = "outv",
) -> tuple[str, float]:
"""构建 xfade 转场滤镜链。
对每步 xfade 自动钳制 transition duration,确保
``offset + td ≤ first_input_duration``,避免 FFmpeg exit 234。
Args:
clip_durations: 每个片段的时长(必须与 trim 后的实际时长一致)
clip_video_labels: 每个片段的视频流标签(如 "v0", "v1")
transitions: 每个片段对应的转场效果(第一个片段的转场被忽略)
transition_duration: 转场时长(秒)
output_label: 最终输出标签
Returns:
(filter_string, estimated_total_duration)
"""
n = len(clip_durations)
parts: list[str] = []
if n == 0:
return "", 0.0
if n == 1:
parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]")
return ";".join(parts), clip_durations[0]
# xfade 链 — 每步动态钳制 td,防止 offset + td > first_input_duration
cumulative = 0.0
prev_label = clip_video_labels[0]
total_transition = 0.0 # 累计已使用的转场时长
for i in range(1, n):
cumulative += clip_durations[i - 1]
# 当前 xfade 的第一个输入时长
if i == 1:
first_input_dur = clip_durations[0]
else:
first_input_dur = cumulative - total_transition
# 原始 offset 计算
offset = max(0.0, cumulative - transition_duration * i)
# 安全钳制:offset + td 不能超过第一个输入的时长
available = max(0.0, first_input_dur - offset)
safe_td = min(transition_duration, available)
# 同时不能超过剩余总时长
remaining = max(0.0, sum(clip_durations) - cumulative)
safe_td = min(safe_td, remaining)
# 同时不能超过当前第二个输入(单个片段)的时长
safe_td = min(safe_td, clip_durations[i])
safe_td = max(0.001, safe_td) # 至少 1ms,避免 td=0
transition = transitions[i] if i < len(transitions) else "cut"
xfade_transition = resolve_xfade_transition(transition)
if i == n - 1:
out_label = output_label
else:
out_label = f"xf{i}"
parts.append(
f"[{prev_label}][{clip_video_labels[i]}]"
f"xfade=transition={xfade_transition}"
f":duration={safe_td:.3f}"
f":offset={offset:.3f}"
f"[{out_label}]"
)
prev_label = out_label
total_transition += safe_td
# 总时长减去转场重叠部分
total_duration = sum(clip_durations) - total_transition
return ";".join(parts), max(0.0, total_duration)
@@ -2,126 +2,28 @@
支持对音频进行背景噪音消除、人声增强,适用于语音录制、采访等场景。
使用方式:
config = NoiseReductionConfig(level="medium")
engine = NoiseReductionEngine(config)
filter_str = engine.build_filter(input_label, output_label)
# 结果: [0:a]afftdn=nf=-25[out]
降级策略:
- 参数越界自动钳制
- FFmpeg 不支持 afftdn 时,调用方可捕获异常并跳过
领域模型已抽离至 packages/domain/noise_reduction_config.py,本模块保留薄包装以维持向后兼容。
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from packages.domain.noise_reduction_config import (
NoiseReductionConfig,
NoiseReductionLevel,
)
from packages.domain.noise_reduction_config import (
apply_noise_reduction_if_needed as _apply_noise_reduction_if_needed_base,
)
from packages.domain.noise_reduction_config import build_afftdn_filter as _build_afftdn_filter_base # noqa: F401 — 向后兼容
from packages.domain.noise_reduction_config import build_arnndn_filter as _build_arnndn_filter_base
logger = logging.getLogger(__name__)
# ── 降噪等级 ──────────────────────────────────────────────────────────────────
class NoiseReductionLevel(str, Enum):
"""降噪等级预设。"""
LOW = "low" # 轻度降噪,保留细节,适合轻微背景噪音
MEDIUM = "medium" # 中度降噪,平衡效果和音质
HIGH = "high" # 高度降噪,适合嘈杂环境,可能轻微影响音质
CUSTOM = "custom" # 自定义参数
# 各等级对应的降噪参数(afftdn 的 noise floor,单位 dB)
# 值越大(越接近 0),降噪越强;值越小(越负),降噪越弱
_LEVEL_PARAMS = {
NoiseReductionLevel.LOW: {
"nf": -35, # 噪音阈值(dB),越负越保守
"tn": -10, # 噪音频谱平滑度
"tr": 50, # 时间分辨率(ms)
},
NoiseReductionLevel.MEDIUM: {
"nf": -25,
"tn": -10,
"tr": 50,
},
NoiseReductionLevel.HIGH: {
"nf": -15,
"tn": -5,
"tr": 30,
},
}
# ── 配置模型 ──────────────────────────────────────────────────────────────────
@dataclass
class NoiseReductionConfig:
"""音频降噪配置。
Attributes:
enabled: 是否启用降噪
level: 降噪等级 low/medium/high/custom
noise_floor: 自定义噪音阈值(dB),仅 level=custom 时有效,范围 -60 ~ -5
voice_enhance: 是否启用人声增强
output_format: 输出格式描述(内部使用)
"""
enabled: bool = False
level: NoiseReductionLevel = NoiseReductionLevel.MEDIUM
noise_floor: float = -25.0 # dB
voice_enhance: bool = False
@classmethod
def from_dict(cls, data: dict | None) -> "NoiseReductionConfig":
"""从字典解析配置,参数越界自动钳制。"""
if not data or not data.get("enabled", False):
return cls(enabled=False)
level_str = str(data.get("level", "medium")).lower()
try:
level = NoiseReductionLevel(level_str)
except ValueError:
level = NoiseReductionLevel.MEDIUM
try:
noise_floor = float(data.get("noise_floor", -25.0))
except (TypeError, ValueError):
noise_floor = -25.0
voice_enhance = bool(data.get("voice_enhance", False))
# 钳制到合法范围
noise_floor = max(-60.0, min(-5.0, noise_floor))
return cls(
enabled=True,
level=level,
noise_floor=noise_floor,
voice_enhance=voice_enhance,
)
def has_effect(self) -> bool:
"""判断是否有实际降噪效果。"""
return self.enabled
def get_effective_noise_floor(self) -> float:
"""获取实际生效的噪音阈值(dB)。"""
if self.level == NoiseReductionLevel.CUSTOM:
return self.noise_floor
params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[NoiseReductionLevel.MEDIUM])
return float(params["nf"])
# ── 引擎实现 ──────────────────────────────────────────────────────────────────
class NoiseReductionEngine:
"""音频降噪引擎。
"""音频降噪引擎 — 薄包装,实际逻辑在 domain.noise_reduction_config.
基于 FFmpeg afftdn(Audio FFt Denoiser)滤镜实现:
- 使用短时傅里叶变换分析音频频谱
@@ -133,97 +35,40 @@ class NoiseReductionEngine:
self.config = config
def build_filter(self, input_label: str, output_label: str) -> str:
"""构建音频降噪滤镜字符串。
"""构建音频降噪滤镜字符串.
Args:
input_label: 输入标签,如 "[0:a]" 或 "[a0]"
output_label: 输出标签,如 "[nr0]"
Returns:
FFmpeg 滤镜字符串,如 "[a0]afftdn=nf=-25:tn=-10:tr=50[nr0]"
Raises:
ValueError: 配置无效时抛出(调用方应捕获并降级)
FFmpeg 滤镜字符串
"""
if not self.config.has_effect():
return f"{input_label}anull{output_label}"
# 获取参数
if self.config.level == NoiseReductionLevel.CUSTOM:
nf = self.config.noise_floor
tn = -10 # 默认频谱平滑度
tr = 50 # 默认时间分辨率
else:
params = _LEVEL_PARAMS.get(
self.config.level,
_LEVEL_PARAMS[NoiseReductionLevel.MEDIUM],
)
nf = float(params["nf"])
tn = float(params["tn"])
tr = float(params["tr"])
# 构建 afftdn 滤镜
# nf: noise floor (dB)
# tn: temporal noise floor smoothing (dB)
# tr: time resolution (ms)
filter_parts = [f"afftdn=nf={nf}:tn={tn}:tr={tr}"]
# 人声增强:通过 highpass + 轻微压缩实现
if self.config.voice_enhance:
# 1. 高通滤波,去除低频噪音
filter_parts.append("highpass=f=80")
# 2. 轻微压缩,提升人声清晰度
filter_parts.append("acompressor=threshold=-20:ratio=2:attack=5:release=50")
# 3. 响度归一化
filter_parts.append("loudnorm=I=-16:TP=-1.5:LRA=11")
filter_str = f"{input_label}{','.join(filter_parts)}{output_label}"
return filter_str
return _build_afftdn_filter_base(self.config, input_label, output_label)
def build_filter_arnndn(self, input_label: str, output_label: str, model_file: str) -> str:
"""使用 RNN 降噪滤镜(arnndn,效果更好但需要模型文件)。
注意:需要额外下载 RNNNoise 模型文件,默认使用 afftdn(无需额外依赖)。
"""使用 RNN 降噪滤镜(arnndn,效果更好但需要模型文件).
Args:
input_label: 输入标签
output_label: 输出标签
model_file: RNNNoise 模型文件路径(.rnnn 格式)
model_file: RNNNoise 模型文件路径
Returns:
FFmpeg 滤镜字符串
"""
if not self.config.has_effect():
return f"{input_label}anull{output_label}"
return f"{input_label}arnndn=m={model_file}{output_label}"
return _build_arnndn_filter_base(self.config, input_label, output_label, model_file)
def apply_noise_reduction_if_needed(
config_data: dict | None,
input_label: str,
output_label: str,
) -> Optional[str]:
"""便捷函数:根据配置判断是否需要应用音频降噪。
def apply_noise_reduction_if_needed(config_data, input_label: str, output_label: str):
"""便捷函数:根据配置判断是否需要应用音频降噪.
Args:
config_data: 降噪配置字典(从 plan.config.audio_noise_reduction 或 clip.config.noise_reduction 读取)
config_data: 降噪配置字典
input_label: 输入标签
output_label: 输出标签
Returns:
滤镜字符串,不需要降噪时返回 None
"""
if not config_data:
return None
try:
config = NoiseReductionConfig.from_dict(config_data)
if not config.has_effect():
return None
engine = NoiseReductionEngine(config)
return engine.build_filter(input_label, output_label)
except Exception as e:
logger.warning("[noise-reduction] 应用降噪失败,跳过: %s", e)
return None
return _apply_noise_reduction_if_needed_base(config_data, input_label, output_label)
+37 -203
View File
@@ -1,8 +1,8 @@
"""ASS 字幕生成模块 — 从 unified_render_service.py 拆分.
"""ASS 字幕生成模块 — 薄包装,实际逻辑在 packages/domain/ass_subtitle_builder.py.
职责:
- 将 title / subtitle 配置转换为 ASS 字幕文件
- 提供样式计算(颜色、对齐、描边/阴影)
- 文件IO 在此模块,纯逻辑已抽离到 domain
- 供 UnifiedRenderService._maybe_generate_ass 调用
"""
@@ -12,107 +12,40 @@ import logging
from pathlib import Path
from typing import Any
from packages.domain.ass_subtitle_builder import (
TITLE_MARGIN_BOTTOM,
TITLE_MARGIN_SIDE,
TITLE_MARGIN_TOP,
build_ass_content,
)
from packages.domain.ass_subtitle_builder import build_ass_style as _build_ass_style_base # noqa: F401 — 向后兼容
from packages.domain.ass_subtitle_builder import escape_ass_text as _escape_ass_text_base
from packages.domain.ass_subtitle_builder import format_ass_time as _format_ass_time_base
from packages.domain.ass_subtitle_builder import hex_to_ass_color as _hex_to_ass_color_base
from packages.domain.ass_subtitle_builder import position_to_ass_alignment as _position_to_ass_alignment_base
logger = logging.getLogger(__name__)
# ── 常量 ──────────────────────────────────────────────────────────────────────
# Title/Subtitle 默认边距(像素)
TITLE_MARGIN_TOP = 60
TITLE_MARGIN_BOTTOM = 60
TITLE_MARGIN_SIDE = 40
# ── ASS 字幕工具 ─────────────────────────────────────────────────────────────
# 向后兼容:模块级函数保留为薄包装
def _hex_to_ass_color(hex_color: str) -> str:
"""将 HEX 颜色(#RRGGBB)转换为 ASS &HBBGGRR 格式。"""
hex_color = hex_color.lstrip("#")
if len(hex_color) != 6:
return "&H000000"
r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
return f"&H{b.upper()}{g.upper()}{r.upper()}"
return _hex_to_ass_color_base(hex_color)
def _position_to_ass_alignment(position: str) -> int:
"""将文字位置映射为 ASS \\an 对齐编号。
ASS 对齐编号(数字小键盘布局):
7 8 9
4 5 6
1 2 3
"""
mapping = {
"top": 8, # 顶部居中
"center": 5, # 居中
"bottom": 2, # 底部居中
}
return mapping.get(position, 8)
return _position_to_ass_alignment_base(position)
def _build_ass_style(
style_name: str,
*,
font_name: str = "思源黑体",
font_size: int = 48,
primary_color: str = "&H00FFFFFF",
outline_color: str = "&H00000000",
outline_width: float = 1.0,
shadow_blur: float = 0.0,
shadow_offset: tuple[int, int] = (0, 0),
bold: bool = False,
italic: bool = False,
alignment: int = 8,
margin_v: int = 60,
margin_l: int = 40,
margin_r: int = 40,
) -> str:
"""构建 ASS Style 行。
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour,
Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle,
BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
"""
bold_val = -1 if bold else 0
italic_val = -1 if italic else 0
# BackColour 用于阴影(BorderStyle=1 时 outline + shadow)
back_color = primary_color # 阴影颜色默认同文字色(带透明度由阴影模糊控制)
# Shadow 值:ASS 中 Shadow 字段是阴影偏移距离(像素),
# 我们用 shadow_offset[1] 作为纵向偏移,模糊由 BorderStyle=3 实现
# 简化:BorderStyle=1(outline + drop shadow),Shadow 字段表示阴影深度
shadow_depth = shadow_offset[1] if shadow_blur > 0 else 0
return (
f"Style: {style_name},{font_name},{font_size},{primary_color},"
f"&H000000FF,{outline_color},{back_color},"
f"{bold_val},{italic_val},0,0,100,100,0,0,"
f"1,{outline_width},{shadow_depth},{alignment},"
f"{margin_l},{margin_r},{margin_v},1"
)
def _build_ass_style(*args, **kwargs) -> str:
return _build_ass_style_base(*args, **kwargs)
def _escape_ass_text(text: str) -> str:
r"""转义 ASS 文本中的特殊字符。
ASS 中换行用 \N(硬换行)或 \n(软换行),
大括号 {} 用于覆盖样式,需要转义。
"""
# 将实际换行转为 ASS 硬换行
text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N")
# 转义大括号(ASS 用它做样式覆盖标签)
text = text.replace("{", "(").replace("}", ")")
return text
return _escape_ass_text_base(text)
def _format_ass_time(seconds: float) -> str:
"""将秒数格式化为 ASS 时间格式 H:MM:SS.cc。"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = seconds % 60
return f"{hours}:{minutes:02d}:{secs:05.2f}"
return _format_ass_time_base(seconds)
def generate_ass_subtitles(
@@ -126,130 +59,31 @@ def generate_ass_subtitles(
subtitle_text: str = "",
subtitle_config: dict[str, Any] | None = None,
) -> Path:
"""生成 ASS 字幕文件。
支持 Title(标题)和 Subtitle(字幕)两种字幕类型,
各自可独立配置样式、位置和内容。
"""生成 ASS 字幕文件.
Args:
output_path: 输出 ASS 文件路径
video_width: 视频宽度(用于 ASS PlayResX)
video_height: 视频高度(用于 ASS PlayResY)
video_duration: 视频总时长(秒),字幕显示整个时长
video_width: 视频宽度
video_height: 视频高度
video_duration: 视频总时长(秒)
title_text: 标题文本
title_config: 标题样式配置(TitleConfig dict)
title_config: 标题样式配置
subtitle_text: 字幕文本
subtitle_config: 字幕样式配置(SubtitleConfig dict)
subtitle_config: 字幕样式配置
Returns:
生成的 ASS 文件路径
"""
title_config = title_config or {}
subtitle_config = subtitle_config or {}
title_enabled = title_config.get("enabled", True) and bool(title_text.strip())
subtitle_enabled = subtitle_config.get("enabled", True) and bool(subtitle_text.strip())
if not title_enabled and not subtitle_enabled:
# 没有字幕,生成空文件(仍返回路径,调用方自行判断是否使用)
output_path.write_text("", encoding="utf-8")
return output_path
styles: list[str] = []
events: list[str] = []
# ── Title 样式与事件 ──────────────────────────────────────────────────
if title_enabled:
title_color = _hex_to_ass_color(title_config.get("color", "#ffffff"))
title_stroke = title_config.get("stroke", {}) or {}
title_shadow = title_config.get("shadow", {}) or {}
stroke_color = _hex_to_ass_color(title_stroke.get("color", "#000000"))
stroke_width = float(title_stroke.get("width", 1)) if title_stroke.get("enabled", False) else 0.0
shadow_blur = float(title_shadow.get("blur", 4)) if title_shadow.get("enabled", False) else 0.0
shadow_offset = (
title_shadow.get("offset_x", 2) if title_shadow.get("enabled", False) else 0,
title_shadow.get("offset_y", 2) if title_shadow.get("enabled", False) else 0,
)
title_alignment = _position_to_ass_alignment(title_config.get("position", "top"))
styles.append(
_build_ass_style(
"TitleStyle",
font_name=title_config.get("font", "思源黑体"),
font_size=int(title_config.get("size", 48)),
primary_color=title_color,
outline_color=stroke_color,
outline_width=stroke_width,
shadow_blur=shadow_blur,
shadow_offset=shadow_offset,
bold=bool(title_config.get("bold", True)),
italic=bool(title_config.get("italic", False)),
alignment=title_alignment,
margin_v=TITLE_MARGIN_TOP,
margin_l=TITLE_MARGIN_SIDE,
margin_r=TITLE_MARGIN_SIDE,
)
)
# 转义 ASS 特殊字符
safe_title_text = _escape_ass_text(title_text)
events.append(
"Dialogue: 0,0:00:00.00," f"{_format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}"
)
# ── Subtitle 样式与事件 ───────────────────────────────────────────────
if subtitle_enabled:
sub_color = _hex_to_ass_color(subtitle_config.get("color", "#ffffff"))
sub_alignment = _position_to_ass_alignment(subtitle_config.get("position", "bottom"))
styles.append(
_build_ass_style(
"SubtitleStyle",
font_name=subtitle_config.get("font", "思源黑体"),
font_size=int(subtitle_config.get("size", 24)),
primary_color=sub_color,
outline_color="&H00000000",
outline_width=1.0,
shadow_blur=0.0,
shadow_offset=(0, 0),
bold=False,
italic=False,
alignment=sub_alignment,
margin_v=TITLE_MARGIN_BOTTOM,
margin_l=TITLE_MARGIN_SIDE,
margin_r=TITLE_MARGIN_SIDE,
)
)
safe_subtitle_text = _escape_ass_text(subtitle_text)
events.append(
"Dialogue: 0,0:00:00.00,"
f"{_format_ass_time(video_duration)},"
"SubtitleStyle,,0,0,0,,"
f"{safe_subtitle_text}"
)
# ── 组装 ASS 文件 ─────────────────────────────────────────────────────
ass_content = f"""[Script Info]
ScriptType: v4.00+
PlayResX: {video_width}
PlayResY: {video_height}
ScaledBorderAndShadow: yes
WrapStyle: 2
Encoding: UTF-8
[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding # noqa: E501
{chr(10).join(styles)}
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
{chr(10).join(events)}
"""
content = build_ass_content(
video_width=video_width,
video_height=video_height,
video_duration=video_duration,
title_text=title_text,
title_config=title_config,
subtitle_text=subtitle_text,
subtitle_config=subtitle_config,
)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(ass_content, encoding="utf-8")
output_path.write_text(content, encoding="utf-8")
return output_path
@@ -14,14 +14,16 @@ import logging
from pathlib import Path
from typing import Any
from packages.domain.sticker_config import ( # noqa: F401 向后兼容导出
ImageStickerConfig,
from packages.domain.sticker_config import (
POSITION_PRESETS,
STICKER_CATEGORIES,
ImageStickerConfig,
StickerOverlayResult,
TextStickerConfig,
get_sticker_categories as _get_sticker_categories_base,
parse_stickers_from_config as _parse_stickers_base,
)
from packages.domain.sticker_config import get_sticker_categories as _get_sticker_categories_base # noqa: F401 向后兼容导出
from packages.domain.sticker_config import parse_stickers_from_config as _parse_stickers_base
from packages.domain.sticker_config import (
resolve_sticker_position,
)
+28 -298
View File
@@ -5,169 +5,37 @@
- 边界自动钳制(超出素材时长自动修正,不阻断渲染)
- 多段裁剪(一个素材裁剪出多段)
- 音画同步(视频 + 音频同步裁剪)
注:核心领域模型已抽离到 packages/domain/trim_config.py,
本模块保留薄包装层,确保向后兼容。
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any
from packages.domain.trim_config import (
MIN_TRIM_DURATION,
TrimConfig,
TrimSegment,
)
from packages.domain.trim_config import build_audio_trim_filter as _build_audio_trim_filter # noqa: F401 — 向后兼容
from packages.domain.trim_config import build_video_trim_filter as _build_video_trim_filter
from packages.domain.trim_config import (
extract_trim_from_clip_config,
)
from packages.domain.trim_config import parse_segments_from_config as _parse_segments_from_config
from packages.domain.trim_config import resolve_segments as _resolve_segments
logger = logging.getLogger(__name__)
# 最小裁剪时长(秒),低于此值视为无效
MIN_TRIM_DURATION = 0.1
@dataclass
class TrimConfig:
"""裁剪配置.
三选二规则:start_time / end_time / duration 中必须至少给出两个,
第三个会被自动推导。如果三个都给了,以 start_time + duration 为准。
边界保护:
- start_time < 0 → 钳制到 0
- end_time > 素材时长 → 钳制到素材时长
- 计算出的 duration < 最小阈值 → 标记为无效
"""
start_time: float = 0.0 # 入点(素材内时间,秒)
end_time: float = 0.0 # 出点(素材内时间,秒),0 表示未指定
duration: float = 0.0 # 裁剪时长(秒),0 表示未指定
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> TrimConfig | None:
"""从字典构造,无有效裁剪参数时返回 None(不裁剪)."""
if not data:
return None
start = float(data.get("start_time", 0) or 0)
end = float(data.get("end_time", 0) or 0)
dur = float(data.get("duration", 0) or 0)
# 三个参数都没有 → 不裁剪
if start <= 0 and end <= 0 and dur <= 0:
return None
# 至少有两个参数(或一个合理的 start/duration)
# 兼容:只传了 start_time → 从 start 开始取到末尾
# 兼容:只传了 duration → 从 0 开始取 duration
if start > 0 and end <= 0 and dur <= 0:
# 只有 start,取到末尾 → 这是"从某点开始"的语义,算有效
pass
elif dur > 0 and start <= 0 and end <= 0:
# 只有 duration → 从开头取 duration,算有效
pass
elif start <= 0 and end <= 0 and dur <= 0:
return None
return cls(start_time=start, end_time=end, duration=dur)
def validate_and_resolve(self, asset_duration: float) -> TrimConfig:
"""根据素材实际时长,解析并钳制裁剪参数.
返回一个新的 TrimConfig,其中 start_time / end_time / duration 都已确定。
如果裁剪无效(时长为0或负数),仍返回但调用方应检查 is_valid。
"""
start = self.start_time
end = self.end_time
dur = self.duration
# 边界:start 不能为负
if start < 0:
start = 0.0
# 边界:asset_duration 为 0 时保守处理(不裁剪,取全部)
if asset_duration <= 0:
return TrimConfig(start_time=0.0, end_time=0.0, duration=0.0)
# 三选二推导
# 判断顺序很重要:先判断需要两个显式值的组合,最后判断含默认值的
# 情况1:start + end 都有显式值
if start > 0 and end > 0:
if end <= start:
# 出点 <= 入点,无效 → 返回 start 处一个极短片段(调用方会判无效)
return TrimConfig(start_time=start, end_time=start, duration=0.0)
dur = end - start
# 情况2:end + duration 都有显式值
elif end > 0 and dur > 0:
start = end - dur
if start < 0:
start = 0.0
dur = end # 重新计算
# 情况3:start + duration 都有值(start 可以是 0)
elif dur > 0:
end = start + dur
# 情况4:只有 start → 取到素材末尾
elif start > 0 and end <= 0 and dur <= 0:
end = asset_duration
dur = end - start
# 情况5:只有 end → 从开头取到 end
elif end > 0 and start <= 0 and dur <= 0:
start = 0.0
dur = end
else:
# 都没有 → 不裁剪
return TrimConfig(start_time=0.0, end_time=0.0, duration=0.0)
# 边界钳制:end 不能超过素材时长
if end > asset_duration:
end = asset_duration
dur = end - start
# 边界钳制:start 不能超过素材时长
if start >= asset_duration:
start = max(0.0, asset_duration - MIN_TRIM_DURATION)
dur = asset_duration - start
end = asset_duration
# 保证 duration 不为负
if dur < 0:
dur = 0.0
return TrimConfig(start_time=start, end_time=end, duration=dur)
@property
def is_valid(self) -> bool:
"""裁剪是否有效(时长大于最小阈值)."""
return self.duration >= MIN_TRIM_DURATION
@property
def is_noop(self) -> bool:
"""是否等价于不裁剪(从0开始取全部)."""
return self.start_time <= 0 and self.duration <= 0
@property
def trim_from_start(self) -> bool:
"""是否从开头裁剪(start_time == 0)."""
return self.start_time <= 0
@dataclass
class TrimSegment:
"""多段裁剪中的一段."""
segment_id: str # 段 ID(用于生成唯一标签)
trim: TrimConfig # 裁剪配置
order: int = 0 # 排序
@classmethod
def from_dict(cls, data: dict[str, Any], default_order: int = 0) -> TrimSegment:
"""从字典构造."""
return cls(
segment_id=str(data.get("segment_id", "") or f"seg_{default_order}"),
trim=TrimConfig(
start_time=float(data.get("start_time", 0) or 0),
end_time=float(data.get("end_time", 0) or 0),
duration=float(data.get("duration", 0) or 0),
),
order=int(data.get("order", default_order)),
)
class TrimEngine:
"""裁剪引擎 — 生成 FFmpeg trim / atrim 滤镜."""
"""裁剪引擎 — 生成 FFmpeg trim / atrim 滤镜.
薄包装层,实际逻辑委托给 packages.domain.trim_config。
"""
@staticmethod
def build_video_trim_filter(
@@ -175,38 +43,8 @@ class TrimEngine:
trim: TrimConfig,
output_label: str,
) -> str:
"""构建视频裁剪滤镜链.
Args:
input_label: 输入视频标签,如 "[0:v]"
trim: 裁剪配置(已解析钳制)
output_label: 输出视频标签,如 "[v0_trimmed]"
Returns:
FFmpeg filter 字符串,如 "[0:v]trim=start=10:duration=5,setpts=PTS-STARTPTS[v0_trimmed]"
"""
if trim.is_noop:
# 不裁剪,直接直通(仅重置时间戳)
return f"{input_label}setpts=PTS-STARTPTS{output_label}"
parts: list[str] = []
# trim 滤镜参数
trim_args: list[str] = []
if trim.start_time > 0:
trim_args.append(f"start={trim.start_time:.3f}")
if trim.duration > 0:
trim_args.append(f"duration={trim.duration:.3f}")
elif trim.end_time > 0:
# end 用 duration 表示(start 到 end 的时长)
# 但 validate_and_resolve 后应该已经有 duration 了
pass
parts.append(f"trim={':'.join(trim_args)}")
parts.append("setpts=PTS-STARTPTS")
filter_str = f"{input_label}{','.join(parts)}{output_label}"
return filter_str
"""构建视频裁剪滤镜链."""
return _build_video_trim_filter(input_label, trim, output_label)
@staticmethod
def build_audio_trim_filter(
@@ -214,126 +52,18 @@ class TrimEngine:
trim: TrimConfig,
output_label: str,
) -> str:
"""构建音频裁剪滤镜链.
Args:
input_label: 输入音频标签,如 "[0:a]"
trim: 裁剪配置(已解析钳制)
output_label: 输出音频标签,如 "[a0_trimmed]"
Returns:
FFmpeg filter 字符串,如 "[0:a]atrim=start=10:duration=5,asetpts=PTS-STARTPTS[a0_trimmed]"
"""
if trim.is_noop:
return f"{input_label}asetpts=PTS-STARTPTS{output_label}"
parts: list[str] = []
trim_args: list[str] = []
if trim.start_time > 0:
trim_args.append(f"start={trim.start_time:.3f}")
if trim.duration > 0:
trim_args.append(f"duration={trim.duration:.3f}")
parts.append(f"atrim={':'.join(trim_args)}")
parts.append("asetpts=PTS-STARTPTS")
filter_str = f"{input_label}{','.join(parts)}{output_label}"
return filter_str
"""构建音频裁剪滤镜链."""
return _build_audio_trim_filter(input_label, trim, output_label)
@staticmethod
def resolve_segments(
segments: list[TrimSegment],
asset_duration: float,
) -> list[TrimSegment]:
"""解析并钳制多段裁剪配置,过滤无效段.
Args:
segments: 原始段列表
asset_duration: 素材实际时长
Returns:
解析后的有效段列表,按 order 排序
"""
resolved: list[TrimSegment] = []
for i, seg in enumerate(segments):
resolved_trim = seg.trim.validate_and_resolve(asset_duration)
if not resolved_trim.is_valid:
logger.warning("裁剪段无效,跳过: segment_id=%s duration=%.3f", seg.segment_id, resolved_trim.duration)
continue
resolved.append(
TrimSegment(
segment_id=seg.segment_id,
trim=resolved_trim,
order=seg.order if seg.order >= 0 else i,
)
)
resolved.sort(key=lambda s: s.order)
return resolved
"""解析并钳制多段裁剪配置,过滤无效段."""
return _resolve_segments(segments, asset_duration)
@staticmethod
def parse_segments_from_config(config: dict[str, Any] | None) -> list[TrimSegment]:
"""从 clip config 中解析多段裁剪配置.
config 中支持:
- trim_segments: [ {segment_id, start_time, end_time, duration, order}, ... ]
- trim_start / trim_end / trim_duration: 单段裁剪(兼容旧格式)
"""
if not config:
return []
# 优先解析多段
raw_segments = config.get("trim_segments", [])
if raw_segments and isinstance(raw_segments, list):
segments = []
for i, raw in enumerate(raw_segments):
if isinstance(raw, dict):
segments.append(TrimSegment.from_dict(raw, default_order=i))
return segments
# 单段裁剪兼容:从 trim_start/trim_end/trim_duration 构造
has_single = any(k in config for k in ("trim_start", "trim_end", "trim_duration"))
if has_single:
seg = TrimSegment(
segment_id="main",
trim=TrimConfig(
start_time=float(config.get("trim_start", 0) or 0),
end_time=float(config.get("trim_end", 0) or 0),
duration=float(config.get("trim_duration", 0) or 0),
),
order=0,
)
return [seg]
return []
# ── 工具函数 ──────────────────────────────────────────────────────────────────
def extract_trim_from_clip_config(config: dict[str, Any] | None) -> TrimConfig | None:
"""从 clip config 中提取单段裁剪配置.
兼容以下字段名:
- trim_start / trim_end / trim_duration
- start_time / end_time / duration(在 trim 子字典里)
"""
if not config:
return None
# trim 子字典
if "trim" in config and isinstance(config["trim"], dict):
return TrimConfig.from_dict(config["trim"])
# 扁平字段
has_any = any(k in config for k in ("trim_start", "trim_end", "trim_duration"))
if not has_any:
return None
data = {
"start_time": config.get("trim_start", 0),
"end_time": config.get("trim_end", 0),
"duration": config.get("trim_duration", 0),
}
return TrimConfig.from_dict(data)
"""从 clip config 中解析多段裁剪配置."""
return _parse_segments_from_config(config)
+26 -225
View File
@@ -6,135 +6,35 @@
- 9宫格位置 + 边距配置
- 透明度/大小缩放
- 滚动水印(跑马灯)
注:核心领域模型已抽离到 packages/domain/watermark_config.py,
本模块保留薄包装层,确保向后兼容。
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any
from packages.domain.watermark_config import (
WATERMARK_POSITIONS,
WatermarkConfig,
)
from packages.domain.watermark_config import ( # noqa: F401 — 向后兼容
build_image_watermark_filter as _build_image_watermark_filter,
)
from packages.domain.watermark_config import build_text_watermark_filter as _build_text_watermark_filter
from packages.domain.watermark_config import calc_position as _calc_position_base
from packages.domain.watermark_config import calc_scroll_x as _calc_scroll_x_base
logger = logging.getLogger(__name__)
# 9宫格位置枚举
WATERMARK_POSITIONS = {
"top_left": "左上",
"top_center": "中上",
"top_right": "右上",
"center_left": "左中",
"center": "中心",
"center_right": "右中",
"bottom_left": "左下",
"bottom_center": "中下",
"bottom_right": "右下",
}
@dataclass
class WatermarkConfig:
"""水印配置.
mode: "image" 图片水印 | "text" 文字水印
position: 9宫格位置
opacity: 透明度 0.0-1.0
scale: 缩放比例(图片水印),0.1-1.0
margin: 边距(像素)
scroll: 是否滚动(跑马灯)
scroll_speed: 滚动速度(像素/秒)
"""
mode: str = "text" # image | text
position: str = "bottom_right"
# 图片水印
image_path: str = "" # 本地图片路径
scale: float = 0.2 # 相对输出宽度的比例
opacity: float = 0.8 # 0.0-1.0
# 文字水印
text: str = ""
font_size: int = 24
font_color: str = "white"
font_path: str = "" # 字体文件路径
# 边距
margin_x: int = 20
margin_y: int = 20
# 滚动水印
scroll: bool = False
scroll_speed: int = 50 # 像素/秒
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> WatermarkConfig | None:
"""从字典构造,空配置返回 None(不加水印)."""
if not data:
return None
enabled = data.get("enabled", False)
if not enabled:
return None
mode = data.get("mode", "text")
# 图片模式需要 image_path;文字模式需要 text
if mode == "image":
image_path = data.get("image_path", "") or data.get("image", "") or ""
if not image_path:
logger.warning("图片水印缺少 image_path,跳过水印")
return None
elif mode == "text":
text = data.get("text", "") or ""
if not text:
logger.warning("文字水印缺少 text,跳过水印")
return None
position = data.get("position", "bottom_right")
if position not in WATERMARK_POSITIONS:
position = "bottom_right"
return cls(
mode=mode,
position=position,
image_path=str(data.get("image_path", data.get("image", "")) or ""),
scale=float(data.get("scale", 0.2)),
opacity=float(data.get("opacity", 0.8)),
text=str(data.get("text", "") or ""),
font_size=int(data.get("font_size", 24)),
font_color=str(data.get("font_color", "white")),
font_path=str(data.get("font_path", "") or ""),
margin_x=int(data.get("margin_x", 20)),
margin_y=int(data.get("margin_y", 20)),
scroll=bool(data.get("scroll", False)),
scroll_speed=int(data.get("scroll_speed", 50)),
)
def validate(self) -> tuple[bool, str]:
"""校验配置是否有效."""
if self.position not in WATERMARK_POSITIONS:
return False, f"不支持的位置: {self.position}"
if not (0.0 <= self.opacity <= 1.0):
return False, "透明度必须在 0-1 之间"
if self.mode == "image":
if not self.image_path:
return False, "图片水印缺少图片路径"
if not (0.01 <= self.scale <= 1.0):
return False, "缩放比例必须在 0.01-1.0 之间"
elif self.mode == "text":
if not self.text:
return False, "文字水印缺少文字内容"
if self.font_size <= 0:
return False, "字体大小必须大于 0"
else:
return False, f"不支持的水印模式: {self.mode}"
return True, ""
class WatermarkEngine:
"""水印引擎 — 生成 FFmpeg 水印滤镜."""
"""水印引擎 — 生成 FFmpeg 水印滤镜.
薄包装层,实际逻辑委托给 packages.domain.watermark_config。
"""
@staticmethod
def calc_position(
@@ -150,27 +50,7 @@ class WatermarkEngine:
坐标系:左上角为 (0, 0)
"""
if position == "top_left":
return margin_x, margin_y
elif position == "top_center":
return (output_width - wm_width) // 2, margin_y
elif position == "top_right":
return output_width - wm_width - margin_x, margin_y
elif position == "center_left":
return margin_x, (output_height - wm_height) // 2
elif position == "center":
return (output_width - wm_width) // 2, (output_height - wm_height) // 2
elif position == "center_right":
return output_width - wm_width - margin_x, (output_height - wm_height) // 2
elif position == "bottom_left":
return margin_x, output_height - wm_height - margin_y
elif position == "bottom_center":
return (output_width - wm_width) // 2, output_height - wm_height - margin_y
elif position == "bottom_right":
return output_width - wm_width - margin_x, output_height - wm_height - margin_y
else:
# 默认右下角
return output_width - wm_width - margin_x, output_height - wm_height - margin_y
return _calc_position_base(position, output_width, output_height, wm_width, wm_height, margin_x, margin_y)
@staticmethod
def calc_scroll_x(position: str, output_width: int, wm_width: int, speed: int) -> str:
@@ -178,12 +58,7 @@ class WatermarkEngine:
从右向左滚动(跑马灯效果)
"""
# x 从 W 到 -wm_width,整个宽度 + wm_width 的距离
# 使用 overlay 的 enable 表达式
# x = 'W - (t * speed)' → 不对,应该是持续滚动
# 标准跑马灯:x = -w + (t * speed) % (W + w)
# 但 FFmpeg overlay 支持表达式
return f"mod({output_width}-mod({speed}*t\\,{output_width}+{wm_width})"
return _calc_scroll_x_base(position, output_width, wm_width, speed)
@staticmethod
def build_image_watermark_filter(
@@ -208,51 +83,15 @@ class WatermarkEngine:
(filter_complex_str, input_args_list)
input_args 是 ["-i", wm_image_path] 格式
"""
# 计算水印尺寸(按输出宽度比例缩放)
wm_width = int(output_width * config.scale)
wm_height = -1 # 保持比例
wm_filter = f"scale={wm_width}:{wm_height}"
# 透明度处理
if config.opacity < 1.0:
wm_filter += f",format=rgba,colorchannelmixer=aa={config.opacity}"
# 水印预处理标签
wm_pre_label = "[wm_scaled]"
# 计算位置
x, y = WatermarkEngine.calc_position(
config.position,
return _build_image_watermark_filter(
input_video_label,
wm_image_path,
output_width,
output_height,
wm_width,
wm_width, # 高度未知,先用宽度估算
config.margin_x,
config.margin_y,
output_label,
config,
)
# 滚动水印
if config.scroll:
# 从右向左滚动:x = W - (t * speed) mod (W + wm_w)
# 使用 overlay 表达式
x_expr = f"{output_width}-mod({config.scroll_speed}*t\\,{output_width}+{wm_width}"
y_expr = str(y)
overlay_expr = f"x={x_expr}:y={y_expr}"
else:
overlay_expr = f"x={x}:y={y}"
# 构建滤镜
# 先缩放水印图
filter_parts = [
f"[1:v]{wm_filter}{wm_pre_label}",
f"{input_video_label}{wm_pre_label}overlay={overlay_expr}{output_label}",
]
filter_complex = ";".join(filter_parts)
input_args = ["-i", wm_image_path]
return filter_complex, input_args
@staticmethod
def build_text_watermark_filter(
input_video_label: str,
@@ -273,42 +112,4 @@ class WatermarkEngine:
Returns:
FFmpeg filter 字符串
"""
# 转义文字中的特殊字符
text = config.text.replace(":", "\\:").replace("'", "\\'")
# 字体配置
font_config = []
if config.font_path:
font_path_escaped = config.font_path.replace(":", "\\:").replace("'", "\\'")
font_config.append(f"fontfile='{font_path_escaped}'")
font_config.append(f"fontsize={config.font_size}")
font_config.append(f"fontcolor={config.font_color}@{config.opacity}")
# 估算文字宽高(粗略估算,用于位置计算)
# 每个汉字约等于 font_size 宽高
approx_w = len(config.text) * config.font_size
approx_h = config.font_size
# 位置计算
x, y = WatermarkEngine.calc_position(
config.position,
output_width,
output_height,
approx_w,
approx_h,
config.margin_x,
config.margin_y,
)
# 滚动水印
if config.scroll:
x_expr = f"w-mod({config.scroll_speed}*t\\,W+w)"
pos_config = [f"x={x_expr}", f"y={y}"]
else:
pos_config = [f"x={x}", f"y={y}"]
# 组装 drawtext
drawtext_parts = [f"text='{text}'"] + font_config + pos_config
drawtext = "drawtext=" + ":".join(drawtext_parts)
return f"{input_video_label}{drawtext}{output_label}"
return _build_text_watermark_filter(input_video_label, output_label, config, output_width, output_height)
+306
View File
@@ -0,0 +1,306 @@
"""ASS 字幕构建领域模型 — 纯逻辑,无文件IO依赖.
抽离自 render_subtitles.py,包含:
- 颜色转换(hex → ASS &HBBGGRR)
- 位置对齐映射
- ASS Style 行构建
- 文本转义
- 时间格式化
- 完整 ASS 内容生成(返回字符串,不写文件)
"""
from __future__ import annotations
import logging
from typing import Any
logger = logging.getLogger(__name__)
# ── 常量 ──────────────────────────────────────────────────────────────────────
# Title/Subtitle 默认边距(像素)
TITLE_MARGIN_TOP = 60
TITLE_MARGIN_BOTTOM = 60
TITLE_MARGIN_SIDE = 40
# ── 颜色转换 ──────────────────────────────────────────────────────────────────
def hex_to_ass_color(hex_color: str) -> str:
"""将 HEX 颜色(#RRGGBB)转换为 ASS &HBBGGRR 格式.
Args:
hex_color: HEX 颜色字符串,支持 #RRGGBB 或 RRGGBB 格式
Returns:
ASS 格式颜色,如 &H0000FF(红色)
"""
hex_color = hex_color.lstrip("#")
if len(hex_color) != 6:
return "&H000000"
r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
return f"&H{b.upper()}{g.upper()}{r.upper()}"
# ── 位置对齐 ──────────────────────────────────────────────────────────────────
def position_to_ass_alignment(position: str) -> int:
"""将文字位置映射为 ASS \\an 对齐编号.
ASS 对齐编号(数字小键盘布局):
7 8 9
4 5 6
1 2 3
Args:
position: 位置字符串 top/center/bottom
Returns:
ASS 对齐编号,默认 8(顶部居中)
"""
mapping = {
"top": 8,
"center": 5,
"bottom": 2,
}
return mapping.get(position, 8)
# ── Style 行构建 ──────────────────────────────────────────────────────────────
def build_ass_style(
style_name: str,
*,
font_name: str = "思源黑体",
font_size: int = 48,
primary_color: str = "&H00FFFFFF",
outline_color: str = "&H00000000",
outline_width: float = 1.0,
shadow_blur: float = 0.0,
shadow_offset: tuple[int, int] = (0, 0),
bold: bool = False,
italic: bool = False,
alignment: int = 8,
margin_v: int = 60,
margin_l: int = 40,
margin_r: int = 40,
) -> str:
"""构建 ASS Style 行.
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour,
Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle,
BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Args:
style_name: 样式名称
font_name: 字体名称
font_size: 字体大小
primary_color: 主色(文字颜色)
outline_color: 描边颜色
outline_width: 描边宽度
shadow_blur: 阴影模糊度(>0 时启用阴影)
shadow_offset: 阴影偏移 (x, y)
bold: 是否粗体
italic: 是否斜体
alignment: 对齐方式(ASS \an 编号)
margin_v: 垂直边距
margin_l: 左边距
margin_r: 右边距
Returns:
完整的 Style: 行字符串
"""
bold_val = -1 if bold else 0
italic_val = -1 if italic else 0
# BackColour 用于阴影(BorderStyle=1 时 outline + shadow)
back_color = primary_color
# Shadow 深度:shadow_offset[1] 作为纵向偏移
shadow_depth = shadow_offset[1] if shadow_blur > 0 else 0
return (
f"Style: {style_name},{font_name},{font_size},{primary_color},"
f"&H000000FF,{outline_color},{back_color},"
f"{bold_val},{italic_val},0,0,100,100,0,0,"
f"1,{outline_width},{shadow_depth},{alignment},"
f"{margin_l},{margin_r},{margin_v},1"
)
# ── 文本转义 ──────────────────────────────────────────────────────────────────
def escape_ass_text(text: str) -> str:
r"""转义 ASS 文本中的特殊字符.
ASS 中换行用 \N(硬换行)或 \n(软换行),
大括号 {} 用于覆盖样式,需要转义.
Args:
text: 原始文本
Returns:
转义后的 ASS 文本
"""
# 将实际换行转为 ASS 硬换行
text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N")
# 转义大括号(ASS 用它做样式覆盖标签)
text = text.replace("{", "(").replace("}", ")")
return text
# ── 时间格式化 ────────────────────────────────────────────────────────────────
def format_ass_time(seconds: float) -> str:
"""将秒数格式化为 ASS 时间格式 H:MM:SS.cc.
Args:
seconds: 秒数
Returns:
ASS 格式时间,如 "1:23:45.67"
"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = seconds % 60
return f"{hours}:{minutes:02d}:{secs:05.2f}"
# ── 完整 ASS 内容生成 ─────────────────────────────────────────────────────────
def build_ass_content(
*,
video_width: int,
video_height: int,
video_duration: float,
title_text: str = "",
title_config: dict[str, Any] | None = None,
subtitle_text: str = "",
subtitle_config: dict[str, Any] | None = None,
) -> str:
"""生成 ASS 字幕文件内容(纯字符串,不写文件).
支持 Title(标题)和 Subtitle(字幕)两种字幕类型,
各自可独立配置样式、位置和内容.
Args:
video_width: 视频宽度(用于 ASS PlayResX)
video_height: 视频高度(用于 ASS PlayResY)
video_duration: 视频总时长(秒),字幕显示整个时长
title_text: 标题文本
title_config: 标题样式配置
subtitle_text: 字幕文本
subtitle_config: 字幕样式配置
Returns:
完整的 ASS 文件内容字符串;无字幕时返回空字符串
"""
title_config = title_config or {}
subtitle_config = subtitle_config or {}
title_enabled = title_config.get("enabled", True) and bool(title_text.strip())
subtitle_enabled = subtitle_config.get("enabled", True) and bool(subtitle_text.strip())
if not title_enabled and not subtitle_enabled:
return ""
styles: list[str] = []
events: list[str] = []
# ── Title 样式与事件 ──────────────────────────────────────────────────
if title_enabled:
title_color = hex_to_ass_color(title_config.get("color", "#ffffff"))
title_stroke = title_config.get("stroke", {}) or {}
title_shadow = title_config.get("shadow", {}) or {}
stroke_color = hex_to_ass_color(title_stroke.get("color", "#000000"))
stroke_width = float(title_stroke.get("width", 1)) if title_stroke.get("enabled", False) else 0.0
shadow_blur = float(title_shadow.get("blur", 4)) if title_shadow.get("enabled", False) else 0.0
shadow_offset = (
title_shadow.get("offset_x", 2) if title_shadow.get("enabled", False) else 0,
title_shadow.get("offset_y", 2) if title_shadow.get("enabled", False) else 0,
)
title_alignment = position_to_ass_alignment(title_config.get("position", "top"))
styles.append(
build_ass_style(
"TitleStyle",
font_name=title_config.get("font", "思源黑体"),
font_size=int(title_config.get("size", 48)),
primary_color=title_color,
outline_color=stroke_color,
outline_width=stroke_width,
shadow_blur=shadow_blur,
shadow_offset=shadow_offset,
bold=bool(title_config.get("bold", True)),
italic=bool(title_config.get("italic", False)),
alignment=title_alignment,
margin_v=TITLE_MARGIN_TOP,
margin_l=TITLE_MARGIN_SIDE,
margin_r=TITLE_MARGIN_SIDE,
)
)
safe_title_text = escape_ass_text(title_text)
events.append(
"Dialogue: 0,0:00:00.00," f"{format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}"
)
# ── Subtitle 样式与事件 ───────────────────────────────────────────────
if subtitle_enabled:
sub_color = hex_to_ass_color(subtitle_config.get("color", "#ffffff"))
sub_alignment = position_to_ass_alignment(subtitle_config.get("position", "bottom"))
styles.append(
build_ass_style(
"SubtitleStyle",
font_name=subtitle_config.get("font", "思源黑体"),
font_size=int(subtitle_config.get("size", 24)),
primary_color=sub_color,
outline_color="&H00000000",
outline_width=1.0,
shadow_blur=0.0,
shadow_offset=(0, 0),
bold=False,
italic=False,
alignment=sub_alignment,
margin_v=TITLE_MARGIN_BOTTOM,
margin_l=TITLE_MARGIN_SIDE,
margin_r=TITLE_MARGIN_SIDE,
)
)
safe_subtitle_text = escape_ass_text(subtitle_text)
events.append(
"Dialogue: 0,0:00:00.00,"
f"{format_ass_time(video_duration)},"
"SubtitleStyle,,0,0,0,,"
f"{safe_subtitle_text}"
)
# ── 组装 ASS 文件 ─────────────────────────────────────────────────────
return f"""[Script Info]
ScriptType: v4.00+
PlayResX: {video_width}
PlayResY: {video_height}
ScaledBorderAndShadow: yes
WrapStyle: 2
Encoding: UTF-8
[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding # noqa: E501
{chr(10).join(styles)}
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
{chr(10).join(events)}
"""
+287
View File
@@ -0,0 +1,287 @@
"""绿幕抠像配置领域模型 — 纯逻辑,无FFmpeg依赖.
抽离自 chroma_key_engine.py,包含:
- ChromaKeyConfig 数据类(解析/钳制/效果判断)
- 预设配置(绿幕/蓝幕/红幕等)
- 颜色归一化
- colorkey / chromakey 滤镜构建
- 便捷函数(apply_chroma_key_if_needed)
"""
from __future__ import annotations
import logging
import re
from dataclasses import dataclass
from typing import Any
logger = logging.getLogger(__name__)
# ── 预设配置 ──────────────────────────────────────────────────────────────────
# 常见绿幕/蓝幕预设
CHROMA_KEY_PRESETS: dict[str, dict[str, Any]] = {
"green_screen": {
"key_color": "#00FF00",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.5,
},
"blue_screen": {
"key_color": "#0000FF",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.5,
},
"red_screen": {
"key_color": "#FF0000",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.0,
},
"precise_green": {
"key_color": "#00FF00",
"similarity": 0.2,
"blend": 0.05,
"spill_suppress": 0.3,
},
"soft_green": {
"key_color": "#00FF00",
"similarity": 0.45,
"blend": 0.2,
"spill_suppress": 0.5,
},
}
VALID_PRESETS = set(CHROMA_KEY_PRESETS.keys())
# 参数范围
MIN_SIMILARITY = 0.01
MAX_SIMILARITY = 1.0
MIN_BLEND = 0.0
MAX_BLEND = 1.0
MIN_SPILL_SUPPRESS = 0.0
MAX_SPILL_SUPPRESS = 1.0
# 默认值
DEFAULT_KEY_COLOR = "#00FF00"
DEFAULT_SIMILARITY = 0.3
DEFAULT_BLEND = 0.1
DEFAULT_SPILL_SUPPRESS = 0.0
# ── 配置模型 ──────────────────────────────────────────────────────────────────
@dataclass
class ChromaKeyConfig:
"""绿幕抠像配置.
Attributes:
enabled: 是否启用抠像
key_color: 要抠除的颜色,支持 hex 格式(如 "#00FF00")或颜色名
similarity: 颜色相似度阈值 0.01~1.0,值越大抠除范围越大
blend: 边缘平滑/混合度 0.0~1.0,值越大边缘越柔和
spill_suppress: 溢色抑制 0.0~1.0,减少边缘的绿幕反光
"""
enabled: bool = False
key_color: str = DEFAULT_KEY_COLOR
similarity: float = DEFAULT_SIMILARITY
blend: float = DEFAULT_BLEND
spill_suppress: float = DEFAULT_SPILL_SUPPRESS
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> ChromaKeyConfig:
"""从字典解析配置,参数越界自动钳制."""
if not data or not data.get("enabled", False):
return cls(enabled=False)
key_color = str(data.get("key_color", DEFAULT_KEY_COLOR)).strip()
def _safe_float(val: Any, default: float) -> float:
try:
return float(val)
except (TypeError, ValueError):
return default
similarity = _safe_float(data.get("similarity", DEFAULT_SIMILARITY), DEFAULT_SIMILARITY)
blend = _safe_float(data.get("blend", DEFAULT_BLEND), DEFAULT_BLEND)
spill_suppress = _safe_float(data.get("spill_suppress", DEFAULT_SPILL_SUPPRESS), DEFAULT_SPILL_SUPPRESS)
# 钳制到合法范围
similarity = max(MIN_SIMILARITY, min(MAX_SIMILARITY, similarity))
blend = max(MIN_BLEND, min(MAX_BLEND, blend))
spill_suppress = max(MIN_SPILL_SUPPRESS, min(MAX_SPILL_SUPPRESS, spill_suppress))
return cls(
enabled=True,
key_color=key_color,
similarity=similarity,
blend=blend,
spill_suppress=spill_suppress,
)
@classmethod
def from_preset(cls, preset_name: str) -> ChromaKeyConfig | None:
"""从预设名称创建配置."""
preset = CHROMA_KEY_PRESETS.get(preset_name)
if not preset:
return None
return cls(
enabled=True,
key_color=preset["key_color"],
similarity=preset["similarity"],
blend=preset["blend"],
spill_suppress=preset["spill_suppress"],
)
def has_effect(self) -> bool:
"""判断是否有实际抠像效果."""
return self.enabled and self.similarity > 0
def validate(self) -> tuple[bool, str]:
"""校验配置是否有效."""
if not self.enabled:
return True, ""
if not self.key_color:
return False, "key_color 不能为空"
if not (MIN_SIMILARITY <= self.similarity <= MAX_SIMILARITY):
return False, f"similarity 必须在 {MIN_SIMILARITY}~{MAX_SIMILARITY} 之间"
if not (MIN_BLEND <= self.blend <= MAX_BLEND):
return False, f"blend 必须在 {MIN_BLEND}~{MAX_BLEND} 之间"
if not (MIN_SPILL_SUPPRESS <= self.spill_suppress <= MAX_SPILL_SUPPRESS):
return False, f"spill_suppress 必须在 {MIN_SPILL_SUPPRESS}~{MAX_SPILL_SUPPRESS} 之间"
return True, ""
# ── 颜色归一化 ────────────────────────────────────────────────────────────────
def normalize_color(color_str: str) -> str:
"""将颜色字符串转为 FFmpeg colorkey 接受的格式.
支持:
- "#RRGGBB" / "#RRGGBBAA" → 0xRRGGBB
- "0xRRGGBB" → 直接使用
- 颜色名(green/blue/red/black/white 等)→ 直接透传
"""
color = color_str.strip()
# hex 格式
hex_match = re.match(r"^#?([0-9a-fA-F]{6})([0-9a-fA-F]{2})?$", color)
if hex_match:
return f"0x{hex_match.group(1).upper()}"
# 已经是 0x 格式
if color.lower().startswith("0x"):
return color.upper()
# 颜色名直接透传(FFmpeg 支持常见颜色名)
return color
# ── 滤镜构建 ────────────────────────────────────────────────────────────────
def build_colorkey_filter(
config: ChromaKeyConfig,
input_label: str,
output_label: str,
) -> str:
"""构建 colorkey 滤镜字符串.
Args:
config: 抠像配置
input_label: 输入标签,如 "[0:v]" 或 "[v0]"
output_label: 输出标签,如 "[ck0]"
Returns:
FFmpeg 滤镜字符串
"""
if not config.has_effect():
return f"{input_label}copy{output_label}"
color = normalize_color(config.key_color)
similarity = config.similarity
blend = config.blend
# 基础 colorkey 滤镜
parts = [f"colorkey=color={color}:similarity={similarity}:blend={blend}"]
# 溢色抑制(通过 colorchannelmixer 降低绿色通道增益)
if config.spill_suppress > 0:
spill = config.spill_suppress
g_gain = max(0.3, 1.0 - spill * 0.7)
r_gain = 1.0 + spill * 0.15
b_gain = 1.0 + spill * 0.15
parts.append(f"colorchannelmixer=rr={r_gain}:gg={g_gain}:bb={b_gain}:aa=1")
return f"{input_label}{','.join(parts)}{output_label}"
def build_chromakey_filter(
config: ChromaKeyConfig,
input_label: str,
output_label: str,
) -> str:
"""使用 chromakey 滤镜(更高级的版本,支持更多参数).
注意:并非所有 FFmpeg 版本都支持 chromakey 滤镜,
优先使用 colorkey(兼容性更好)。
"""
if not config.has_effect():
return f"{input_label}copy{output_label}"
color = normalize_color(config.key_color)
similarity = config.similarity
blend = config.blend
return f"{input_label}chromakey=color={color}:similarity={similarity}:blend={blend}{output_label}"
# ── 工具函数 ────────────────────────────────────────────────────────────────
def apply_chroma_key_if_needed(
clip_config: dict[str, Any] | None,
input_label: str,
output_label: str,
) -> str | None:
"""便捷函数:根据 clip 配置判断是否需要应用绿幕抠像.
Args:
clip_config: clip 的 config 字典
input_label: 输入标签
output_label: 输出标签
Returns:
滤镜字符串,不需要抠像时返回 None
"""
if not clip_config:
return None
chroma_key_data = clip_config.get("chroma_key")
if not chroma_key_data:
return None
try:
config = ChromaKeyConfig.from_dict(chroma_key_data)
if not config.has_effect():
return None
return build_colorkey_filter(config, input_label, output_label)
except Exception as e:
logger.warning("[chroma-key] 应用抠像失败,跳过: %s", e)
return None
def get_preset_names() -> list[str]:
"""获取所有预设名称列表."""
return sorted(list(CHROMA_KEY_PRESETS.keys()))
+231
View File
@@ -0,0 +1,231 @@
"""音频降噪配置领域模型 — 纯逻辑,无FFmpeg依赖.
抽离自 noise_reduction_engine.py,包含:
- NoiseReductionLevel 枚举(low/medium/high/custom)
- NoiseReductionConfig 数据类(解析/钳制/效果判断)
- 等级预设参数
- afftdn / arnndn 滤镜构建
- 便捷函数(apply_noise_reduction_if_needed)
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from enum import Enum
from typing import Any
logger = logging.getLogger(__name__)
# ── 降噪等级 ──────────────────────────────────────────────────────────────────
class NoiseReductionLevel(str, Enum):
"""降噪等级预设."""
LOW = "low" # 轻度降噪,保留细节,适合轻微背景噪音
MEDIUM = "medium" # 中度降噪,平衡效果和音质
HIGH = "high" # 高度降噪,适合嘈杂环境,可能轻微影响音质
CUSTOM = "custom" # 自定义参数
# 各等级对应的降噪参数(afftdn 的 noise floor,单位 dB)
# 值越大(越接近 0),降噪越强;值越小(越负),降噪越弱
_LEVEL_PARAMS: dict[NoiseReductionLevel, dict[str, float]] = {
NoiseReductionLevel.LOW: {
"nf": -35, # 噪音阈值(dB),越负越保守
"tn": -10, # 噪音频谱平滑度
"tr": 50, # 时间分辨率(ms)
},
NoiseReductionLevel.MEDIUM: {
"nf": -25,
"tn": -10,
"tr": 50,
},
NoiseReductionLevel.HIGH: {
"nf": -15,
"tn": -5,
"tr": 30,
},
}
# 参数范围
MIN_NOISE_FLOOR = -60.0
MAX_NOISE_FLOOR = -5.0
# 默认值
DEFAULT_LEVEL = NoiseReductionLevel.MEDIUM
DEFAULT_NOISE_FLOOR = -25.0
# ── 配置模型 ──────────────────────────────────────────────────────────────────
@dataclass
class NoiseReductionConfig:
"""音频降噪配置.
Attributes:
enabled: 是否启用降噪
level: 降噪等级 low/medium/high/custom
noise_floor: 自定义噪音阈值(dB),仅 level=custom 时有效,范围 -60 ~ -5
voice_enhance: 是否启用人声增强
"""
enabled: bool = False
level: NoiseReductionLevel = DEFAULT_LEVEL
noise_floor: float = DEFAULT_NOISE_FLOOR # dB
voice_enhance: bool = False
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> NoiseReductionConfig:
"""从字典解析配置,参数越界自动钳制."""
if not data or not data.get("enabled", False):
return cls(enabled=False)
level_str = str(data.get("level", "medium")).lower()
try:
level = NoiseReductionLevel(level_str)
except ValueError:
level = DEFAULT_LEVEL
try:
noise_floor = float(data.get("noise_floor", DEFAULT_NOISE_FLOOR))
except (TypeError, ValueError):
noise_floor = DEFAULT_NOISE_FLOOR
voice_enhance = bool(data.get("voice_enhance", False))
# 钳制到合法范围
noise_floor = max(MIN_NOISE_FLOOR, min(MAX_NOISE_FLOOR, noise_floor))
return cls(
enabled=True,
level=level,
noise_floor=noise_floor,
voice_enhance=voice_enhance,
)
def has_effect(self) -> bool:
"""判断是否有实际降噪效果."""
return self.enabled
def get_effective_noise_floor(self) -> float:
"""获取实际生效的噪音阈值(dB)."""
if self.level == NoiseReductionLevel.CUSTOM:
return self.noise_floor
params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[DEFAULT_LEVEL])
return float(params["nf"])
def get_level_params(self) -> dict[str, float]:
"""获取当前等级的完整参数字典."""
if self.level == NoiseReductionLevel.CUSTOM:
return {
"nf": self.noise_floor,
"tn": -10.0,
"tr": 50.0,
}
params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[DEFAULT_LEVEL])
return {k: float(v) for k, v in params.items()}
def validate(self) -> tuple[bool, str]:
"""校验配置是否有效."""
if not self.enabled:
return True, ""
if not (MIN_NOISE_FLOOR <= self.noise_floor <= MAX_NOISE_FLOOR):
return False, f"noise_floor 必须在 {MIN_NOISE_FLOOR}~{MAX_NOISE_FLOOR} dB 之间"
return True, ""
# ── 滤镜构建 ────────────────────────────────────────────────────────────────
def build_afftdn_filter(
config: NoiseReductionConfig,
input_label: str,
output_label: str,
) -> str:
"""构建 afftdn 音频降噪滤镜字符串.
Args:
config: 降噪配置
input_label: 输入标签,如 "[0:a]" 或 "[a0]"
output_label: 输出标签,如 "[nr0]"
Returns:
FFmpeg 滤镜字符串
"""
if not config.has_effect():
return f"{input_label}anull{output_label}"
params = config.get_level_params()
nf = params["nf"]
tn = params["tn"]
tr = params["tr"]
# 构建 afftdn 滤镜
filter_parts = [f"afftdn=nf={nf}:tn={tn}:tr={tr}"]
# 人声增强:通过 highpass + 压缩 + 响度归一化实现
if config.voice_enhance:
filter_parts.append("highpass=f=80")
filter_parts.append("acompressor=threshold=-20:ratio=2:attack=5:release=50")
filter_parts.append("loudnorm=I=-16:TP=-1.5:LRA=11")
return f"{input_label}{','.join(filter_parts)}{output_label}"
def build_arnndn_filter(
config: NoiseReductionConfig,
input_label: str,
output_label: str,
model_file: str,
) -> str:
"""使用 RNN 降噪滤镜(arnndn,效果更好但需要模型文件).
注意:需要额外下载 RNNNoise 模型文件,默认使用 afftdn(无需额外依赖)。
"""
if not config.has_effect():
return f"{input_label}anull{output_label}"
return f"{input_label}arnndn=m={model_file}{output_label}"
# ── 便捷函数 ────────────────────────────────────────────────────────────────
def apply_noise_reduction_if_needed(
config_data: dict[str, Any] | None,
input_label: str,
output_label: str,
) -> str | None:
"""便捷函数:根据配置判断是否需要应用音频降噪.
Args:
config_data: 降噪配置字典
input_label: 输入标签
output_label: 输出标签
Returns:
滤镜字符串,不需要降噪时返回 None
"""
if not config_data:
return None
try:
config = NoiseReductionConfig.from_dict(config_data)
if not config.has_effect():
return None
return build_afftdn_filter(config, input_label, output_label)
except Exception as e:
logger.warning("[noise-reduction] 应用降噪失败,跳过: %s", e)
return None
def get_level_names() -> list[str]:
"""获取所有降噪等级名称列表."""
return [level.value for level in NoiseReductionLevel]
+352
View File
@@ -0,0 +1,352 @@
"""裁剪配置领域模型 — 纯逻辑,无FFmpeg依赖.
抽离自 trim_engine.py,包含:
- TrimConfig 数据类(三选二推导 + 边界钳制 + 有效性判断)
- TrimSegment 数据类(多段裁剪)
- 滤镜字符串构建(build_video_trim_filter / build_audio_trim_filter)
- 多段解析(resolve_segments / parse_segments_from_config)
- 工具函数(extract_trim_from_clip_config)
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from typing import Any
logger = logging.getLogger(__name__)
# ── 常量 ────────────────────────────────────────────────────────────────────
# 最小裁剪时长(秒),低于此值视为无效
MIN_TRIM_DURATION = 0.1
# ── 数据类 ──────────────────────────────────────────────────────────────────
@dataclass
class TrimConfig:
"""裁剪配置.
三选二规则:start_time / end_time / duration 中必须至少给出两个,
第三个会被自动推导。如果三个都给了,以 start_time + duration 为准。
边界保护:
- start_time < 0 → 钳制到 0
- end_time > 素材时长 → 钳制到素材时长
- 计算出的 duration < 最小阈值 → 标记为无效
"""
start_time: float = 0.0 # 入点(素材内时间,秒)
end_time: float = 0.0 # 出点(素材内时间,秒),0 表示未指定
duration: float = 0.0 # 裁剪时长(秒),0 表示未指定
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> TrimConfig | None:
"""从字典构造,无有效裁剪参数时返回 None(不裁剪)."""
if not data:
return None
start = float(data.get("start_time", 0) or 0)
end = float(data.get("end_time", 0) or 0)
dur = float(data.get("duration", 0) or 0)
# 三个参数都没有 → 不裁剪
if start <= 0 and end <= 0 and dur <= 0:
return None
# 至少有两个参数(或一个合理的 start/duration)
# 兼容:只传了 start_time → 从 start 开始取到末尾
# 兼容:只传了 duration → 从 0 开始取 duration
if start > 0 and end <= 0 and dur <= 0:
# 只有 start,取到末尾 → 这是"从某点开始"的语义,算有效
pass
elif dur > 0 and start <= 0 and end <= 0:
# 只有 duration → 从开头取 duration,算有效
pass
elif start <= 0 and end <= 0 and dur <= 0:
return None
return cls(start_time=start, end_time=end, duration=dur)
def validate_and_resolve(self, asset_duration: float) -> TrimConfig:
"""根据素材实际时长,解析并钳制裁剪参数.
返回一个新的 TrimConfig,其中 start_time / end_time / duration 都已确定。
如果裁剪无效(时长为0或负数),仍返回但调用方应检查 is_valid。
"""
start = self.start_time
end = self.end_time
dur = self.duration
# 边界:start 不能为负
if start < 0:
start = 0.0
# 边界:asset_duration 为 0 时保守处理(不裁剪,取全部)
if asset_duration <= 0:
return TrimConfig(start_time=0.0, end_time=0.0, duration=0.0)
# 三选二推导
# 判断顺序很重要:先判断需要两个显式值的组合,最后判断含默认值的
# 情况1:start + end 都有显式值
if start > 0 and end > 0:
if end <= start:
# 出点 <= 入点,无效 → 返回 start 处一个极短片段(调用方会判无效)
return TrimConfig(start_time=start, end_time=start, duration=0.0)
dur = end - start
# 情况2:end + duration 都有显式值
elif end > 0 and dur > 0:
start = end - dur
if start < 0:
start = 0.0
dur = end # 重新计算
# 情况3:start + duration 都有值(start 可以是 0)
elif dur > 0:
end = start + dur
# 情况4:只有 start → 取到素材末尾
elif start > 0 and end <= 0 and dur <= 0:
end = asset_duration
dur = end - start
# 情况5:只有 end → 从开头取到 end
elif end > 0 and start <= 0 and dur <= 0:
start = 0.0
dur = end
else:
# 都没有 → 不裁剪
return TrimConfig(start_time=0.0, end_time=0.0, duration=0.0)
# 边界钳制:end 不能超过素材时长
if end > asset_duration:
end = asset_duration
dur = end - start
# 边界钳制:start 不能超过素材时长
if start >= asset_duration:
start = max(0.0, asset_duration - MIN_TRIM_DURATION)
dur = asset_duration - start
end = asset_duration
# 保证 duration 不为负
if dur < 0:
dur = 0.0
return TrimConfig(start_time=start, end_time=end, duration=dur)
@property
def is_valid(self) -> bool:
"""裁剪是否有效(时长大于最小阈值)."""
return self.duration >= MIN_TRIM_DURATION
@property
def is_noop(self) -> bool:
"""是否等价于不裁剪(从0开始取全部)."""
return self.start_time <= 0 and self.duration <= 0
@property
def trim_from_start(self) -> bool:
"""是否从开头裁剪(start_time == 0)."""
return self.start_time <= 0
@dataclass
class TrimSegment:
"""多段裁剪中的一段."""
segment_id: str # 段 ID(用于生成唯一标签)
trim: TrimConfig # 裁剪配置
order: int = 0 # 排序
@classmethod
def from_dict(cls, data: dict[str, Any], default_order: int = 0) -> TrimSegment:
"""从字典构造."""
return cls(
segment_id=str(data.get("segment_id", "") or f"seg_{default_order}"),
trim=TrimConfig(
start_time=float(data.get("start_time", 0) or 0),
end_time=float(data.get("end_time", 0) or 0),
duration=float(data.get("duration", 0) or 0),
),
order=int(data.get("order", default_order)),
)
# ── 滤镜构建 ────────────────────────────────────────────────────────────────
def build_video_trim_filter(
input_label: str,
trim: TrimConfig,
output_label: str,
) -> str:
"""构建视频裁剪滤镜链.
Args:
input_label: 输入视频标签,如 "[0:v]"
trim: 裁剪配置(已解析钳制)
output_label: 输出视频标签,如 "[v0_trimmed]"
Returns:
FFmpeg filter 字符串,如 "[0:v]trim=start=10:duration=5,setpts=PTS-STARTPTS[v0_trimmed]"
"""
if trim.is_noop:
# 不裁剪,直接直通(仅重置时间戳)
return f"{input_label}setpts=PTS-STARTPTS{output_label}"
parts: list[str] = []
# trim 滤镜参数
trim_args: list[str] = []
if trim.start_time > 0:
trim_args.append(f"start={trim.start_time:.3f}")
if trim.duration > 0:
trim_args.append(f"duration={trim.duration:.3f}")
elif trim.end_time > 0:
# end 用 duration 表示(start 到 end 的时长)
# 但 validate_and_resolve 后应该已经有 duration 了
pass
parts.append(f"trim={':'.join(trim_args)}")
parts.append("setpts=PTS-STARTPTS")
filter_str = f"{input_label}{','.join(parts)}{output_label}"
return filter_str
def build_audio_trim_filter(
input_label: str,
trim: TrimConfig,
output_label: str,
) -> str:
"""构建音频裁剪滤镜链.
Args:
input_label: 输入音频标签,如 "[0:a]"
trim: 裁剪配置(已解析钳制)
output_label: 输出音频标签,如 "[a0_trimmed]"
Returns:
FFmpeg filter 字符串,如 "[0:a]atrim=start=10:duration=5,asetpts=PTS-STARTPTS[a0_trimmed]"
"""
if trim.is_noop:
return f"{input_label}asetpts=PTS-STARTPTS{output_label}"
parts: list[str] = []
trim_args: list[str] = []
if trim.start_time > 0:
trim_args.append(f"start={trim.start_time:.3f}")
if trim.duration > 0:
trim_args.append(f"duration={trim.duration:.3f}")
parts.append(f"atrim={':'.join(trim_args)}")
parts.append("asetpts=PTS-STARTPTS")
filter_str = f"{input_label}{','.join(parts)}{output_label}"
return filter_str
# ── 多段裁剪 ────────────────────────────────────────────────────────────────
def resolve_segments(
segments: list[TrimSegment],
asset_duration: float,
) -> list[TrimSegment]:
"""解析并钳制多段裁剪配置,过滤无效段.
Args:
segments: 原始段列表
asset_duration: 素材实际时长
Returns:
解析后的有效段列表,按 order 排序
"""
resolved: list[TrimSegment] = []
for i, seg in enumerate(segments):
resolved_trim = seg.trim.validate_and_resolve(asset_duration)
if not resolved_trim.is_valid:
logger.warning(
"裁剪段无效,跳过: segment_id=%s duration=%.3f",
seg.segment_id,
resolved_trim.duration,
)
continue
resolved.append(
TrimSegment(
segment_id=seg.segment_id,
trim=resolved_trim,
order=seg.order if seg.order >= 0 else i,
)
)
resolved.sort(key=lambda s: s.order)
return resolved
def parse_segments_from_config(config: dict[str, Any] | None) -> list[TrimSegment]:
"""从 clip config 中解析多段裁剪配置.
config 中支持:
- trim_segments: [ {segment_id, start_time, end_time, duration, order}, ... ]
- trim_start / trim_end / trim_duration: 单段裁剪(兼容旧格式)
"""
if not config:
return []
# 优先解析多段
raw_segments = config.get("trim_segments", [])
if raw_segments and isinstance(raw_segments, list):
segments = []
for i, raw in enumerate(raw_segments):
if isinstance(raw, dict):
segments.append(TrimSegment.from_dict(raw, default_order=i))
return segments
# 单段裁剪兼容:从 trim_start/trim_end/trim_duration 构造
has_single = any(k in config for k in ("trim_start", "trim_end", "trim_duration"))
if has_single:
seg = TrimSegment(
segment_id="main",
trim=TrimConfig(
start_time=float(config.get("trim_start", 0) or 0),
end_time=float(config.get("trim_end", 0) or 0),
duration=float(config.get("trim_duration", 0) or 0),
),
order=0,
)
return [seg]
return []
# ── 工具函数 ────────────────────────────────────────────────────────────────
def extract_trim_from_clip_config(config: dict[str, Any] | None) -> TrimConfig | None:
"""从 clip config 中提取单段裁剪配置.
兼容以下字段名:
- trim_start / trim_end / trim_duration
- start_time / end_time / duration(在 trim 子字典里)
"""
if not config:
return None
# trim 子字典
if "trim" in config and isinstance(config["trim"], dict):
return TrimConfig.from_dict(config["trim"])
# 扁平字段
has_any = any(k in config for k in ("trim_start", "trim_end", "trim_duration"))
if not has_any:
return None
data = {
"start_time": config.get("trim_start", 0),
"end_time": config.get("trim_end", 0),
"duration": config.get("trim_duration", 0),
}
return TrimConfig.from_dict(data)
+360
View File
@@ -0,0 +1,360 @@
"""水印配置领域模型 — 纯逻辑,无FFmpeg依赖.
抽离自 watermark_engine.py,包含:
- 水印位置常量(9宫格)
- WatermarkConfig 数据类(from_dict / validate)
- 位置计算(calc_position / calc_scroll_x)
- 滤镜字符串构建(build_image_watermark_filter / build_text_watermark_filter)
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from typing import Any
logger = logging.getLogger(__name__)
# ── 常量 ────────────────────────────────────────────────────────────────────
# 9宫格位置枚举
WATERMARK_POSITIONS: dict[str, str] = {
"top_left": "左上",
"top_center": "中上",
"top_right": "右上",
"center_left": "左中",
"center": "中心",
"center_right": "右中",
"bottom_left": "左下",
"bottom_center": "中下",
"bottom_right": "右下",
}
VALID_POSITIONS = set(WATERMARK_POSITIONS.keys())
# 默认值常量
DEFAULT_POSITION = "bottom_right"
DEFAULT_MODE = "text"
DEFAULT_SCALE = 0.2
DEFAULT_OPACITY = 0.8
DEFAULT_FONT_SIZE = 24
DEFAULT_FONT_COLOR = "white"
DEFAULT_MARGIN_X = 20
DEFAULT_MARGIN_Y = 20
DEFAULT_SCROLL_SPEED = 50
# ── 数据类 ──────────────────────────────────────────────────────────────────
@dataclass
class WatermarkConfig:
"""水印配置.
mode: "image" 图片水印 | "text" 文字水印
position: 9宫格位置
opacity: 透明度 0.0-1.0
scale: 缩放比例(图片水印),0.1-1.0
margin: 边距(像素)
scroll: 是否滚动(跑马灯)
scroll_speed: 滚动速度(像素/秒)
"""
mode: str = DEFAULT_MODE # image | text
position: str = DEFAULT_POSITION
# 图片水印
image_path: str = "" # 本地图片路径
scale: float = DEFAULT_SCALE # 相对输出宽度的比例
opacity: float = DEFAULT_OPACITY # 0.0-1.0
# 文字水印
text: str = ""
font_size: int = DEFAULT_FONT_SIZE
font_color: str = DEFAULT_FONT_COLOR
font_path: str = "" # 字体文件路径
# 边距
margin_x: int = DEFAULT_MARGIN_X
margin_y: int = DEFAULT_MARGIN_Y
# 滚动水印
scroll: bool = False
scroll_speed: int = DEFAULT_SCROLL_SPEED # 像素/秒
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> WatermarkConfig | None:
"""从字典构造,空配置返回 None(不加水印)."""
if not data:
return None
enabled = data.get("enabled", False)
if not enabled:
return None
mode = data.get("mode", DEFAULT_MODE)
# 图片模式需要 image_path;文字模式需要 text
if mode == "image":
image_path = data.get("image_path", "") or data.get("image", "") or ""
if not image_path:
logger.warning("图片水印缺少 image_path,跳过水印")
return None
elif mode == "text":
text = data.get("text", "") or ""
if not text:
logger.warning("文字水印缺少 text,跳过水印")
return None
position = data.get("position", DEFAULT_POSITION)
if position not in VALID_POSITIONS:
position = DEFAULT_POSITION
return cls(
mode=mode,
position=position,
image_path=str(data.get("image_path", data.get("image", "")) or ""),
scale=float(data.get("scale", DEFAULT_SCALE)),
opacity=float(data.get("opacity", DEFAULT_OPACITY)),
text=str(data.get("text", "") or ""),
font_size=int(data.get("font_size", DEFAULT_FONT_SIZE)),
font_color=str(data.get("font_color", DEFAULT_FONT_COLOR)),
font_path=str(data.get("font_path", "") or ""),
margin_x=int(data.get("margin_x", DEFAULT_MARGIN_X)),
margin_y=int(data.get("margin_y", DEFAULT_MARGIN_Y)),
scroll=bool(data.get("scroll", False)),
scroll_speed=int(data.get("scroll_speed", DEFAULT_SCROLL_SPEED)),
)
def validate(self) -> tuple[bool, str]:
"""校验配置是否有效."""
if self.position not in VALID_POSITIONS:
return False, f"不支持的位置: {self.position}"
if not (0.0 <= self.opacity <= 1.0):
return False, "透明度必须在 0-1 之间"
if self.mode == "image":
if not self.image_path:
return False, "图片水印缺少图片路径"
if not (0.01 <= self.scale <= 1.0):
return False, "缩放比例必须在 0.01-1.0 之间"
elif self.mode == "text":
if not self.text:
return False, "文字水印缺少文字内容"
if self.font_size <= 0:
return False, "字体大小必须大于 0"
else:
return False, f"不支持的水印模式: {self.mode}"
return True, ""
def has_effect(self) -> bool:
"""判断水印是否有实际效果(非空配置)."""
if self.mode == "image":
return bool(self.image_path) and self.opacity > 0
elif self.mode == "text":
return bool(self.text) and self.opacity > 0 and self.font_size > 0
return False
# ── 位置计算 ────────────────────────────────────────────────────────────────
def calc_position(
position: str,
output_width: int,
output_height: int,
wm_width: int,
wm_height: int,
margin_x: int,
margin_y: int,
) -> tuple[int, int]:
"""根据9宫格位置计算水印坐标 (x, y).
坐标系:左上角为 (0, 0)
"""
if position == "top_left":
return margin_x, margin_y
elif position == "top_center":
return (output_width - wm_width) // 2, margin_y
elif position == "top_right":
return output_width - wm_width - margin_x, margin_y
elif position == "center_left":
return margin_x, (output_height - wm_height) // 2
elif position == "center":
return (output_width - wm_width) // 2, (output_height - wm_height) // 2
elif position == "center_right":
return output_width - wm_width - margin_x, (output_height - wm_height) // 2
elif position == "bottom_left":
return margin_x, output_height - wm_height - margin_y
elif position == "bottom_center":
return (output_width - wm_width) // 2, output_height - wm_height - margin_y
elif position == "bottom_right":
return output_width - wm_width - margin_x, output_height - wm_height - margin_y
else:
# 默认右下角
return output_width - wm_width - margin_x, output_height - wm_height - margin_y
def calc_scroll_x(position: str, output_width: int, wm_width: int, speed: int) -> str:
"""生成滚动水印的 x 坐标表达式.
从右向左滚动(跑马灯效果)
"""
# 标准跑马灯:x = -w + (t * speed) % (W + w)
# FFmpeg overlay 表达式写法
return f"mod({output_width}-mod({speed}*t\\,{output_width}+{wm_width})"
# ── 滤镜构建 ────────────────────────────────────────────────────────────────
def build_image_watermark_filter(
input_video_label: str,
wm_image_path: str,
output_width: int,
output_height: int,
output_label: str,
config: WatermarkConfig,
) -> tuple[str, list[str]]:
"""构建图片水印滤镜链.
Args:
input_video_label: 输入视频标签,如 "[final_video]"
wm_image_path: 水印图片本地路径
output_width: 输出视频宽度
output_height: 输出视频高度
output_label: 输出标签
config: 水印配置
Returns:
(filter_complex_str, input_args_list)
input_args 是 ["-i", wm_image_path] 格式
"""
# 计算水印尺寸(按输出宽度比例缩放)
wm_width = int(output_width * config.scale)
wm_height = -1 # 保持比例
wm_filter = f"scale={wm_width}:{wm_height}"
# 透明度处理
if config.opacity < 1.0:
wm_filter += f",format=rgba,colorchannelmixer=aa={config.opacity}"
# 水印预处理标签
wm_pre_label = "[wm_scaled]"
# 计算位置
x, y = calc_position(
config.position,
output_width,
output_height,
wm_width,
wm_width, # 高度未知,先用宽度估算
config.margin_x,
config.margin_y,
)
# 滚动水印
if config.scroll:
# 从右向左滚动:x = W - (t * speed) mod (W + wm_w)
x_expr = f"{output_width}-mod({config.scroll_speed}*t\\,{output_width}+{wm_width}"
y_expr = str(y)
overlay_expr = f"x={x_expr}:y={y_expr}"
else:
overlay_expr = f"x={x}:y={y}"
# 构建滤镜
filter_parts = [
f"[1:v]{wm_filter}{wm_pre_label}",
f"{input_video_label}{wm_pre_label}overlay={overlay_expr}{output_label}",
]
filter_complex = ";".join(filter_parts)
input_args = ["-i", wm_image_path]
return filter_complex, input_args
def build_text_watermark_filter(
input_video_label: str,
output_label: str,
config: WatermarkConfig,
output_width: int,
output_height: int,
) -> str:
"""构建文字水印滤镜(drawtext).
Args:
input_video_label: 输入视频标签
output_label: 输出标签
config: 水印配置
output_width: 输出宽度
output_height: 输出高度
Returns:
FFmpeg filter 字符串
"""
# 转义文字中的特殊字符
text = config.text.replace(":", "\\:").replace("'", "\\'")
# 字体配置
font_config = []
if config.font_path:
font_path_escaped = config.font_path.replace(":", "\\:").replace("'", "\\'")
font_config.append(f"fontfile='{font_path_escaped}'")
font_config.append(f"fontsize={config.font_size}")
font_config.append(f"fontcolor={config.font_color}@{config.opacity}")
# 估算文字宽高(粗略估算,用于位置计算)
# 每个汉字约等于 font_size 宽高
approx_w = len(config.text) * config.font_size
approx_h = config.font_size
# 位置计算
x, y = calc_position(
config.position,
output_width,
output_height,
approx_w,
approx_h,
config.margin_x,
config.margin_y,
)
# 滚动水印
if config.scroll:
x_expr = f"w-mod({config.scroll_speed}*t\\,W+w)"
pos_config = [f"x={x_expr}", f"y={y}"]
else:
pos_config = [f"x={x}", f"y={y}"]
# 组装 drawtext
drawtext_parts = [f"text='{text}'"] + font_config + pos_config
drawtext = "drawtext=" + ":".join(drawtext_parts)
return f"{input_video_label}{drawtext}{output_label}"
# ── 工具函数 ────────────────────────────────────────────────────────────────
def get_position_names() -> list[str]:
"""获取所有合法位置名称列表(按从上到下、从左到右顺序)."""
return [
"top_left",
"top_center",
"top_right",
"center_left",
"center",
"center_right",
"bottom_left",
"bottom_center",
"bottom_right",
]
def get_position_display_name(position: str) -> str:
"""获取位置的中文显示名."""
return WATERMARK_POSITIONS.get(position, position)
+187
View File
@@ -0,0 +1,187 @@
"""XFade 转场滤镜构建 — 纯逻辑,无 FFmpeg 依赖.
抽离自 apps/worker/video_processing/ffmpeg_utils.py,包含:
- xfade 转场效果名称映射
- 滤镜链串联工具
- xfade 转场滤镜链构建(带时长钳制)
"""
from __future__ import annotations
import logging
from typing import Any
logger = logging.getLogger(__name__)
# ── 常量 ──────────────────────────────────────────────────────────────────────
DEFAULT_TRANSITION_DURATION = 0.5
# xfade 转场映射:transition_effect 名称 → FFmpeg xfade transition 名称
# 键同时支持 TransitionEffect 枚举值和字符串名称(向后兼容)
# "cut" 为特殊值:硬切,不使用 xfade(由调用方特殊处理)
XFADE_TRANSITION_MAP: dict[str, str] = {
# 基础
"fade": "fade",
"dissolve": "dissolve",
"crossfade": "dissolve",
"crossdissolve": "dissolve",
# 滑入系列
"slideleft": "slideleft",
"slide_left": "slideleft",
"slideright": "slideright",
"slide_right": "slideright",
"slideup": "slideup",
"slide_up": "slideup",
"slidedown": "slidedown",
"slide_down": "slidedown",
"slide": "slideleft", # 默认向左滑
# 缩放
"zoom": "zoomin",
"zoomin": "zoomin",
"zoomout": "zoomout",
# 擦除系列
"wipe": "wipeleft", # 默认向左擦
"wipeleft": "wipeleft",
"wiperight": "wiperight",
"wipeup": "wipeup",
"wipedown": "wipedown",
# 特殊效果
"circlecrop": "circlecrop",
"circle": "circlecrop",
"rectcrop": "rectcrop",
"rect": "rectcrop",
}
# 所有支持的转场效果名称(用户侧输入)
SUPPORTED_TRANSITIONS: set[str] = set(XFADE_TRANSITION_MAP.keys())
# 所有 FFmpeg xfade transition 名称(输出侧)
XFade_TRANSITION_NAMES: set[str] = set(XFADE_TRANSITION_MAP.values())
# ── 工具函数 ─────────────────────────────────────────────────────────────────
def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
"""将滤镜列表串联为 FFmpeg 滤镜字符串.
例:chain_filters(["scale=1280:720", "fps=25"], "v0")
→ "[0:v]scale=1280:720,fps=25[v0]"
Args:
filters: 滤镜字符串列表
output_label: 输出标签(不带方括号)
input_label: 输入标签(不带方括号),默认 "0:v"
Returns:
完整的滤镜字符串
"""
filter_body = ",".join(filters)
return f"[{input_label}]{filter_body}[{output_label}]"
def resolve_xfade_transition(transition_name: Any) -> str:
"""将转场效果名称映射为 FFmpeg xfade transition 名称.
支持 TransitionEffect 枚举值和字符串名称,未知值回退到 "fade"。
Args:
transition_name: 转场名称(字符串或带 .value 属性的枚举)
Returns:
FFmpeg xfade transition 名称
"""
# 兼容 TransitionEffect 枚举(有 .value 属性)
if hasattr(transition_name, "value"):
transition_name = transition_name.value
return XFADE_TRANSITION_MAP.get(transition_name, "fade")
# ── xfade 滤镜链构建 ─────────────────────────────────────────────────────────
def build_xfade_filter_chain(
clip_durations: list[float],
clip_video_labels: list[str],
transitions: list[str],
*,
transition_duration: float = DEFAULT_TRANSITION_DURATION,
output_label: str = "outv",
) -> tuple[str, float]:
"""构建 xfade 转场滤镜链.
对每步 xfade 自动钳制 transition duration,确保
``offset + td ≤ first_input_duration``,避免 FFmpeg exit 234。
Args:
clip_durations: 每个片段的时长(必须与 trim 后的实际时长一致)
clip_video_labels: 每个片段的视频流标签(如 "v0", "v1")
transitions: 每个片段对应的转场效果(第一个片段的转场被忽略)
transition_duration: 转场时长(秒)
output_label: 最终输出标签
Returns:
(filter_string, estimated_total_duration)
"""
n = len(clip_durations)
parts: list[str] = []
if n == 0:
return "", 0.0
if n == 1:
parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]")
return ";".join(parts), clip_durations[0]
# xfade 链 — 每步动态钳制 td,防止 offset + td > first_input_duration
cumulative = 0.0
prev_label = clip_video_labels[0]
total_transition = 0.0 # 累计已使用的转场时长
for i in range(1, n):
cumulative += clip_durations[i - 1]
# 当前 xfade 的第一个输入时长
if i == 1:
first_input_dur = clip_durations[0]
else:
first_input_dur = cumulative - total_transition
# 原始 offset 计算
offset = max(0.0, cumulative - transition_duration * i)
# 安全钳制:offset + td 不能超过第一个输入的时长
available = max(0.0, first_input_dur - offset)
safe_td = min(transition_duration, available)
# 同时不能超过剩余总时长
remaining = max(0.0, sum(clip_durations) - cumulative)
safe_td = min(safe_td, remaining)
# 同时不能超过当前第二个输入(单个片段)的时长
safe_td = min(safe_td, clip_durations[i])
safe_td = max(0.001, safe_td) # 至少 1ms,避免 td=0
transition = transitions[i] if i < len(transitions) else "cut"
xfade_transition = resolve_xfade_transition(transition)
if i == n - 1:
out_label = output_label
else:
out_label = f"xf{i}"
parts.append(
f"[{prev_label}][{clip_video_labels[i]}]"
f"xfade=transition={xfade_transition}"
f":duration={safe_td:.3f}"
f":offset={offset:.3f}"
f"[{out_label}]"
)
prev_label = out_label
total_transition += safe_td
# 总时长减去转场重叠部分
total_duration = sum(clip_durations) - total_transition
return ";".join(parts), max(0.0, total_duration)
+448
View File
@@ -0,0 +1,448 @@
"""ASS 字幕构建领域模型单元测试 — 纯逻辑,无文件IO."""
from __future__ import annotations
import pytest
from packages.domain.ass_subtitle_builder import (
TITLE_MARGIN_BOTTOM,
TITLE_MARGIN_SIDE,
TITLE_MARGIN_TOP,
build_ass_content,
build_ass_style,
escape_ass_text,
format_ass_time,
hex_to_ass_color,
position_to_ass_alignment,
)
# ── 颜色转换 ──────────────────────────────────────────────────────────────────
class TestHexToAssColor:
def test_red(self):
assert hex_to_ass_color("#FF0000") == "&H0000FF"
def test_green(self):
assert hex_to_ass_color("#00FF00") == "&H00FF00"
def test_blue(self):
assert hex_to_ass_color("#0000FF") == "&HFF0000"
def test_white(self):
assert hex_to_ass_color("#FFFFFF") == "&HFFFFFF"
def test_black(self):
assert hex_to_ass_color("#000000") == "&H000000"
def test_without_hash(self):
assert hex_to_ass_color("FF0000") == "&H0000FF"
def test_lowercase(self):
assert hex_to_ass_color("#ff0000") == "&H0000FF"
def test_invalid_length_short(self):
assert hex_to_ass_color("#FFF") == "&H000000"
def test_invalid_length_long(self):
assert hex_to_ass_color("#FFFFFFFF") == "&H000000"
def test_empty(self):
assert hex_to_ass_color("") == "&H000000"
# ── 位置对齐 ──────────────────────────────────────────────────────────────────
class TestPositionToAssAlignment:
def test_top(self):
assert position_to_ass_alignment("top") == 8
def test_center(self):
assert position_to_ass_alignment("center") == 5
def test_bottom(self):
assert position_to_ass_alignment("bottom") == 2
def test_unknown_default_top(self):
assert position_to_ass_alignment("unknown") == 8
def test_empty_default_top(self):
assert position_to_ass_alignment("") == 8
# ── Style 行构建 ─────────────────────────────────────────────────────────────
class TestBuildAssStyle:
def test_minimal_style(self):
result = build_ass_style("TestStyle")
assert result.startswith("Style: TestStyle,")
assert "思源黑体" in result
assert ",48," in result
def test_custom_font_size(self):
result = build_ass_style("Title", font_size=64)
assert ",64," in result
def test_bold_enabled(self):
result = build_ass_style("BoldStyle", bold=True)
parts = result.split(",")
# Bold 是第 8 个字段(index 7)
assert parts[7] == "-1"
def test_bold_disabled(self):
result = build_ass_style("NormalStyle", bold=False)
parts = result.split(",")
assert parts[7] == "0"
def test_italic_enabled(self):
result = build_ass_style("ItalicStyle", italic=True)
parts = result.split(",")
assert parts[8] == "-1"
def test_alignment(self):
result = build_ass_style("AlignBottom", alignment=2)
parts = result.split(",")
# Alignment 是第 19 个字段(index 18)
assert parts[18] == "2"
def test_margins(self):
result = build_ass_style("MarginStyle", margin_v=100, margin_l=50, margin_r=50)
parts = result.split(",")
# MarginL, MarginR, MarginV 分别是 index 19, 20, 21
assert parts[19] == "50"
assert parts[20] == "50"
assert parts[21] == "100"
def test_outline_width(self):
result = build_ass_style("OutlineStyle", outline_width=3.5)
# Outline 是 index 16
parts = result.split(",")
assert parts[16] == "3.5"
def test_shadow_with_blur(self):
result = build_ass_style("ShadowStyle", shadow_blur=4.0, shadow_offset=(2, 3))
parts = result.split(",")
# Shadow 深度(纵向偏移)是 index 17
assert parts[17] == "3"
def test_shadow_without_blur(self):
result = build_ass_style("NoShadowStyle", shadow_blur=0.0, shadow_offset=(2, 3))
parts = result.split(",")
assert parts[17] == "0"
def test_primary_color(self):
result = build_ass_style("ColorStyle", primary_color="&H00FFFFFF")
# PrimaryColour 是 index 3
parts = result.split(",")
assert parts[3] == "&H00FFFFFF"
def test_outline_color(self):
result = build_ass_style("StrokeStyle", outline_color="&H00000000")
# OutlineColour 是 index 5
parts = result.split(",")
assert parts[5] == "&H00000000"
def test_field_count(self):
"""验证 Style 行有正确的字段数(23 个字段)."""
result = build_ass_style("FullStyle")
parts = result.split(",")
# Style: 行有 23 个字段(去掉 "Style: " 前缀后)
assert len(parts) == 23
# ── 文本转义 ──────────────────────────────────────────────────────────────────
class TestEscapeAssText:
def test_plain_text(self):
assert escape_ass_text("Hello World") == "Hello World"
def test_newline_lf(self):
assert escape_ass_text("line1\nline2") == "line1\\Nline2"
def test_newline_crlf(self):
assert escape_ass_text("line1\r\nline2") == "line1\\Nline2"
def test_newline_cr(self):
assert escape_ass_text("line1\rline2") == "line1\\Nline2"
def test_curly_braces(self):
assert escape_ass_text("text {tag} text") == "text (tag) text"
def test_left_brace_only(self):
assert escape_ass_text("{start") == "(start"
def test_right_brace_only(self):
assert escape_ass_text("end}") == "end)"
def test_multiple_braces(self):
assert escape_ass_text("{a}{b}{c}") == "(a)(b)(c)"
def test_mixed_newline_and_braces(self):
assert escape_ass_text("line1\n{tag}\nline2") == "line1\\N(tag)\\Nline2"
def test_empty_string(self):
assert escape_ass_text("") == ""
def test_chinese_text(self):
assert escape_ass_text("你好世界") == "你好世界"
# ── 时间格式化 ────────────────────────────────────────────────────────────────
class TestFormatAssTime:
def test_zero(self):
assert format_ass_time(0) == "0:00:00.00"
def test_seconds_only(self):
assert format_ass_time(5.5) == "0:00:05.50"
def test_minutes(self):
assert format_ass_time(125.0) == "0:02:05.00"
def test_hours(self):
assert format_ass_time(3661.5) == "1:01:01.50"
def test_one_hour_exact(self):
assert format_ass_time(3600) == "1:00:00.00"
def test_sub_second_precision(self):
result = format_ass_time(1.23)
assert result == "0:00:01.23"
def test_59_seconds(self):
assert format_ass_time(59.99) == "0:00:59.99"
def test_60_seconds(self):
assert format_ass_time(60.0) == "0:01:00.00"
def test_90_minutes(self):
assert format_ass_time(5400.0) == "1:30:00.00"
# ── 完整 ASS 内容生成 ─────────────────────────────────────────────────────────
class TestBuildAssContent:
def test_no_subtitles_returns_empty(self):
result = build_ass_content(video_width=1920, video_height=1080, video_duration=10.0)
assert result == ""
def test_title_disabled_returns_empty(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Title",
title_config={"enabled": False},
)
assert result == ""
def test_empty_title_text_returns_empty(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text=" ",
title_config={"enabled": True},
)
assert result == ""
def test_with_title(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=30.0,
title_text="My Title",
title_config={"enabled": True, "color": "#FFFFFF"},
)
assert "[Script Info]" in result
assert "PlayResX: 1920" in result
assert "PlayResY: 1080" in result
assert "[V4+ Styles]" in result
assert "TitleStyle" in result
assert "[Events]" in result
assert "Dialogue:" in result
assert "My Title" in result
def test_with_subtitle(self):
result = build_ass_content(
video_width=1280,
video_height=720,
video_duration=15.0,
subtitle_text="Subtitle Text",
subtitle_config={"enabled": True},
)
assert "PlayResX: 1280" in result
assert "PlayResY: 720" in result
assert "SubtitleStyle" in result
assert "Subtitle Text" in result
def test_with_both_title_and_subtitle(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=60.0,
title_text="Big Title",
title_config={"enabled": True},
subtitle_text="Small subtitle",
subtitle_config={"enabled": True},
)
assert "TitleStyle" in result
assert "SubtitleStyle" in result
assert "Big Title" in result
assert "Small subtitle" in result
# 两个 Dialogue 行
assert result.count("Dialogue:") == 2
def test_title_position_bottom(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Bottom Title",
title_config={"enabled": True, "position": "bottom"},
)
# 对齐方式为 2(底部居中)
assert "TitleStyle" in result
def test_title_with_stroke(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Stroke Title",
title_config={
"enabled": True,
"stroke": {"enabled": True, "color": "#000000", "width": 3},
},
)
assert "Stroke Title" in result
def test_title_with_shadow(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Shadow Title",
title_config={
"enabled": True,
"shadow": {"enabled": True, "blur": 4, "offset_x": 2, "offset_y": 3},
},
)
assert "Shadow Title" in result
def test_title_bold_default(self):
"""标题默认启用粗体."""
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Bold Title",
title_config={"enabled": True},
)
# 在 TitleStyle 行中找 bold=-1
for line in result.split("\n"):
if line.startswith("Style: TitleStyle"):
parts = line.split(",")
assert parts[7] == "-1"
break
else:
pytest.fail("TitleStyle not found")
def test_subtitle_not_bold(self):
"""字幕默认不启用粗体."""
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
subtitle_text="Normal Subtitle",
subtitle_config={"enabled": True},
)
for line in result.split("\n"):
if line.startswith("Style: SubtitleStyle"):
parts = line.split(",")
assert parts[7] == "0"
break
else:
pytest.fail("SubtitleStyle not found")
def test_duration_format_in_dialogue(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=125.5,
title_text="Timed",
title_config={"enabled": True},
)
# 结束时间应该是 0:02:05.50
assert "0:02:05.50" in result
def test_title_text_escaped(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Line1\n{tag}Line2",
title_config={"enabled": True},
)
assert "Line1\\N(tag)Line2" in result
def test_default_title_enabled(self):
"""不传 enabled 时默认为 True."""
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Default Enabled",
title_config={},
)
assert result != ""
assert "Default Enabled" in result
def test_subtitle_position_top(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
subtitle_text="Top Subtitle",
subtitle_config={"enabled": True, "position": "top"},
)
assert "Top Subtitle" in result
def test_scaled_border_and_shadow(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Test",
title_config={"enabled": True},
)
assert "ScaledBorderAndShadow: yes" in result
def test_wrap_style(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Test",
title_config={"enabled": True},
)
assert "WrapStyle: 2" in result
# ── 常量 ──────────────────────────────────────────────────────────────────────
class TestConstants:
def test_title_margin_top(self):
assert TITLE_MARGIN_TOP == 60
def test_title_margin_bottom(self):
assert TITLE_MARGIN_BOTTOM == 60
def test_title_margin_side(self):
assert TITLE_MARGIN_SIDE == 40
+281
View File
@@ -0,0 +1,281 @@
"""chroma_key_config 领域模型单测."""
from __future__ import annotations
import pytest
from packages.domain.chroma_key_config import (
CHROMA_KEY_PRESETS,
ChromaKeyConfig,
apply_chroma_key_if_needed,
build_chromakey_filter,
build_colorkey_filter,
get_preset_names,
normalize_color,
)
# ── ChromaKeyConfig.from_dict 测试 ────────────────────────────────────────
class TestChromaKeyConfigFromDict:
def test_none_returns_disabled(self):
cfg = ChromaKeyConfig.from_dict(None)
assert cfg.enabled is False
def test_empty_dict_returns_disabled(self):
cfg = ChromaKeyConfig.from_dict({})
assert cfg.enabled is False
def test_disabled_returns_disabled(self):
cfg = ChromaKeyConfig.from_dict({"enabled": False})
assert cfg.enabled is False
def test_enabled_default_params(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True})
assert cfg.enabled is True
assert cfg.key_color == "#00FF00"
assert cfg.similarity == 0.3
assert cfg.blend == 0.1
assert cfg.spill_suppress == 0.0
def test_custom_params(self):
cfg = ChromaKeyConfig.from_dict(
{
"enabled": True,
"key_color": "#0000FF",
"similarity": 0.5,
"blend": 0.2,
"spill_suppress": 0.4,
}
)
assert cfg.key_color == "#0000FF"
assert cfg.similarity == 0.5
assert cfg.blend == 0.2
assert cfg.spill_suppress == 0.4
def test_similarity_clamped_low(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "similarity": 0.001})
assert cfg.similarity == 0.01
def test_similarity_clamped_high(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "similarity": 2.0})
assert cfg.similarity == 1.0
def test_blend_clamped_low(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "blend": -0.5})
assert cfg.blend == 0.0
def test_blend_clamped_high(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "blend": 1.5})
assert cfg.blend == 1.0
def test_spill_suppress_clamped(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "spill_suppress": 2.0})
assert cfg.spill_suppress == 1.0
def test_invalid_similarity_type_uses_default(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "similarity": "high"})
assert cfg.similarity == 0.3
def test_key_color_stripped(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "key_color": " #00FF00 "})
assert cfg.key_color == "#00FF00"
# ── from_preset 测试 ──────────────────────────────────────────────────────
class TestFromPreset:
def test_green_screen_preset(self):
cfg = ChromaKeyConfig.from_preset("green_screen")
assert cfg is not None
assert cfg.enabled is True
assert cfg.key_color == "#00FF00"
assert cfg.similarity == 0.3
def test_blue_screen_preset(self):
cfg = ChromaKeyConfig.from_preset("blue_screen")
assert cfg is not None
assert cfg.key_color == "#0000FF"
def test_invalid_preset_returns_none(self):
assert ChromaKeyConfig.from_preset("nonexistent") is None
def test_all_presets_valid(self):
for name in CHROMA_KEY_PRESETS:
cfg = ChromaKeyConfig.from_preset(name)
assert cfg is not None
assert cfg.enabled is True
# ── has_effect / validate 测试 ────────────────────────────────────────────
class TestHasEffectAndValidate:
def test_disabled_no_effect(self):
cfg = ChromaKeyConfig(enabled=False)
assert cfg.has_effect() is False
def test_enabled_has_effect(self):
cfg = ChromaKeyConfig(enabled=True, similarity=0.3)
assert cfg.has_effect() is True
def test_zero_similarity_no_effect(self):
cfg = ChromaKeyConfig(enabled=True, similarity=0.0)
# similarity 被钳制后为 0.01,所以应该有效果
# 等等,from_dict 才会钳制,直接构造不会
assert cfg.has_effect() is False
def test_validate_disabled_valid(self):
cfg = ChromaKeyConfig(enabled=False)
ok, msg = cfg.validate()
assert ok is True
assert msg == ""
def test_validate_enabled_valid(self):
cfg = ChromaKeyConfig(enabled=True, key_color="#00FF00")
ok, msg = cfg.validate()
assert ok is True
def test_validate_empty_color_invalid(self):
cfg = ChromaKeyConfig(enabled=True, key_color="")
ok, msg = cfg.validate()
assert ok is False
assert "key_color" in msg
def test_validate_similarity_out_of_range(self):
cfg = ChromaKeyConfig(enabled=True, similarity=2.0)
ok, msg = cfg.validate()
assert ok is False
assert "similarity" in msg
# ── normalize_color 测试 ──────────────────────────────────────────────────
class TestNormalizeColor:
def test_hex_with_hash(self):
assert normalize_color("#00FF00") == "0x00FF00"
def test_hex_lowercase(self):
assert normalize_color("#00ff00") == "0x00FF00"
def test_hex_without_hash(self):
assert normalize_color("00FF00") == "0x00FF00"
def test_hex_with_alpha(self):
assert normalize_color("#00FF00FF") == "0x00FF00"
def test_already_0x_format(self):
assert normalize_color("0x00FF00") == "0X00FF00"
def test_0x_lowercase(self):
assert normalize_color("0x00ff00") == "0X00FF00"
def test_color_name_passthrough(self):
assert normalize_color("green") == "green"
assert normalize_color("blue") == "blue"
def test_whitespace_stripped(self):
assert normalize_color(" #FF0000 ") == "0xFF0000"
# ── build_colorkey_filter 测试 ────────────────────────────────────────────
class TestBuildColorkeyFilter:
def test_disabled_returns_copy(self):
cfg = ChromaKeyConfig(enabled=False)
result = build_colorkey_filter(cfg, "[in]", "[out]")
assert "copy" in result
assert "[in]" in result
assert "[out]" in result
def test_basic_colorkey(self):
cfg = ChromaKeyConfig(enabled=True, key_color="#00FF00", similarity=0.3, blend=0.1)
result = build_colorkey_filter(cfg, "[v]", "[ck]")
assert "colorkey=" in result
assert "color=0x00FF00" in result
assert "similarity=0.3" in result
assert "blend=0.1" in result
assert "[v]" in result
assert "[ck]" in result
def test_with_spill_suppress(self):
cfg = ChromaKeyConfig(enabled=True, key_color="#00FF00", spill_suppress=0.5)
result = build_colorkey_filter(cfg, "[in]", "[out]")
assert "colorchannelmixer=" in result
assert "rr=" in result
assert "gg=" in result
assert "bb=" in result
def test_no_spill_suppress_no_colorchannelmixer(self):
cfg = ChromaKeyConfig(enabled=True, spill_suppress=0.0)
result = build_colorkey_filter(cfg, "[in]", "[out]")
assert "colorchannelmixer" not in result
# ── build_chromakey_filter 测试 ───────────────────────────────────────────
class TestBuildChromakeyFilter:
def test_disabled_returns_copy(self):
cfg = ChromaKeyConfig(enabled=False)
result = build_chromakey_filter(cfg, "[in]", "[out]")
assert "copy" in result
def test_basic_chromakey(self):
cfg = ChromaKeyConfig(enabled=True, key_color="#00FF00", similarity=0.3, blend=0.1)
result = build_chromakey_filter(cfg, "[v]", "[ck]")
assert "chromakey=" in result
assert "color=0x00FF00" in result
assert "similarity=0.3" in result
assert "blend=0.1" in result
def test_contains_input_and_output_labels(self):
cfg = ChromaKeyConfig(enabled=True)
result = build_chromakey_filter(cfg, "[in_v]", "[out_v]")
assert "[in_v]" in result
assert "[out_v]" in result
# ── apply_chroma_key_if_needed 测试 ───────────────────────────────────────
class TestApplyChromaKeyIfNeeded:
def test_none_config_returns_none(self):
assert apply_chroma_key_if_needed(None, "[in]", "[out]") is None
def test_no_chroma_key_returns_none(self):
assert apply_chroma_key_if_needed({}, "[in]", "[out]") is None
def test_disabled_chroma_key_returns_none(self):
config = {"chroma_key": {"enabled": False}}
assert apply_chroma_key_if_needed(config, "[in]", "[out]") is None
def test_enabled_chroma_key_returns_filter(self):
config = {"chroma_key": {"enabled": True, "key_color": "#00FF00"}}
result = apply_chroma_key_if_needed(config, "[in]", "[out]")
assert result is not None
assert "colorkey" in result
def test_invalid_config_handles_exception(self):
# 传入无效配置触发异常,应该返回 None 而不是抛出
config = {"chroma_key": "invalid_string"}
result = apply_chroma_key_if_needed(config, "[in]", "[out]")
assert result is None
# ── 预设工具函数测试 ───────────────────────────────────────────────────────
class TestPresetUtils:
def test_get_preset_names_returns_sorted_list(self):
names = get_preset_names()
assert isinstance(names, list)
assert len(names) == len(CHROMA_KEY_PRESETS)
assert names == sorted(names)
def test_all_preset_names_in_presets_dict(self):
for name in get_preset_names():
assert name in CHROMA_KEY_PRESETS
+258
View File
@@ -0,0 +1,258 @@
"""noise_reduction_config 领域模型单测."""
from __future__ import annotations
import pytest
from packages.domain.noise_reduction_config import (
DEFAULT_LEVEL,
DEFAULT_NOISE_FLOOR,
MAX_NOISE_FLOOR,
MIN_NOISE_FLOOR,
NoiseReductionConfig,
NoiseReductionLevel,
apply_noise_reduction_if_needed,
build_afftdn_filter,
build_arnndn_filter,
get_level_names,
)
# ── NoiseReductionLevel 枚举测试 ───────────────────────────────────────────
class TestNoiseReductionLevel:
def test_four_levels(self):
assert len(NoiseReductionLevel) == 4
def test_level_values(self):
assert NoiseReductionLevel.LOW.value == "low"
assert NoiseReductionLevel.MEDIUM.value == "medium"
assert NoiseReductionLevel.HIGH.value == "high"
assert NoiseReductionLevel.CUSTOM.value == "custom"
def test_from_string(self):
assert NoiseReductionLevel("low") == NoiseReductionLevel.LOW
assert NoiseReductionLevel("medium") == NoiseReductionLevel.MEDIUM
assert NoiseReductionLevel("high") == NoiseReductionLevel.HIGH
assert NoiseReductionLevel("custom") == NoiseReductionLevel.CUSTOM
# ── NoiseReductionConfig.from_dict 测试 ───────────────────────────────────
class TestNoiseReductionConfigFromDict:
def test_none_returns_disabled(self):
cfg = NoiseReductionConfig.from_dict(None)
assert cfg.enabled is False
def test_empty_dict_returns_disabled(self):
cfg = NoiseReductionConfig.from_dict({})
assert cfg.enabled is False
def test_disabled_returns_disabled(self):
cfg = NoiseReductionConfig.from_dict({"enabled": False})
assert cfg.enabled is False
def test_enabled_default_params(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True})
assert cfg.enabled is True
assert cfg.level == NoiseReductionLevel.MEDIUM
assert cfg.noise_floor == DEFAULT_NOISE_FLOOR
assert cfg.voice_enhance is False
def test_custom_level(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": -30.0})
assert cfg.level == NoiseReductionLevel.CUSTOM
assert cfg.noise_floor == -30.0
def test_invalid_level_defaults_medium(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "invalid"})
assert cfg.level == NoiseReductionLevel.MEDIUM
def test_noise_floor_clamped_low(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": -100.0})
assert cfg.noise_floor == MIN_NOISE_FLOOR
def test_noise_floor_clamped_high(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": 0.0})
assert cfg.noise_floor == MAX_NOISE_FLOOR
def test_invalid_noise_floor_type_uses_default(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": "not_a_number"})
assert cfg.noise_floor == DEFAULT_NOISE_FLOOR
def test_voice_enhance_true(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "voice_enhance": True})
assert cfg.voice_enhance is True
def test_case_insensitive_level(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "HIGH"})
assert cfg.level == NoiseReductionLevel.HIGH
# ── has_effect / get_effective_noise_floor 测试 ───────────────────────────
class TestConfigProperties:
def test_disabled_no_effect(self):
cfg = NoiseReductionConfig(enabled=False)
assert cfg.has_effect() is False
def test_enabled_has_effect(self):
cfg = NoiseReductionConfig(enabled=True)
assert cfg.has_effect() is True
def test_effective_noise_floor_low(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.LOW)
assert cfg.get_effective_noise_floor() == -35.0
def test_effective_noise_floor_medium(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
assert cfg.get_effective_noise_floor() == -25.0
def test_effective_noise_floor_high(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.HIGH)
assert cfg.get_effective_noise_floor() == -15.0
def test_effective_noise_floor_custom(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-40.0)
assert cfg.get_effective_noise_floor() == -40.0
def test_get_level_params_medium(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
params = cfg.get_level_params()
assert params["nf"] == -25.0
assert params["tn"] == -10.0
assert params["tr"] == 50.0
def test_get_level_params_custom(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-30.0)
params = cfg.get_level_params()
assert params["nf"] == -30.0
assert "tn" in params
assert "tr" in params
# ── validate 测试 ─────────────────────────────────────────────────────────
class TestValidate:
def test_disabled_valid(self):
cfg = NoiseReductionConfig(enabled=False)
ok, msg = cfg.validate()
assert ok is True
assert msg == ""
def test_enabled_valid(self):
cfg = NoiseReductionConfig(enabled=True, noise_floor=-25.0)
ok, msg = cfg.validate()
assert ok is True
def test_noise_floor_out_of_range(self):
cfg = NoiseReductionConfig(enabled=True, noise_floor=-100.0)
ok, msg = cfg.validate()
assert ok is False
assert "noise_floor" in msg
# ── build_afftdn_filter 测试 ───────────────────────────────────────────────
class TestBuildAfftdnFilter:
def test_disabled_returns_anull(self):
cfg = NoiseReductionConfig(enabled=False)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "anull" in result
assert "[in]" in result
assert "[out]" in result
def test_medium_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
result = build_afftdn_filter(cfg, "[a]", "[nr]")
assert "afftdn=" in result
assert "nf=-25.0" in result or "nf=-25" in result
assert "[a]" in result
assert "[nr]" in result
def test_high_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.HIGH)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "afftdn=" in result
assert "nf=-15.0" in result or "nf=-15" in result
def test_low_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.LOW)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "afftdn=" in result
assert "nf=-35.0" in result or "nf=-35" in result
def test_custom_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-40.0)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "afftdn=" in result
assert "nf=-40.0" in result or "nf=-40" in result
def test_voice_enhance_adds_filters(self):
cfg = NoiseReductionConfig(enabled=True, voice_enhance=True)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "highpass" in result
assert "acompressor" in result
assert "loudnorm" in result
def test_no_voice_enhance_no_extra_filters(self):
cfg = NoiseReductionConfig(enabled=True, voice_enhance=False)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "highpass" not in result
assert "acompressor" not in result
# ── build_arnndn_filter 测试 ───────────────────────────────────────────────
class TestBuildArnndnFilter:
def test_disabled_returns_anull(self):
cfg = NoiseReductionConfig(enabled=False)
result = build_arnndn_filter(cfg, "[in]", "[out]", "model.rnnn")
assert "anull" in result
def test_enabled_returns_arnndn(self):
cfg = NoiseReductionConfig(enabled=True)
result = build_arnndn_filter(cfg, "[a]", "[nr]", "/path/to/model.rnnn")
assert "arnndn=" in result
assert "m=/path/to/model.rnnn" in result
assert "[a]" in result
assert "[nr]" in result
# ── apply_noise_reduction_if_needed 测试 ─────────────────────────────────
class TestApplyNoiseReductionIfNeeded:
def test_none_config_returns_none(self):
assert apply_noise_reduction_if_needed(None, "[in]", "[out]") is None
def test_disabled_returns_none(self):
assert apply_noise_reduction_if_needed({"enabled": False}, "[in]", "[out]") is None
def test_enabled_returns_filter(self):
result = apply_noise_reduction_if_needed({"enabled": True, "level": "medium"}, "[in]", "[out]")
assert result is not None
assert "afftdn" in result
def test_invalid_config_handles_exception(self):
# 异常情况应该返回 None 而不是抛出
result = apply_noise_reduction_if_needed("invalid", "[in]", "[out]")
assert result is None
# ── 工具函数测试 ───────────────────────────────────────────────────────────
class TestUtils:
def test_get_level_names_returns_four(self):
names = get_level_names()
assert len(names) == 4
assert "low" in names
assert "medium" in names
assert "high" in names
assert "custom" in names
+423
View File
@@ -0,0 +1,423 @@
"""trim_config 领域模型单测."""
from __future__ import annotations
import pytest
from packages.domain.trim_config import (
MIN_TRIM_DURATION,
TrimConfig,
TrimSegment,
build_audio_trim_filter,
build_video_trim_filter,
extract_trim_from_clip_config,
parse_segments_from_config,
resolve_segments,
)
# ── TrimConfig.from_dict 测试 ─────────────────────────────────────────────
class TestTrimConfigFromDict:
def test_none_returns_none(self):
assert TrimConfig.from_dict(None) is None
def test_empty_dict_returns_none(self):
assert TrimConfig.from_dict({}) is None
def test_all_zero_returns_none(self):
assert TrimConfig.from_dict({"start_time": 0, "end_time": 0, "duration": 0}) is None
def test_start_only_valid(self):
cfg = TrimConfig.from_dict({"start_time": 5.0})
assert cfg is not None
assert cfg.start_time == 5.0
assert cfg.end_time == 0
assert cfg.duration == 0
def test_duration_only_valid(self):
cfg = TrimConfig.from_dict({"duration": 10.0})
assert cfg is not None
assert cfg.duration == 10.0
assert cfg.start_time == 0
def test_start_and_duration(self):
cfg = TrimConfig.from_dict({"start_time": 2.0, "duration": 5.0})
assert cfg is not None
assert cfg.start_time == 2.0
assert cfg.duration == 5.0
def test_start_and_end(self):
cfg = TrimConfig.from_dict({"start_time": 1.0, "end_time": 5.0})
assert cfg is not None
assert cfg.start_time == 1.0
assert cfg.end_time == 5.0
def test_end_only(self):
cfg = TrimConfig.from_dict({"end_time": 8.0})
assert cfg is not None
assert cfg.end_time == 8.0
def test_string_values_coerced(self):
cfg = TrimConfig.from_dict({"start_time": "3.5", "duration": "2.0"})
assert cfg is not None
assert cfg.start_time == 3.5
assert cfg.duration == 2.0
def test_falsy_values_treated_as_zero(self):
cfg = TrimConfig.from_dict({"start_time": None, "duration": None})
assert cfg is None
def test_default_values(self):
cfg = TrimConfig()
assert cfg.start_time == 0.0
assert cfg.end_time == 0.0
assert cfg.duration == 0.0
# ── validate_and_resolve 测试 ─────────────────────────────────────────────
class TestValidateAndResolve:
def test_start_and_end_resolves_duration(self):
cfg = TrimConfig(start_time=2.0, end_time=7.0)
resolved = cfg.validate_and_resolve(100.0)
assert resolved.start_time == 2.0
assert resolved.end_time == 7.0
assert resolved.duration == 5.0
def test_start_and_duration_resolves_end(self):
cfg = TrimConfig(start_time=3.0, duration=10.0)
resolved = cfg.validate_and_resolve(100.0)
assert resolved.start_time == 3.0
assert resolved.duration == 10.0
assert resolved.end_time == 13.0
def test_end_and_duration_resolves_start(self):
cfg = TrimConfig(end_time=15.0, duration=5.0)
resolved = cfg.validate_and_resolve(100.0)
assert resolved.end_time == 15.0
assert resolved.duration == 5.0
assert resolved.start_time == 10.0
def test_start_only_takes_to_end(self):
cfg = TrimConfig(start_time=5.0)
resolved = cfg.validate_and_resolve(30.0)
assert resolved.start_time == 5.0
assert resolved.end_time == 30.0
assert resolved.duration == 25.0
def test_end_only_takes_from_start(self):
cfg = TrimConfig(end_time=8.0)
resolved = cfg.validate_and_resolve(30.0)
assert resolved.start_time == 0.0
assert resolved.end_time == 8.0
assert resolved.duration == 8.0
def test_duration_only_from_zero(self):
cfg = TrimConfig(duration=10.0)
resolved = cfg.validate_and_resolve(30.0)
assert resolved.start_time == 0.0
assert resolved.duration == 10.0
assert resolved.end_time == 10.0
def test_negative_start_clamped(self):
cfg = TrimConfig(start_time=-5.0, duration=10.0)
resolved = cfg.validate_and_resolve(30.0)
assert resolved.start_time == 0.0
def test_end_exceeds_asset_clamped(self):
cfg = TrimConfig(start_time=5.0, duration=50.0)
resolved = cfg.validate_and_resolve(30.0)
assert resolved.end_time == 30.0
assert resolved.duration == 25.0
def test_start_exceeds_asset_clamped(self):
cfg = TrimConfig(start_time=50.0, duration=10.0)
resolved = cfg.validate_and_resolve(30.0)
assert resolved.start_time < 30.0
assert resolved.end_time == 30.0
def test_end_before_start_invalid(self):
cfg = TrimConfig(start_time=10.0, end_time=5.0)
resolved = cfg.validate_and_resolve(30.0)
assert resolved.duration == 0.0
assert resolved.is_valid is False
def test_zero_asset_duration(self):
cfg = TrimConfig(start_time=1.0, duration=5.0)
resolved = cfg.validate_and_resolve(0.0)
assert resolved.is_noop
def test_negative_asset_duration(self):
cfg = TrimConfig(start_time=1.0, duration=5.0)
resolved = cfg.validate_and_resolve(-1.0)
assert resolved.is_noop
def test_end_and_duration_with_negative_start(self):
cfg = TrimConfig(end_time=3.0, duration=10.0)
resolved = cfg.validate_and_resolve(30.0)
assert resolved.start_time == 0.0
assert resolved.end_time == 3.0
assert resolved.duration == 3.0
def test_all_three_params_uses_start_duration(self):
cfg = TrimConfig(start_time=2.0, end_time=8.0, duration=3.0)
resolved = cfg.validate_and_resolve(30.0)
# 有 start + end 时应该用 start+end 推导 duration
assert resolved.start_time == 2.0
assert resolved.end_time == 8.0
assert resolved.duration == 6.0
def test_empty_config_returns_noop(self):
cfg = TrimConfig()
resolved = cfg.validate_and_resolve(30.0)
assert resolved.is_noop
# ── is_valid / is_noop / trim_from_start 测试 ─────────────────────────────
class TestProperties:
def test_is_valid_true_for_normal(self):
cfg = TrimConfig(start_time=0, end_time=0, duration=5.0)
assert cfg.is_valid is True
def test_is_valid_false_for_zero(self):
cfg = TrimConfig(duration=0.0)
assert cfg.is_valid is False
def test_is_valid_false_for_very_small(self):
cfg = TrimConfig(duration=0.01)
assert cfg.is_valid is False
def test_is_valid_true_at_boundary(self):
cfg = TrimConfig(duration=MIN_TRIM_DURATION)
assert cfg.is_valid is True
def test_is_noop_true_for_default(self):
cfg = TrimConfig()
assert cfg.is_noop is True
def test_is_noop_false_with_start(self):
cfg = TrimConfig(start_time=1.0)
assert cfg.is_noop is False
def test_is_noop_false_with_duration(self):
cfg = TrimConfig(duration=1.0)
assert cfg.is_noop is False
def test_trim_from_start_true(self):
cfg = TrimConfig(start_time=0.0, duration=5.0)
assert cfg.trim_from_start is True
def test_trim_from_start_false(self):
cfg = TrimConfig(start_time=2.0, duration=5.0)
assert cfg.trim_from_start is False
# ── TrimSegment 测试 ───────────────────────────────────────────────────────
class TestTrimSegment:
def test_from_dict_basic(self):
seg = TrimSegment.from_dict({"segment_id": "s1", "start_time": 1.0, "duration": 3.0})
assert seg.segment_id == "s1"
assert seg.trim.start_time == 1.0
assert seg.trim.duration == 3.0
assert seg.order == 0
def test_from_dict_with_order(self):
seg = TrimSegment.from_dict({"segment_id": "s2", "start_time": 0, "end_time": 5.0, "order": 2})
assert seg.order == 2
def test_from_dict_default_order(self):
seg = TrimSegment.from_dict({"start_time": 1.0}, default_order=5)
assert seg.order == 5
def test_from_dict_default_segment_id(self):
seg = TrimSegment.from_dict({"start_time": 1.0}, default_order=3)
assert seg.segment_id == "seg_3"
# ── build_video_trim_filter 测试 ───────────────────────────────────────────
class TestBuildVideoTrimFilter:
def test_noop_returns_setpts(self):
cfg = TrimConfig()
result = build_video_trim_filter("[0:v]", cfg, "[v]")
assert "setpts=PTS-STARTPTS" in result
assert "trim=" not in result
assert "[0:v]" in result
assert "[v]" in result
def test_with_start_and_duration(self):
cfg = TrimConfig(start_time=5.0, end_time=10.0, duration=5.0)
result = build_video_trim_filter("[0:v]", cfg, "[out]")
assert "trim=" in result
assert "start=5.000" in result
assert "duration=5.000" in result
assert "setpts=PTS-STARTPTS" in result
def test_contains_input_and_output_labels(self):
cfg = TrimConfig(start_time=1.0, duration=2.0)
result = build_video_trim_filter("[in_v]", cfg, "[out_v]")
assert "[in_v]" in result
assert "[out_v]" in result
def test_duration_only(self):
cfg = TrimConfig(duration=3.5)
result = build_video_trim_filter("[0:v]", cfg, "[v]")
assert "duration=3.500" in result
assert "start=" not in result
# ── build_audio_trim_filter 测试 ───────────────────────────────────────────
class TestBuildAudioTrimFilter:
def test_noop_returns_asetpts(self):
cfg = TrimConfig()
result = build_audio_trim_filter("[0:a]", cfg, "[a]")
assert "asetpts=PTS-STARTPTS" in result
assert "atrim=" not in result
def test_with_start_and_duration(self):
cfg = TrimConfig(start_time=2.0, end_time=7.0, duration=5.0)
result = build_audio_trim_filter("[0:a]", cfg, "[out]")
assert "atrim=" in result
assert "start=2.000" in result
assert "duration=5.000" in result
assert "asetpts=PTS-STARTPTS" in result
def test_contains_input_and_output_labels(self):
cfg = TrimConfig(start_time=1.0, duration=2.0)
result = build_audio_trim_filter("[in_a]", cfg, "[out_a]")
assert "[in_a]" in result
assert "[out_a]" in result
# ── resolve_segments 测试 ──────────────────────────────────────────────────
class TestResolveSegments:
def test_empty_list_returns_empty(self):
result = resolve_segments([], 30.0)
assert result == []
def test_single_segment(self):
segs = [TrimSegment(segment_id="s1", trim=TrimConfig(start_time=1.0, duration=5.0), order=0)]
result = resolve_segments(segs, 30.0)
assert len(result) == 1
assert result[0].segment_id == "s1"
assert result[0].trim.duration == 5.0
def test_invalid_segment_filters_out(self):
segs = [
TrimSegment(segment_id="good", trim=TrimConfig(start_time=0, duration=5.0), order=0),
TrimSegment(
segment_id="bad",
trim=TrimConfig(start_time=5.0, end_time=5.0), # end == start → duration 0
order=1,
),
]
result = resolve_segments(segs, 30.0)
assert len(result) == 1
assert result[0].segment_id == "good"
def test_sorted_by_order(self):
segs = [
TrimSegment(segment_id="s2", trim=TrimConfig(start_time=5.0, duration=3.0), order=2),
TrimSegment(segment_id="s1", trim=TrimConfig(start_time=0, duration=3.0), order=1),
TrimSegment(segment_id="s0", trim=TrimConfig(start_time=10.0, duration=3.0), order=0),
]
result = resolve_segments(segs, 30.0)
assert [s.segment_id for s in result] == ["s0", "s1", "s2"]
def test_negative_order_uses_index(self):
segs = [
TrimSegment(segment_id="s0", trim=TrimConfig(duration=3.0), order=-1),
]
result = resolve_segments(segs, 30.0)
assert len(result) == 1
assert result[0].order == 0
# ── parse_segments_from_config 测试 ────────────────────────────────────────
class TestParseSegmentsFromConfig:
def test_none_returns_empty(self):
assert parse_segments_from_config(None) == []
def test_empty_dict_returns_empty(self):
assert parse_segments_from_config({}) == []
def test_trim_segments_list(self):
config = {
"trim_segments": [
{"segment_id": "s1", "start_time": 0, "duration": 3.0, "order": 0},
{"segment_id": "s2", "start_time": 5.0, "duration": 2.0, "order": 1},
]
}
result = parse_segments_from_config(config)
assert len(result) == 2
assert result[0].segment_id == "s1"
assert result[1].segment_id == "s2"
def test_trim_segments_skips_non_dict(self):
config = {"trim_segments": [{"segment_id": "s1", "duration": 3.0}, "invalid", None]}
result = parse_segments_from_config(config)
assert len(result) == 1
def test_single_trim_compat(self):
config = {"trim_start": 1.0, "trim_duration": 5.0}
result = parse_segments_from_config(config)
assert len(result) == 1
assert result[0].segment_id == "main"
assert result[0].trim.start_time == 1.0
assert result[0].trim.duration == 5.0
def test_no_trim_fields_returns_empty(self):
config = {"other_field": "value"}
assert parse_segments_from_config(config) == []
# ── extract_trim_from_clip_config 测试 ────────────────────────────────────
class TestExtractTrimFromClipConfig:
def test_none_returns_none(self):
assert extract_trim_from_clip_config(None) is None
def test_empty_dict_returns_none(self):
assert extract_trim_from_clip_config({}) is None
def test_trim_subdict(self):
config = {"trim": {"start_time": 2.0, "duration": 5.0}}
cfg = extract_trim_from_clip_config(config)
assert cfg is not None
assert cfg.start_time == 2.0
assert cfg.duration == 5.0
def test_flat_trim_fields(self):
config = {"trim_start": 1.0, "trim_end": 6.0}
cfg = extract_trim_from_clip_config(config)
assert cfg is not None
assert cfg.start_time == 1.0
assert cfg.end_time == 6.0
def test_trim_subdict_empty(self):
config = {"trim": {}}
assert extract_trim_from_clip_config(config) is None
def test_no_trim_fields(self):
config = {"foo": "bar"}
assert extract_trim_from_clip_config(config) is None
def test_flat_trim_duration_only(self):
config = {"trim_duration": 10.0}
cfg = extract_trim_from_clip_config(config)
assert cfg is not None
assert cfg.duration == 10.0
+452
View File
@@ -0,0 +1,452 @@
"""watermark_config 领域模型单测."""
from __future__ import annotations
import pytest
from packages.domain.watermark_config import (
DEFAULT_FONT_COLOR,
DEFAULT_FONT_SIZE,
DEFAULT_MODE,
DEFAULT_OPACITY,
DEFAULT_POSITION,
DEFAULT_SCALE,
VALID_POSITIONS,
WATERMARK_POSITIONS,
WatermarkConfig,
build_image_watermark_filter,
build_text_watermark_filter,
calc_position,
calc_scroll_x,
get_position_display_name,
get_position_names,
)
# ── 常量测试 ────────────────────────────────────────────────────────────────
class TestConstants:
def test_nine_positions(self):
assert len(WATERMARK_POSITIONS) == 9
def test_all_position_keys_valid(self):
for key in WATERMARK_POSITIONS:
assert key in VALID_POSITIONS
def test_valid_positions_match(self):
assert set(WATERMARK_POSITIONS.keys()) == VALID_POSITIONS
def test_default_values(self):
assert DEFAULT_POSITION == "bottom_right"
assert DEFAULT_MODE == "text"
assert DEFAULT_SCALE == 0.2
assert DEFAULT_OPACITY == 0.8
assert DEFAULT_FONT_SIZE == 24
assert DEFAULT_FONT_COLOR == "white"
# ── WatermarkConfig.from_dict 测试 ─────────────────────────────────────────
class TestWatermarkConfigFromDict:
def test_none_returns_none(self):
assert WatermarkConfig.from_dict(None) is None
def test_empty_dict_returns_none(self):
assert WatermarkConfig.from_dict({}) is None
def test_disabled_returns_none(self):
assert WatermarkConfig.from_dict({"enabled": False}) is None
def test_text_mode_basic(self):
cfg = WatermarkConfig.from_dict({"enabled": True, "mode": "text", "text": "hello"})
assert cfg is not None
assert cfg.mode == "text"
assert cfg.text == "hello"
assert cfg.position == DEFAULT_POSITION
def test_image_mode_basic(self):
cfg = WatermarkConfig.from_dict({"enabled": True, "mode": "image", "image_path": "/tmp/wm.png"})
assert cfg is not None
assert cfg.mode == "image"
assert cfg.image_path == "/tmp/wm.png"
def test_image_mode_accepts_image_key(self):
cfg = WatermarkConfig.from_dict({"enabled": True, "mode": "image", "image": "/tmp/wm.png"})
assert cfg is not None
assert cfg.image_path == "/tmp/wm.png"
def test_image_mode_missing_path_returns_none(self):
assert WatermarkConfig.from_dict({"enabled": True, "mode": "image"}) is None
def test_text_mode_missing_text_returns_none(self):
assert WatermarkConfig.from_dict({"enabled": True, "mode": "text"}) is None
def test_text_mode_empty_text_returns_none(self):
assert WatermarkConfig.from_dict({"enabled": True, "mode": "text", "text": ""}) is None
def test_invalid_position_defaults(self):
cfg = WatermarkConfig.from_dict({"enabled": True, "mode": "text", "text": "hi", "position": "invalid"})
assert cfg.position == DEFAULT_POSITION
def test_custom_all_params(self):
cfg = WatermarkConfig.from_dict(
{
"enabled": True,
"mode": "text",
"text": "测试水印",
"position": "top_left",
"font_size": 32,
"font_color": "red",
"opacity": 0.5,
"margin_x": 30,
"margin_y": 40,
"scroll": True,
"scroll_speed": 100,
}
)
assert cfg is not None
assert cfg.text == "测试水印"
assert cfg.position == "top_left"
assert cfg.font_size == 32
assert cfg.font_color == "red"
assert cfg.opacity == 0.5
assert cfg.margin_x == 30
assert cfg.margin_y == 40
assert cfg.scroll is True
assert cfg.scroll_speed == 100
def test_default_mode_is_text(self):
cfg = WatermarkConfig.from_dict({"enabled": True, "text": "hi"})
assert cfg is not None
assert cfg.mode == "text"
# ── WatermarkConfig.validate 测试 ──────────────────────────────────────────
class TestWatermarkConfigValidate:
def test_valid_text_config(self):
cfg = WatermarkConfig(mode="text", text="hello")
ok, msg = cfg.validate()
assert ok is True
assert msg == ""
def test_valid_image_config(self):
cfg = WatermarkConfig(mode="image", image_path="/tmp/wm.png", scale=0.3)
ok, msg = cfg.validate()
assert ok is True
def test_invalid_position(self):
cfg = WatermarkConfig(mode="text", text="hi", position="nowhere")
ok, msg = cfg.validate()
assert ok is False
assert "位置" in msg
def test_opacity_negative(self):
cfg = WatermarkConfig(mode="text", text="hi", opacity=-0.1)
ok, msg = cfg.validate()
assert ok is False
assert "透明度" in msg
def test_opacity_over_one(self):
cfg = WatermarkConfig(mode="text", text="hi", opacity=1.5)
ok, msg = cfg.validate()
assert ok is False
def test_opacity_boundary_zero(self):
cfg = WatermarkConfig(mode="text", text="hi", opacity=0.0)
ok, _ = cfg.validate()
assert ok is True
def test_opacity_boundary_one(self):
cfg = WatermarkConfig(mode="text", text="hi", opacity=1.0)
ok, _ = cfg.validate()
assert ok is True
def test_image_missing_path(self):
cfg = WatermarkConfig(mode="image")
ok, msg = cfg.validate()
assert ok is False
assert "图片路径" in msg
def test_image_scale_too_small(self):
cfg = WatermarkConfig(mode="image", image_path="/a.png", scale=0.001)
ok, msg = cfg.validate()
assert ok is False
assert "缩放比例" in msg
def test_image_scale_too_large(self):
cfg = WatermarkConfig(mode="image", image_path="/a.png", scale=2.0)
ok, msg = cfg.validate()
assert ok is False
def test_image_scale_boundary_low(self):
cfg = WatermarkConfig(mode="image", image_path="/a.png", scale=0.01)
ok, _ = cfg.validate()
assert ok is True
def test_image_scale_boundary_high(self):
cfg = WatermarkConfig(mode="image", image_path="/a.png", scale=1.0)
ok, _ = cfg.validate()
assert ok is True
def test_text_missing_content(self):
cfg = WatermarkConfig(mode="text", text="")
ok, msg = cfg.validate()
assert ok is False
assert "文字内容" in msg
def test_text_font_size_zero(self):
cfg = WatermarkConfig(mode="text", text="hi", font_size=0)
ok, msg = cfg.validate()
assert ok is False
assert "字体大小" in msg
def test_text_font_size_negative(self):
cfg = WatermarkConfig(mode="text", text="hi", font_size=-5)
ok, msg = cfg.validate()
assert ok is False
def test_unknown_mode(self):
cfg = WatermarkConfig(mode="video")
ok, msg = cfg.validate()
assert ok is False
assert "模式" in msg
# ── has_effect 测试 ────────────────────────────────────────────────────────
class TestHasEffect:
def test_text_with_content_has_effect(self):
cfg = WatermarkConfig(mode="text", text="hello")
assert cfg.has_effect() is True
def test_text_empty_no_effect(self):
cfg = WatermarkConfig(mode="text", text="")
assert cfg.has_effect() is False
def test_text_zero_opacity_no_effect(self):
cfg = WatermarkConfig(mode="text", text="hello", opacity=0.0)
assert cfg.has_effect() is False
def test_image_with_path_has_effect(self):
cfg = WatermarkConfig(mode="image", image_path="/a.png")
assert cfg.has_effect() is True
def test_image_no_path_no_effect(self):
cfg = WatermarkConfig(mode="image")
assert cfg.has_effect() is False
def test_unknown_mode_no_effect(self):
cfg = WatermarkConfig(mode="unknown")
assert cfg.has_effect() is False
# ── calc_position 测试 ─────────────────────────────────────────────────────
class TestCalcPosition:
def test_top_left(self):
x, y = calc_position("top_left", 1000, 2000, 100, 50, 10, 20)
assert (x, y) == (10, 20)
def test_top_center(self):
x, y = calc_position("top_center", 1000, 2000, 100, 50, 10, 20)
assert (x, y) == (450, 20)
def test_top_right(self):
x, y = calc_position("top_right", 1000, 2000, 100, 50, 10, 20)
assert (x, y) == (890, 20)
def test_center_left(self):
x, y = calc_position("center_left", 1000, 2000, 100, 50, 10, 20)
assert (x, y) == (10, 975)
def test_center(self):
x, y = calc_position("center", 1000, 2000, 100, 50, 10, 20)
assert (x, y) == (450, 975)
def test_center_right(self):
x, y = calc_position("center_right", 1000, 2000, 100, 50, 10, 20)
assert (x, y) == (890, 975)
def test_bottom_left(self):
x, y = calc_position("bottom_left", 1000, 2000, 100, 50, 10, 20)
assert (x, y) == (10, 1930)
def test_bottom_center(self):
x, y = calc_position("bottom_center", 1000, 2000, 100, 50, 10, 20)
assert (x, y) == (450, 1930)
def test_bottom_right(self):
x, y = calc_position("bottom_right", 1000, 2000, 100, 50, 10, 20)
assert (x, y) == (890, 1930)
def test_unknown_position_defaults_bottom_right(self):
x, y = calc_position("invalid", 1000, 2000, 100, 50, 10, 20)
assert (x, y) == (890, 1930)
def test_zero_margin(self):
x, y = calc_position("top_left", 1000, 2000, 100, 50, 0, 0)
assert (x, y) == (0, 0)
def test_small_output(self):
x, y = calc_position("center", 100, 100, 50, 30, 5, 5)
assert (x, y) == (25, 35)
# ── calc_scroll_x 测试 ─────────────────────────────────────────────────────
class TestCalcScrollX:
def test_returns_string_expression(self):
result = calc_scroll_x("bottom", 1000, 200, 50)
assert isinstance(result, str)
def test_contains_mod_function(self):
result = calc_scroll_x("bottom", 1000, 200, 50)
assert "mod" in result
def test_contains_speed_and_width(self):
result = calc_scroll_x("bottom", 1080, 300, 60)
assert "1080" in result
assert "60" in result
assert "300" in result
# ── build_image_watermark_filter 测试 ──────────────────────────────────────
class TestBuildImageWatermarkFilter:
def test_returns_tuple(self):
cfg = WatermarkConfig(mode="image", image_path="/wm.png")
result = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
assert isinstance(result, tuple)
assert len(result) == 2
def test_filter_contains_overlay(self):
cfg = WatermarkConfig(mode="image", image_path="/wm.png")
filter_str, inputs = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
assert "overlay" in filter_str
def test_filter_contains_scale(self):
cfg = WatermarkConfig(mode="image", image_path="/wm.png", scale=0.5)
filter_str, _ = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
assert "scale=" in filter_str
def test_full_opacity_no_alpha_filter(self):
cfg = WatermarkConfig(mode="image", image_path="/wm.png", opacity=1.0)
filter_str, _ = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
assert "colorchannelmixer" not in filter_str
def test_partial_opacity_has_alpha_filter(self):
cfg = WatermarkConfig(mode="image", image_path="/wm.png", opacity=0.5)
filter_str, _ = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
assert "colorchannelmixer" in filter_str
assert "aa=0.5" in filter_str
def test_input_args_contains_image_path(self):
cfg = WatermarkConfig(mode="image", image_path="/path/to/wm.png")
_, inputs = build_image_watermark_filter("[in]", "/path/to/wm.png", 1080, 1920, "[out]", cfg)
assert inputs == ["-i", "/path/to/wm.png"]
def test_scroll_mode_has_t_variable(self):
cfg = WatermarkConfig(mode="image", image_path="/wm.png", scroll=True, scroll_speed=50)
filter_str, _ = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[out]", cfg)
assert "t" in filter_str
def test_output_label_appears(self):
cfg = WatermarkConfig(mode="image", image_path="/wm.png")
filter_str, _ = build_image_watermark_filter("[in]", "/wm.png", 1080, 1920, "[final]", cfg)
assert "[final]" in filter_str
# ── build_text_watermark_filter 测试 ───────────────────────────────────────
class TestBuildTextWatermarkFilter:
def test_returns_string(self):
cfg = WatermarkConfig(mode="text", text="hello")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
assert isinstance(result, str)
def test_contains_drawtext(self):
cfg = WatermarkConfig(mode="text", text="hello")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
assert "drawtext=" in result
def test_contains_text_content(self):
cfg = WatermarkConfig(mode="text", text="watermark_test")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
assert "watermark_test" in result
def test_contains_font_size(self):
cfg = WatermarkConfig(mode="text", text="hi", font_size=48)
result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
assert "fontsize=48" in result
def test_contains_font_color(self):
cfg = WatermarkConfig(mode="text", text="hi", font_color="red")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
assert "fontcolor=red" in result
def test_font_path_included_when_set(self):
cfg = WatermarkConfig(mode="text", text="hi", font_path="/fonts/a.ttf")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
assert "fontfile=" in result
assert "a.ttf" in result
def test_font_path_not_included_when_empty(self):
cfg = WatermarkConfig(mode="text", text="hi")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
assert "fontfile=" not in result
def test_scroll_mode_has_t_variable(self):
cfg = WatermarkConfig(mode="text", text="hello", scroll=True, scroll_speed=30)
result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
assert "t" in result
def test_no_scroll_uses_fixed_position(self):
cfg = WatermarkConfig(mode="text", text="hello", scroll=False)
result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
assert "x=" in result
# 非滚动模式 x= 后面应该是数字,不是表达式
# 找 x= 后的第一个字符
import re
match = re.search(r"x=(\d+)", result)
assert match is not None
def test_output_label_appears(self):
cfg = WatermarkConfig(mode="text", text="hi")
result = build_text_watermark_filter("[in]", "[text_out]", cfg, 1080, 1920)
assert "[text_out]" in result
def test_special_chars_escaped(self):
cfg = WatermarkConfig(mode="text", text="hello:world")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1080, 1920)
# 冒号应该被转义
assert "hello\\:world" in result or "hello\\\\\\:world" in result or "hello\\:" in result
# ── 工具函数测试 ────────────────────────────────────────────────────────────
class TestUtils:
def test_get_position_names_returns_nine(self):
names = get_position_names()
assert len(names) == 9
def test_get_position_names_all_valid(self):
names = get_position_names()
for name in names:
assert name in VALID_POSITIONS
def test_get_position_display_name_valid(self):
assert get_position_display_name("top_left") == "左上"
assert get_position_display_name("bottom_right") == "右下"
def test_get_position_display_name_invalid(self):
assert get_position_display_name("invalid") == "invalid"
+371
View File
@@ -0,0 +1,371 @@
"""XFade 转场滤镜构建领域模型单元测试."""
from __future__ import annotations
import pytest
from packages.domain.xfade_builder import (
DEFAULT_TRANSITION_DURATION,
SUPPORTED_TRANSITIONS,
XFADE_TRANSITION_MAP,
XFade_TRANSITION_NAMES,
build_xfade_filter_chain,
chain_filters,
resolve_xfade_transition,
)
# ── 常量测试 ─────────────────────────────────────────────────────────────────
class TestConstants:
def test_default_transition_duration(self):
assert DEFAULT_TRANSITION_DURATION == 0.5
def test_xfade_transition_map_not_empty(self):
assert len(XFADE_TRANSITION_MAP) > 0
def test_supported_transitions(self):
assert "fade" in SUPPORTED_TRANSITIONS
assert "dissolve" in SUPPORTED_TRANSITIONS
def test_xfade_transition_names(self):
assert "fade" in XFade_TRANSITION_NAMES
assert "dissolve" in XFade_TRANSITION_NAMES
# ── chain_filters 测试 ──────────────────────────────────────────────────────
class TestChainFilters:
def test_single_filter(self):
result = chain_filters(["scale=1280:720"], "v0")
assert result == "[0:v]scale=1280:720[v0]"
def test_multiple_filters(self):
result = chain_filters(["scale=1280:720", "fps=25"], "v0")
assert result == "[0:v]scale=1280:720,fps=25[v0]"
def test_empty_filters(self):
result = chain_filters([], "out")
assert result == "[0:v][out]"
def test_custom_input_label(self):
result = chain_filters(["scale=640:480"], "out", input_label="1:v")
assert result == "[1:v]scale=640:480[out]"
def test_three_filters(self):
result = chain_filters(["trim=0:5", "setpts=PTS-STARTPTS", "fps=30"], "v1")
assert result == "[0:v]trim=0:5,setpts=PTS-STARTPTS,fps=30[v1]"
# ── resolve_xfade_transition 测试 ───────────────────────────────────────────
class TestResolveXfadeTransition:
def test_fade(self):
assert resolve_xfade_transition("fade") == "fade"
def test_dissolve(self):
assert resolve_xfade_transition("dissolve") == "dissolve"
def test_crossfade_maps_to_dissolve(self):
assert resolve_xfade_transition("crossfade") == "dissolve"
def test_slideleft(self):
assert resolve_xfade_transition("slideleft") == "slideleft"
def test_slide_left_maps_to_slideleft(self):
assert resolve_xfade_transition("slide_left") == "slideleft"
def test_slide_default_left(self):
assert resolve_xfade_transition("slide") == "slideleft"
def test_slideup(self):
assert resolve_xfade_transition("slideup") == "slideup"
def test_zoom_maps_to_zoomin(self):
assert resolve_xfade_transition("zoom") == "zoomin"
def test_zoomin(self):
assert resolve_xfade_transition("zoomin") == "zoomin"
def test_wipe_default_left(self):
assert resolve_xfade_transition("wipe") == "wipeleft"
def test_wipeup(self):
assert resolve_xfade_transition("wipeup") == "wipeup"
def test_circle_maps_to_circlecrop(self):
assert resolve_xfade_transition("circle") == "circlecrop"
def test_rect_maps_to_rectcrop(self):
assert resolve_xfade_transition("rect") == "rectcrop"
def test_unknown_falls_back_to_fade(self):
assert resolve_xfade_transition("nonexistent_effect") == "fade"
def test_empty_string_falls_back_to_fade(self):
assert resolve_xfade_transition("") == "fade"
def test_enum_with_value_attribute(self):
"""测试带 .value 属性的枚举对象."""
class FakeEnum:
def __init__(self, val):
self.value = val
assert resolve_xfade_transition(FakeEnum("fade")) == "fade"
assert resolve_xfade_transition(FakeEnum("slideleft")) == "slideleft"
assert resolve_xfade_transition(FakeEnum("unknown")) == "fade"
# ── build_xfade_filter_chain 测试 ───────────────────────────────────────────
class TestBuildXfadeFilterChain:
# ── 边界情况 ──────────────────────────────────────────────────────
def test_empty_clips(self):
result, duration = build_xfade_filter_chain([], [], [])
assert result == ""
assert duration == 0.0
def test_single_clip(self):
result, duration = build_xfade_filter_chain([10.0], ["v0"], ["none"])
assert "copy" in result
assert "[v0]copy[outv]" in result
assert duration == 10.0
def test_single_clip_custom_output_label(self):
result, duration = build_xfade_filter_chain([5.0], ["a0"], ["none"], output_label="final")
assert "[a0]copy[final]" in result
assert duration == 5.0
# ── 两片段基础测试 ────────────────────────────────────────────────
def test_two_clips_basic(self):
result, duration = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "xfade=transition=fade" in result
assert "[v0][v1]" in result
assert "[outv]" in result
# 总时长 = 10 + 10 - 0.5 = 19.5
assert abs(duration - 19.5) < 0.01
def test_two_clips_custom_duration(self):
result, duration = build_xfade_filter_chain(
[10.0, 10.0],
["v0", "v1"],
["none", "fade"],
transition_duration=1.0,
)
assert "duration=1.000" in result
# 总时长 = 10 + 10 - 1.0 = 19.0
assert abs(duration - 19.0) < 0.01
def test_two_clips_offset(self):
"""两片段时 offset 应该为 0(cumulative - td * 1 = 10 - 0.5 = 9.5?不对)。
对于两个片段:
- cumulative = clip_durations[0] = 10.0
- offset = max(0, cumulative - td * i) = max(0, 10.0 - 0.5 * 1) = 9.5
- duration=0.5, offset=9.5
"""
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "offset=9.500" in result
# ── 多片段测试 ────────────────────────────────────────────────────
def test_three_clips(self):
result, duration = build_xfade_filter_chain(
[5.0, 5.0, 5.0],
["v0", "v1", "v2"],
["none", "fade", "dissolve"],
)
# 应该有两个 xfade
assert result.count("xfade=") == 2
# 第一个 xfade 输出标签 xf1,第二个 xfade 输出 outv
assert "xf1" in result
assert "[outv]" in result
# 总时长 ≈ 5 + 5 + 5 - 0.5 - 0.5 = 14.0
assert abs(duration - 14.0) < 0.1
def test_five_clips(self):
result, duration = build_xfade_filter_chain(
[3.0, 3.0, 3.0, 3.0, 3.0],
["v0", "v1", "v2", "v3", "v4"],
["none", "fade", "fade", "fade", "fade"],
)
assert result.count("xfade=") == 4
# 总时长 ≈ 15 - 4 * 0.5 = 13.0
assert abs(duration - 13.0) < 0.2
# ── 转场效果测试 ──────────────────────────────────────────────────
def test_dissolve_transition(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "dissolve"])
assert "transition=dissolve" in result
def test_slideleft_transition(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "slideleft"])
assert "transition=slideleft" in result
def test_cut_uses_fade(self):
"""cut 转场效果应该回退到 fade."""
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "cut"])
# cut 不是 XFADE_TRANSITION_MAP 的键,所以会回退到 fade
assert "transition=fade" in result
def test_transitions_shorter_than_clips(self):
"""如果 transitions 列表比 clips 短,剩余的用 'cut'(→ fade)."""
result, _ = build_xfade_filter_chain(
[5.0, 5.0, 5.0],
["v0", "v1", "v2"],
["none"], # 只有一个
)
# 第二个转场(index 2)会回退到 cut → fade
assert result.count("transition=fade") == 2
# ── 时长钳制测试 ──────────────────────────────────────────────────
def test_short_first_clip_truncates_transition(self):
"""第一个片段非常短,转场时长应该被钳制."""
result, duration = build_xfade_filter_chain(
[0.3, 10.0],
["v0", "v1"],
["none", "fade"],
transition_duration=1.0,
)
# offset = max(0, 0.3 - 1.0 * 1) = 0.0
# available = max(0, 0.3 - 0.0) = 0.3
# safe_td = min(1.0, 0.3, 剩余 10.0, clip_durations[1] 10.0) = 0.3
assert "duration=0.300" in result
assert abs(duration - 10.0) < 0.01 # 0.3 + 10.0 - 0.3 = 10.0
def test_zero_duration_clips(self):
"""零时长片段的边界情况."""
result, duration = build_xfade_filter_chain([0.0, 5.0], ["v0", "v1"], ["none", "fade"])
# 第一个片段 0 时长,转场时长应该被钳制到最小值 0.001
# offset = max(0, 0 - 0.5) = 0
# available = max(0, 0 - 0) = 0
# safe_td = min(0.5, 0, ...) = min(0.5, 0, 5.0, 5.0) = 0 → max(0.001, 0) = 0.001
assert "duration=0.001" in result
def test_very_long_transition_duration(self):
"""转场时长超过所有片段时长."""
result, duration = build_xfade_filter_chain(
[2.0, 2.0],
["v0", "v1"],
["none", "fade"],
transition_duration=5.0,
)
# offset = max(0, 2.0 - 5.0) = 0
# available = max(0, 2.0 - 0) = 2.0
# safe_td = min(5.0, 2.0, 剩余 2.0, 2.0) = 2.0
assert "duration=2.000" in result
assert abs(duration - 2.0) < 0.01 # 2 + 2 - 2 = 2
# ── 标签测试 ──────────────────────────────────────────────────────
def test_custom_labels(self):
result, _ = build_xfade_filter_chain(
[10.0, 10.0],
["clip_a", "clip_b"],
["none", "fade"],
output_label="final_v",
)
assert "[clip_a][clip_b]" in result
assert "[final_v]" in result
def test_intermediate_labels_three_clips(self):
result, _ = build_xfade_filter_chain([5.0, 5.0, 5.0], ["v0", "v1", "v2"], ["none", "fade", "fade"])
# 第一个 xfade 输出 xf1
assert "[xf1][v2]" in result or result.count("[xf1]") >= 1
# ── 总时长计算验证 ────────────────────────────────────────────────
def test_total_duration_two_equal_clips(self):
_, duration = build_xfade_filter_chain([8.0, 8.0], ["v0", "v1"], ["none", "fade"])
# 8 + 8 - 0.5 = 15.5
assert abs(duration - 15.5) < 0.01
def test_total_duration_no_transition_impossible(self):
"""即使 transition_duration=0,也有最小 0.001 的钳制."""
_, duration = build_xfade_filter_chain(
[10.0, 10.0],
["v0", "v1"],
["none", "fade"],
transition_duration=0.0,
)
# transition_duration=0,但 safe_td 有下限 0.001
assert duration < 20.0 # 应该小于 20(有重叠)
assert duration > 19.9 # 但接近 20
# ── 滤镜字符串格式验证 ────────────────────────────────────────────
def test_filter_format_contains_xfade_keyword(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "xfade=" in result
def test_filter_uses_semicolon_separator(self):
"""多步 xfade 之间用分号分隔."""
result, _ = build_xfade_filter_chain([5.0, 5.0, 5.0], ["v0", "v1", "v2"], ["none", "fade", "fade"])
assert ";" in result
# 3个片段 → 2个xfade → 1个分号
assert result.count("xfade=") == 2
assert result.count(";") == 1
def test_filter_has_transition_param(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "transition=fade" in result
def test_filter_has_duration_param(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "duration=" in result
def test_filter_has_offset_param(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "offset=" in result
# ── 各种转场效果遍历测试 ──────────────────────────────────────────
@pytest.mark.parametrize(
"transition_name",
list(XFADE_TRANSITION_MAP.keys()),
)
def test_all_supported_transitions(self, transition_name):
"""所有支持的转场效果都应该能正确生成滤镜."""
result, duration = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", transition_name])
expected = XFADE_TRANSITION_MAP[transition_name]
assert f"transition={expected}" in result
assert duration > 0
# ── 四片段复杂场景 ────────────────────────────────────────────────
def test_four_clips_different_durations(self):
durations = [3.0, 5.0, 2.0, 7.0]
result, duration = build_xfade_filter_chain(
durations,
["v0", "v1", "v2", "v3"],
["none", "fade", "dissolve", "slideleft"],
)
assert result.count("xfade=") == 3
# 总时长 = sum(durations) - 3 * 0.5 ≈ 17 - 1.5 = 15.5
assert abs(duration - 15.5) < 0.2
# ── transition_duration = 0 的边界 ───────────────────────────────
def test_zero_transition_duration_minimum_clamped(self):
result, _ = build_xfade_filter_chain(
[10.0, 10.0],
["v0", "v1"],
["none", "fade"],
transition_duration=0.0,
)
# 至少 0.001
assert "duration=0.001" in result
# ── 单片段自定义输出标签 ─────────────────────────────────────────
def test_single_clip_output_label(self):
result, _ = build_xfade_filter_chain([5.0], ["v0"], ["none"], output_label="result")
assert "[v0]copy[result]" in result