Merge branch 'develop' into refactor/duplication-results-split
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 20s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m43s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 2m2s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 1m55s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 43s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 2m34s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 2m48s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 1m3s
Preview Deploy / Deploy Preview Environment (pull_request) Failing after 37s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 58s
AI Code Review / AI Code Review (pull_request) Successful in 2m0s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 6m14s
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1226h50m49s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1226h50m51s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1227h23m36s

This commit is contained in:
2026-07-27 12:33:35 +08:00
19 changed files with 2908 additions and 1035 deletions
+24 -167
View File
@@ -3,126 +3,39 @@
* 卡片视图展示用户已保存的剪辑模板
* 支持搜索、分类筛选、编辑/复制/删除/使用模板生成
*/
import React, { useState } from "react"
import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"
import {
Typography,
Card,
Input,
Select,
Tag,
Button,
Space,
Empty,
Spin,
Tooltip,
message,
Popconfirm,
Row,
Col,
} from "antd"
import {
SearchOutlined,
EditOutlined,
CopyOutlined,
DeleteOutlined,
VideoCameraOutlined,
AppstoreOutlined,
PlusOutlined,
} from "@ant-design/icons"
import React from "react"
import { Typography, Input, Select, Button, Empty, Spin, Row, Col } from "antd"
import { SearchOutlined, AppstoreOutlined, PlusOutlined } from "@ant-design/icons"
import { useNavigate } from "react-router-dom"
import {
getEditingTemplates,
getTemplateCategories,
deleteEditingTemplate,
createEditingTemplate,
MODE_LABELS,
MODE_COLORS,
type EditingTemplate,
type TemplateMode,
} from "@/api/editing-planner"
import type { EditingTemplate } from "@/api/editing-planner"
import { useMyTemplates } from "./hooks/useMyTemplates"
import { TemplateCard } from "./components/TemplateCard"
import "./MyTemplates.css"
const { Title, Text } = Typography
const MyTemplates: React.FC = () => {
const navigate = useNavigate()
const queryClient = useQueryClient()
const {
searchText,
setSearchText,
filterCategory,
setFilterCategory,
templates,
categories,
isLoading,
handleCopy,
handleDelete,
} = useMyTemplates()
const [searchText, setSearchText] = useState("")
const [filterCategory, setFilterCategory] = useState("")
/* ── 数据查询 ── */
const { data: templates = [], isLoading } = useQuery({
queryKey: ["editing-templates", filterCategory, searchText],
queryFn: () =>
getEditingTemplates({
category: filterCategory || undefined,
tag: searchText || undefined,
}),
})
const { data: categories = [] } = useQuery({
queryKey: ["template-categories"],
queryFn: getTemplateCategories,
})
/* ── Mutations ── */
const deleteMutation = useMutation({
mutationFn: deleteEditingTemplate,
onSuccess: () => {
message.success("模板已删除")
queryClient.invalidateQueries({ queryKey: ["editing-templates"] })
},
onError: (err: unknown) => {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("删除失败")
},
})
const copyMutation = useMutation({
mutationFn: (tpl: EditingTemplate) =>
createEditingTemplate({
name: `${tpl.name}(副本)`,
mode: tpl.mode,
category: tpl.category,
tags: tpl.tags,
title_config: tpl.title_config,
subtitle_config: tpl.subtitle_config,
bgm_config: tpl.bgm_config,
estimated_duration:
tpl.estimated_duration ??
Math.round(
tpl.segments.reduce((s, seg) => s + (seg.duration_min + seg.duration_max) / 2, 0),
),
segments: tpl.segments.map(({ id: _id, ...rest }) => rest),
}),
onSuccess: () => {
message.success("模板已复制")
queryClient.invalidateQueries({ queryKey: ["editing-templates"] })
},
onError: (err: unknown) => {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("复制失败")
},
})
/* ── 操作 ── */
const handleEdit = (tpl: EditingTemplate) => {
navigate(`/editing-planner?template=${tpl.id}`)
}
const handleGenerate = (tpl: EditingTemplate) => {
// 跳转到智能剪辑页面,统一从智能剪辑出片
navigate(`/generate?templateId=${tpl.id}`)
}
const handleCopy = (tpl: EditingTemplate) => {
copyMutation.mutate(tpl)
}
const handleDelete = (id: string) => {
deleteMutation.mutate(id)
}
return (
<div className="mt-page">
{/* 页面头部 */}
@@ -179,69 +92,13 @@ const MyTemplates: React.FC = () => {
<Row gutter={[16, 16]}>
{templates.map((tpl) => (
<Col key={tpl.id} xs={24} sm={12} md={8} lg={6}>
<Card
className="mt-card"
hoverable
actions={[
<Tooltip title="编辑" key="edit">
<EditOutlined onClick={() => handleEdit(tpl)} />
</Tooltip>,
<Tooltip title="复制" key="copy">
<CopyOutlined onClick={() => handleCopy(tpl)} />
</Tooltip>,
<Tooltip title="使用模板生成" key="generate">
<VideoCameraOutlined onClick={() => handleGenerate(tpl)} />
</Tooltip>,
<Popconfirm
key="delete"
title="确定删除此模板?"
onConfirm={() => handleDelete(tpl.id)}
okText="删除"
cancelText="取消"
>
<Tooltip title="删除">
<DeleteOutlined style={{ color: "#ff4d4f" }} />
</Tooltip>
</Popconfirm>,
]}
>
<div className="mt-card-head">
<Text strong ellipsis style={{ fontSize: 15 }}>
{tpl.name}
</Text>
<Tag color={MODE_COLORS[tpl.mode as TemplateMode] || "default"}>
{MODE_LABELS[tpl.mode as TemplateMode] || tpl.mode}
</Tag>
<Tag color="green">用户自制</Tag>
</div>
<div className="mt-card-meta">
<Text type="secondary" style={{ fontSize: 12 }}>
{tpl.segments.length} 片段 · 预估 ~{tpl.estimated_duration}s
</Text>
{tpl.category && (
<Tag style={{ fontSize: 11, marginTop: 4 }}>{tpl.category}</Tag>
)}
</div>
{tpl.tags.length > 0 && (
<div className="mt-card-tags">
{tpl.tags.map((tag) => (
<Tag key={tag} style={{ fontSize: 11 }}>
{tag}
</Tag>
))}
</div>
)}
<div className="mt-card-config">
<Space size={4} wrap>
{tpl.title_config.ai_auto_select && <Tag color="cyan">AI标题</Tag>}
{tpl.subtitle_config.enabled && <Tag color="geekblue">字幕</Tag>}
{tpl.bgm_config.enabled && <Tag color="pink">BGM</Tag>}
</Space>
</div>
</Card>
<TemplateCard
tpl={tpl}
onEdit={handleEdit}
onCopy={handleCopy}
onGenerate={handleGenerate}
onDelete={handleDelete}
/>
</Col>
))}
</Row>
@@ -0,0 +1,92 @@
import React from "react"
import { Card, Tag, Tooltip, Popconfirm, Space, Typography } from "antd"
import { EditOutlined, CopyOutlined, DeleteOutlined, VideoCameraOutlined } from "@ant-design/icons"
import {
MODE_LABELS,
MODE_COLORS,
type EditingTemplate,
type TemplateMode,
} from "@/api/editing-planner"
const { Text } = Typography
interface TemplateCardProps {
tpl: EditingTemplate
onEdit: (tpl: EditingTemplate) => void
onCopy: (tpl: EditingTemplate) => void
onGenerate: (tpl: EditingTemplate) => void
onDelete: (id: string) => void
}
/**
* 单个模板卡片组件
*/
export const TemplateCard: React.FC<TemplateCardProps> = ({
tpl,
onEdit,
onCopy,
onGenerate,
onDelete,
}) => (
<Card
className="mt-card"
hoverable
actions={[
<Tooltip title="编辑" key="edit">
<EditOutlined onClick={() => onEdit(tpl)} />
</Tooltip>,
<Tooltip title="复制" key="copy">
<CopyOutlined onClick={() => onCopy(tpl)} />
</Tooltip>,
<Tooltip title="使用模板生成" key="generate">
<VideoCameraOutlined onClick={() => onGenerate(tpl)} />
</Tooltip>,
<Popconfirm
key="delete"
title="确定删除此模板?"
onConfirm={() => onDelete(tpl.id)}
okText="删除"
cancelText="取消"
>
<Tooltip title="删除">
<DeleteOutlined style={{ color: "#ff4d4f" }} />
</Tooltip>
</Popconfirm>,
]}
>
<div className="mt-card-head">
<Text strong ellipsis style={{ fontSize: 15 }}>
{tpl.name}
</Text>
<Tag color={MODE_COLORS[tpl.mode as TemplateMode] || "default"}>
{MODE_LABELS[tpl.mode as TemplateMode] || tpl.mode}
</Tag>
<Tag color="green">用户自制</Tag>
</div>
<div className="mt-card-meta">
<Text type="secondary" style={{ fontSize: 12 }}>
{tpl.segments.length} 片段 · 预估 ~{tpl.estimated_duration}s
</Text>
{tpl.category && <Tag style={{ fontSize: 11, marginTop: 4 }}>{tpl.category}</Tag>}
</div>
{tpl.tags.length > 0 && (
<div className="mt-card-tags">
{tpl.tags.map((tag) => (
<Tag key={tag} style={{ fontSize: 11 }}>
{tag}
</Tag>
))}
</div>
)}
<div className="mt-card-config">
<Space size={4} wrap>
{tpl.title_config.ai_auto_select && <Tag color="cyan">AI标题</Tag>}
{tpl.subtitle_config.enabled && <Tag color="geekblue">字幕</Tag>}
{tpl.bgm_config.enabled && <Tag color="pink">BGM</Tag>}
</Space>
</div>
</Card>
)
@@ -0,0 +1,100 @@
import { useState } from "react"
import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"
import { message } from "antd"
import {
getEditingTemplates,
getTemplateCategories,
deleteEditingTemplate,
createEditingTemplate,
type EditingTemplate,
} from "@/api/editing-planner"
/**
* 我的模板数据 Hook
* 封装模板列表查询、筛选、删除、复制等数据操作
*/
export function useMyTemplates() {
const queryClient = useQueryClient()
const [searchText, setSearchText] = useState("")
const [filterCategory, setFilterCategory] = useState("")
/* 模板列表 */
const { data: templates = [], isLoading } = useQuery({
queryKey: ["editing-templates", filterCategory, searchText],
queryFn: () =>
getEditingTemplates({
category: filterCategory || undefined,
tag: searchText || undefined,
}),
})
/* 分类列表 */
const { data: categories = [] } = useQuery({
queryKey: ["template-categories"],
queryFn: getTemplateCategories,
})
/* 删除 mutation */
const deleteMutation = useMutation({
mutationFn: deleteEditingTemplate,
onSuccess: () => {
message.success("模板已删除")
queryClient.invalidateQueries({ queryKey: ["editing-templates"] })
},
onError: (err: unknown) => {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("删除失败")
},
})
/* 复制 mutation */
const copyMutation = useMutation({
mutationFn: (tpl: EditingTemplate) =>
createEditingTemplate({
name: `${tpl.name}(副本)`,
mode: tpl.mode,
category: tpl.category,
tags: tpl.tags,
title_config: tpl.title_config,
subtitle_config: tpl.subtitle_config,
bgm_config: tpl.bgm_config,
estimated_duration:
tpl.estimated_duration ??
Math.round(
tpl.segments.reduce((s, seg) => s + (seg.duration_min + seg.duration_max) / 2, 0),
),
segments: tpl.segments.map(({ id: _id, ...rest }) => rest),
}),
onSuccess: () => {
message.success("模板已复制")
queryClient.invalidateQueries({ queryKey: ["editing-templates"] })
},
onError: (err: unknown) => {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("复制失败")
},
})
const handleCopy = (tpl: EditingTemplate) => {
copyMutation.mutate(tpl)
}
const handleDelete = (id: string) => {
deleteMutation.mutate(id)
}
return {
// 状态
searchText,
setSearchText,
filterCategory,
setFilterCategory,
// 数据
templates,
categories,
isLoading,
// 操作
handleCopy,
handleDelete,
isDeleting: deleteMutation.isPending,
isCopying: copyMutation.isPending,
}
}
@@ -2,151 +2,59 @@
* 升级/降级/续费页面
* P1-3: antd Button/Modal/Radio/Spin → 自定义 UI 组件
*/
import React, { useState, useEffect } from "react"
import { message } from "antd"
import { Button, Modal } from "@/components/ui"
import React from "react"
import { Modal } from "@/components/ui"
import { useNavigate } from "react-router-dom"
import {
getCurrentSubscription,
changePlan,
toggleAutoRenew,
cancelSubscription,
} from "@/api/subscription"
import type { SubscriptionInfo, PlanType, BillingCycle } from "@/api/subscription"
import type { PlanType } from "@/api/subscription"
import PageHead from "@/components/layout/PageHead"
import { Button } from "@/components/ui"
import { PLANS_META, getPlanName, getPlanPrice } from "./constants"
import { BillingCycleSwitch, Spinner } from "./components/SubscriptionUI"
import { useSubscription } from "./hooks/useSubscription"
import "./UpgradeSubscription.css"
const PLANS_META: Record<string, { name: string; price: number; yearlyPrice: number }> = {
free: { name: "体验版", price: 0, yearlyPrice: 0 },
standard: { name: "标准版", price: 99, yearlyPrice: 990 },
pro: { name: "专业版", price: 299, yearlyPrice: 2990 },
enterprise: { name: "企业版", price: 0, yearlyPrice: 0 },
}
/** 自定义计费周期切换组件 */
const BillingCycleSwitch: React.FC<{
value: BillingCycle
onChange: (cycle: BillingCycle) => void
monthlyPrice: number
yearlyPrice: number
}> = ({ value, onChange, monthlyPrice, yearlyPrice }) => (
<div className="xx-billing-cycle-switch">
<button
type="button"
className={`xx-billing-cycle-btn ${value === "monthly" ? "active" : ""}`}
onClick={() => onChange("monthly")}
>
¥{monthlyPrice}/月
</button>
<button
type="button"
className={`xx-billing-cycle-btn ${value === "yearly" ? "active" : ""}`}
onClick={() => onChange("yearly")}
>
¥{yearlyPrice}/年
{yearlyPrice > 0 && monthlyPrice > 0 && (
<span className="xx-save">省 ¥{monthlyPrice * 12 - yearlyPrice}</span>
)}
</button>
</div>
)
/** 自定义 Spinner 组件 */
const Spinner: React.FC<{ size?: "small" | "large" }> = ({ size = "large" }) => (
<div className={`xx-spinner xx-spinner--${size}`}>
<div className="xx-spinner-dot" />
<div className="xx-spinner-dot" />
<div className="xx-spinner-dot" />
</div>
)
const UpgradeSubscription: React.FC = () => {
const navigate = useNavigate()
const [subscription, setSubscription] = useState<SubscriptionInfo | null>(null)
const [loading, setLoading] = useState(true)
const [submitting, setSubmitting] = useState(false)
const [selectedPlan, setSelectedPlan] = useState<PlanType>("standard")
const [billingCycle, setBillingCycle] = useState<BillingCycle>("monthly")
const {
subscription,
loading,
submitting,
selectedPlan,
billingCycle,
setSelectedPlan,
setBillingCycle,
executeChangePlan,
handleToggleAutoRenew,
handleCancel,
} = useSubscription()
useEffect(() => {
loadSubscription()
}, [])
const loadSubscription = async () => {
try {
const data = await getCurrentSubscription()
setSubscription(data)
setSelectedPlan(data.plan_id)
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("获取订阅信息失败")
} finally {
setLoading(false)
}
}
const handleUpgrade = async () => {
const handleUpgradeClick = () => {
if (!subscription) return
if (selectedPlan === subscription.plan_id && billingCycle === subscription.billing_cycle) {
message.info("当前已是该套餐")
return
}
const plan = PLANS_META[selectedPlan]
const price = billingCycle === "yearly" ? plan.yearlyPrice : plan.price
const price = getPlanPrice(selectedPlan, billingCycle)
Modal.confirm({
title: "确认变更套餐",
content: `即将变更为「${plan.name}」(${billingCycle === "monthly" ? "月付" : "年付"}),${price > 0 ? `费用 ¥${price}${billingCycle === "monthly" ? "/月" : "/年"}` : "免费"}。变更立即生效。`,
okText: "确认变更",
cancelText: "取消",
onOk: async () => {
try {
setSubmitting(true)
const res = await changePlan({
target_plan_id: selectedPlan,
billing_cycle: billingCycle,
})
if (res.success) {
message.success(res.message)
setSubscription(res.new_subscription ?? null)
} else {
message.error(res.message)
}
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("套餐变更失败,请重试")
} finally {
setSubmitting(false)
}
},
onOk: executeChangePlan,
})
}
const handleToggleAutoRenew = async (enabled: boolean) => {
try {
const res = await toggleAutoRenew(enabled)
message.success(res.message)
if (subscription) {
setSubscription({ ...subscription, auto_renew: enabled })
}
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("操作失败")
}
}
const handleCancel = () => {
const handleCancelClick = () => {
Modal.confirm({
title: "确认取消订阅",
content: "取消后,当前周期结束前仍可正常使用,到期后降级为体验版。",
okText: "确认取消",
cancelText: "再想想",
onOk: async () => {
try {
const res = await cancelSubscription()
message.success(res.message)
navigate("/app/subscription")
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("取消失败")
}
const ok = await handleCancel()
if (ok) navigate("/app/subscription")
},
})
}
@@ -163,10 +71,7 @@ const UpgradeSubscription: React.FC = () => {
return (
<div className="xx-upgrade-page">
<PageHead
title="变更订阅方案"
description={`当前套餐:${PLANS_META[currentPlan]?.name ?? "体验版"}`}
/>
<PageHead title="变更订阅方案" description={`当前套餐:${getPlanName(currentPlan)}`} />
<div className="xx-upgrade-plans">
{(["standard", "pro", "enterprise"] as PlanType[]).map((planId) => {
@@ -198,7 +103,7 @@ const UpgradeSubscription: React.FC = () => {
buttonType="primary"
buttonSize="lg"
disabled={submitting || selectedPlan === currentPlan}
onClick={handleUpgrade}
onClick={handleUpgradeClick}
>
{submitting ? "处理中..." : "确认变更"}
</Button>
@@ -220,7 +125,7 @@ const UpgradeSubscription: React.FC = () => {
<Button
buttonType="danger"
buttonSize="sm"
onClick={handleCancel}
onClick={handleCancelClick}
className="xx-cancel-btn"
>
取消订阅
@@ -0,0 +1,50 @@
import React from "react"
import type { BillingCycle } from "@/api/subscription"
interface BillingCycleSwitchProps {
value: BillingCycle
onChange: (cycle: BillingCycle) => void
monthlyPrice: number
yearlyPrice: number
}
/** 自定义计费周期切换组件 */
export const BillingCycleSwitch: React.FC<BillingCycleSwitchProps> = ({
value,
onChange,
monthlyPrice,
yearlyPrice,
}) => (
<div className="xx-billing-cycle-switch">
<button
type="button"
className={`xx-billing-cycle-btn ${value === "monthly" ? "active" : ""}`}
onClick={() => onChange("monthly")}
>
¥{monthlyPrice}/月
</button>
<button
type="button"
className={`xx-billing-cycle-btn ${value === "yearly" ? "active" : ""}`}
onClick={() => onChange("yearly")}
>
¥{yearlyPrice}/年
{yearlyPrice > 0 && monthlyPrice > 0 && (
<span className="xx-save">省 ¥{monthlyPrice * 12 - yearlyPrice}</span>
)}
</button>
</div>
)
interface SpinnerProps {
size?: "small" | "large"
}
/** 自定义 Spinner 组件 */
export const Spinner: React.FC<SpinnerProps> = ({ size = "large" }) => (
<div className={`xx-spinner xx-spinner--${size}`}>
<div className="xx-spinner-dot" />
<div className="xx-spinner-dot" />
<div className="xx-spinner-dot" />
</div>
)
@@ -0,0 +1,16 @@
import type { PlanType, BillingCycle } from "@/api/subscription"
export const PLANS_META: Record<string, { name: string; price: number; yearlyPrice: number }> = {
free: { name: "体验版", price: 0, yearlyPrice: 0 },
standard: { name: "标准版", price: 99, yearlyPrice: 990 },
pro: { name: "专业版", price: 299, yearlyPrice: 2990 },
enterprise: { name: "企业版", price: 0, yearlyPrice: 0 },
}
export const getPlanName = (planId: PlanType | string) => PLANS_META[planId]?.name ?? "体验版"
export const getPlanPrice = (planId: PlanType | string, cycle: BillingCycle) => {
const plan = PLANS_META[planId]
if (!plan) return 0
return cycle === "yearly" ? plan.yearlyPrice : plan.price
}
@@ -0,0 +1,111 @@
import { useState, useEffect, useCallback } from "react"
import { message } from "antd"
import {
getCurrentSubscription,
changePlan,
toggleAutoRenew,
cancelSubscription,
type SubscriptionInfo,
type PlanType,
type BillingCycle,
} from "@/api/subscription"
/**
* 订阅管理 Hook
* 封装订阅信息查询、套餐变更、自动续费切换、取消订阅等逻辑
*/
export function useSubscription() {
const [subscription, setSubscription] = useState<SubscriptionInfo | null>(null)
const [loading, setLoading] = useState(true)
const [submitting, setSubmitting] = useState(false)
const [selectedPlan, setSelectedPlan] = useState<PlanType>("standard")
const [billingCycle, setBillingCycle] = useState<BillingCycle>("monthly")
const loadSubscription = useCallback(async () => {
try {
const data = await getCurrentSubscription()
setSubscription(data)
setSelectedPlan(data.plan_id)
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("获取订阅信息失败")
} finally {
setLoading(false)
}
}, [])
useEffect(() => {
loadSubscription()
}, [loadSubscription])
const handleUpgrade = useCallback(async () => {
if (!subscription) return
if (selectedPlan === subscription.plan_id && billingCycle === subscription.billing_cycle) {
message.info("当前已是该套餐")
return
}
// 由调用方决定是否弹确认框
}, [subscription, selectedPlan, billingCycle])
const executeChangePlan = useCallback(async () => {
try {
setSubmitting(true)
const res = await changePlan({
target_plan_id: selectedPlan,
billing_cycle: billingCycle,
})
if (res.success) {
message.success(res.message)
setSubscription(res.new_subscription ?? null)
} else {
message.error(res.message)
}
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("套餐变更失败,请重试")
} finally {
setSubmitting(false)
}
}, [selectedPlan, billingCycle])
const handleToggleAutoRenew = useCallback(
async (enabled: boolean) => {
try {
const res = await toggleAutoRenew(enabled)
message.success(res.message)
if (subscription) {
setSubscription({ ...subscription, auto_renew: enabled })
}
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("操作失败")
}
},
[subscription],
)
const handleCancel = useCallback(async () => {
try {
const res = await cancelSubscription()
message.success(res.message)
return true
} catch (err: unknown) {
if (!(err as { __msgShown?: boolean })?.__msgShown) message.error("取消失败")
return false
}
}, [])
return {
// 状态
subscription,
loading,
submitting,
selectedPlan,
billingCycle,
setSelectedPlan,
setBillingCycle,
// 操作
loadSubscription,
handleUpgrade,
executeChangePlan,
handleToggleAutoRenew,
handleCancel,
}
}
+21 -221
View File
@@ -2,129 +2,33 @@
支持将指定颜色(默认绿色)变为透明,可用于虚拟背景、画中画背景替换等场景。
使用方式:
config = ChromaKeyConfig(key_color="#00FF00", similarity=0.3, blend=0.1)
engine = ChromaKeyEngine(config)
filter_str = engine.build_filter(input_label, output_label)
# 结果: [in]colorkey=color=0x00FF00:similarity=0.3:blend=0.1[out]
降级策略:
- 参数越界自动钳制
- 素材格式不支持时跳过(调用方捕获异常)
注:核心领域模型已抽离到 packages/domain/chroma_key_config.py,
本模块保留薄包装层,确保向后兼容。
"""
from __future__ import annotations
import logging
import re
from dataclasses import dataclass
from typing import Optional
from packages.domain.chroma_key_config import (
CHROMA_KEY_PRESETS,
ChromaKeyConfig,
apply_chroma_key_if_needed,
)
from packages.domain.chroma_key_config import ( # noqa: F401 — 向后兼容
build_chromakey_filter as _build_chromakey_filter_base,
)
from packages.domain.chroma_key_config import build_colorkey_filter as _build_colorkey_filter_base
from packages.domain.chroma_key_config import normalize_color as _normalize_color_base
logger = logging.getLogger(__name__)
# ── 配置模型 ──────────────────────────────────────────────────────────────────
@dataclass
class ChromaKeyConfig:
"""绿幕抠像配置。
Attributes:
enabled: 是否启用抠像
key_color: 要抠除的颜色,支持 hex 格式(如 "#00FF00")或颜色名
similarity: 颜色相似度阈值 0.01~1.0,值越大抠除范围越大
blend: 边缘平滑/混合度 0.0~1.0,值越大边缘越柔和
spill_suppress: 溢色抑制 0.0~1.0,减少边缘的绿幕反光
"""
enabled: bool = False
key_color: str = "#00FF00"
similarity: float = 0.3
blend: float = 0.1
spill_suppress: float = 0.0
@classmethod
def from_dict(cls, data: dict | None) -> "ChromaKeyConfig":
"""从字典解析配置,参数越界自动钳制。"""
if not data or not data.get("enabled", False):
return cls(enabled=False)
key_color = str(data.get("key_color", "#00FF00")).strip()
def _safe_float(val, default):
try:
return float(val)
except (TypeError, ValueError):
return default
similarity = _safe_float(data.get("similarity", 0.3), 0.3)
blend = _safe_float(data.get("blend", 0.1), 0.1)
spill_suppress = _safe_float(data.get("spill_suppress", 0.0), 0.0)
# 钳制到合法范围
similarity = max(0.01, min(1.0, similarity))
blend = max(0.0, min(1.0, blend))
spill_suppress = max(0.0, min(1.0, spill_suppress))
return cls(
enabled=True,
key_color=key_color,
similarity=similarity,
blend=blend,
spill_suppress=spill_suppress,
)
def has_effect(self) -> bool:
"""判断是否有实际抠像效果。"""
return self.enabled and self.similarity > 0
# ── 预设配置 ──────────────────────────────────────────────────────────────────
# 常见绿幕/蓝幕预设
CHROMA_KEY_PRESETS = {
"green_screen": {
"key_color": "#00FF00",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.5,
},
"blue_screen": {
"key_color": "#0000FF",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.5,
},
"red_screen": {
"key_color": "#FF0000",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.0,
},
"precise_green": {
"key_color": "#00FF00",
"similarity": 0.2,
"blend": 0.05,
"spill_suppress": 0.3,
},
"soft_green": {
"key_color": "#00FF00",
"similarity": 0.45,
"blend": 0.2,
"spill_suppress": 0.5,
},
}
# ── 引擎实现 ──────────────────────────────────────────────────────────────────
class ChromaKeyEngine:
"""绿幕抠像引擎。
"""绿幕抠像引擎.
基于 FFmpeg colorkey 滤镜实现,将指定颜色变为透明。
适用于绿幕/蓝幕视频的背景去除,配合画中画或 overlay 实现虚拟背景。
薄包装层,实际逻辑委托给 packages.domain.chroma_key_config。
"""
def __init__(self, config: ChromaKeyConfig):
@@ -132,117 +36,13 @@ class ChromaKeyEngine:
@staticmethod
def _normalize_color(color_str: str) -> str:
"""将颜色字符串转为 FFmpeg colorkey 接受的格式。
支持:
- "#RRGGBB" / "#RRGGBBAA" → 0xRRGGBB
- "0xRRGGBB" → 直接使用
- 颜色名(green/blue/red/black/white 等)→ 直接透传
"""
color = color_str.strip()
# hex 格式
hex_match = re.match(r"^#?([0-9a-fA-F]{6})([0-9a-fA-F]{2})?$", color)
if hex_match:
return f"0x{hex_match.group(1).upper()}"
# 已经是 0x 格式
if color.lower().startswith("0x"):
return color.upper()
# 颜色名直接透传(FFmpeg 支持常见颜色名)
return color
"""将颜色字符串转为 FFmpeg colorkey 接受的格式."""
return _normalize_color_base(color_str)
def build_filter(self, input_label: str, output_label: str) -> str:
"""构建 colorkey 滤镜字符串。
Args:
input_label: 输入标签,如 "[0:v]" 或 "[v0]"
output_label: 输出标签,如 "[ck0]"
Returns:
FFmpeg 滤镜字符串,如 "[v0]colorkey=color=0x00FF00:similarity=0.3:blend=0.1[ck0]"
Raises:
ValueError: 配置无效时抛出(调用方应捕获并降级)
"""
if not self.config.has_effect():
# 无效果,直接直通
return f"{input_label}copy{output_label}"
color = self._normalize_color(self.config.key_color)
similarity = self.config.similarity
blend = self.config.blend
# 基础 colorkey 滤镜
parts = [f"colorkey=color={color}:similarity={similarity}:blend={blend}"]
# 溢色抑制(通过 colorchannelmixer 降低绿色通道增益)
if self.config.spill_suppress > 0:
# 降低绿通道增益,减少绿幕反光溢出
spill = self.config.spill_suppress
# 绿通道增益 = 1 - spill_factor
g_gain = max(0.3, 1.0 - spill * 0.7)
# 同时稍微提升红和蓝来补偿色偏
r_gain = 1.0 + spill * 0.15
b_gain = 1.0 + spill * 0.15
parts.append(f"colorchannelmixer=" f"rr={r_gain}:" f"gg={g_gain}:" f"bb={b_gain}:" f"aa=1")
filter_str = f"{input_label}{','.join(parts)}{output_label}"
return filter_str
"""构建 colorkey 滤镜字符串."""
return _build_colorkey_filter_base(self.config, input_label, output_label)
def build_filter_chromakey(self, input_label: str, output_label: str) -> str:
"""使用 chromakey 滤镜(更高级的版本,支持更多参数)。
注意:并非所有 FFmpeg 版本都支持 chromakey 滤镜,
优先使用 colorkey(兼容性更好)。
Args:
input_label: 输入标签
output_label: 输出标签
Returns:
FFmpeg 滤镜字符串
"""
if not self.config.has_effect():
return f"{input_label}copy{output_label}"
color = self._normalize_color(self.config.key_color)
similarity = self.config.similarity
blend = self.config.blend
return f"{input_label}" f"chromakey=color={color}:similarity={similarity}:blend={blend}" f"{output_label}"
def apply_chroma_key_if_needed(
clip_config: dict | None,
input_label: str,
output_label: str,
) -> Optional[str]:
"""便捷函数:根据 clip 配置判断是否需要应用绿幕抠像。
Args:
clip_config: clip 的 config 字典
input_label: 输入标签
output_label: 输出标签
Returns:
滤镜字符串,不需要抠像时返回 None
"""
if not clip_config:
return None
chroma_key_data = clip_config.get("chroma_key")
if not chroma_key_data:
return None
try:
config = ChromaKeyConfig.from_dict(chroma_key_data)
if not config.has_effect():
return None
engine = ChromaKeyEngine(config)
return engine.build_filter(input_label, output_label)
except Exception as e:
logger.warning("[chroma-key] 应用抠像失败,跳过: %s", e)
return None
"""使用 chromakey 滤镜(更高级的版本,支持更多参数)."""
return _build_chromakey_filter_base(self.config, input_label, output_label)
+38 -144
View File
@@ -20,6 +20,17 @@ from shared.ffmpeg_utils import ( # noqa: F401
run_ffmpeg,
)
# xfade 转场纯逻辑已抽离到 domain 层,这里 re-export 保持向后兼容
from packages.domain.xfade_builder import DEFAULT_TRANSITION_DURATION as _default_transition_duration_base # noqa: F401
from packages.domain.xfade_builder import (
SUPPORTED_TRANSITIONS,
XFADE_TRANSITION_MAP,
XFade_TRANSITION_NAMES,
)
from packages.domain.xfade_builder import build_xfade_filter_chain as _build_xfade_filter_chain_base
from packages.domain.xfade_builder import chain_filters as _chain_filters_base
from packages.domain.xfade_builder import resolve_xfade_transition as _resolve_xfade_transition_base
logger = logging.getLogger(__name__)
# ── 常量(Worker 层业务相关) ────────────────────────────────────────────────
@@ -28,43 +39,34 @@ DEFAULT_OUTPUT_WIDTH = 1280
DEFAULT_OUTPUT_HEIGHT = 720
DEFAULT_FPS = 25
# xfade 转场映射:transition_effect 名称 → FFmpeg xfade transition 名称
# 键同时支持 TransitionEffect 枚举值和字符串名称(向后兼容)
# "cut" 为特殊值:硬切,不使用 xfade(由调用方特殊处理)
XFADE_TRANSITION_MAP: dict[str, str] = {
# 基础
"fade": "fade",
"dissolve": "dissolve",
"crossfade": "dissolve",
"crossdissolve": "dissolve",
# 滑入系列
"slideleft": "slideleft",
"slide_left": "slideleft",
"slideright": "slideright",
"slide_right": "slideright",
"slideup": "slideup",
"slide_up": "slideup",
"slidedown": "slidedown",
"slide_down": "slidedown",
"slide": "slideleft", # 默认向左滑
# 缩放
"zoom": "zoomin",
"zoomin": "zoomin",
"zoomout": "zoomout",
# 擦除系列
"wipe": "wipeleft", # 默认向左擦
"wipeleft": "wipeleft",
"wiperight": "wiperight",
"wipeup": "wipeup",
"wipedown": "wipedown",
# 特殊效果
"circlecrop": "circlecrop",
"circle": "circlecrop",
"rectcrop": "rectcrop",
"rect": "rectcrop",
}
# 向后兼容:DEFAULT_TRANSITION_DURATION 从 domain 层导出
DEFAULT_TRANSITION_DURATION = _default_transition_duration_base
DEFAULT_TRANSITION_DURATION = 0.5
# 向后兼容:薄包装函数
def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
return _chain_filters_base(filters, output_label, input_label=input_label)
def resolve_xfade_transition(transition_name: Any) -> str:
return _resolve_xfade_transition_base(transition_name)
def build_xfade_filter_chain(
clip_durations: list[float],
clip_video_labels: list[str],
transitions: list[str],
*,
transition_duration: float = DEFAULT_TRANSITION_DURATION,
output_label: str = "outv",
) -> tuple[str, float]:
return _build_xfade_filter_chain_base(
clip_durations,
clip_video_labels,
transitions,
transition_duration=transition_duration,
output_label=output_label,
)
# ── FFprobe 探测 ──────────────────────────────────────────────────────────────
@@ -304,111 +306,3 @@ def normalize_video(
]
run_ffmpeg(command)
return {"width": width, "height": height, "path": output_path}
# ── xfade / concat 滤镜构建 ──────────────────────────────────────────────────
def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
"""将滤镜列表串联为 FFmpeg 滤镜字符串。
例:chain_filters(["scale=1280:720", "fps=25"], "v0")
→ "[0:v]scale=1280:720,fps=25[v0]"
"""
filter_body = ",".join(filters)
return f"[{input_label}]{filter_body}[{output_label}]"
def resolve_xfade_transition(transition_name: str) -> str:
"""将转场效果名称映射为 FFmpeg xfade transition 名称。
支持 TransitionEffect 枚举值和字符串名称,未知值回退到 "fade"。
"""
# 兼容 TransitionEffect 枚举(有 .value 属性)
if hasattr(transition_name, "value"):
transition_name = transition_name.value
return XFADE_TRANSITION_MAP.get(transition_name, "fade")
def build_xfade_filter_chain(
clip_durations: list[float],
clip_video_labels: list[str],
transitions: list[str],
*,
transition_duration: float = DEFAULT_TRANSITION_DURATION,
output_label: str = "outv",
) -> tuple[str, float]:
"""构建 xfade 转场滤镜链。
对每步 xfade 自动钳制 transition duration,确保
``offset + td ≤ first_input_duration``,避免 FFmpeg exit 234。
Args:
clip_durations: 每个片段的时长(必须与 trim 后的实际时长一致)
clip_video_labels: 每个片段的视频流标签(如 "v0", "v1")
transitions: 每个片段对应的转场效果(第一个片段的转场被忽略)
transition_duration: 转场时长(秒)
output_label: 最终输出标签
Returns:
(filter_string, estimated_total_duration)
"""
n = len(clip_durations)
parts: list[str] = []
if n == 0:
return "", 0.0
if n == 1:
parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]")
return ";".join(parts), clip_durations[0]
# xfade 链 — 每步动态钳制 td,防止 offset + td > first_input_duration
cumulative = 0.0
prev_label = clip_video_labels[0]
total_transition = 0.0 # 累计已使用的转场时长
for i in range(1, n):
cumulative += clip_durations[i - 1]
# 当前 xfade 的第一个输入时长
if i == 1:
first_input_dur = clip_durations[0]
else:
first_input_dur = cumulative - total_transition
# 原始 offset 计算
offset = max(0.0, cumulative - transition_duration * i)
# 安全钳制:offset + td 不能超过第一个输入的时长
available = max(0.0, first_input_dur - offset)
safe_td = min(transition_duration, available)
# 同时不能超过剩余总时长
remaining = max(0.0, sum(clip_durations) - cumulative)
safe_td = min(safe_td, remaining)
# 同时不能超过当前第二个输入(单个片段)的时长
safe_td = min(safe_td, clip_durations[i])
safe_td = max(0.001, safe_td) # 至少 1ms,避免 td=0
transition = transitions[i] if i < len(transitions) else "cut"
xfade_transition = resolve_xfade_transition(transition)
if i == n - 1:
out_label = output_label
else:
out_label = f"xf{i}"
parts.append(
f"[{prev_label}][{clip_video_labels[i]}]"
f"xfade=transition={xfade_transition}"
f":duration={safe_td:.3f}"
f":offset={offset:.3f}"
f"[{out_label}]"
)
prev_label = out_label
total_transition += safe_td
# 总时长减去转场重叠部分
total_duration = sum(clip_durations) - total_transition
return ";".join(parts), max(0.0, total_duration)
@@ -2,126 +2,28 @@
支持对音频进行背景噪音消除、人声增强,适用于语音录制、采访等场景。
使用方式:
config = NoiseReductionConfig(level="medium")
engine = NoiseReductionEngine(config)
filter_str = engine.build_filter(input_label, output_label)
# 结果: [0:a]afftdn=nf=-25[out]
降级策略:
- 参数越界自动钳制
- FFmpeg 不支持 afftdn 时,调用方可捕获异常并跳过
领域模型已抽离至 packages/domain/noise_reduction_config.py,本模块保留薄包装以维持向后兼容。
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from packages.domain.noise_reduction_config import (
NoiseReductionConfig,
NoiseReductionLevel,
)
from packages.domain.noise_reduction_config import (
apply_noise_reduction_if_needed as _apply_noise_reduction_if_needed_base,
)
from packages.domain.noise_reduction_config import build_afftdn_filter as _build_afftdn_filter_base # noqa: F401 — 向后兼容
from packages.domain.noise_reduction_config import build_arnndn_filter as _build_arnndn_filter_base
logger = logging.getLogger(__name__)
# ── 降噪等级 ──────────────────────────────────────────────────────────────────
class NoiseReductionLevel(str, Enum):
"""降噪等级预设。"""
LOW = "low" # 轻度降噪,保留细节,适合轻微背景噪音
MEDIUM = "medium" # 中度降噪,平衡效果和音质
HIGH = "high" # 高度降噪,适合嘈杂环境,可能轻微影响音质
CUSTOM = "custom" # 自定义参数
# 各等级对应的降噪参数(afftdn 的 noise floor,单位 dB)
# 值越大(越接近 0),降噪越强;值越小(越负),降噪越弱
_LEVEL_PARAMS = {
NoiseReductionLevel.LOW: {
"nf": -35, # 噪音阈值(dB),越负越保守
"tn": -10, # 噪音频谱平滑度
"tr": 50, # 时间分辨率(ms)
},
NoiseReductionLevel.MEDIUM: {
"nf": -25,
"tn": -10,
"tr": 50,
},
NoiseReductionLevel.HIGH: {
"nf": -15,
"tn": -5,
"tr": 30,
},
}
# ── 配置模型 ──────────────────────────────────────────────────────────────────
@dataclass
class NoiseReductionConfig:
"""音频降噪配置。
Attributes:
enabled: 是否启用降噪
level: 降噪等级 low/medium/high/custom
noise_floor: 自定义噪音阈值(dB),仅 level=custom 时有效,范围 -60 ~ -5
voice_enhance: 是否启用人声增强
output_format: 输出格式描述(内部使用)
"""
enabled: bool = False
level: NoiseReductionLevel = NoiseReductionLevel.MEDIUM
noise_floor: float = -25.0 # dB
voice_enhance: bool = False
@classmethod
def from_dict(cls, data: dict | None) -> "NoiseReductionConfig":
"""从字典解析配置,参数越界自动钳制。"""
if not data or not data.get("enabled", False):
return cls(enabled=False)
level_str = str(data.get("level", "medium")).lower()
try:
level = NoiseReductionLevel(level_str)
except ValueError:
level = NoiseReductionLevel.MEDIUM
try:
noise_floor = float(data.get("noise_floor", -25.0))
except (TypeError, ValueError):
noise_floor = -25.0
voice_enhance = bool(data.get("voice_enhance", False))
# 钳制到合法范围
noise_floor = max(-60.0, min(-5.0, noise_floor))
return cls(
enabled=True,
level=level,
noise_floor=noise_floor,
voice_enhance=voice_enhance,
)
def has_effect(self) -> bool:
"""判断是否有实际降噪效果。"""
return self.enabled
def get_effective_noise_floor(self) -> float:
"""获取实际生效的噪音阈值(dB)。"""
if self.level == NoiseReductionLevel.CUSTOM:
return self.noise_floor
params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[NoiseReductionLevel.MEDIUM])
return float(params["nf"])
# ── 引擎实现 ──────────────────────────────────────────────────────────────────
class NoiseReductionEngine:
"""音频降噪引擎。
"""音频降噪引擎 — 薄包装,实际逻辑在 domain.noise_reduction_config.
基于 FFmpeg afftdn(Audio FFt Denoiser)滤镜实现:
- 使用短时傅里叶变换分析音频频谱
@@ -133,97 +35,40 @@ class NoiseReductionEngine:
self.config = config
def build_filter(self, input_label: str, output_label: str) -> str:
"""构建音频降噪滤镜字符串。
"""构建音频降噪滤镜字符串.
Args:
input_label: 输入标签,如 "[0:a]" 或 "[a0]"
output_label: 输出标签,如 "[nr0]"
Returns:
FFmpeg 滤镜字符串,如 "[a0]afftdn=nf=-25:tn=-10:tr=50[nr0]"
Raises:
ValueError: 配置无效时抛出(调用方应捕获并降级)
FFmpeg 滤镜字符串
"""
if not self.config.has_effect():
return f"{input_label}anull{output_label}"
# 获取参数
if self.config.level == NoiseReductionLevel.CUSTOM:
nf = self.config.noise_floor
tn = -10 # 默认频谱平滑度
tr = 50 # 默认时间分辨率
else:
params = _LEVEL_PARAMS.get(
self.config.level,
_LEVEL_PARAMS[NoiseReductionLevel.MEDIUM],
)
nf = float(params["nf"])
tn = float(params["tn"])
tr = float(params["tr"])
# 构建 afftdn 滤镜
# nf: noise floor (dB)
# tn: temporal noise floor smoothing (dB)
# tr: time resolution (ms)
filter_parts = [f"afftdn=nf={nf}:tn={tn}:tr={tr}"]
# 人声增强:通过 highpass + 轻微压缩实现
if self.config.voice_enhance:
# 1. 高通滤波,去除低频噪音
filter_parts.append("highpass=f=80")
# 2. 轻微压缩,提升人声清晰度
filter_parts.append("acompressor=threshold=-20:ratio=2:attack=5:release=50")
# 3. 响度归一化
filter_parts.append("loudnorm=I=-16:TP=-1.5:LRA=11")
filter_str = f"{input_label}{','.join(filter_parts)}{output_label}"
return filter_str
return _build_afftdn_filter_base(self.config, input_label, output_label)
def build_filter_arnndn(self, input_label: str, output_label: str, model_file: str) -> str:
"""使用 RNN 降噪滤镜(arnndn,效果更好但需要模型文件)。
注意:需要额外下载 RNNNoise 模型文件,默认使用 afftdn(无需额外依赖)。
"""使用 RNN 降噪滤镜(arnndn,效果更好但需要模型文件).
Args:
input_label: 输入标签
output_label: 输出标签
model_file: RNNNoise 模型文件路径(.rnnn 格式)
model_file: RNNNoise 模型文件路径
Returns:
FFmpeg 滤镜字符串
"""
if not self.config.has_effect():
return f"{input_label}anull{output_label}"
return f"{input_label}arnndn=m={model_file}{output_label}"
return _build_arnndn_filter_base(self.config, input_label, output_label, model_file)
def apply_noise_reduction_if_needed(
config_data: dict | None,
input_label: str,
output_label: str,
) -> Optional[str]:
"""便捷函数:根据配置判断是否需要应用音频降噪。
def apply_noise_reduction_if_needed(config_data, input_label: str, output_label: str):
"""便捷函数:根据配置判断是否需要应用音频降噪.
Args:
config_data: 降噪配置字典(从 plan.config.audio_noise_reduction 或 clip.config.noise_reduction 读取)
config_data: 降噪配置字典
input_label: 输入标签
output_label: 输出标签
Returns:
滤镜字符串,不需要降噪时返回 None
"""
if not config_data:
return None
try:
config = NoiseReductionConfig.from_dict(config_data)
if not config.has_effect():
return None
engine = NoiseReductionEngine(config)
return engine.build_filter(input_label, output_label)
except Exception as e:
logger.warning("[noise-reduction] 应用降噪失败,跳过: %s", e)
return None
return _apply_noise_reduction_if_needed_base(config_data, input_label, output_label)
+37 -203
View File
@@ -1,8 +1,8 @@
"""ASS 字幕生成模块 — 从 unified_render_service.py 拆分.
"""ASS 字幕生成模块 — 薄包装,实际逻辑在 packages/domain/ass_subtitle_builder.py.
职责:
- 将 title / subtitle 配置转换为 ASS 字幕文件
- 提供样式计算(颜色、对齐、描边/阴影)
- 文件IO 在此模块,纯逻辑已抽离到 domain
- 供 UnifiedRenderService._maybe_generate_ass 调用
"""
@@ -12,107 +12,40 @@ import logging
from pathlib import Path
from typing import Any
from packages.domain.ass_subtitle_builder import (
TITLE_MARGIN_BOTTOM,
TITLE_MARGIN_SIDE,
TITLE_MARGIN_TOP,
build_ass_content,
)
from packages.domain.ass_subtitle_builder import build_ass_style as _build_ass_style_base # noqa: F401 — 向后兼容
from packages.domain.ass_subtitle_builder import escape_ass_text as _escape_ass_text_base
from packages.domain.ass_subtitle_builder import format_ass_time as _format_ass_time_base
from packages.domain.ass_subtitle_builder import hex_to_ass_color as _hex_to_ass_color_base
from packages.domain.ass_subtitle_builder import position_to_ass_alignment as _position_to_ass_alignment_base
logger = logging.getLogger(__name__)
# ── 常量 ──────────────────────────────────────────────────────────────────────
# Title/Subtitle 默认边距(像素)
TITLE_MARGIN_TOP = 60
TITLE_MARGIN_BOTTOM = 60
TITLE_MARGIN_SIDE = 40
# ── ASS 字幕工具 ─────────────────────────────────────────────────────────────
# 向后兼容:模块级函数保留为薄包装
def _hex_to_ass_color(hex_color: str) -> str:
"""将 HEX 颜色(#RRGGBB)转换为 ASS &HBBGGRR 格式。"""
hex_color = hex_color.lstrip("#")
if len(hex_color) != 6:
return "&H000000"
r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
return f"&H{b.upper()}{g.upper()}{r.upper()}"
return _hex_to_ass_color_base(hex_color)
def _position_to_ass_alignment(position: str) -> int:
"""将文字位置映射为 ASS \\an 对齐编号。
ASS 对齐编号(数字小键盘布局):
7 8 9
4 5 6
1 2 3
"""
mapping = {
"top": 8, # 顶部居中
"center": 5, # 居中
"bottom": 2, # 底部居中
}
return mapping.get(position, 8)
return _position_to_ass_alignment_base(position)
def _build_ass_style(
style_name: str,
*,
font_name: str = "思源黑体",
font_size: int = 48,
primary_color: str = "&H00FFFFFF",
outline_color: str = "&H00000000",
outline_width: float = 1.0,
shadow_blur: float = 0.0,
shadow_offset: tuple[int, int] = (0, 0),
bold: bool = False,
italic: bool = False,
alignment: int = 8,
margin_v: int = 60,
margin_l: int = 40,
margin_r: int = 40,
) -> str:
"""构建 ASS Style 行。
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour,
Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle,
BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
"""
bold_val = -1 if bold else 0
italic_val = -1 if italic else 0
# BackColour 用于阴影(BorderStyle=1 时 outline + shadow)
back_color = primary_color # 阴影颜色默认同文字色(带透明度由阴影模糊控制)
# Shadow 值:ASS 中 Shadow 字段是阴影偏移距离(像素),
# 我们用 shadow_offset[1] 作为纵向偏移,模糊由 BorderStyle=3 实现
# 简化:BorderStyle=1(outline + drop shadow),Shadow 字段表示阴影深度
shadow_depth = shadow_offset[1] if shadow_blur > 0 else 0
return (
f"Style: {style_name},{font_name},{font_size},{primary_color},"
f"&H000000FF,{outline_color},{back_color},"
f"{bold_val},{italic_val},0,0,100,100,0,0,"
f"1,{outline_width},{shadow_depth},{alignment},"
f"{margin_l},{margin_r},{margin_v},1"
)
def _build_ass_style(*args, **kwargs) -> str:
return _build_ass_style_base(*args, **kwargs)
def _escape_ass_text(text: str) -> str:
r"""转义 ASS 文本中的特殊字符。
ASS 中换行用 \N(硬换行)或 \n(软换行),
大括号 {} 用于覆盖样式,需要转义。
"""
# 将实际换行转为 ASS 硬换行
text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N")
# 转义大括号(ASS 用它做样式覆盖标签)
text = text.replace("{", "(").replace("}", ")")
return text
return _escape_ass_text_base(text)
def _format_ass_time(seconds: float) -> str:
"""将秒数格式化为 ASS 时间格式 H:MM:SS.cc。"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = seconds % 60
return f"{hours}:{minutes:02d}:{secs:05.2f}"
return _format_ass_time_base(seconds)
def generate_ass_subtitles(
@@ -126,130 +59,31 @@ def generate_ass_subtitles(
subtitle_text: str = "",
subtitle_config: dict[str, Any] | None = None,
) -> Path:
"""生成 ASS 字幕文件。
支持 Title(标题)和 Subtitle(字幕)两种字幕类型,
各自可独立配置样式、位置和内容。
"""生成 ASS 字幕文件.
Args:
output_path: 输出 ASS 文件路径
video_width: 视频宽度(用于 ASS PlayResX)
video_height: 视频高度(用于 ASS PlayResY)
video_duration: 视频总时长(秒),字幕显示整个时长
video_width: 视频宽度
video_height: 视频高度
video_duration: 视频总时长(秒)
title_text: 标题文本
title_config: 标题样式配置(TitleConfig dict)
title_config: 标题样式配置
subtitle_text: 字幕文本
subtitle_config: 字幕样式配置(SubtitleConfig dict)
subtitle_config: 字幕样式配置
Returns:
生成的 ASS 文件路径
"""
title_config = title_config or {}
subtitle_config = subtitle_config or {}
title_enabled = title_config.get("enabled", True) and bool(title_text.strip())
subtitle_enabled = subtitle_config.get("enabled", True) and bool(subtitle_text.strip())
if not title_enabled and not subtitle_enabled:
# 没有字幕,生成空文件(仍返回路径,调用方自行判断是否使用)
output_path.write_text("", encoding="utf-8")
return output_path
styles: list[str] = []
events: list[str] = []
# ── Title 样式与事件 ──────────────────────────────────────────────────
if title_enabled:
title_color = _hex_to_ass_color(title_config.get("color", "#ffffff"))
title_stroke = title_config.get("stroke", {}) or {}
title_shadow = title_config.get("shadow", {}) or {}
stroke_color = _hex_to_ass_color(title_stroke.get("color", "#000000"))
stroke_width = float(title_stroke.get("width", 1)) if title_stroke.get("enabled", False) else 0.0
shadow_blur = float(title_shadow.get("blur", 4)) if title_shadow.get("enabled", False) else 0.0
shadow_offset = (
title_shadow.get("offset_x", 2) if title_shadow.get("enabled", False) else 0,
title_shadow.get("offset_y", 2) if title_shadow.get("enabled", False) else 0,
)
title_alignment = _position_to_ass_alignment(title_config.get("position", "top"))
styles.append(
_build_ass_style(
"TitleStyle",
font_name=title_config.get("font", "思源黑体"),
font_size=int(title_config.get("size", 48)),
primary_color=title_color,
outline_color=stroke_color,
outline_width=stroke_width,
shadow_blur=shadow_blur,
shadow_offset=shadow_offset,
bold=bool(title_config.get("bold", True)),
italic=bool(title_config.get("italic", False)),
alignment=title_alignment,
margin_v=TITLE_MARGIN_TOP,
margin_l=TITLE_MARGIN_SIDE,
margin_r=TITLE_MARGIN_SIDE,
)
)
# 转义 ASS 特殊字符
safe_title_text = _escape_ass_text(title_text)
events.append(
"Dialogue: 0,0:00:00.00," f"{_format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}"
)
# ── Subtitle 样式与事件 ───────────────────────────────────────────────
if subtitle_enabled:
sub_color = _hex_to_ass_color(subtitle_config.get("color", "#ffffff"))
sub_alignment = _position_to_ass_alignment(subtitle_config.get("position", "bottom"))
styles.append(
_build_ass_style(
"SubtitleStyle",
font_name=subtitle_config.get("font", "思源黑体"),
font_size=int(subtitle_config.get("size", 24)),
primary_color=sub_color,
outline_color="&H00000000",
outline_width=1.0,
shadow_blur=0.0,
shadow_offset=(0, 0),
bold=False,
italic=False,
alignment=sub_alignment,
margin_v=TITLE_MARGIN_BOTTOM,
margin_l=TITLE_MARGIN_SIDE,
margin_r=TITLE_MARGIN_SIDE,
)
)
safe_subtitle_text = _escape_ass_text(subtitle_text)
events.append(
"Dialogue: 0,0:00:00.00,"
f"{_format_ass_time(video_duration)},"
"SubtitleStyle,,0,0,0,,"
f"{safe_subtitle_text}"
)
# ── 组装 ASS 文件 ─────────────────────────────────────────────────────
ass_content = f"""[Script Info]
ScriptType: v4.00+
PlayResX: {video_width}
PlayResY: {video_height}
ScaledBorderAndShadow: yes
WrapStyle: 2
Encoding: UTF-8
[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding # noqa: E501
{chr(10).join(styles)}
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
{chr(10).join(events)}
"""
content = build_ass_content(
video_width=video_width,
video_height=video_height,
video_duration=video_duration,
title_text=title_text,
title_config=title_config,
subtitle_text=subtitle_text,
subtitle_config=subtitle_config,
)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(ass_content, encoding="utf-8")
output_path.write_text(content, encoding="utf-8")
return output_path
+306
View File
@@ -0,0 +1,306 @@
"""ASS 字幕构建领域模型 — 纯逻辑,无文件IO依赖.
抽离自 render_subtitles.py,包含:
- 颜色转换(hex → ASS &HBBGGRR)
- 位置对齐映射
- ASS Style 行构建
- 文本转义
- 时间格式化
- 完整 ASS 内容生成(返回字符串,不写文件)
"""
from __future__ import annotations
import logging
from typing import Any
logger = logging.getLogger(__name__)
# ── 常量 ──────────────────────────────────────────────────────────────────────
# Title/Subtitle 默认边距(像素)
TITLE_MARGIN_TOP = 60
TITLE_MARGIN_BOTTOM = 60
TITLE_MARGIN_SIDE = 40
# ── 颜色转换 ──────────────────────────────────────────────────────────────────
def hex_to_ass_color(hex_color: str) -> str:
"""将 HEX 颜色(#RRGGBB)转换为 ASS &HBBGGRR 格式.
Args:
hex_color: HEX 颜色字符串,支持 #RRGGBB 或 RRGGBB 格式
Returns:
ASS 格式颜色,如 &H0000FF(红色)
"""
hex_color = hex_color.lstrip("#")
if len(hex_color) != 6:
return "&H000000"
r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
return f"&H{b.upper()}{g.upper()}{r.upper()}"
# ── 位置对齐 ──────────────────────────────────────────────────────────────────
def position_to_ass_alignment(position: str) -> int:
"""将文字位置映射为 ASS \\an 对齐编号.
ASS 对齐编号(数字小键盘布局):
7 8 9
4 5 6
1 2 3
Args:
position: 位置字符串 top/center/bottom
Returns:
ASS 对齐编号,默认 8(顶部居中)
"""
mapping = {
"top": 8,
"center": 5,
"bottom": 2,
}
return mapping.get(position, 8)
# ── Style 行构建 ──────────────────────────────────────────────────────────────
def build_ass_style(
style_name: str,
*,
font_name: str = "思源黑体",
font_size: int = 48,
primary_color: str = "&H00FFFFFF",
outline_color: str = "&H00000000",
outline_width: float = 1.0,
shadow_blur: float = 0.0,
shadow_offset: tuple[int, int] = (0, 0),
bold: bool = False,
italic: bool = False,
alignment: int = 8,
margin_v: int = 60,
margin_l: int = 40,
margin_r: int = 40,
) -> str:
"""构建 ASS Style 行.
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour,
Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle,
BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Args:
style_name: 样式名称
font_name: 字体名称
font_size: 字体大小
primary_color: 主色(文字颜色)
outline_color: 描边颜色
outline_width: 描边宽度
shadow_blur: 阴影模糊度(>0 时启用阴影)
shadow_offset: 阴影偏移 (x, y)
bold: 是否粗体
italic: 是否斜体
alignment: 对齐方式(ASS \an 编号)
margin_v: 垂直边距
margin_l: 左边距
margin_r: 右边距
Returns:
完整的 Style: 行字符串
"""
bold_val = -1 if bold else 0
italic_val = -1 if italic else 0
# BackColour 用于阴影(BorderStyle=1 时 outline + shadow)
back_color = primary_color
# Shadow 深度:shadow_offset[1] 作为纵向偏移
shadow_depth = shadow_offset[1] if shadow_blur > 0 else 0
return (
f"Style: {style_name},{font_name},{font_size},{primary_color},"
f"&H000000FF,{outline_color},{back_color},"
f"{bold_val},{italic_val},0,0,100,100,0,0,"
f"1,{outline_width},{shadow_depth},{alignment},"
f"{margin_l},{margin_r},{margin_v},1"
)
# ── 文本转义 ──────────────────────────────────────────────────────────────────
def escape_ass_text(text: str) -> str:
r"""转义 ASS 文本中的特殊字符.
ASS 中换行用 \N(硬换行)或 \n(软换行),
大括号 {} 用于覆盖样式,需要转义.
Args:
text: 原始文本
Returns:
转义后的 ASS 文本
"""
# 将实际换行转为 ASS 硬换行
text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N")
# 转义大括号(ASS 用它做样式覆盖标签)
text = text.replace("{", "(").replace("}", ")")
return text
# ── 时间格式化 ────────────────────────────────────────────────────────────────
def format_ass_time(seconds: float) -> str:
"""将秒数格式化为 ASS 时间格式 H:MM:SS.cc.
Args:
seconds: 秒数
Returns:
ASS 格式时间,如 "1:23:45.67"
"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = seconds % 60
return f"{hours}:{minutes:02d}:{secs:05.2f}"
# ── 完整 ASS 内容生成 ─────────────────────────────────────────────────────────
def build_ass_content(
*,
video_width: int,
video_height: int,
video_duration: float,
title_text: str = "",
title_config: dict[str, Any] | None = None,
subtitle_text: str = "",
subtitle_config: dict[str, Any] | None = None,
) -> str:
"""生成 ASS 字幕文件内容(纯字符串,不写文件).
支持 Title(标题)和 Subtitle(字幕)两种字幕类型,
各自可独立配置样式、位置和内容.
Args:
video_width: 视频宽度(用于 ASS PlayResX)
video_height: 视频高度(用于 ASS PlayResY)
video_duration: 视频总时长(秒),字幕显示整个时长
title_text: 标题文本
title_config: 标题样式配置
subtitle_text: 字幕文本
subtitle_config: 字幕样式配置
Returns:
完整的 ASS 文件内容字符串;无字幕时返回空字符串
"""
title_config = title_config or {}
subtitle_config = subtitle_config or {}
title_enabled = title_config.get("enabled", True) and bool(title_text.strip())
subtitle_enabled = subtitle_config.get("enabled", True) and bool(subtitle_text.strip())
if not title_enabled and not subtitle_enabled:
return ""
styles: list[str] = []
events: list[str] = []
# ── Title 样式与事件 ──────────────────────────────────────────────────
if title_enabled:
title_color = hex_to_ass_color(title_config.get("color", "#ffffff"))
title_stroke = title_config.get("stroke", {}) or {}
title_shadow = title_config.get("shadow", {}) or {}
stroke_color = hex_to_ass_color(title_stroke.get("color", "#000000"))
stroke_width = float(title_stroke.get("width", 1)) if title_stroke.get("enabled", False) else 0.0
shadow_blur = float(title_shadow.get("blur", 4)) if title_shadow.get("enabled", False) else 0.0
shadow_offset = (
title_shadow.get("offset_x", 2) if title_shadow.get("enabled", False) else 0,
title_shadow.get("offset_y", 2) if title_shadow.get("enabled", False) else 0,
)
title_alignment = position_to_ass_alignment(title_config.get("position", "top"))
styles.append(
build_ass_style(
"TitleStyle",
font_name=title_config.get("font", "思源黑体"),
font_size=int(title_config.get("size", 48)),
primary_color=title_color,
outline_color=stroke_color,
outline_width=stroke_width,
shadow_blur=shadow_blur,
shadow_offset=shadow_offset,
bold=bool(title_config.get("bold", True)),
italic=bool(title_config.get("italic", False)),
alignment=title_alignment,
margin_v=TITLE_MARGIN_TOP,
margin_l=TITLE_MARGIN_SIDE,
margin_r=TITLE_MARGIN_SIDE,
)
)
safe_title_text = escape_ass_text(title_text)
events.append(
"Dialogue: 0,0:00:00.00," f"{format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}"
)
# ── Subtitle 样式与事件 ───────────────────────────────────────────────
if subtitle_enabled:
sub_color = hex_to_ass_color(subtitle_config.get("color", "#ffffff"))
sub_alignment = position_to_ass_alignment(subtitle_config.get("position", "bottom"))
styles.append(
build_ass_style(
"SubtitleStyle",
font_name=subtitle_config.get("font", "思源黑体"),
font_size=int(subtitle_config.get("size", 24)),
primary_color=sub_color,
outline_color="&H00000000",
outline_width=1.0,
shadow_blur=0.0,
shadow_offset=(0, 0),
bold=False,
italic=False,
alignment=sub_alignment,
margin_v=TITLE_MARGIN_BOTTOM,
margin_l=TITLE_MARGIN_SIDE,
margin_r=TITLE_MARGIN_SIDE,
)
)
safe_subtitle_text = escape_ass_text(subtitle_text)
events.append(
"Dialogue: 0,0:00:00.00,"
f"{format_ass_time(video_duration)},"
"SubtitleStyle,,0,0,0,,"
f"{safe_subtitle_text}"
)
# ── 组装 ASS 文件 ─────────────────────────────────────────────────────
return f"""[Script Info]
ScriptType: v4.00+
PlayResX: {video_width}
PlayResY: {video_height}
ScaledBorderAndShadow: yes
WrapStyle: 2
Encoding: UTF-8
[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding # noqa: E501
{chr(10).join(styles)}
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
{chr(10).join(events)}
"""
+287
View File
@@ -0,0 +1,287 @@
"""绿幕抠像配置领域模型 — 纯逻辑,无FFmpeg依赖.
抽离自 chroma_key_engine.py,包含:
- ChromaKeyConfig 数据类(解析/钳制/效果判断)
- 预设配置(绿幕/蓝幕/红幕等)
- 颜色归一化
- colorkey / chromakey 滤镜构建
- 便捷函数(apply_chroma_key_if_needed)
"""
from __future__ import annotations
import logging
import re
from dataclasses import dataclass
from typing import Any
logger = logging.getLogger(__name__)
# ── 预设配置 ──────────────────────────────────────────────────────────────────
# 常见绿幕/蓝幕预设
CHROMA_KEY_PRESETS: dict[str, dict[str, Any]] = {
"green_screen": {
"key_color": "#00FF00",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.5,
},
"blue_screen": {
"key_color": "#0000FF",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.5,
},
"red_screen": {
"key_color": "#FF0000",
"similarity": 0.3,
"blend": 0.1,
"spill_suppress": 0.0,
},
"precise_green": {
"key_color": "#00FF00",
"similarity": 0.2,
"blend": 0.05,
"spill_suppress": 0.3,
},
"soft_green": {
"key_color": "#00FF00",
"similarity": 0.45,
"blend": 0.2,
"spill_suppress": 0.5,
},
}
VALID_PRESETS = set(CHROMA_KEY_PRESETS.keys())
# 参数范围
MIN_SIMILARITY = 0.01
MAX_SIMILARITY = 1.0
MIN_BLEND = 0.0
MAX_BLEND = 1.0
MIN_SPILL_SUPPRESS = 0.0
MAX_SPILL_SUPPRESS = 1.0
# 默认值
DEFAULT_KEY_COLOR = "#00FF00"
DEFAULT_SIMILARITY = 0.3
DEFAULT_BLEND = 0.1
DEFAULT_SPILL_SUPPRESS = 0.0
# ── 配置模型 ──────────────────────────────────────────────────────────────────
@dataclass
class ChromaKeyConfig:
"""绿幕抠像配置.
Attributes:
enabled: 是否启用抠像
key_color: 要抠除的颜色,支持 hex 格式(如 "#00FF00")或颜色名
similarity: 颜色相似度阈值 0.01~1.0,值越大抠除范围越大
blend: 边缘平滑/混合度 0.0~1.0,值越大边缘越柔和
spill_suppress: 溢色抑制 0.0~1.0,减少边缘的绿幕反光
"""
enabled: bool = False
key_color: str = DEFAULT_KEY_COLOR
similarity: float = DEFAULT_SIMILARITY
blend: float = DEFAULT_BLEND
spill_suppress: float = DEFAULT_SPILL_SUPPRESS
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> ChromaKeyConfig:
"""从字典解析配置,参数越界自动钳制."""
if not data or not data.get("enabled", False):
return cls(enabled=False)
key_color = str(data.get("key_color", DEFAULT_KEY_COLOR)).strip()
def _safe_float(val: Any, default: float) -> float:
try:
return float(val)
except (TypeError, ValueError):
return default
similarity = _safe_float(data.get("similarity", DEFAULT_SIMILARITY), DEFAULT_SIMILARITY)
blend = _safe_float(data.get("blend", DEFAULT_BLEND), DEFAULT_BLEND)
spill_suppress = _safe_float(data.get("spill_suppress", DEFAULT_SPILL_SUPPRESS), DEFAULT_SPILL_SUPPRESS)
# 钳制到合法范围
similarity = max(MIN_SIMILARITY, min(MAX_SIMILARITY, similarity))
blend = max(MIN_BLEND, min(MAX_BLEND, blend))
spill_suppress = max(MIN_SPILL_SUPPRESS, min(MAX_SPILL_SUPPRESS, spill_suppress))
return cls(
enabled=True,
key_color=key_color,
similarity=similarity,
blend=blend,
spill_suppress=spill_suppress,
)
@classmethod
def from_preset(cls, preset_name: str) -> ChromaKeyConfig | None:
"""从预设名称创建配置."""
preset = CHROMA_KEY_PRESETS.get(preset_name)
if not preset:
return None
return cls(
enabled=True,
key_color=preset["key_color"],
similarity=preset["similarity"],
blend=preset["blend"],
spill_suppress=preset["spill_suppress"],
)
def has_effect(self) -> bool:
"""判断是否有实际抠像效果."""
return self.enabled and self.similarity > 0
def validate(self) -> tuple[bool, str]:
"""校验配置是否有效."""
if not self.enabled:
return True, ""
if not self.key_color:
return False, "key_color 不能为空"
if not (MIN_SIMILARITY <= self.similarity <= MAX_SIMILARITY):
return False, f"similarity 必须在 {MIN_SIMILARITY}~{MAX_SIMILARITY} 之间"
if not (MIN_BLEND <= self.blend <= MAX_BLEND):
return False, f"blend 必须在 {MIN_BLEND}~{MAX_BLEND} 之间"
if not (MIN_SPILL_SUPPRESS <= self.spill_suppress <= MAX_SPILL_SUPPRESS):
return False, f"spill_suppress 必须在 {MIN_SPILL_SUPPRESS}~{MAX_SPILL_SUPPRESS} 之间"
return True, ""
# ── 颜色归一化 ────────────────────────────────────────────────────────────────
def normalize_color(color_str: str) -> str:
"""将颜色字符串转为 FFmpeg colorkey 接受的格式.
支持:
- "#RRGGBB" / "#RRGGBBAA" → 0xRRGGBB
- "0xRRGGBB" → 直接使用
- 颜色名(green/blue/red/black/white 等)→ 直接透传
"""
color = color_str.strip()
# hex 格式
hex_match = re.match(r"^#?([0-9a-fA-F]{6})([0-9a-fA-F]{2})?$", color)
if hex_match:
return f"0x{hex_match.group(1).upper()}"
# 已经是 0x 格式
if color.lower().startswith("0x"):
return color.upper()
# 颜色名直接透传(FFmpeg 支持常见颜色名)
return color
# ── 滤镜构建 ────────────────────────────────────────────────────────────────
def build_colorkey_filter(
config: ChromaKeyConfig,
input_label: str,
output_label: str,
) -> str:
"""构建 colorkey 滤镜字符串.
Args:
config: 抠像配置
input_label: 输入标签,如 "[0:v]" 或 "[v0]"
output_label: 输出标签,如 "[ck0]"
Returns:
FFmpeg 滤镜字符串
"""
if not config.has_effect():
return f"{input_label}copy{output_label}"
color = normalize_color(config.key_color)
similarity = config.similarity
blend = config.blend
# 基础 colorkey 滤镜
parts = [f"colorkey=color={color}:similarity={similarity}:blend={blend}"]
# 溢色抑制(通过 colorchannelmixer 降低绿色通道增益)
if config.spill_suppress > 0:
spill = config.spill_suppress
g_gain = max(0.3, 1.0 - spill * 0.7)
r_gain = 1.0 + spill * 0.15
b_gain = 1.0 + spill * 0.15
parts.append(f"colorchannelmixer=rr={r_gain}:gg={g_gain}:bb={b_gain}:aa=1")
return f"{input_label}{','.join(parts)}{output_label}"
def build_chromakey_filter(
config: ChromaKeyConfig,
input_label: str,
output_label: str,
) -> str:
"""使用 chromakey 滤镜(更高级的版本,支持更多参数).
注意:并非所有 FFmpeg 版本都支持 chromakey 滤镜,
优先使用 colorkey(兼容性更好)。
"""
if not config.has_effect():
return f"{input_label}copy{output_label}"
color = normalize_color(config.key_color)
similarity = config.similarity
blend = config.blend
return f"{input_label}chromakey=color={color}:similarity={similarity}:blend={blend}{output_label}"
# ── 工具函数 ────────────────────────────────────────────────────────────────
def apply_chroma_key_if_needed(
clip_config: dict[str, Any] | None,
input_label: str,
output_label: str,
) -> str | None:
"""便捷函数:根据 clip 配置判断是否需要应用绿幕抠像.
Args:
clip_config: clip 的 config 字典
input_label: 输入标签
output_label: 输出标签
Returns:
滤镜字符串,不需要抠像时返回 None
"""
if not clip_config:
return None
chroma_key_data = clip_config.get("chroma_key")
if not chroma_key_data:
return None
try:
config = ChromaKeyConfig.from_dict(chroma_key_data)
if not config.has_effect():
return None
return build_colorkey_filter(config, input_label, output_label)
except Exception as e:
logger.warning("[chroma-key] 应用抠像失败,跳过: %s", e)
return None
def get_preset_names() -> list[str]:
"""获取所有预设名称列表."""
return sorted(list(CHROMA_KEY_PRESETS.keys()))
+231
View File
@@ -0,0 +1,231 @@
"""音频降噪配置领域模型 — 纯逻辑,无FFmpeg依赖.
抽离自 noise_reduction_engine.py,包含:
- NoiseReductionLevel 枚举(low/medium/high/custom)
- NoiseReductionConfig 数据类(解析/钳制/效果判断)
- 等级预设参数
- afftdn / arnndn 滤镜构建
- 便捷函数(apply_noise_reduction_if_needed)
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from enum import Enum
from typing import Any
logger = logging.getLogger(__name__)
# ── 降噪等级 ──────────────────────────────────────────────────────────────────
class NoiseReductionLevel(str, Enum):
"""降噪等级预设."""
LOW = "low" # 轻度降噪,保留细节,适合轻微背景噪音
MEDIUM = "medium" # 中度降噪,平衡效果和音质
HIGH = "high" # 高度降噪,适合嘈杂环境,可能轻微影响音质
CUSTOM = "custom" # 自定义参数
# 各等级对应的降噪参数(afftdn 的 noise floor,单位 dB)
# 值越大(越接近 0),降噪越强;值越小(越负),降噪越弱
_LEVEL_PARAMS: dict[NoiseReductionLevel, dict[str, float]] = {
NoiseReductionLevel.LOW: {
"nf": -35, # 噪音阈值(dB),越负越保守
"tn": -10, # 噪音频谱平滑度
"tr": 50, # 时间分辨率(ms)
},
NoiseReductionLevel.MEDIUM: {
"nf": -25,
"tn": -10,
"tr": 50,
},
NoiseReductionLevel.HIGH: {
"nf": -15,
"tn": -5,
"tr": 30,
},
}
# 参数范围
MIN_NOISE_FLOOR = -60.0
MAX_NOISE_FLOOR = -5.0
# 默认值
DEFAULT_LEVEL = NoiseReductionLevel.MEDIUM
DEFAULT_NOISE_FLOOR = -25.0
# ── 配置模型 ──────────────────────────────────────────────────────────────────
@dataclass
class NoiseReductionConfig:
"""音频降噪配置.
Attributes:
enabled: 是否启用降噪
level: 降噪等级 low/medium/high/custom
noise_floor: 自定义噪音阈值(dB),仅 level=custom 时有效,范围 -60 ~ -5
voice_enhance: 是否启用人声增强
"""
enabled: bool = False
level: NoiseReductionLevel = DEFAULT_LEVEL
noise_floor: float = DEFAULT_NOISE_FLOOR # dB
voice_enhance: bool = False
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> NoiseReductionConfig:
"""从字典解析配置,参数越界自动钳制."""
if not data or not data.get("enabled", False):
return cls(enabled=False)
level_str = str(data.get("level", "medium")).lower()
try:
level = NoiseReductionLevel(level_str)
except ValueError:
level = DEFAULT_LEVEL
try:
noise_floor = float(data.get("noise_floor", DEFAULT_NOISE_FLOOR))
except (TypeError, ValueError):
noise_floor = DEFAULT_NOISE_FLOOR
voice_enhance = bool(data.get("voice_enhance", False))
# 钳制到合法范围
noise_floor = max(MIN_NOISE_FLOOR, min(MAX_NOISE_FLOOR, noise_floor))
return cls(
enabled=True,
level=level,
noise_floor=noise_floor,
voice_enhance=voice_enhance,
)
def has_effect(self) -> bool:
"""判断是否有实际降噪效果."""
return self.enabled
def get_effective_noise_floor(self) -> float:
"""获取实际生效的噪音阈值(dB)."""
if self.level == NoiseReductionLevel.CUSTOM:
return self.noise_floor
params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[DEFAULT_LEVEL])
return float(params["nf"])
def get_level_params(self) -> dict[str, float]:
"""获取当前等级的完整参数字典."""
if self.level == NoiseReductionLevel.CUSTOM:
return {
"nf": self.noise_floor,
"tn": -10.0,
"tr": 50.0,
}
params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[DEFAULT_LEVEL])
return {k: float(v) for k, v in params.items()}
def validate(self) -> tuple[bool, str]:
"""校验配置是否有效."""
if not self.enabled:
return True, ""
if not (MIN_NOISE_FLOOR <= self.noise_floor <= MAX_NOISE_FLOOR):
return False, f"noise_floor 必须在 {MIN_NOISE_FLOOR}~{MAX_NOISE_FLOOR} dB 之间"
return True, ""
# ── 滤镜构建 ────────────────────────────────────────────────────────────────
def build_afftdn_filter(
config: NoiseReductionConfig,
input_label: str,
output_label: str,
) -> str:
"""构建 afftdn 音频降噪滤镜字符串.
Args:
config: 降噪配置
input_label: 输入标签,如 "[0:a]" 或 "[a0]"
output_label: 输出标签,如 "[nr0]"
Returns:
FFmpeg 滤镜字符串
"""
if not config.has_effect():
return f"{input_label}anull{output_label}"
params = config.get_level_params()
nf = params["nf"]
tn = params["tn"]
tr = params["tr"]
# 构建 afftdn 滤镜
filter_parts = [f"afftdn=nf={nf}:tn={tn}:tr={tr}"]
# 人声增强:通过 highpass + 压缩 + 响度归一化实现
if config.voice_enhance:
filter_parts.append("highpass=f=80")
filter_parts.append("acompressor=threshold=-20:ratio=2:attack=5:release=50")
filter_parts.append("loudnorm=I=-16:TP=-1.5:LRA=11")
return f"{input_label}{','.join(filter_parts)}{output_label}"
def build_arnndn_filter(
config: NoiseReductionConfig,
input_label: str,
output_label: str,
model_file: str,
) -> str:
"""使用 RNN 降噪滤镜(arnndn,效果更好但需要模型文件).
注意:需要额外下载 RNNNoise 模型文件,默认使用 afftdn(无需额外依赖)。
"""
if not config.has_effect():
return f"{input_label}anull{output_label}"
return f"{input_label}arnndn=m={model_file}{output_label}"
# ── 便捷函数 ────────────────────────────────────────────────────────────────
def apply_noise_reduction_if_needed(
config_data: dict[str, Any] | None,
input_label: str,
output_label: str,
) -> str | None:
"""便捷函数:根据配置判断是否需要应用音频降噪.
Args:
config_data: 降噪配置字典
input_label: 输入标签
output_label: 输出标签
Returns:
滤镜字符串,不需要降噪时返回 None
"""
if not config_data:
return None
try:
config = NoiseReductionConfig.from_dict(config_data)
if not config.has_effect():
return None
return build_afftdn_filter(config, input_label, output_label)
except Exception as e:
logger.warning("[noise-reduction] 应用降噪失败,跳过: %s", e)
return None
def get_level_names() -> list[str]:
"""获取所有降噪等级名称列表."""
return [level.value for level in NoiseReductionLevel]
+187
View File
@@ -0,0 +1,187 @@
"""XFade 转场滤镜构建 — 纯逻辑,无 FFmpeg 依赖.
抽离自 apps/worker/video_processing/ffmpeg_utils.py,包含:
- xfade 转场效果名称映射
- 滤镜链串联工具
- xfade 转场滤镜链构建(带时长钳制)
"""
from __future__ import annotations
import logging
from typing import Any
logger = logging.getLogger(__name__)
# ── 常量 ──────────────────────────────────────────────────────────────────────
DEFAULT_TRANSITION_DURATION = 0.5
# xfade 转场映射:transition_effect 名称 → FFmpeg xfade transition 名称
# 键同时支持 TransitionEffect 枚举值和字符串名称(向后兼容)
# "cut" 为特殊值:硬切,不使用 xfade(由调用方特殊处理)
XFADE_TRANSITION_MAP: dict[str, str] = {
# 基础
"fade": "fade",
"dissolve": "dissolve",
"crossfade": "dissolve",
"crossdissolve": "dissolve",
# 滑入系列
"slideleft": "slideleft",
"slide_left": "slideleft",
"slideright": "slideright",
"slide_right": "slideright",
"slideup": "slideup",
"slide_up": "slideup",
"slidedown": "slidedown",
"slide_down": "slidedown",
"slide": "slideleft", # 默认向左滑
# 缩放
"zoom": "zoomin",
"zoomin": "zoomin",
"zoomout": "zoomout",
# 擦除系列
"wipe": "wipeleft", # 默认向左擦
"wipeleft": "wipeleft",
"wiperight": "wiperight",
"wipeup": "wipeup",
"wipedown": "wipedown",
# 特殊效果
"circlecrop": "circlecrop",
"circle": "circlecrop",
"rectcrop": "rectcrop",
"rect": "rectcrop",
}
# 所有支持的转场效果名称(用户侧输入)
SUPPORTED_TRANSITIONS: set[str] = set(XFADE_TRANSITION_MAP.keys())
# 所有 FFmpeg xfade transition 名称(输出侧)
XFade_TRANSITION_NAMES: set[str] = set(XFADE_TRANSITION_MAP.values())
# ── 工具函数 ─────────────────────────────────────────────────────────────────
def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
"""将滤镜列表串联为 FFmpeg 滤镜字符串.
例:chain_filters(["scale=1280:720", "fps=25"], "v0")
→ "[0:v]scale=1280:720,fps=25[v0]"
Args:
filters: 滤镜字符串列表
output_label: 输出标签(不带方括号)
input_label: 输入标签(不带方括号),默认 "0:v"
Returns:
完整的滤镜字符串
"""
filter_body = ",".join(filters)
return f"[{input_label}]{filter_body}[{output_label}]"
def resolve_xfade_transition(transition_name: Any) -> str:
"""将转场效果名称映射为 FFmpeg xfade transition 名称.
支持 TransitionEffect 枚举值和字符串名称,未知值回退到 "fade"。
Args:
transition_name: 转场名称(字符串或带 .value 属性的枚举)
Returns:
FFmpeg xfade transition 名称
"""
# 兼容 TransitionEffect 枚举(有 .value 属性)
if hasattr(transition_name, "value"):
transition_name = transition_name.value
return XFADE_TRANSITION_MAP.get(transition_name, "fade")
# ── xfade 滤镜链构建 ─────────────────────────────────────────────────────────
def build_xfade_filter_chain(
clip_durations: list[float],
clip_video_labels: list[str],
transitions: list[str],
*,
transition_duration: float = DEFAULT_TRANSITION_DURATION,
output_label: str = "outv",
) -> tuple[str, float]:
"""构建 xfade 转场滤镜链.
对每步 xfade 自动钳制 transition duration,确保
``offset + td ≤ first_input_duration``,避免 FFmpeg exit 234。
Args:
clip_durations: 每个片段的时长(必须与 trim 后的实际时长一致)
clip_video_labels: 每个片段的视频流标签(如 "v0", "v1")
transitions: 每个片段对应的转场效果(第一个片段的转场被忽略)
transition_duration: 转场时长(秒)
output_label: 最终输出标签
Returns:
(filter_string, estimated_total_duration)
"""
n = len(clip_durations)
parts: list[str] = []
if n == 0:
return "", 0.0
if n == 1:
parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]")
return ";".join(parts), clip_durations[0]
# xfade 链 — 每步动态钳制 td,防止 offset + td > first_input_duration
cumulative = 0.0
prev_label = clip_video_labels[0]
total_transition = 0.0 # 累计已使用的转场时长
for i in range(1, n):
cumulative += clip_durations[i - 1]
# 当前 xfade 的第一个输入时长
if i == 1:
first_input_dur = clip_durations[0]
else:
first_input_dur = cumulative - total_transition
# 原始 offset 计算
offset = max(0.0, cumulative - transition_duration * i)
# 安全钳制:offset + td 不能超过第一个输入的时长
available = max(0.0, first_input_dur - offset)
safe_td = min(transition_duration, available)
# 同时不能超过剩余总时长
remaining = max(0.0, sum(clip_durations) - cumulative)
safe_td = min(safe_td, remaining)
# 同时不能超过当前第二个输入(单个片段)的时长
safe_td = min(safe_td, clip_durations[i])
safe_td = max(0.001, safe_td) # 至少 1ms,避免 td=0
transition = transitions[i] if i < len(transitions) else "cut"
xfade_transition = resolve_xfade_transition(transition)
if i == n - 1:
out_label = output_label
else:
out_label = f"xf{i}"
parts.append(
f"[{prev_label}][{clip_video_labels[i]}]"
f"xfade=transition={xfade_transition}"
f":duration={safe_td:.3f}"
f":offset={offset:.3f}"
f"[{out_label}]"
)
prev_label = out_label
total_transition += safe_td
# 总时长减去转场重叠部分
total_duration = sum(clip_durations) - total_transition
return ";".join(parts), max(0.0, total_duration)
+448
View File
@@ -0,0 +1,448 @@
"""ASS 字幕构建领域模型单元测试 — 纯逻辑,无文件IO."""
from __future__ import annotations
import pytest
from packages.domain.ass_subtitle_builder import (
TITLE_MARGIN_BOTTOM,
TITLE_MARGIN_SIDE,
TITLE_MARGIN_TOP,
build_ass_content,
build_ass_style,
escape_ass_text,
format_ass_time,
hex_to_ass_color,
position_to_ass_alignment,
)
# ── 颜色转换 ──────────────────────────────────────────────────────────────────
class TestHexToAssColor:
def test_red(self):
assert hex_to_ass_color("#FF0000") == "&H0000FF"
def test_green(self):
assert hex_to_ass_color("#00FF00") == "&H00FF00"
def test_blue(self):
assert hex_to_ass_color("#0000FF") == "&HFF0000"
def test_white(self):
assert hex_to_ass_color("#FFFFFF") == "&HFFFFFF"
def test_black(self):
assert hex_to_ass_color("#000000") == "&H000000"
def test_without_hash(self):
assert hex_to_ass_color("FF0000") == "&H0000FF"
def test_lowercase(self):
assert hex_to_ass_color("#ff0000") == "&H0000FF"
def test_invalid_length_short(self):
assert hex_to_ass_color("#FFF") == "&H000000"
def test_invalid_length_long(self):
assert hex_to_ass_color("#FFFFFFFF") == "&H000000"
def test_empty(self):
assert hex_to_ass_color("") == "&H000000"
# ── 位置对齐 ──────────────────────────────────────────────────────────────────
class TestPositionToAssAlignment:
def test_top(self):
assert position_to_ass_alignment("top") == 8
def test_center(self):
assert position_to_ass_alignment("center") == 5
def test_bottom(self):
assert position_to_ass_alignment("bottom") == 2
def test_unknown_default_top(self):
assert position_to_ass_alignment("unknown") == 8
def test_empty_default_top(self):
assert position_to_ass_alignment("") == 8
# ── Style 行构建 ─────────────────────────────────────────────────────────────
class TestBuildAssStyle:
def test_minimal_style(self):
result = build_ass_style("TestStyle")
assert result.startswith("Style: TestStyle,")
assert "思源黑体" in result
assert ",48," in result
def test_custom_font_size(self):
result = build_ass_style("Title", font_size=64)
assert ",64," in result
def test_bold_enabled(self):
result = build_ass_style("BoldStyle", bold=True)
parts = result.split(",")
# Bold 是第 8 个字段(index 7)
assert parts[7] == "-1"
def test_bold_disabled(self):
result = build_ass_style("NormalStyle", bold=False)
parts = result.split(",")
assert parts[7] == "0"
def test_italic_enabled(self):
result = build_ass_style("ItalicStyle", italic=True)
parts = result.split(",")
assert parts[8] == "-1"
def test_alignment(self):
result = build_ass_style("AlignBottom", alignment=2)
parts = result.split(",")
# Alignment 是第 19 个字段(index 18)
assert parts[18] == "2"
def test_margins(self):
result = build_ass_style("MarginStyle", margin_v=100, margin_l=50, margin_r=50)
parts = result.split(",")
# MarginL, MarginR, MarginV 分别是 index 19, 20, 21
assert parts[19] == "50"
assert parts[20] == "50"
assert parts[21] == "100"
def test_outline_width(self):
result = build_ass_style("OutlineStyle", outline_width=3.5)
# Outline 是 index 16
parts = result.split(",")
assert parts[16] == "3.5"
def test_shadow_with_blur(self):
result = build_ass_style("ShadowStyle", shadow_blur=4.0, shadow_offset=(2, 3))
parts = result.split(",")
# Shadow 深度(纵向偏移)是 index 17
assert parts[17] == "3"
def test_shadow_without_blur(self):
result = build_ass_style("NoShadowStyle", shadow_blur=0.0, shadow_offset=(2, 3))
parts = result.split(",")
assert parts[17] == "0"
def test_primary_color(self):
result = build_ass_style("ColorStyle", primary_color="&H00FFFFFF")
# PrimaryColour 是 index 3
parts = result.split(",")
assert parts[3] == "&H00FFFFFF"
def test_outline_color(self):
result = build_ass_style("StrokeStyle", outline_color="&H00000000")
# OutlineColour 是 index 5
parts = result.split(",")
assert parts[5] == "&H00000000"
def test_field_count(self):
"""验证 Style 行有正确的字段数(23 个字段)."""
result = build_ass_style("FullStyle")
parts = result.split(",")
# Style: 行有 23 个字段(去掉 "Style: " 前缀后)
assert len(parts) == 23
# ── 文本转义 ──────────────────────────────────────────────────────────────────
class TestEscapeAssText:
def test_plain_text(self):
assert escape_ass_text("Hello World") == "Hello World"
def test_newline_lf(self):
assert escape_ass_text("line1\nline2") == "line1\\Nline2"
def test_newline_crlf(self):
assert escape_ass_text("line1\r\nline2") == "line1\\Nline2"
def test_newline_cr(self):
assert escape_ass_text("line1\rline2") == "line1\\Nline2"
def test_curly_braces(self):
assert escape_ass_text("text {tag} text") == "text (tag) text"
def test_left_brace_only(self):
assert escape_ass_text("{start") == "(start"
def test_right_brace_only(self):
assert escape_ass_text("end}") == "end)"
def test_multiple_braces(self):
assert escape_ass_text("{a}{b}{c}") == "(a)(b)(c)"
def test_mixed_newline_and_braces(self):
assert escape_ass_text("line1\n{tag}\nline2") == "line1\\N(tag)\\Nline2"
def test_empty_string(self):
assert escape_ass_text("") == ""
def test_chinese_text(self):
assert escape_ass_text("你好世界") == "你好世界"
# ── 时间格式化 ────────────────────────────────────────────────────────────────
class TestFormatAssTime:
def test_zero(self):
assert format_ass_time(0) == "0:00:00.00"
def test_seconds_only(self):
assert format_ass_time(5.5) == "0:00:05.50"
def test_minutes(self):
assert format_ass_time(125.0) == "0:02:05.00"
def test_hours(self):
assert format_ass_time(3661.5) == "1:01:01.50"
def test_one_hour_exact(self):
assert format_ass_time(3600) == "1:00:00.00"
def test_sub_second_precision(self):
result = format_ass_time(1.23)
assert result == "0:00:01.23"
def test_59_seconds(self):
assert format_ass_time(59.99) == "0:00:59.99"
def test_60_seconds(self):
assert format_ass_time(60.0) == "0:01:00.00"
def test_90_minutes(self):
assert format_ass_time(5400.0) == "1:30:00.00"
# ── 完整 ASS 内容生成 ─────────────────────────────────────────────────────────
class TestBuildAssContent:
def test_no_subtitles_returns_empty(self):
result = build_ass_content(video_width=1920, video_height=1080, video_duration=10.0)
assert result == ""
def test_title_disabled_returns_empty(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Title",
title_config={"enabled": False},
)
assert result == ""
def test_empty_title_text_returns_empty(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text=" ",
title_config={"enabled": True},
)
assert result == ""
def test_with_title(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=30.0,
title_text="My Title",
title_config={"enabled": True, "color": "#FFFFFF"},
)
assert "[Script Info]" in result
assert "PlayResX: 1920" in result
assert "PlayResY: 1080" in result
assert "[V4+ Styles]" in result
assert "TitleStyle" in result
assert "[Events]" in result
assert "Dialogue:" in result
assert "My Title" in result
def test_with_subtitle(self):
result = build_ass_content(
video_width=1280,
video_height=720,
video_duration=15.0,
subtitle_text="Subtitle Text",
subtitle_config={"enabled": True},
)
assert "PlayResX: 1280" in result
assert "PlayResY: 720" in result
assert "SubtitleStyle" in result
assert "Subtitle Text" in result
def test_with_both_title_and_subtitle(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=60.0,
title_text="Big Title",
title_config={"enabled": True},
subtitle_text="Small subtitle",
subtitle_config={"enabled": True},
)
assert "TitleStyle" in result
assert "SubtitleStyle" in result
assert "Big Title" in result
assert "Small subtitle" in result
# 两个 Dialogue 行
assert result.count("Dialogue:") == 2
def test_title_position_bottom(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Bottom Title",
title_config={"enabled": True, "position": "bottom"},
)
# 对齐方式为 2(底部居中)
assert "TitleStyle" in result
def test_title_with_stroke(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Stroke Title",
title_config={
"enabled": True,
"stroke": {"enabled": True, "color": "#000000", "width": 3},
},
)
assert "Stroke Title" in result
def test_title_with_shadow(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Shadow Title",
title_config={
"enabled": True,
"shadow": {"enabled": True, "blur": 4, "offset_x": 2, "offset_y": 3},
},
)
assert "Shadow Title" in result
def test_title_bold_default(self):
"""标题默认启用粗体."""
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Bold Title",
title_config={"enabled": True},
)
# 在 TitleStyle 行中找 bold=-1
for line in result.split("\n"):
if line.startswith("Style: TitleStyle"):
parts = line.split(",")
assert parts[7] == "-1"
break
else:
pytest.fail("TitleStyle not found")
def test_subtitle_not_bold(self):
"""字幕默认不启用粗体."""
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
subtitle_text="Normal Subtitle",
subtitle_config={"enabled": True},
)
for line in result.split("\n"):
if line.startswith("Style: SubtitleStyle"):
parts = line.split(",")
assert parts[7] == "0"
break
else:
pytest.fail("SubtitleStyle not found")
def test_duration_format_in_dialogue(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=125.5,
title_text="Timed",
title_config={"enabled": True},
)
# 结束时间应该是 0:02:05.50
assert "0:02:05.50" in result
def test_title_text_escaped(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Line1\n{tag}Line2",
title_config={"enabled": True},
)
assert "Line1\\N(tag)Line2" in result
def test_default_title_enabled(self):
"""不传 enabled 时默认为 True."""
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Default Enabled",
title_config={},
)
assert result != ""
assert "Default Enabled" in result
def test_subtitle_position_top(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
subtitle_text="Top Subtitle",
subtitle_config={"enabled": True, "position": "top"},
)
assert "Top Subtitle" in result
def test_scaled_border_and_shadow(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Test",
title_config={"enabled": True},
)
assert "ScaledBorderAndShadow: yes" in result
def test_wrap_style(self):
result = build_ass_content(
video_width=1920,
video_height=1080,
video_duration=10.0,
title_text="Test",
title_config={"enabled": True},
)
assert "WrapStyle: 2" in result
# ── 常量 ──────────────────────────────────────────────────────────────────────
class TestConstants:
def test_title_margin_top(self):
assert TITLE_MARGIN_TOP == 60
def test_title_margin_bottom(self):
assert TITLE_MARGIN_BOTTOM == 60
def test_title_margin_side(self):
assert TITLE_MARGIN_SIDE == 40
+281
View File
@@ -0,0 +1,281 @@
"""chroma_key_config 领域模型单测."""
from __future__ import annotations
import pytest
from packages.domain.chroma_key_config import (
CHROMA_KEY_PRESETS,
ChromaKeyConfig,
apply_chroma_key_if_needed,
build_chromakey_filter,
build_colorkey_filter,
get_preset_names,
normalize_color,
)
# ── ChromaKeyConfig.from_dict 测试 ────────────────────────────────────────
class TestChromaKeyConfigFromDict:
def test_none_returns_disabled(self):
cfg = ChromaKeyConfig.from_dict(None)
assert cfg.enabled is False
def test_empty_dict_returns_disabled(self):
cfg = ChromaKeyConfig.from_dict({})
assert cfg.enabled is False
def test_disabled_returns_disabled(self):
cfg = ChromaKeyConfig.from_dict({"enabled": False})
assert cfg.enabled is False
def test_enabled_default_params(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True})
assert cfg.enabled is True
assert cfg.key_color == "#00FF00"
assert cfg.similarity == 0.3
assert cfg.blend == 0.1
assert cfg.spill_suppress == 0.0
def test_custom_params(self):
cfg = ChromaKeyConfig.from_dict(
{
"enabled": True,
"key_color": "#0000FF",
"similarity": 0.5,
"blend": 0.2,
"spill_suppress": 0.4,
}
)
assert cfg.key_color == "#0000FF"
assert cfg.similarity == 0.5
assert cfg.blend == 0.2
assert cfg.spill_suppress == 0.4
def test_similarity_clamped_low(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "similarity": 0.001})
assert cfg.similarity == 0.01
def test_similarity_clamped_high(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "similarity": 2.0})
assert cfg.similarity == 1.0
def test_blend_clamped_low(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "blend": -0.5})
assert cfg.blend == 0.0
def test_blend_clamped_high(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "blend": 1.5})
assert cfg.blend == 1.0
def test_spill_suppress_clamped(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "spill_suppress": 2.0})
assert cfg.spill_suppress == 1.0
def test_invalid_similarity_type_uses_default(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "similarity": "high"})
assert cfg.similarity == 0.3
def test_key_color_stripped(self):
cfg = ChromaKeyConfig.from_dict({"enabled": True, "key_color": " #00FF00 "})
assert cfg.key_color == "#00FF00"
# ── from_preset 测试 ──────────────────────────────────────────────────────
class TestFromPreset:
def test_green_screen_preset(self):
cfg = ChromaKeyConfig.from_preset("green_screen")
assert cfg is not None
assert cfg.enabled is True
assert cfg.key_color == "#00FF00"
assert cfg.similarity == 0.3
def test_blue_screen_preset(self):
cfg = ChromaKeyConfig.from_preset("blue_screen")
assert cfg is not None
assert cfg.key_color == "#0000FF"
def test_invalid_preset_returns_none(self):
assert ChromaKeyConfig.from_preset("nonexistent") is None
def test_all_presets_valid(self):
for name in CHROMA_KEY_PRESETS:
cfg = ChromaKeyConfig.from_preset(name)
assert cfg is not None
assert cfg.enabled is True
# ── has_effect / validate 测试 ────────────────────────────────────────────
class TestHasEffectAndValidate:
def test_disabled_no_effect(self):
cfg = ChromaKeyConfig(enabled=False)
assert cfg.has_effect() is False
def test_enabled_has_effect(self):
cfg = ChromaKeyConfig(enabled=True, similarity=0.3)
assert cfg.has_effect() is True
def test_zero_similarity_no_effect(self):
cfg = ChromaKeyConfig(enabled=True, similarity=0.0)
# similarity 被钳制后为 0.01,所以应该有效果
# 等等,from_dict 才会钳制,直接构造不会
assert cfg.has_effect() is False
def test_validate_disabled_valid(self):
cfg = ChromaKeyConfig(enabled=False)
ok, msg = cfg.validate()
assert ok is True
assert msg == ""
def test_validate_enabled_valid(self):
cfg = ChromaKeyConfig(enabled=True, key_color="#00FF00")
ok, msg = cfg.validate()
assert ok is True
def test_validate_empty_color_invalid(self):
cfg = ChromaKeyConfig(enabled=True, key_color="")
ok, msg = cfg.validate()
assert ok is False
assert "key_color" in msg
def test_validate_similarity_out_of_range(self):
cfg = ChromaKeyConfig(enabled=True, similarity=2.0)
ok, msg = cfg.validate()
assert ok is False
assert "similarity" in msg
# ── normalize_color 测试 ──────────────────────────────────────────────────
class TestNormalizeColor:
def test_hex_with_hash(self):
assert normalize_color("#00FF00") == "0x00FF00"
def test_hex_lowercase(self):
assert normalize_color("#00ff00") == "0x00FF00"
def test_hex_without_hash(self):
assert normalize_color("00FF00") == "0x00FF00"
def test_hex_with_alpha(self):
assert normalize_color("#00FF00FF") == "0x00FF00"
def test_already_0x_format(self):
assert normalize_color("0x00FF00") == "0X00FF00"
def test_0x_lowercase(self):
assert normalize_color("0x00ff00") == "0X00FF00"
def test_color_name_passthrough(self):
assert normalize_color("green") == "green"
assert normalize_color("blue") == "blue"
def test_whitespace_stripped(self):
assert normalize_color(" #FF0000 ") == "0xFF0000"
# ── build_colorkey_filter 测试 ────────────────────────────────────────────
class TestBuildColorkeyFilter:
def test_disabled_returns_copy(self):
cfg = ChromaKeyConfig(enabled=False)
result = build_colorkey_filter(cfg, "[in]", "[out]")
assert "copy" in result
assert "[in]" in result
assert "[out]" in result
def test_basic_colorkey(self):
cfg = ChromaKeyConfig(enabled=True, key_color="#00FF00", similarity=0.3, blend=0.1)
result = build_colorkey_filter(cfg, "[v]", "[ck]")
assert "colorkey=" in result
assert "color=0x00FF00" in result
assert "similarity=0.3" in result
assert "blend=0.1" in result
assert "[v]" in result
assert "[ck]" in result
def test_with_spill_suppress(self):
cfg = ChromaKeyConfig(enabled=True, key_color="#00FF00", spill_suppress=0.5)
result = build_colorkey_filter(cfg, "[in]", "[out]")
assert "colorchannelmixer=" in result
assert "rr=" in result
assert "gg=" in result
assert "bb=" in result
def test_no_spill_suppress_no_colorchannelmixer(self):
cfg = ChromaKeyConfig(enabled=True, spill_suppress=0.0)
result = build_colorkey_filter(cfg, "[in]", "[out]")
assert "colorchannelmixer" not in result
# ── build_chromakey_filter 测试 ───────────────────────────────────────────
class TestBuildChromakeyFilter:
def test_disabled_returns_copy(self):
cfg = ChromaKeyConfig(enabled=False)
result = build_chromakey_filter(cfg, "[in]", "[out]")
assert "copy" in result
def test_basic_chromakey(self):
cfg = ChromaKeyConfig(enabled=True, key_color="#00FF00", similarity=0.3, blend=0.1)
result = build_chromakey_filter(cfg, "[v]", "[ck]")
assert "chromakey=" in result
assert "color=0x00FF00" in result
assert "similarity=0.3" in result
assert "blend=0.1" in result
def test_contains_input_and_output_labels(self):
cfg = ChromaKeyConfig(enabled=True)
result = build_chromakey_filter(cfg, "[in_v]", "[out_v]")
assert "[in_v]" in result
assert "[out_v]" in result
# ── apply_chroma_key_if_needed 测试 ───────────────────────────────────────
class TestApplyChromaKeyIfNeeded:
def test_none_config_returns_none(self):
assert apply_chroma_key_if_needed(None, "[in]", "[out]") is None
def test_no_chroma_key_returns_none(self):
assert apply_chroma_key_if_needed({}, "[in]", "[out]") is None
def test_disabled_chroma_key_returns_none(self):
config = {"chroma_key": {"enabled": False}}
assert apply_chroma_key_if_needed(config, "[in]", "[out]") is None
def test_enabled_chroma_key_returns_filter(self):
config = {"chroma_key": {"enabled": True, "key_color": "#00FF00"}}
result = apply_chroma_key_if_needed(config, "[in]", "[out]")
assert result is not None
assert "colorkey" in result
def test_invalid_config_handles_exception(self):
# 传入无效配置触发异常,应该返回 None 而不是抛出
config = {"chroma_key": "invalid_string"}
result = apply_chroma_key_if_needed(config, "[in]", "[out]")
assert result is None
# ── 预设工具函数测试 ───────────────────────────────────────────────────────
class TestPresetUtils:
def test_get_preset_names_returns_sorted_list(self):
names = get_preset_names()
assert isinstance(names, list)
assert len(names) == len(CHROMA_KEY_PRESETS)
assert names == sorted(names)
def test_all_preset_names_in_presets_dict(self):
for name in get_preset_names():
assert name in CHROMA_KEY_PRESETS
+258
View File
@@ -0,0 +1,258 @@
"""noise_reduction_config 领域模型单测."""
from __future__ import annotations
import pytest
from packages.domain.noise_reduction_config import (
DEFAULT_LEVEL,
DEFAULT_NOISE_FLOOR,
MAX_NOISE_FLOOR,
MIN_NOISE_FLOOR,
NoiseReductionConfig,
NoiseReductionLevel,
apply_noise_reduction_if_needed,
build_afftdn_filter,
build_arnndn_filter,
get_level_names,
)
# ── NoiseReductionLevel 枚举测试 ───────────────────────────────────────────
class TestNoiseReductionLevel:
def test_four_levels(self):
assert len(NoiseReductionLevel) == 4
def test_level_values(self):
assert NoiseReductionLevel.LOW.value == "low"
assert NoiseReductionLevel.MEDIUM.value == "medium"
assert NoiseReductionLevel.HIGH.value == "high"
assert NoiseReductionLevel.CUSTOM.value == "custom"
def test_from_string(self):
assert NoiseReductionLevel("low") == NoiseReductionLevel.LOW
assert NoiseReductionLevel("medium") == NoiseReductionLevel.MEDIUM
assert NoiseReductionLevel("high") == NoiseReductionLevel.HIGH
assert NoiseReductionLevel("custom") == NoiseReductionLevel.CUSTOM
# ── NoiseReductionConfig.from_dict 测试 ───────────────────────────────────
class TestNoiseReductionConfigFromDict:
def test_none_returns_disabled(self):
cfg = NoiseReductionConfig.from_dict(None)
assert cfg.enabled is False
def test_empty_dict_returns_disabled(self):
cfg = NoiseReductionConfig.from_dict({})
assert cfg.enabled is False
def test_disabled_returns_disabled(self):
cfg = NoiseReductionConfig.from_dict({"enabled": False})
assert cfg.enabled is False
def test_enabled_default_params(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True})
assert cfg.enabled is True
assert cfg.level == NoiseReductionLevel.MEDIUM
assert cfg.noise_floor == DEFAULT_NOISE_FLOOR
assert cfg.voice_enhance is False
def test_custom_level(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": -30.0})
assert cfg.level == NoiseReductionLevel.CUSTOM
assert cfg.noise_floor == -30.0
def test_invalid_level_defaults_medium(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "invalid"})
assert cfg.level == NoiseReductionLevel.MEDIUM
def test_noise_floor_clamped_low(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": -100.0})
assert cfg.noise_floor == MIN_NOISE_FLOOR
def test_noise_floor_clamped_high(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": 0.0})
assert cfg.noise_floor == MAX_NOISE_FLOOR
def test_invalid_noise_floor_type_uses_default(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": "not_a_number"})
assert cfg.noise_floor == DEFAULT_NOISE_FLOOR
def test_voice_enhance_true(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "voice_enhance": True})
assert cfg.voice_enhance is True
def test_case_insensitive_level(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "HIGH"})
assert cfg.level == NoiseReductionLevel.HIGH
# ── has_effect / get_effective_noise_floor 测试 ───────────────────────────
class TestConfigProperties:
def test_disabled_no_effect(self):
cfg = NoiseReductionConfig(enabled=False)
assert cfg.has_effect() is False
def test_enabled_has_effect(self):
cfg = NoiseReductionConfig(enabled=True)
assert cfg.has_effect() is True
def test_effective_noise_floor_low(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.LOW)
assert cfg.get_effective_noise_floor() == -35.0
def test_effective_noise_floor_medium(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
assert cfg.get_effective_noise_floor() == -25.0
def test_effective_noise_floor_high(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.HIGH)
assert cfg.get_effective_noise_floor() == -15.0
def test_effective_noise_floor_custom(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-40.0)
assert cfg.get_effective_noise_floor() == -40.0
def test_get_level_params_medium(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
params = cfg.get_level_params()
assert params["nf"] == -25.0
assert params["tn"] == -10.0
assert params["tr"] == 50.0
def test_get_level_params_custom(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-30.0)
params = cfg.get_level_params()
assert params["nf"] == -30.0
assert "tn" in params
assert "tr" in params
# ── validate 测试 ─────────────────────────────────────────────────────────
class TestValidate:
def test_disabled_valid(self):
cfg = NoiseReductionConfig(enabled=False)
ok, msg = cfg.validate()
assert ok is True
assert msg == ""
def test_enabled_valid(self):
cfg = NoiseReductionConfig(enabled=True, noise_floor=-25.0)
ok, msg = cfg.validate()
assert ok is True
def test_noise_floor_out_of_range(self):
cfg = NoiseReductionConfig(enabled=True, noise_floor=-100.0)
ok, msg = cfg.validate()
assert ok is False
assert "noise_floor" in msg
# ── build_afftdn_filter 测试 ───────────────────────────────────────────────
class TestBuildAfftdnFilter:
def test_disabled_returns_anull(self):
cfg = NoiseReductionConfig(enabled=False)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "anull" in result
assert "[in]" in result
assert "[out]" in result
def test_medium_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
result = build_afftdn_filter(cfg, "[a]", "[nr]")
assert "afftdn=" in result
assert "nf=-25.0" in result or "nf=-25" in result
assert "[a]" in result
assert "[nr]" in result
def test_high_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.HIGH)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "afftdn=" in result
assert "nf=-15.0" in result or "nf=-15" in result
def test_low_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.LOW)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "afftdn=" in result
assert "nf=-35.0" in result or "nf=-35" in result
def test_custom_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-40.0)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "afftdn=" in result
assert "nf=-40.0" in result or "nf=-40" in result
def test_voice_enhance_adds_filters(self):
cfg = NoiseReductionConfig(enabled=True, voice_enhance=True)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "highpass" in result
assert "acompressor" in result
assert "loudnorm" in result
def test_no_voice_enhance_no_extra_filters(self):
cfg = NoiseReductionConfig(enabled=True, voice_enhance=False)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "highpass" not in result
assert "acompressor" not in result
# ── build_arnndn_filter 测试 ───────────────────────────────────────────────
class TestBuildArnndnFilter:
def test_disabled_returns_anull(self):
cfg = NoiseReductionConfig(enabled=False)
result = build_arnndn_filter(cfg, "[in]", "[out]", "model.rnnn")
assert "anull" in result
def test_enabled_returns_arnndn(self):
cfg = NoiseReductionConfig(enabled=True)
result = build_arnndn_filter(cfg, "[a]", "[nr]", "/path/to/model.rnnn")
assert "arnndn=" in result
assert "m=/path/to/model.rnnn" in result
assert "[a]" in result
assert "[nr]" in result
# ── apply_noise_reduction_if_needed 测试 ─────────────────────────────────
class TestApplyNoiseReductionIfNeeded:
def test_none_config_returns_none(self):
assert apply_noise_reduction_if_needed(None, "[in]", "[out]") is None
def test_disabled_returns_none(self):
assert apply_noise_reduction_if_needed({"enabled": False}, "[in]", "[out]") is None
def test_enabled_returns_filter(self):
result = apply_noise_reduction_if_needed({"enabled": True, "level": "medium"}, "[in]", "[out]")
assert result is not None
assert "afftdn" in result
def test_invalid_config_handles_exception(self):
# 异常情况应该返回 None 而不是抛出
result = apply_noise_reduction_if_needed("invalid", "[in]", "[out]")
assert result is None
# ── 工具函数测试 ───────────────────────────────────────────────────────────
class TestUtils:
def test_get_level_names_returns_four(self):
names = get_level_names()
assert len(names) == 4
assert "low" in names
assert "medium" in names
assert "high" in names
assert "custom" in names
+371
View File
@@ -0,0 +1,371 @@
"""XFade 转场滤镜构建领域模型单元测试."""
from __future__ import annotations
import pytest
from packages.domain.xfade_builder import (
DEFAULT_TRANSITION_DURATION,
SUPPORTED_TRANSITIONS,
XFADE_TRANSITION_MAP,
XFade_TRANSITION_NAMES,
build_xfade_filter_chain,
chain_filters,
resolve_xfade_transition,
)
# ── 常量测试 ─────────────────────────────────────────────────────────────────
class TestConstants:
def test_default_transition_duration(self):
assert DEFAULT_TRANSITION_DURATION == 0.5
def test_xfade_transition_map_not_empty(self):
assert len(XFADE_TRANSITION_MAP) > 0
def test_supported_transitions(self):
assert "fade" in SUPPORTED_TRANSITIONS
assert "dissolve" in SUPPORTED_TRANSITIONS
def test_xfade_transition_names(self):
assert "fade" in XFade_TRANSITION_NAMES
assert "dissolve" in XFade_TRANSITION_NAMES
# ── chain_filters 测试 ──────────────────────────────────────────────────────
class TestChainFilters:
def test_single_filter(self):
result = chain_filters(["scale=1280:720"], "v0")
assert result == "[0:v]scale=1280:720[v0]"
def test_multiple_filters(self):
result = chain_filters(["scale=1280:720", "fps=25"], "v0")
assert result == "[0:v]scale=1280:720,fps=25[v0]"
def test_empty_filters(self):
result = chain_filters([], "out")
assert result == "[0:v][out]"
def test_custom_input_label(self):
result = chain_filters(["scale=640:480"], "out", input_label="1:v")
assert result == "[1:v]scale=640:480[out]"
def test_three_filters(self):
result = chain_filters(["trim=0:5", "setpts=PTS-STARTPTS", "fps=30"], "v1")
assert result == "[0:v]trim=0:5,setpts=PTS-STARTPTS,fps=30[v1]"
# ── resolve_xfade_transition 测试 ───────────────────────────────────────────
class TestResolveXfadeTransition:
def test_fade(self):
assert resolve_xfade_transition("fade") == "fade"
def test_dissolve(self):
assert resolve_xfade_transition("dissolve") == "dissolve"
def test_crossfade_maps_to_dissolve(self):
assert resolve_xfade_transition("crossfade") == "dissolve"
def test_slideleft(self):
assert resolve_xfade_transition("slideleft") == "slideleft"
def test_slide_left_maps_to_slideleft(self):
assert resolve_xfade_transition("slide_left") == "slideleft"
def test_slide_default_left(self):
assert resolve_xfade_transition("slide") == "slideleft"
def test_slideup(self):
assert resolve_xfade_transition("slideup") == "slideup"
def test_zoom_maps_to_zoomin(self):
assert resolve_xfade_transition("zoom") == "zoomin"
def test_zoomin(self):
assert resolve_xfade_transition("zoomin") == "zoomin"
def test_wipe_default_left(self):
assert resolve_xfade_transition("wipe") == "wipeleft"
def test_wipeup(self):
assert resolve_xfade_transition("wipeup") == "wipeup"
def test_circle_maps_to_circlecrop(self):
assert resolve_xfade_transition("circle") == "circlecrop"
def test_rect_maps_to_rectcrop(self):
assert resolve_xfade_transition("rect") == "rectcrop"
def test_unknown_falls_back_to_fade(self):
assert resolve_xfade_transition("nonexistent_effect") == "fade"
def test_empty_string_falls_back_to_fade(self):
assert resolve_xfade_transition("") == "fade"
def test_enum_with_value_attribute(self):
"""测试带 .value 属性的枚举对象."""
class FakeEnum:
def __init__(self, val):
self.value = val
assert resolve_xfade_transition(FakeEnum("fade")) == "fade"
assert resolve_xfade_transition(FakeEnum("slideleft")) == "slideleft"
assert resolve_xfade_transition(FakeEnum("unknown")) == "fade"
# ── build_xfade_filter_chain 测试 ───────────────────────────────────────────
class TestBuildXfadeFilterChain:
# ── 边界情况 ──────────────────────────────────────────────────────
def test_empty_clips(self):
result, duration = build_xfade_filter_chain([], [], [])
assert result == ""
assert duration == 0.0
def test_single_clip(self):
result, duration = build_xfade_filter_chain([10.0], ["v0"], ["none"])
assert "copy" in result
assert "[v0]copy[outv]" in result
assert duration == 10.0
def test_single_clip_custom_output_label(self):
result, duration = build_xfade_filter_chain([5.0], ["a0"], ["none"], output_label="final")
assert "[a0]copy[final]" in result
assert duration == 5.0
# ── 两片段基础测试 ────────────────────────────────────────────────
def test_two_clips_basic(self):
result, duration = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "xfade=transition=fade" in result
assert "[v0][v1]" in result
assert "[outv]" in result
# 总时长 = 10 + 10 - 0.5 = 19.5
assert abs(duration - 19.5) < 0.01
def test_two_clips_custom_duration(self):
result, duration = build_xfade_filter_chain(
[10.0, 10.0],
["v0", "v1"],
["none", "fade"],
transition_duration=1.0,
)
assert "duration=1.000" in result
# 总时长 = 10 + 10 - 1.0 = 19.0
assert abs(duration - 19.0) < 0.01
def test_two_clips_offset(self):
"""两片段时 offset 应该为 0(cumulative - td * 1 = 10 - 0.5 = 9.5?不对)。
对于两个片段:
- cumulative = clip_durations[0] = 10.0
- offset = max(0, cumulative - td * i) = max(0, 10.0 - 0.5 * 1) = 9.5
- duration=0.5, offset=9.5
"""
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "offset=9.500" in result
# ── 多片段测试 ────────────────────────────────────────────────────
def test_three_clips(self):
result, duration = build_xfade_filter_chain(
[5.0, 5.0, 5.0],
["v0", "v1", "v2"],
["none", "fade", "dissolve"],
)
# 应该有两个 xfade
assert result.count("xfade=") == 2
# 第一个 xfade 输出标签 xf1,第二个 xfade 输出 outv
assert "xf1" in result
assert "[outv]" in result
# 总时长 ≈ 5 + 5 + 5 - 0.5 - 0.5 = 14.0
assert abs(duration - 14.0) < 0.1
def test_five_clips(self):
result, duration = build_xfade_filter_chain(
[3.0, 3.0, 3.0, 3.0, 3.0],
["v0", "v1", "v2", "v3", "v4"],
["none", "fade", "fade", "fade", "fade"],
)
assert result.count("xfade=") == 4
# 总时长 ≈ 15 - 4 * 0.5 = 13.0
assert abs(duration - 13.0) < 0.2
# ── 转场效果测试 ──────────────────────────────────────────────────
def test_dissolve_transition(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "dissolve"])
assert "transition=dissolve" in result
def test_slideleft_transition(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "slideleft"])
assert "transition=slideleft" in result
def test_cut_uses_fade(self):
"""cut 转场效果应该回退到 fade."""
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "cut"])
# cut 不是 XFADE_TRANSITION_MAP 的键,所以会回退到 fade
assert "transition=fade" in result
def test_transitions_shorter_than_clips(self):
"""如果 transitions 列表比 clips 短,剩余的用 'cut'(→ fade)."""
result, _ = build_xfade_filter_chain(
[5.0, 5.0, 5.0],
["v0", "v1", "v2"],
["none"], # 只有一个
)
# 第二个转场(index 2)会回退到 cut → fade
assert result.count("transition=fade") == 2
# ── 时长钳制测试 ──────────────────────────────────────────────────
def test_short_first_clip_truncates_transition(self):
"""第一个片段非常短,转场时长应该被钳制."""
result, duration = build_xfade_filter_chain(
[0.3, 10.0],
["v0", "v1"],
["none", "fade"],
transition_duration=1.0,
)
# offset = max(0, 0.3 - 1.0 * 1) = 0.0
# available = max(0, 0.3 - 0.0) = 0.3
# safe_td = min(1.0, 0.3, 剩余 10.0, clip_durations[1] 10.0) = 0.3
assert "duration=0.300" in result
assert abs(duration - 10.0) < 0.01 # 0.3 + 10.0 - 0.3 = 10.0
def test_zero_duration_clips(self):
"""零时长片段的边界情况."""
result, duration = build_xfade_filter_chain([0.0, 5.0], ["v0", "v1"], ["none", "fade"])
# 第一个片段 0 时长,转场时长应该被钳制到最小值 0.001
# offset = max(0, 0 - 0.5) = 0
# available = max(0, 0 - 0) = 0
# safe_td = min(0.5, 0, ...) = min(0.5, 0, 5.0, 5.0) = 0 → max(0.001, 0) = 0.001
assert "duration=0.001" in result
def test_very_long_transition_duration(self):
"""转场时长超过所有片段时长."""
result, duration = build_xfade_filter_chain(
[2.0, 2.0],
["v0", "v1"],
["none", "fade"],
transition_duration=5.0,
)
# offset = max(0, 2.0 - 5.0) = 0
# available = max(0, 2.0 - 0) = 2.0
# safe_td = min(5.0, 2.0, 剩余 2.0, 2.0) = 2.0
assert "duration=2.000" in result
assert abs(duration - 2.0) < 0.01 # 2 + 2 - 2 = 2
# ── 标签测试 ──────────────────────────────────────────────────────
def test_custom_labels(self):
result, _ = build_xfade_filter_chain(
[10.0, 10.0],
["clip_a", "clip_b"],
["none", "fade"],
output_label="final_v",
)
assert "[clip_a][clip_b]" in result
assert "[final_v]" in result
def test_intermediate_labels_three_clips(self):
result, _ = build_xfade_filter_chain([5.0, 5.0, 5.0], ["v0", "v1", "v2"], ["none", "fade", "fade"])
# 第一个 xfade 输出 xf1
assert "[xf1][v2]" in result or result.count("[xf1]") >= 1
# ── 总时长计算验证 ────────────────────────────────────────────────
def test_total_duration_two_equal_clips(self):
_, duration = build_xfade_filter_chain([8.0, 8.0], ["v0", "v1"], ["none", "fade"])
# 8 + 8 - 0.5 = 15.5
assert abs(duration - 15.5) < 0.01
def test_total_duration_no_transition_impossible(self):
"""即使 transition_duration=0,也有最小 0.001 的钳制."""
_, duration = build_xfade_filter_chain(
[10.0, 10.0],
["v0", "v1"],
["none", "fade"],
transition_duration=0.0,
)
# transition_duration=0,但 safe_td 有下限 0.001
assert duration < 20.0 # 应该小于 20(有重叠)
assert duration > 19.9 # 但接近 20
# ── 滤镜字符串格式验证 ────────────────────────────────────────────
def test_filter_format_contains_xfade_keyword(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "xfade=" in result
def test_filter_uses_semicolon_separator(self):
"""多步 xfade 之间用分号分隔."""
result, _ = build_xfade_filter_chain([5.0, 5.0, 5.0], ["v0", "v1", "v2"], ["none", "fade", "fade"])
assert ";" in result
# 3个片段 → 2个xfade → 1个分号
assert result.count("xfade=") == 2
assert result.count(";") == 1
def test_filter_has_transition_param(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "transition=fade" in result
def test_filter_has_duration_param(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "duration=" in result
def test_filter_has_offset_param(self):
result, _ = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", "fade"])
assert "offset=" in result
# ── 各种转场效果遍历测试 ──────────────────────────────────────────
@pytest.mark.parametrize(
"transition_name",
list(XFADE_TRANSITION_MAP.keys()),
)
def test_all_supported_transitions(self, transition_name):
"""所有支持的转场效果都应该能正确生成滤镜."""
result, duration = build_xfade_filter_chain([10.0, 10.0], ["v0", "v1"], ["none", transition_name])
expected = XFADE_TRANSITION_MAP[transition_name]
assert f"transition={expected}" in result
assert duration > 0
# ── 四片段复杂场景 ────────────────────────────────────────────────
def test_four_clips_different_durations(self):
durations = [3.0, 5.0, 2.0, 7.0]
result, duration = build_xfade_filter_chain(
durations,
["v0", "v1", "v2", "v3"],
["none", "fade", "dissolve", "slideleft"],
)
assert result.count("xfade=") == 3
# 总时长 = sum(durations) - 3 * 0.5 ≈ 17 - 1.5 = 15.5
assert abs(duration - 15.5) < 0.2
# ── transition_duration = 0 的边界 ───────────────────────────────
def test_zero_transition_duration_minimum_clamped(self):
result, _ = build_xfade_filter_chain(
[10.0, 10.0],
["v0", "v1"],
["none", "fade"],
transition_duration=0.0,
)
# 至少 0.001
assert "duration=0.001" in result
# ── 单片段自定义输出标签 ─────────────────────────────────────────
def test_single_clip_output_label(self):
result, _ = build_xfade_filter_chain([5.0], ["v0"], ["none"], output_label="result")
assert "[v0]copy[result]" in result