diff --git a/apps/web/src/pages/my-templates/MyTemplates.tsx b/apps/web/src/pages/my-templates/MyTemplates.tsx index cf60fafea..f1119f5ff 100644 --- a/apps/web/src/pages/my-templates/MyTemplates.tsx +++ b/apps/web/src/pages/my-templates/MyTemplates.tsx @@ -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 (
{/* 页面头部 */} @@ -179,69 +92,13 @@ const MyTemplates: React.FC = () => { {templates.map((tpl) => ( - - handleEdit(tpl)} /> - , - - handleCopy(tpl)} /> - , - - handleGenerate(tpl)} /> - , - handleDelete(tpl.id)} - okText="删除" - cancelText="取消" - > - - - - , - ]} - > -
- - {tpl.name} - - - {MODE_LABELS[tpl.mode as TemplateMode] || tpl.mode} - - 用户自制 -
- -
- - {tpl.segments.length} 片段 · 预估 ~{tpl.estimated_duration}s - - {tpl.category && ( - {tpl.category} - )} -
- - {tpl.tags.length > 0 && ( -
- {tpl.tags.map((tag) => ( - - {tag} - - ))} -
- )} - -
- - {tpl.title_config.ai_auto_select && AI标题} - {tpl.subtitle_config.enabled && 字幕} - {tpl.bgm_config.enabled && BGM} - -
- + ))} diff --git a/apps/web/src/pages/my-templates/components/TemplateCard.tsx b/apps/web/src/pages/my-templates/components/TemplateCard.tsx new file mode 100644 index 000000000..2979228c3 --- /dev/null +++ b/apps/web/src/pages/my-templates/components/TemplateCard.tsx @@ -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 = ({ + tpl, + onEdit, + onCopy, + onGenerate, + onDelete, +}) => ( + + onEdit(tpl)} /> + , + + onCopy(tpl)} /> + , + + onGenerate(tpl)} /> + , + onDelete(tpl.id)} + okText="删除" + cancelText="取消" + > + + + + , + ]} + > +
+ + {tpl.name} + + + {MODE_LABELS[tpl.mode as TemplateMode] || tpl.mode} + + 用户自制 +
+ +
+ + {tpl.segments.length} 片段 · 预估 ~{tpl.estimated_duration}s + + {tpl.category && {tpl.category}} +
+ + {tpl.tags.length > 0 && ( +
+ {tpl.tags.map((tag) => ( + + {tag} + + ))} +
+ )} + +
+ + {tpl.title_config.ai_auto_select && AI标题} + {tpl.subtitle_config.enabled && 字幕} + {tpl.bgm_config.enabled && BGM} + +
+ +) diff --git a/apps/web/src/pages/my-templates/hooks/useMyTemplates.ts b/apps/web/src/pages/my-templates/hooks/useMyTemplates.ts new file mode 100644 index 000000000..76610297c --- /dev/null +++ b/apps/web/src/pages/my-templates/hooks/useMyTemplates.ts @@ -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, + } +} diff --git a/apps/web/src/pages/subscription/UpgradeSubscription.tsx b/apps/web/src/pages/subscription/UpgradeSubscription.tsx index 720beeb2d..36033d6b0 100644 --- a/apps/web/src/pages/subscription/UpgradeSubscription.tsx +++ b/apps/web/src/pages/subscription/UpgradeSubscription.tsx @@ -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 = { - 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 }) => ( -
- - -
-) - -/** 自定义 Spinner 组件 */ -const Spinner: React.FC<{ size?: "small" | "large" }> = ({ size = "large" }) => ( -
-
-
-
-
-) - const UpgradeSubscription: React.FC = () => { const navigate = useNavigate() - const [subscription, setSubscription] = useState(null) - const [loading, setLoading] = useState(true) - const [submitting, setSubmitting] = useState(false) - const [selectedPlan, setSelectedPlan] = useState("standard") - const [billingCycle, setBillingCycle] = useState("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 (
- +
{(["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 ? "处理中..." : "确认变更"} @@ -220,7 +125,7 @@ const UpgradeSubscription: React.FC = () => { + +
+) + +interface SpinnerProps { + size?: "small" | "large" +} + +/** 自定义 Spinner 组件 */ +export const Spinner: React.FC = ({ size = "large" }) => ( +
+
+
+
+
+) diff --git a/apps/web/src/pages/subscription/constants.ts b/apps/web/src/pages/subscription/constants.ts new file mode 100644 index 000000000..0e1b1843e --- /dev/null +++ b/apps/web/src/pages/subscription/constants.ts @@ -0,0 +1,16 @@ +import type { PlanType, BillingCycle } from "@/api/subscription" + +export const PLANS_META: Record = { + 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 +} diff --git a/apps/web/src/pages/subscription/hooks/useSubscription.ts b/apps/web/src/pages/subscription/hooks/useSubscription.ts new file mode 100644 index 000000000..eeeb4e8c2 --- /dev/null +++ b/apps/web/src/pages/subscription/hooks/useSubscription.ts @@ -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(null) + const [loading, setLoading] = useState(true) + const [submitting, setSubmitting] = useState(false) + const [selectedPlan, setSelectedPlan] = useState("standard") + const [billingCycle, setBillingCycle] = useState("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, + } +} diff --git a/apps/worker/video_processing/chroma_key_engine.py b/apps/worker/video_processing/chroma_key_engine.py index 5176f95f4..7499b2f61 100755 --- a/apps/worker/video_processing/chroma_key_engine.py +++ b/apps/worker/video_processing/chroma_key_engine.py @@ -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) diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py index 8ac5f493a..6cac1228b 100755 --- a/apps/worker/video_processing/ffmpeg_utils.py +++ b/apps/worker/video_processing/ffmpeg_utils.py @@ -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) diff --git a/apps/worker/video_processing/noise_reduction_engine.py b/apps/worker/video_processing/noise_reduction_engine.py index d1d5e053f..e55220d62 100755 --- a/apps/worker/video_processing/noise_reduction_engine.py +++ b/apps/worker/video_processing/noise_reduction_engine.py @@ -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) diff --git a/apps/worker/video_processing/render_subtitles.py b/apps/worker/video_processing/render_subtitles.py old mode 100644 new mode 100755 index 0ace09a38..9b8f58e63 --- a/apps/worker/video_processing/render_subtitles.py +++ b/apps/worker/video_processing/render_subtitles.py @@ -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 diff --git a/packages/domain/ass_subtitle_builder.py b/packages/domain/ass_subtitle_builder.py new file mode 100755 index 000000000..143d56f80 --- /dev/null +++ b/packages/domain/ass_subtitle_builder.py @@ -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)} +""" diff --git a/packages/domain/chroma_key_config.py b/packages/domain/chroma_key_config.py new file mode 100755 index 000000000..3d112b117 --- /dev/null +++ b/packages/domain/chroma_key_config.py @@ -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())) diff --git a/packages/domain/noise_reduction_config.py b/packages/domain/noise_reduction_config.py new file mode 100755 index 000000000..6ab44a452 --- /dev/null +++ b/packages/domain/noise_reduction_config.py @@ -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] diff --git a/packages/domain/xfade_builder.py b/packages/domain/xfade_builder.py new file mode 100755 index 000000000..c5ec2a505 --- /dev/null +++ b/packages/domain/xfade_builder.py @@ -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) diff --git a/tests/unit/test_ass_subtitle_builder.py b/tests/unit/test_ass_subtitle_builder.py new file mode 100755 index 000000000..971255014 --- /dev/null +++ b/tests/unit/test_ass_subtitle_builder.py @@ -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 diff --git a/tests/unit/test_chroma_key_config.py b/tests/unit/test_chroma_key_config.py new file mode 100755 index 000000000..1c543d340 --- /dev/null +++ b/tests/unit/test_chroma_key_config.py @@ -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 diff --git a/tests/unit/test_noise_reduction_config.py b/tests/unit/test_noise_reduction_config.py new file mode 100755 index 000000000..0a41a231d --- /dev/null +++ b/tests/unit/test_noise_reduction_config.py @@ -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 diff --git a/tests/unit/test_xfade_builder.py b/tests/unit/test_xfade_builder.py new file mode 100755 index 000000000..b69b75d4d --- /dev/null +++ b/tests/unit/test_xfade_builder.py @@ -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