Compare commits

..

5 Commits

Author SHA1 Message Date
xiaoxia 0717aea36e style(tasks): 修复 smoke.test.tsx Prettier 格式
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 12s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m21s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m17s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m27s
AI Code Review / AI Code Review (pull_request) Successful in 1m36s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m37s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m13s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 5m44s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 55s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m15s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 2m46s
CI/CD Pipeline / CI Gate (pull_request) Successful in 23s
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 35s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Waiting to run
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1171h57m15s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1172h3m52s
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 1172h15m3s
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 1172h15m5s
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 1172h15m7s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 1172h15m57s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1172h15m59s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1172h18m19s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1172h24m49s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 1172h25m0s
CI/CD Pipeline / PR Build Worker Image (pull_request) Failing after 1172h24m51s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1172h32m3s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1172h32m3s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1172h32m5s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1172h49m0s
CI/CD Pipeline / PR Build API Image (pull_request) Failing after 1172h57m51s
2026-07-29 18:48:41 +08:00
xiaoxia db83222dd7 test: 添加tasks模块smoke测试
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 48s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m12s
AI Code Review / AI Code Review (pull_request) Successful in 1m26s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m36s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m44s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 2m13s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m13s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m41s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 28s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 47s
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Web Image (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 / CI Gate (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Failing after 1172h41m5s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 1172h41m12s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1172h41m16s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1172h47m30s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1172h47m31s
CI/CD Pipeline / PR Build API Image (pull_request) Failing after 1173h14m8s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1173h20m31s
2026-07-29 18:37:38 +08:00
xiaoxia d3373144aa style: prettier格式化
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 57s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m43s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 2m2s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m14s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m5s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m15s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 2m18s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 1m29s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 1m58s
AI Code Review / AI Code Review (pull_request) Successful in 4m28s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m59s
CI/CD Pipeline / 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 / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 1172h52m57s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1172h53m1s
CI/CD Pipeline / PR Build Worker Image (pull_request) Failing after 1172h55m52s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 1172h55m56s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1172h55m56s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1172h57m1s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1172h57m3s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1173h26m0s
CI/CD Pipeline / PR Build API Image (pull_request) Failing after 1173h28m55s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1173h30m0s
2026-07-29 18:30:36 +08:00
xiaoxia 9aab0f5dca fix: 移除未使用的React import
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 39s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 22s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 36s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m10s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 54s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 1m22s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m57s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m57s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 1m33s
AI Code Review / AI Code Review (pull_request) Successful in 3m7s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 6m11s
CI/CD Pipeline / CI Gate (pull_request) Failing after 12s
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1173h58m45s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1173h58m45s
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 1173h58m46s
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 1173h58m47s
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 1173h58m47s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1173h58m47s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 1173h59m55s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1173h59m55s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1174h2m48s
CI/CD Pipeline / PR Build Worker Image (pull_request) Failing after 1174h2m50s
CI/CD Pipeline / PR Build API Image (pull_request) Failing after 1174h2m57s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1173h59m57s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1174h4m19s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1174h4m21s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1174h4m23s
CI/CD Pipeline / Canary Release to Production (pull_request) Failing after 1174h31m45s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 1174h36m29s
2026-07-29 17:22:47 +08:00
xiaoxia 542ea59869 refactor(tasks): 拆分 TaskTable 列定义和空状态到子模块(190→72行, -62%)
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 30s
Preview Deploy / Deploy Preview Environment (pull_request) Failing after 48s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 46s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 55s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 1m19s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m35s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 2m7s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 2m10s
AI Code Review / AI Code Review (pull_request) Successful in 4m13s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 3m11s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 6m38s
CI/CD Pipeline / CI Gate (pull_request) Failing after 5s
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1174h8m56s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1174h8m57s
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 1174h8m57s
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 1174h8m58s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1174h8m58s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 1174h14m15s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1174h14m17s
CI/CD Pipeline / PR Build Worker Image (pull_request) Failing after 1174h14m19s
CI/CD Pipeline / PR Build API Image (pull_request) Failing after 1174h14m21s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1174h15m9s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1174h15m28s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1174h15m30s
CI/CD Pipeline / Canary Release to Production (pull_request) Failing after 1174h41m57s
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 1174h41m58s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1174h47m17s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 1174h48m7s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1174h48m28s
2026-07-29 17:11:47 +08:00
18 changed files with 1660 additions and 2456 deletions
+58 -14
View File
@@ -1,16 +1,18 @@
/**
* 字幕样式配置面板 — Drawer 形式
* 字幕开关(手动 / ASR 自动识别)、字体大小、颜色、描边/阴影、位置、ASR 语言
*/
import React from "react"
import { Drawer, Slider, ColorPicker, Select } from "antd"
import type { Color } from "antd/es/color-picker"
import type { SubtitleStyleConfig } from "@/pages/editing-planner/types/subtitle"
import {
POSITION_OPTIONS,
FONT_OPTIONS,
ANIMATION_OPTIONS,
ASR_LANGUAGE_OPTIONS,
} from "@/pages/editing-planner/constants/subtitleStyle"
import SubtitlePreview from "./subtitle-style/SubtitlePreview"
import { SubtitleModeSwitch } from "./subtitle-style/SubtitleModeSwitch"
import { SubtitlePositionSelector } from "./subtitle-style/SubtitlePositionSelector"
import { SubtitleEffectButtons } from "./subtitle-style/SubtitleEffectButtons"
interface SubtitleStylePanelProps {
open: boolean
@@ -38,6 +40,7 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
onClose={onClose}
className="subtitle-style-drawer"
>
{/* ── 字幕开关 ── */}
<div className="sub-field">
<div className="sub-toggle-row">
<span className="sub-label"></span>
@@ -52,11 +55,26 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
{config.enabled && (
<>
{/* ── 模式切换 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<SubtitleModeSwitch mode={config.mode} onModeChange={(mode) => update({ mode })} />
<div className="sub-mode-switch">
<button
className={`sub-mode-btn${config.mode === "manual" ? " active" : ""}`}
onClick={() => update({ mode: "manual" })}
>
</button>
<button
className={`sub-mode-btn${config.mode === "asr" ? " active" : ""}`}
onClick={() => update({ mode: "asr" })}
>
🤖 ASR
</button>
</div>
</div>
{/* ── ASR 语言(仅 ASR 模式) ── */}
{config.mode === "asr" && (
<div className="sub-field">
<label className="sub-label"></label>
@@ -70,6 +88,7 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
</div>
)}
{/* ── 字体大小 ── */}
<div className="sub-field">
<label className="sub-label">
<span className="sub-value">{config.fontSize}px</span>
@@ -82,6 +101,7 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
/>
</div>
{/* ── 字体颜色 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<div className="sub-color-row">
@@ -93,6 +113,7 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
</div>
</div>
{/* ── 字体 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<Select
@@ -104,24 +125,46 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
/>
</div>
{/* ── 字幕位置 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<SubtitlePositionSelector
position={config.position}
onPositionChange={(position) => update({ position })}
/>
<div className="sub-position-group">
{POSITION_OPTIONS.map((opt) => (
<button
key={opt.value}
className={`sub-position-btn${config.position === opt.value ? " active" : ""}`}
onClick={() =>
update({
position: opt.value as SubtitleStyleConfig["position"],
})
}
>
{opt.label}
</button>
))}
</div>
</div>
{/* ── 描边 / 阴影 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<SubtitleEffectButtons
stroke={config.stroke}
shadow={config.shadow}
onStrokeChange={(stroke) => update({ stroke })}
onShadowChange={(shadow) => update({ shadow })}
/>
<div className="sub-effect-btns">
<button
className={`sub-effect-btn${config.stroke ? " active" : ""}`}
onClick={() => update({ stroke: !config.stroke })}
>
S
</button>
<button
className={`sub-effect-btn${config.shadow ? " active" : ""}`}
onClick={() => update({ shadow: !config.shadow })}
>
</button>
</div>
</div>
{/* ── 动画 ── */}
<div className="sub-field">
<label className="sub-label"></label>
<Select
@@ -136,6 +179,7 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
/>
</div>
{/* ── 预览 ── */}
<SubtitlePreview config={config} />
</>
)}
@@ -1,32 +0,0 @@
import React from "react"
interface SubtitleEffectButtonsProps {
stroke: boolean
shadow: boolean
onStrokeChange: (enabled: boolean) => void
onShadowChange: (enabled: boolean) => void
}
export const SubtitleEffectButtons: React.FC<SubtitleEffectButtonsProps> = ({
stroke,
shadow,
onStrokeChange,
onShadowChange,
}) => {
return (
<div className="sub-effect-btns">
<button
className={`sub-effect-btn${stroke ? " active" : ""}`}
onClick={() => onStrokeChange(!stroke)}
>
S
</button>
<button
className={`sub-effect-btn${shadow ? " active" : ""}`}
onClick={() => onShadowChange(!shadow)}
>
</button>
</div>
)
}
@@ -1,25 +0,0 @@
import React from "react"
interface SubtitleModeSwitchProps {
mode: "manual" | "asr"
onModeChange: (mode: "manual" | "asr") => void
}
export const SubtitleModeSwitch: React.FC<SubtitleModeSwitchProps> = ({ mode, onModeChange }) => {
return (
<div className="sub-mode-switch">
<button
className={`sub-mode-btn${mode === "manual" ? " active" : ""}`}
onClick={() => onModeChange("manual")}
>
</button>
<button
className={`sub-mode-btn${mode === "asr" ? " active" : ""}`}
onClick={() => onModeChange("asr")}
>
🤖 ASR
</button>
</div>
)
}
@@ -1,27 +0,0 @@
import React from "react"
import type { SubtitleStyleConfig } from "@/pages/editing-planner/types/subtitle"
import { POSITION_OPTIONS } from "@/pages/editing-planner/constants/subtitleStyle"
interface SubtitlePositionSelectorProps {
position: SubtitleStyleConfig["position"]
onPositionChange: (position: SubtitleStyleConfig["position"]) => void
}
export const SubtitlePositionSelector: React.FC<SubtitlePositionSelectorProps> = ({
position,
onPositionChange,
}) => {
return (
<div className="sub-position-group">
{POSITION_OPTIONS.map((opt) => (
<button
key={opt.value}
className={`sub-position-btn${position === opt.value ? " active" : ""}`}
onClick={() => onPositionChange(opt.value as SubtitleStyleConfig["position"])}
>
{opt.label}
</button>
))}
</div>
)
}
+5 -123
View File
@@ -1,11 +1,8 @@
import React from "react"
import { Table, Tag, Button, Popconfirm, Tooltip } from "antd"
import { RedoOutlined, InfoCircleOutlined, ClockCircleOutlined } from "@ant-design/icons"
import type { ColumnsType } from "antd/es/table"
import type { TaskItem, TaskStatus } from "@/api/tasks"
import { STATUS_CONFIG, TYPE_LABELS } from "../constants"
import { formatDuration, formatTime } from "../utils"
import { Table } from "antd"
import type { TaskItem } from "@/api/tasks"
import { TaskErrorDetail } from "./TaskErrorDetail"
import { useTaskTableColumns, TaskEmptyState } from "./task-table"
interface TaskTableProps {
dataSource: TaskItem[]
@@ -24,7 +21,6 @@ interface TaskTableProps {
/**
* 任务列表表格
* 含列定义、分页、展开行
*/
export const TaskTable: React.FC<TaskTableProps> = ({
dataSource,
@@ -40,116 +36,7 @@ export const TaskTable: React.FC<TaskTableProps> = ({
onRetry,
onViewDetail,
}) => {
// 表格列定义
const columns: ColumnsType<TaskItem> = [
{
title: "任务ID",
dataIndex: "id",
key: "id",
width: 120,
ellipsis: true,
render: (id: string) => (
<Tooltip title={id}>
<span className="task-id">{id.slice(0, 8)}...</span>
</Tooltip>
),
},
{
title: "类型",
dataIndex: "task_type",
key: "task_type",
width: 100,
render: (type: string) => {
const config = TYPE_LABELS[type] || { label: type, color: "default" }
return <Tag color={config.color}>{config.label}</Tag>
},
},
{
title: "状态",
dataIndex: "status",
key: "status",
width: 120,
render: (status: TaskStatus, record: TaskItem) => {
const config = STATUS_CONFIG[status] || {
label: status,
color: "default",
icon: null,
}
return (
<Tag color={config.color} icon={config.icon} className="task-status-tag">
{config.label}
{status === "running" && record.progress > 0 && (
<span className="task-progress"> {record.progress}%</span>
)}
</Tag>
)
},
},
{
title: "当前步骤",
dataIndex: "current_step",
key: "current_step",
width: 150,
ellipsis: true,
render: (step: string) => <span className="task-step">{step || "-"}</span>,
},
{
title: "耗时",
dataIndex: "duration_seconds",
key: "duration_seconds",
width: 100,
render: (seconds: number) => <span className="task-duration">{formatDuration(seconds)}</span>,
},
{
title: "创建时间",
dataIndex: "created_at",
key: "created_at",
width: 120,
render: (time: string) => <span className="task-time">{formatTime(time)}</span>,
},
{
title: "操作",
key: "action",
width: 100,
fixed: "right",
render: (_: unknown, record: TaskItem) => {
if (record.status === "failed" && record.retryable) {
return (
<Popconfirm
title="确认重试"
description="确定要重试这个失败的任务吗?"
onConfirm={() => onRetry(record.id)}
okText="确定"
cancelText="取消"
>
<Button
type="link"
size="small"
icon={<RedoOutlined />}
loading={retryLoading}
className="task-retry-btn"
>
</Button>
</Popconfirm>
)
}
if (record.status === "failed") {
return (
<Button
type="link"
size="small"
icon={<InfoCircleOutlined />}
onClick={() => onViewDetail(record)}
>
</Button>
)
}
return <span className="task-action-placeholder">-</span>
},
},
]
const columns = useTaskTableColumns({ retryLoading, onRetry, onViewDetail })
return (
<Table
@@ -178,12 +65,7 @@ export const TaskTable: React.FC<TaskTableProps> = ({
scroll={{ x: 800 }}
className="task-table"
locale={{
emptyText: (
<div className="task-empty">
<ClockCircleOutlined />
<p></p>
</div>
),
emptyText: <TaskEmptyState />,
}}
/>
)
@@ -0,0 +1,9 @@
import React from "react"
import { ClockCircleOutlined } from "@ant-design/icons"
export const TaskEmptyState: React.FC = () => (
<div className="task-empty">
<ClockCircleOutlined />
<p></p>
</div>
)
+2
View File
@@ -0,0 +1,2 @@
export { useTaskTableColumns } from "./useTaskTableColumns"
export { TaskEmptyState } from "./TaskEmptyState"
@@ -0,0 +1,128 @@
import { Tag, Button, Popconfirm, Tooltip } from "antd"
import { RedoOutlined, InfoCircleOutlined } from "@ant-design/icons"
import type { ColumnsType } from "antd/es/table"
import type { TaskItem, TaskStatus } from "@/api/tasks"
import { STATUS_CONFIG, TYPE_LABELS } from "../../constants"
import { formatDuration, formatTime } from "../../utils"
interface UseTaskTableColumnsOptions {
retryLoading: boolean
onRetry: (id: string) => void
onViewDetail: (record: TaskItem) => void
}
export function useTaskTableColumns({
retryLoading,
onRetry,
onViewDetail,
}: UseTaskTableColumnsOptions): ColumnsType<TaskItem> {
return [
{
title: "任务ID",
dataIndex: "id",
key: "id",
width: 120,
ellipsis: true,
render: (id: string) => (
<Tooltip title={id}>
<span className="task-id">{id.slice(0, 8)}...</span>
</Tooltip>
),
},
{
title: "类型",
dataIndex: "task_type",
key: "task_type",
width: 100,
render: (type: string) => {
const config = TYPE_LABELS[type] || { label: type, color: "default" }
return <Tag color={config.color}>{config.label}</Tag>
},
},
{
title: "状态",
dataIndex: "status",
key: "status",
width: 120,
render: (status: TaskStatus, record: TaskItem) => {
const config = STATUS_CONFIG[status] || {
label: status,
color: "default",
icon: null,
}
return (
<Tag color={config.color} icon={config.icon} className="task-status-tag">
{config.label}
{status === "running" && record.progress > 0 && (
<span className="task-progress"> {record.progress}%</span>
)}
</Tag>
)
},
},
{
title: "当前步骤",
dataIndex: "current_step",
key: "current_step",
width: 150,
ellipsis: true,
render: (step: string) => <span className="task-step">{step || "-"}</span>,
},
{
title: "耗时",
dataIndex: "duration_seconds",
key: "duration_seconds",
width: 100,
render: (seconds: number) => <span className="task-duration">{formatDuration(seconds)}</span>,
},
{
title: "创建时间",
dataIndex: "created_at",
key: "created_at",
width: 120,
render: (time: string) => <span className="task-time">{formatTime(time)}</span>,
},
{
title: "操作",
key: "action",
width: 100,
fixed: "right",
render: (_: unknown, record: TaskItem) => {
if (record.status === "failed" && record.retryable) {
return (
<Popconfirm
title="确认重试"
description="确定要重试这个失败的任务吗?"
onConfirm={() => onRetry(record.id)}
okText="确定"
cancelText="取消"
>
<Button
type="link"
size="small"
icon={<RedoOutlined />}
loading={retryLoading}
className="task-retry-btn"
>
</Button>
</Popconfirm>
)
}
if (record.status === "failed") {
return (
<Button
type="link"
size="small"
icon={<InfoCircleOutlined />}
onClick={() => onViewDetail(record)}
>
</Button>
)
}
return <span className="task-action-placeholder">-</span>
},
},
]
}
@@ -1,4 +1,4 @@
import { useRef, useCallback, useEffect } from "react"
import { useRef, useCallback } from "react"
interface UseRowProgressOptions {
duration: number
@@ -7,22 +7,6 @@ interface UseRowProgressOptions {
export function useRowProgress({ duration, onSeek }: UseRowProgressOptions) {
const progressRef = useRef<HTMLDivElement>(null)
const listenersRef = useRef<{ move: ((e: MouseEvent) => void) | null; up: (() => void) | null }>({
move: null,
up: null,
})
const cleanupListeners = useCallback(() => {
const { move, up } = listenersRef.current
if (move) {
document.removeEventListener("mousemove", move)
listenersRef.current.move = null
}
if (up) {
document.removeEventListener("mouseup", up)
listenersRef.current.up = null
}
}, [])
const handleMouseDown = useCallback(
(e: React.MouseEvent<HTMLDivElement>) => {
@@ -32,7 +16,6 @@ export function useRowProgress({ duration, onSeek }: UseRowProgressOptions) {
const doSeek = (ev: MouseEvent) => {
if (!progressRef.current) return
const rect = progressRef.current.getBoundingClientRect()
if (rect.width <= 0) return
const percent = Math.max(0, Math.min(1, (ev.clientX - rect.left) / rect.width))
onSeek(percent * duration)
}
@@ -41,26 +24,15 @@ export function useRowProgress({ duration, onSeek }: UseRowProgressOptions) {
const handleMove = (ev: MouseEvent) => doSeek(ev)
const handleUp = () => {
cleanupListeners()
document.removeEventListener("mousemove", handleMove)
document.removeEventListener("mouseup", handleUp)
}
// 先清理旧的,再添加新的
cleanupListeners()
listenersRef.current.move = handleMove
listenersRef.current.up = handleUp
document.addEventListener("mousemove", handleMove)
document.addEventListener("mouseup", handleUp)
},
[duration, onSeek, cleanupListeners],
[duration, onSeek],
)
// 组件卸载时清理事件监听器
useEffect(() => {
return () => {
cleanupListeners()
}
}, [cleanupListeners])
return { progressRef, handleMouseDown }
}
@@ -65,9 +65,6 @@ import "@/pages/editing-planner/components/filter/FilterPresetGrid"
import "@/pages/editing-planner/components/filter/FilterManualAdjust"
import "@/pages/editing-planner/components/intro-outro/IntroOutroBlock"
import "@/pages/editing-planner/components/subtitle-style/SubtitlePreview"
import "@/pages/editing-planner/components/subtitle-style/SubtitleModeSwitch"
import "@/pages/editing-planner/components/subtitle-style/SubtitlePositionSelector"
import "@/pages/editing-planner/components/subtitle-style/SubtitleEffectButtons"
import "@/pages/editing-planner/components/tts/VoiceSelector"
import "@/pages/editing-planner/components/tts/TtsSlider"
import "@/pages/editing-planner/components/watermark/WatermarkTypeTabs"
@@ -0,0 +1,17 @@
/**
* Tasks 模块 smoke test
* 建立依赖链,确保 vitest related 能匹配到 tasks 目录下的改动
*/
import { describe, it, expect } from "vitest"
import "@/pages/tasks/components/TaskTable"
import "@/pages/tasks/components/task-table/useTaskTableColumns"
import "@/pages/tasks/components/task-table/TaskEmptyState"
import "@/pages/tasks/components/TaskFilterBar"
import "@/pages/tasks/components/TaskErrorDetail"
describe("Tasks module smoke test", () => {
it("should load all task modules", () => {
expect(true).toBe(true)
})
})
@@ -20,8 +20,6 @@ import "@/pages/voice-materials/components/voice-material-card/CardActions"
import "@/pages/voice-materials/components/voice-material-card/BatchCheckbox"
import "@/pages/voice-materials/components/voice-material-card/types"
import "@/pages/voice-materials/components/VoiceMaterialRow"
import "@/pages/voice-materials/components/voice-material-row/useRowProgress"
import "@/pages/voice-materials/components/voice-material-row/TagDisplay"
import "@/pages/voice-materials/components/Toolbar"
import "@/pages/voice-materials/components/TagFilterBar"
import "@/pages/voice-materials/components/BatchBar"
+1 -3
View File
@@ -569,9 +569,7 @@ class CosyVoiceService:
清洗后的 prefix
"""
# 只保留字母和数字
import re
cleaned = re.sub(r"[^a-zA-Z0-9]", "", name)
cleaned = "".join(c for c in name if c.isalnum())
# 最多10字符
cleaned = cleaned[:10]
# 如果清洗后为空,用默认值
@@ -1,496 +0,0 @@
"""模板片段转换器单测.
纯函数模块,覆盖:枚举安全解析、config过滤、
clip→template转换、snapshot双向转换、名称校验。
"""
from __future__ import annotations
from dataclasses import dataclass
from packages.domain.template_clip_config import (
ClipType,
TemplateClipConfig,
TransitionEffect,
)
from packages.domain.template_clip_converter import (
clip_config_to_snapshot,
clip_configs_to_snapshots,
clip_to_template_clip_config,
clips_to_template_clip_configs,
filter_clip_config,
filter_plan_config_to_template,
safe_parse_clip_type,
safe_parse_transition_effect,
snapshot_to_template_clip_config,
snapshots_to_template_clip_configs,
validate_template_name,
)
class TestSafeParseTransitionEffect:
def test_enum_passthrough(self):
result = safe_parse_transition_effect(TransitionEffect.FADE)
assert result == TransitionEffect.FADE
assert isinstance(result, TransitionEffect)
def test_valid_string(self):
result = safe_parse_transition_effect("fade")
assert result == TransitionEffect.FADE
def test_cut_string(self):
result = safe_parse_transition_effect("cut")
assert result == TransitionEffect.CUT
def test_invalid_string_returns_default(self):
result = safe_parse_transition_effect("invalid_effect")
assert result == TransitionEffect.CUT # 默认
def test_invalid_string_custom_default(self):
result = safe_parse_transition_effect("bad", default=TransitionEffect.DISSOLVE)
assert result == TransitionEffect.DISSOLVE
def test_none_returns_default(self):
result = safe_parse_transition_effect(None)
assert result == TransitionEffect.CUT
def test_int_value_returns_default(self):
result = safe_parse_transition_effect(123)
assert result == TransitionEffect.CUT
def test_empty_string_returns_default(self):
result = safe_parse_transition_effect("")
assert result == TransitionEffect.CUT
class TestSafeParseClipType:
def test_enum_passthrough(self):
result = safe_parse_clip_type(ClipType.SUBTITLE)
assert result == ClipType.SUBTITLE
assert isinstance(result, ClipType)
def test_valid_string_main(self):
result = safe_parse_clip_type("main")
assert result == ClipType.MAIN
def test_valid_string_text(self):
result = safe_parse_clip_type("subtitle")
assert result == ClipType.SUBTITLE
def test_invalid_string_returns_default(self):
result = safe_parse_clip_type("unknown_type")
assert result == ClipType.MAIN
def test_invalid_string_custom_default(self):
result = safe_parse_clip_type("bad", default=ClipType.TITLE)
assert result == ClipType.TITLE
def test_none_returns_default(self):
result = safe_parse_clip_type(None)
assert result == ClipType.MAIN
def test_dict_returns_default(self):
result = safe_parse_clip_type({"key": "val"})
assert result == ClipType.MAIN
class TestFilterClipConfig:
def test_none_config(self):
result = filter_clip_config(None)
assert result == {}
def test_empty_dict(self):
result = filter_clip_config({})
assert result == {}
def test_basic_config_passthrough(self):
cfg = {"font_size": 24, "color": "red"}
result = filter_clip_config(cfg)
assert result == {"font_size": 24, "color": "red"}
def test_filters_asset_info(self):
cfg = {"font_size": 24, "asset_info": {"id": "123"}}
result = filter_clip_config(cfg)
assert "asset_info" not in result
assert result["font_size"] == 24
def test_filters_source_asset_id(self):
cfg = {"source_asset_id": "asset_1", "text_key": "hi"}
result = filter_clip_config(cfg)
assert "source_asset_id" not in result
assert result["text_key"] == "hi"
def test_playback_speed_added_when_not_one(self):
result = filter_clip_config({}, playback_speed=1.5)
assert result["playback_speed"] == 1.5
def test_playback_speed_one_not_added(self):
result = filter_clip_config({}, playback_speed=1.0)
assert "playback_speed" not in result
def test_playback_speed_none_not_added(self):
result = filter_clip_config({}, playback_speed=None)
assert "playback_speed" not in result
def test_playback_speed_config_takes_priority(self):
"""clip_config中的playback_speed会覆盖参数传入的(因为update在后面)."""
cfg = {"playback_speed": 0.5, "other": "val"}
result = filter_clip_config(cfg, playback_speed=2.0)
assert result["playback_speed"] == 0.5 # config里的覆盖参数的
assert result["other"] == "val"
def test_custom_skip_keys(self):
cfg = {"keep_me": 1, "drop_me": 2, "also_drop": 3}
skip = frozenset({"drop_me", "also_drop"})
result = filter_clip_config(cfg, skip_keys=skip)
assert result == {"keep_me": 1}
def test_does_not_mutate_input(self):
cfg = {"a": 1, "asset_info": "x"}
original = dict(cfg)
filter_clip_config(cfg)
assert cfg == original # 原dict不变
class TestFilterPlanConfigToTemplate:
def test_none_config(self):
result = filter_plan_config_to_template(None)
assert result == {}
def test_empty_dict(self):
result = filter_plan_config_to_template({})
assert result == {}
def test_keeps_template_fields(self):
cfg = {"title": "My Template", "aspect_ratio": "9:16"}
result = filter_plan_config_to_template(cfg)
assert result == cfg
def test_filters_runtime_fields(self):
cfg = {
"title": "T",
"is_template_draft": True,
"asset_ids": ["a1"],
"source_edit_plan_id": "ep1",
"generation_task_id": "gt1",
}
result = filter_plan_config_to_template(cfg)
assert "is_template_draft" not in result
assert "asset_ids" not in result
assert "source_edit_plan_id" not in result
assert "generation_task_id" not in result
assert result["title"] == "T"
def test_custom_skip_keys(self):
cfg = {"keep": 1, "skip_a": 2, "skip_b": 3}
skip = frozenset({"skip_a", "skip_b"})
result = filter_plan_config_to_template(cfg, skip_keys=skip)
assert result == {"keep": 1}
class TestClipToTemplateClipConfig:
@dataclass
class FakeClip:
clip_type: str = "main"
order: int = 0
duration: float = 5.0
text_content: str = ""
transition_effect: str = "cut"
playback_speed: float | None = None
config: dict | None = None
def test_basic_conversion(self):
clip = self.FakeClip(
clip_type="subtitle",
order=2,
duration=3.5,
text_content="Hello",
transition_effect="fade",
)
result = clip_to_template_clip_config("tpl_1", clip)
assert isinstance(result, TemplateClipConfig)
assert result.template_id == "tpl_1"
assert result.clip_type == ClipType.SUBTITLE
assert result.order == 2
assert result.min_duration == 3.5
assert result.max_duration == 3.5
assert result.text_template == "Hello"
assert result.transition_effect == TransitionEffect.FADE
def test_duration_fixed_min_max_equal(self):
"""转换后 min_duration == max_duration == clip.duration."""
clip = self.FakeClip(duration=7.2)
result = clip_to_template_clip_config("t1", clip)
assert result.min_duration == 7.2
assert result.max_duration == 7.2
def test_zero_duration(self):
clip = self.FakeClip(duration=0.0)
result = clip_to_template_clip_config("t1", clip)
assert result.min_duration == 0.0
assert result.max_duration == 0.0
def test_none_duration_defaults_to_zero(self):
clip = self.FakeClip()
clip.duration = None # type: ignore
result = clip_to_template_clip_config("t1", clip)
assert result.min_duration == 0.0
assert result.max_duration == 0.0
def test_empty_text_content_becomes_empty_string(self):
clip = self.FakeClip(text_content="")
result = clip_to_template_clip_config("t1", clip)
assert result.text_template == ""
def test_none_text_content_becomes_empty_string(self):
clip = self.FakeClip()
clip.text_content = None # type: ignore
result = clip_to_template_clip_config("t1", clip)
assert result.text_template == ""
def test_playback_speed_in_config(self):
clip = self.FakeClip(playback_speed=1.5, config={"font": "bold"})
result = clip_to_template_clip_config("t1", clip)
assert result.config["playback_speed"] == 1.5
assert result.config["font"] == "bold"
def test_playback_speed_one_not_in_config(self):
clip = self.FakeClip(playback_speed=1.0)
result = clip_to_template_clip_config("t1", clip)
assert "playback_speed" not in result.config
def test_config_asset_info_filtered(self):
clip = self.FakeClip(config={"text_key": "hi", "asset_info": {"id": "a"}})
result = clip_to_template_clip_config("t1", clip)
assert "asset_info" not in result.config
assert result.config["text_key"] == "hi"
def test_invalid_clip_type_falls_back(self):
clip = self.FakeClip(clip_type="invalid_type")
result = clip_to_template_clip_config("t1", clip)
assert result.clip_type == ClipType.MAIN
def test_missing_attributes(self):
"""对象没有某些属性时使用默认值."""
class MinimalClip:
pass
result = clip_to_template_clip_config("t1", MinimalClip())
assert result.clip_type == ClipType.MAIN
assert result.order == 0
assert result.min_duration == 0.0
assert result.text_template == ""
assert result.transition_effect == TransitionEffect.CUT
class TestClipsToTemplateClipConfigs:
def test_empty_list(self):
result = clips_to_template_clip_configs("t1", [])
assert result == []
def test_multiple_clips(self):
clip_a = TestClipToTemplateClipConfig.FakeClip(clip_type="subtitle", order=0, duration=3.0, text_content="A")
clip_b = TestClipToTemplateClipConfig.FakeClip(clip_type="title", order=1, duration=5.0, text_content="")
result = clips_to_template_clip_configs("t1", [clip_a, clip_b])
assert len(result) == 2
assert result[0].clip_type == ClipType.SUBTITLE
assert result[0].order == 0
assert result[1].clip_type == ClipType.TITLE
assert result[1].order == 1
assert all(isinstance(r, TemplateClipConfig) for r in result)
class TestClipConfigToSnapshot:
def test_basic_snapshot(self):
cfg = TemplateClipConfig.create(
template_id="t1",
clip_type=ClipType.SUBTITLE,
order=2,
min_duration=3.0,
max_duration=5.0,
text_template="Hello",
transition_effect=TransitionEffect.FADE,
config={"font_size": 20},
)
snap = clip_config_to_snapshot(cfg)
assert snap["clip_type"] == "subtitle"
assert snap["order"] == 2
assert snap["min_duration"] == 3.0
assert snap["max_duration"] == 5.0
assert snap["text_template"] == "Hello"
assert snap["transition_effect"] == "fade"
assert snap["config"] == {"font_size": 20}
def test_enum_values_are_strings(self):
cfg = TemplateClipConfig.create(template_id="t1", clip_type=ClipType.MAIN, order=0)
snap = clip_config_to_snapshot(cfg)
assert snap["clip_type"] == "main"
assert isinstance(snap["clip_type"], str)
assert snap["transition_effect"] == "cut"
assert isinstance(snap["transition_effect"], str)
def test_config_is_copy_not_reference(self):
config = {"key": "val"}
cfg = TemplateClipConfig.create(template_id="t1", clip_type=ClipType.MAIN, order=0, config=config)
snap = clip_config_to_snapshot(cfg)
snap["config"]["key"] = "changed"
assert config["key"] == "val" # 原config不变
def test_empty_config(self):
cfg = TemplateClipConfig.create(template_id="t1", clip_type=ClipType.MAIN, order=0, config={})
snap = clip_config_to_snapshot(cfg)
assert snap["config"] == {}
def test_none_text_becomes_empty(self):
cfg = TemplateClipConfig.create(template_id="t1", clip_type=ClipType.MAIN, order=0)
cfg.text_template = None # type: ignore
snap = clip_config_to_snapshot(cfg)
assert snap["text_template"] == ""
class TestClipConfigsToSnapshots:
def test_empty_list(self):
assert clip_configs_to_snapshots([]) == []
def test_multiple_configs(self):
cfg1 = TemplateClipConfig.create(
template_id="t1",
clip_type=ClipType.SUBTITLE,
order=0,
min_duration=2.0,
max_duration=2.0,
)
cfg2 = TemplateClipConfig.create(
template_id="t1",
clip_type=ClipType.TITLE,
order=1,
min_duration=3.0,
max_duration=3.0,
)
snaps = clip_configs_to_snapshots([cfg1, cfg2])
assert len(snaps) == 2
assert snaps[0]["clip_type"] == "subtitle"
assert snaps[1]["clip_type"] == "title"
class TestSnapshotToTemplateClipConfig:
def test_basic_conversion(self):
snap = {
"clip_type": "subtitle",
"order": 3,
"min_duration": 2.5,
"max_duration": 4.5,
"text_template": "World",
"transition_effect": "dissolve",
"config": {"color": "blue"},
}
result = snapshot_to_template_clip_config("tpl_2", snap)
assert isinstance(result, TemplateClipConfig)
assert result.template_id == "tpl_2"
assert result.clip_type == ClipType.SUBTITLE
assert result.order == 3
assert result.min_duration == 2.5
assert result.max_duration == 4.5
assert result.text_template == "World"
assert result.transition_effect == TransitionEffect.DISSOLVE
assert result.config == {"color": "blue"}
def test_empty_snapshot_uses_defaults(self):
result = snapshot_to_template_clip_config("t1", {})
assert result.clip_type == ClipType.MAIN
assert result.order == 0
assert result.min_duration == 0.0
assert result.max_duration == 0.0
assert result.text_template == ""
assert result.transition_effect == TransitionEffect.CUT
assert result.config == {}
def test_invalid_clip_type_defaults(self):
snap = {"clip_type": "unknown"}
result = snapshot_to_template_clip_config("t1", snap)
assert result.clip_type == ClipType.MAIN
def test_invalid_transition_defaults(self):
snap = {"transition_effect": "bad_effect"}
result = snapshot_to_template_clip_config("t1", snap)
assert result.transition_effect == TransitionEffect.CUT
def test_none_config_becomes_empty(self):
snap = {"config": None}
result = snapshot_to_template_clip_config("t1", snap)
assert result.config == {}
class TestSnapshotsToTemplateClipConfigs:
def test_empty_list(self):
result = snapshots_to_template_clip_configs("t1", [])
assert result == []
def test_multiple_snapshots(self):
snaps = [
{"clip_type": "subtitle", "order": 0, "text_template": "A"},
{"clip_type": "title", "order": 1},
]
result = snapshots_to_template_clip_configs("t1", snaps)
assert len(result) == 2
assert result[0].clip_type == ClipType.SUBTITLE
assert result[0].text_template == "A"
assert result[1].clip_type == ClipType.TITLE
class TestRoundTrip:
"""clip → config → snapshot → config 双向转换一致性."""
def test_snapshot_config_round_trip(self):
original = TemplateClipConfig.create(
template_id="t1",
clip_type=ClipType.SUBTITLE,
order=5,
min_duration=3.0,
max_duration=6.0,
text_template="Round trip",
transition_effect=TransitionEffect.FADE,
config={"key": "value"},
)
snap = clip_config_to_snapshot(original)
restored = snapshot_to_template_clip_config("t1", snap)
assert restored.clip_type == original.clip_type
assert restored.order == original.order
assert restored.min_duration == original.min_duration
assert restored.max_duration == original.max_duration
assert restored.text_template == original.text_template
assert restored.transition_effect == original.transition_effect
assert restored.config == original.config
class TestValidateTemplateName:
def test_valid_name(self):
assert validate_template_name("我的模板") == "我的模板"
def test_strips_whitespace(self):
assert validate_template_name(" Hello ") == "Hello"
def test_empty_string_raises(self):
try:
validate_template_name("")
except ValueError as e:
assert "不能为空" in str(e)
else:
raise AssertionError("expected ValueError")
def test_whitespace_only_raises(self):
try:
validate_template_name(" ")
except ValueError as e:
assert "不能为空" in str(e)
else:
raise AssertionError("expected ValueError")
def test_none_raises(self):
try:
validate_template_name(None)
except ValueError as e:
assert "不能为空" in str(e)
else:
raise AssertionError("expected ValueError")
+289 -334
View File
@@ -1,12 +1,12 @@
"""concat_engine_pure 单元测试."""
"""视频拼接引擎纯逻辑单元测试."""
from pathlib import Path
from __future__ import annotations
from apps.worker.video_processing.concat_engine_pure import (
import pytest
from video_processing.concat_engine_pure import (
build_concat_filter,
build_fps_filter,
build_scale_pad_filter,
build_setpts_filter,
build_single_segment_filter_chain,
calculate_scaled_size,
can_use_stream_copy,
@@ -20,203 +20,200 @@ from apps.worker.video_processing.concat_engine_pure import (
validate_video_path,
)
# ── parse_fps ────────────────────────────────────────────────────────────────
# ─────────────────────────────────────────────────────────────────────────────
# 帧率解析测试
# ─────────────────────────────────────────────────────────────────────────────
class TestParseFps:
def test_none_returns_default(self):
assert parse_fps(None) == 30.0
"""parse_fps 测试."""
def test_integer_value(self):
def test_integer_fps(self):
"""整数帧率."""
assert parse_fps(30) == 30.0
assert parse_fps(24) == 24.0
def test_float_value(self):
assert parse_fps(29.97) == 29.97
def test_float_fps(self):
"""浮点帧率."""
assert parse_fps(29.97) == pytest.approx(29.97)
def test_string_integer(self):
"""字符串整数."""
assert parse_fps("30") == 30.0
assert parse_fps(" 60 ") == 60.0 # 带空格
def test_string_fraction(self):
"""分数字符串(30/1."""
assert parse_fps("30/1") == 30.0
assert abs(parse_fps("24000/1001") - 23.976) < 0.01
def test_zero_denominator(self):
assert parse_fps("30/0") == 30.0
def test_fraction_24000_1001(self):
"""23.976 帧率."""
result = parse_fps("24000/1001")
assert result == pytest.approx(23.976, rel=0.01)
def test_none_input(self):
"""None 输入返回默认值."""
assert parse_fps(None) == 30.0
def test_empty_string(self):
"""空字符串返回默认值."""
assert parse_fps("") == 30.0
assert parse_fps(" ") == 30.0
def test_invalid_string(self):
"""无效字符串."""
assert parse_fps("abc") == 30.0
assert parse_fps("30fps") == 30.0
def test_zero_denominator(self):
"""分母为 0."""
assert parse_fps("30/0") == 30.0
def test_negative_fps(self):
"""负帧率."""
assert parse_fps(-30) == -30.0
def test_zero_fps(self):
assert parse_fps(0) == 0.0
# ── format_fps_filter ───────────────────────────────────────────────────────
class TestFormatFpsFilter:
"""format_fps_filter 测试."""
def test_integer_fps(self):
"""整数帧率."""
assert format_fps_filter(30.0) == "fps=30"
def test_near_integer_fps(self):
# 接近整数时用整数形式(注意:int(fps)是截断不是四舍五入)
assert format_fps_filter(30.0001) == "fps=30"
assert format_fps_filter(30.0005) == "fps=30" # int(30.0005)=30
def test_non_integer_fps(self):
result = format_fps_filter(23.976)
assert result.startswith("fps=")
assert "23.976" in result
def test_float_precision(self):
def test_float_fps(self):
"""浮点帧率."""
result = format_fps_filter(29.97)
assert result.startswith("fps=")
# 三位小数
parts = result.split("=")[1]
assert len(parts.split(".")[1]) == 3
assert "29.97" in result
def test_one_fps(self):
assert format_fps_filter(1.0) == "fps=1"
def test_near_integer(self):
"""接近整数."""
assert format_fps_filter(30.0001) == "fps=30"
# ── resolve_output_params ───────────────────────────────────────────────────
# ─────────────────────────────────────────────────────────────────────────────
# 输出参数计算测试
# ─────────────────────────────────────────────────────────────────────────────
class TestResolveOutputParams:
def test_config_specified(self):
"""resolve_output_params 测试."""
def test_all_specified(self):
"""全部显式指定."""
w, h, fps = resolve_output_params(1920, 1080, 60.0)
assert w == 1920
assert h == 1080
assert fps == 60.0
def test_fallback_to_first_video_info(self):
def test_no_specified_use_defaults(self):
"""全部未指定,用默认值."""
w, h, fps = resolve_output_params(0, 0, 0)
assert w == 1080
assert h == 1920
assert fps == 30.0
def test_use_first_video_info(self):
"""用第一段视频信息."""
info = {"width": 1280, "height": 720, "r_frame_rate": "24/1"}
w, h, fps = resolve_output_params(0, 0, 0, info)
assert w == 1280
assert h == 720
assert fps == 24.0
def test_fallback_to_defaults(self):
w, h, fps = resolve_output_params(0, 0, 0)
assert w == 1080 # default_width
assert h == 1920 # default_height
assert fps == 30.0
def test_partial_config(self):
# 宽度配置了,高度和帧率用探测的
def test_partial_specified(self):
"""部分指定,未指定的用探测值."""
info = {"width": 1280, "height": 720, "r_frame_rate": "24/1"}
w, h, fps = resolve_output_params(1920, 0, 0, info)
assert w == 1920
assert h == 720
assert w == 1920 # 指定的
assert h == 720 # 探测的
assert fps == 24.0
def test_custom_defaults(self):
w, h, fps = resolve_output_params(
0,
0,
0,
default_width=640,
default_height=480,
default_fps=25.0,
)
assert w == 640
assert h == 480
assert fps == 25.0
def test_minimum_size(self):
w, h, fps = resolve_output_params(0, 0, 0, {"width": 0, "height": 0, "r_frame_rate": "0/1"})
def test_zero_size_clamped(self):
"""零尺寸被钳制."""
w, h, fps = resolve_output_params(0, 0, 0, {})
assert w >= 1
assert h >= 1
assert fps >= 1.0
def test_fps_fraction_in_info(self):
info = {"width": 1920, "height": 1080, "r_frame_rate": "24000/1001"}
_, _, fps = resolve_output_params(0, 0, 0, info)
assert abs(fps - 23.976) < 0.01
# ── calculate_scaled_size ───────────────────────────────────────────────────
def test_custom_defaults(self):
"""自定义默认值."""
w, h, fps = resolve_output_params(0, 0, 0, None, 640, 480, 25.0)
assert w == 640
assert h == 480
assert fps == 25.0
class TestCalculateScaledSize:
"""calculate_scaled_size 测试."""
def test_same_ratio(self):
"""比例相同."""
sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 1920, 1080)
assert sw == 1920
assert sh == 1080
assert ox == 0
assert oy == 0
def test_wider_source_pad_top_bottom(self):
# 源是16:9,目标是9:16竖屏 → 上下填黑边
def test_wider_source(self):
"""源更宽,上下填黑边."""
sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 1080, 1920)
assert sw == 1080 # 以宽度为准
assert sh == 607 # 1080 * 1080 / 1920 = 607.5 → 607
assert sh < 1920 # 高度按比例
assert ox == 0
assert oy > 0 # 垂直居中
def test_taller_source_pad_left_right(self):
# 源是9:16竖屏,目标是16:9横屏 → 左右填黑边
def test_taller_source(self):
"""源更高,左右填黑边."""
sw, sh, ox, oy = calculate_scaled_size(1080, 1920, 1920, 1080)
assert sh == 1080 # 以高度为准
assert sw == 607 # 1080 * 1080 / 1920 = 607.5 → 607
assert sw < 1920 # 宽度按比例
assert ox > 0 # 水平居中
assert oy == 0
def test_zero_source_size(self):
sw, sh, ox, oy = calculate_scaled_size(0, 0, 1920, 1080)
assert sw == 1920
assert sh == 1080
def test_zero_source(self):
"""零尺寸源."""
sw, sh, ox, oy = calculate_scaled_size(0, 0, 100, 100)
assert sw == 100
assert sh == 100
def test_scale_down(self):
"""缩小."""
sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 640, 360)
assert sw == 640
assert sh == 360
assert ox == 0
assert oy == 0
def test_negative_source_size(self):
sw, sh, ox, oy = calculate_scaled_size(-1, -1, 1920, 1080)
assert sw == 1920
assert sh == 1080
assert ox == 0
assert oy == 0
def test_target_same_ratio_different_size(self):
# 比例相同,尺寸不同 → 直接缩放到目标大小
def test_scale_up(self):
"""放大."""
sw, sh, ox, oy = calculate_scaled_size(640, 360, 1920, 1080)
assert sw == 1920
assert sh == 1080
assert ox == 0
assert oy == 0
# ── can_use_stream_copy ─────────────────────────────────────────────────────
# ─────────────────────────────────────────────────────────────────────────────
# stream copy 判断测试
# ─────────────────────────────────────────────────────────────────────────────
class TestCanUseStreamCopy:
def test_force_reencode_false(self):
assert can_use_stream_copy([], 1920, 1080, 30.0, force_reencode=True) is False
"""can_use_stream_copy 测试."""
def test_empty_segments(self):
assert can_use_stream_copy([], 1920, 1080, 30.0) is False
def test_single_segment_matching_params(self):
segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"}]
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is True
def test_multiple_segments_same_params(self):
def test_identical_segments(self):
"""所有段参数相同,可以 stream copy."""
segs = [
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
]
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is True
def test_force_reencode(self):
"""强制重编码."""
segs = [
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
]
assert can_use_stream_copy(segs, 1920, 1080, 30.0, force_reencode=True) is False
def test_different_codec(self):
"""编码不同."""
segs = [
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
{"codec_name": "hevc", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
@@ -224,6 +221,7 @@ class TestCanUseStreamCopy:
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
def test_different_resolution(self):
"""分辨率不同."""
segs = [
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
{"codec_name": "h264", "width": 1280, "height": 720, "r_frame_rate": "30/1"},
@@ -231,349 +229,306 @@ class TestCanUseStreamCopy:
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
def test_different_fps(self):
"""帧率不同."""
segs = [
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "60/1"},
]
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
def test_target_differs_from_source(self):
segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"}]
# 目标分辨率不同
def test_target_differs(self):
"""目标参数与源不同."""
segs = [
{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
]
assert can_use_stream_copy(segs, 1280, 720, 30.0) is False
# 目标帧率不同
assert can_use_stream_copy(segs, 1920, 1080, 60.0) is False
def test_fps_fraction_match(self):
segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "24000/1001"}]
assert can_use_stream_copy(segs, 1920, 1080, 23.976) is True
def test_empty_segments(self):
"""空列表."""
assert can_use_stream_copy([], 1920, 1080, 30.0) is False
def test_single_segment(self):
"""单段."""
segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"}]
assert can_use_stream_copy(segs, 1920, 1080, 30.0) is True
# ── generate_concat_file_list ───────────────────────────────────────────────
# ─────────────────────────────────────────────────────────────────────────────
# 文件列表生成测试
# ─────────────────────────────────────────────────────────────────────────────
class TestGenerateConcatFileList:
"""generate_concat_file_list 测试."""
def test_single_file(self):
result = generate_concat_file_list(["/tmp/video.mp4"])
assert result == "file '/tmp/video.mp4'\n"
"""单个文件."""
result = generate_concat_file_list(["/a.mp4"])
assert "file '/a.mp4'" in result
assert result.endswith("\n")
def test_multiple_files(self):
"""多个文件."""
result = generate_concat_file_list(["/a.mp4", "/b.mp4", "/c.mp4"])
lines = result.strip().split("\n")
assert len(lines) == 3
assert lines[0] == "file '/a.mp4'"
assert lines[1] == "file '/b.mp4'"
assert lines[2] == "file '/c.mp4'"
assert result.endswith("\n")
def test_escapes_single_quotes(self):
result = generate_concat_file_list(["/path/with'quote.mp4"])
# 单引号转义: '\''
assert "'\\''" in result
def test_empty_list(self):
"""空列表."""
result = generate_concat_file_list([])
assert result == "\n"
def test_path_with_single_quote(self):
"""路径包含单引号(转义)."""
result = generate_concat_file_list(["/path/to/file's.mp4"])
# 单引号应该被转义
assert "'\\''" in result or file
assert "file '" in result
def test_path_with_spaces(self):
result = generate_concat_file_list(["/path/to/video file.mp4"])
assert "file '/path/to/video file.mp4'" in result
"""路径包含空格."""
result = generate_concat_file_list(["/path/to/my video.mp4"])
assert "my video" in result
# ── build_scale_pad_filter ──────────────────────────────────────────────────
# ─────────────────────────────────────────────────────────────────────────────
# 滤镜构建测试
# ─────────────────────────────────────────────────────────────────────────────
class TestBuildScalePadFilter:
def test_basic_filter(self):
"""scale+pad 滤镜测试."""
def test_contains_scale(self):
"""包含 scale."""
result = build_scale_pad_filter(1920, 1080)
assert "scale=" in result
def test_contains_pad(self):
"""包含 pad."""
result = build_scale_pad_filter(1920, 1080)
assert "pad=" in result
assert "1920:1080" in result
def test_force_original_aspect_ratio(self):
"""保持宽高比."""
result = build_scale_pad_filter(1920, 1080)
assert "scale=1920:1080" in result
assert "force_original_aspect_ratio=decrease" in result
assert "pad=1920:1080" in result
assert "black" in result
assert "(ow-iw)/2" in result
assert "(oh-ih)/2" in result
def test_different_resolution(self):
result = build_scale_pad_filter(1080, 1920)
assert "scale=1080:1920" in result
assert "pad=1080:1920" in result
def test_ignores_source_size(self):
# src_w/src_h 目前不影响输出,都是用表达式
result1 = build_scale_pad_filter(1920, 1080)
result2 = build_scale_pad_filter(1920, 1080, src_w=1280, src_h=720)
assert result1 == result2
# ── build_fps_filter ────────────────────────────────────────────────────────
def test_black_padding(self):
"""黑边填充."""
result = build_scale_pad_filter(1920, 1080)
assert ":black" in result
class TestBuildFpsFilter:
"""fps 滤镜测试."""
def test_integer_fps(self):
"""整数帧率."""
assert build_fps_filter(30.0) == "fps=30"
def test_float_fps(self):
"""浮点帧率."""
result = build_fps_filter(29.97)
assert result.startswith("fps=")
# ── build_setpts_filter ─────────────────────────────────────────────────────
class TestBuildSetptsFilter:
def test_returns_correct_string(self):
assert build_setpts_filter() == "setpts=PTS-STARTPTS"
# ── build_concat_filter ─────────────────────────────────────────────────────
class TestBuildConcatFilter:
def test_zero_inputs(self):
assert build_concat_filter(0) == ""
"""concat 滤镜测试."""
def test_single_input_with_audio(self):
result = build_concat_filter(1)
assert "[0:v][0:a]" in result
assert "concat=n=1:v=1:a=1" in result
def test_two_inputs_with_audio(self):
"""两路输入,有音频."""
result = build_concat_filter(2, has_audio=True)
assert "[0:v][0:a][1:v][1:a]concat=n=2:v=1:a=1" in result
assert "[concat_v][concat_a]" in result
def test_single_input_no_audio(self):
result = build_concat_filter(1, has_audio=False)
assert "[0:v]" in result
assert "concat=n=1:v=1:a=0" in result
assert "[concat_v]" in result
assert "[concat_a]" not in result
def test_multiple_inputs_with_audio(self):
result = build_concat_filter(3)
assert "[0:v][0:a][1:v][1:a][2:v][2:a]" in result
assert "concat=n=3:v=1:a=1" in result
def test_multiple_inputs_no_audio(self):
def test_three_inputs_video_only(self):
"""三路输入,无音频."""
result = build_concat_filter(3, has_audio=False)
assert "[0:v][1:v][2:v]" in result
assert "concat=n=3:v=1:a=0" in result
assert "[0:v][1:v][2:v]concat=n=3:v=1:a=0" in result
assert "[concat_v]" in result
def test_negative_inputs(self):
assert build_concat_filter(-1) == ""
def test_single_input(self):
"""单路输入."""
result = build_concat_filter(1, has_audio=True)
assert "[0:v][0:a]concat=n=1:v=1:a=1" in result
# ── build_single_segment_filter_chain ───────────────────────────────────────
def test_zero_inputs(self):
"""零输入."""
assert build_concat_filter(0) == ""
class TestBuildSingleSegmentFilterChain:
"""单段滤镜链测试."""
def test_with_audio(self):
"""有音频."""
result = build_single_segment_filter_chain(1920, 1080, 30.0, 0)
# 视频链
assert "[0:v]" in result
assert "[v0]" in result
assert "scale=1920:1080" in result
assert "fps=30" in result
assert "scale=" in result
assert "fps=" in result
assert "setpts=PTS-STARTPTS" in result
# 音频链
assert "[0:a]" in result
assert "[a0]" in result
assert "asetpts=PTS-STARTPTS" in result
# 用分号分隔
assert ";" in result
assert "[v0]" in result
assert "[a0]" in result
def test_without_audio(self):
result = build_single_segment_filter_chain(1920, 1080, 30.0, 2, has_audio=False)
assert "[2:v]" in result
assert "[v2]" in result
assert "[2:a]" not in result
assert ";" not in result # 没有音频就没有分号
def test_video_only(self):
"""无音频."""
result = build_single_segment_filter_chain(1920, 1080, 30.0, 1, has_audio=False)
assert "scale=" in result
assert "setpts=" in result
assert "asetpts" not in result
assert "[v1]" in result
def test_segment_index_propagated(self):
for idx in [0, 5, 10]:
result = build_single_segment_filter_chain(1920, 1080, 30.0, idx)
assert f"[{idx}:v]" in result
assert f"[v{idx}]" in result
def test_segment_index_in_labels(self):
"""段索引在标签中."""
result = build_single_segment_filter_chain(1920, 1080, 30.0, 5)
assert "[5:v]" in result
assert "[v5]" in result
# ── validate_concat_config ──────────────────────────────────────────────────
# ─────────────────────────────────────────────────────────────────────────────
# 配置验证测试
# ─────────────────────────────────────────────────────────────────────────────
class TestValidateConcatConfig:
"""配置验证测试."""
def test_valid_config(self):
"""合法配置."""
config = {
"segments": [
{"video_path": "/a.mp4"},
{"video_path": "/b.mp4"},
],
"segments": [{"video_path": "/a.mp4"}, {"video_path": "/b.mp4"}],
"output_width": 1920,
"output_height": 1080,
"output_fps": 30,
}
valid, errors = validate_concat_config(config)
assert valid is True
assert errors == []
def test_no_segments(self):
valid, errors = validate_concat_config({})
assert valid is False
assert any("至少需要一个" in e for e in errors)
ok, errors = validate_concat_config(config)
assert ok is True
assert len(errors) == 0
def test_empty_segments(self):
valid, errors = validate_concat_config({"segments": []})
assert valid is False
assert len(errors) >= 1
"""空段列表."""
ok, errors = validate_concat_config({"segments": []})
assert ok is False
assert any("至少需要" in e or "视频段" in e for e in errors)
def test_missing_video_path(self):
config = {"segments": [{"video_path": ""}]}
valid, errors = validate_concat_config(config)
assert valid is False
"""缺少 video_path."""
config = {"segments": [{"video_path": "/a.mp4"}, {}]}
ok, errors = validate_concat_config(config)
assert ok is False
assert any("video_path" in e for e in errors)
def test_multiple_missing_paths(self):
config = {
"segments": [
{"video_path": "/a.mp4"},
{"video_path": ""},
{"video_path": ""},
]
}
valid, errors = validate_concat_config(config)
assert valid is False
path_errors = [e for e in errors if "video_path" in e]
assert len(path_errors) == 2
def test_negative_output_width(self):
config = {"segments": [{"video_path": "/a.mp4"}], "output_width": -1}
valid, errors = validate_concat_config(config)
assert valid is False
def test_negative_width(self):
"""负宽度."""
config = {"segments": [{"video_path": "/a.mp4"}], "output_width": -100}
ok, errors = validate_concat_config(config)
assert ok is False
assert any("output_width" in e for e in errors)
def test_negative_output_height(self):
config = {"segments": [{"video_path": "/a.mp4"}], "output_height": -1}
valid, errors = validate_concat_config(config)
assert valid is False
def test_negative_height(self):
"""负高度."""
config = {"segments": [{"video_path": "/a.mp4"}], "output_height": -100}
ok, errors = validate_concat_config(config)
assert ok is False
assert any("output_height" in e for e in errors)
def test_negative_output_fps(self):
config = {"segments": [{"video_path": "/a.mp4"}], "output_fps": -1}
valid, errors = validate_concat_config(config)
assert valid is False
def test_negative_fps(self):
"""负帧率."""
config = {"segments": [{"video_path": "/a.mp4"}], "output_fps": -30}
ok, errors = validate_concat_config(config)
assert ok is False
assert any("output_fps" in e for e in errors)
def test_zero_output_params_valid(self):
# 0值表示未指定,是合法的
config = {
"segments": [{"video_path": "/a.mp4"}],
"output_width": 0,
"output_height": 0,
"output_fps": 0,
}
valid, errors = validate_concat_config(config)
assert valid is True
def test_zero_output_params_ok(self):
"""零输出参数合法(表示自动探测)."""
config = {"segments": [{"video_path": "/a.mp4"}]}
ok, errors = validate_concat_config(config)
assert ok is True
# ── validate_video_path ─────────────────────────────────────────────────────
# ─────────────────────────────────────────────────────────────────────────────
# 路径验证测试
# ─────────────────────────────────────────────────────────────────────────────
class TestValidateVideoPath:
"""视频路径验证测试."""
def test_empty_path(self):
valid, err = validate_video_path("", "/work")
assert valid is False
assert "不能为空" in err
"""空路径."""
ok, msg = validate_video_path("", "/work")
assert ok is False
assert "不能为空" in msg
def test_relative_path_valid(self):
valid, err = validate_video_path("video.mp4", "/work")
assert valid is True
assert err == ""
def test_path_traversal(self):
"""路径遍历."""
ok, msg = validate_video_path("../etc/passwd", "/work")
assert ok is False
assert "回溯" in msg or ".." in msg
def test_relative_path_with_subdir(self):
valid, err = validate_video_path("sub/video.mp4", "/work")
assert valid is True
def test_valid_relative_path(self):
"""相对路径(不检查边界)."""
ok, msg = validate_video_path("video.mp4", "/work")
assert ok is True
def test_path_traversal_rejected(self):
valid, err = validate_video_path("../secret.mp4", "/work")
assert valid is False
assert ".." in err
def test_valid_absolute_path(self):
"""绝对路径在工作目录内."""
ok, msg = validate_video_path("/work/sub/video.mp4", "/work")
assert ok is True
def test_nested_path_traversal_rejected(self):
valid, err = validate_video_path("sub/../../secret.mp4", "/work")
assert valid is False
def test_absolute_path_inside_workdir(self):
valid, err = validate_video_path("/work/sub/video.mp4", "/work")
assert valid is True
def test_absolute_path_outside_workdir(self):
valid, err = validate_video_path("/etc/passwd", "/work")
assert valid is False
assert "工作目录内" in err
def test_path_object_input(self):
valid, err = validate_video_path(Path("video.mp4"), Path("/work"))
assert valid is True
def test_path_outside_work_dir(self):
"""路径在工作目录外."""
ok, msg = validate_video_path("/etc/passwd", "/work")
assert ok is False
assert "工作目录" in msg
# ── estimate_total_duration ─────────────────────────────────────────────────
# ─────────────────────────────────────────────────────────────────────────────
# 工具函数测试
# ─────────────────────────────────────────────────────────────────────────────
class TestEstimateTotalDuration:
def test_single_segment(self):
assert estimate_total_duration([{"duration": 10.5}]) == 10.5
"""总时长估算测试."""
def test_multiple_segments(self):
segs = [
{"duration": 10},
{"duration": 20.5},
{"duration": 5.5},
]
assert estimate_total_duration(segs) == 36.0
"""多段视频."""
segs = [{"duration": 10}, {"duration": 20.5}, {"duration": 5}]
assert estimate_total_duration(segs) == pytest.approx(35.5)
def test_empty_list(self):
"""空列表."""
assert estimate_total_duration([]) == 0.0
def test_missing_duration_field(self):
segs = [{"path": "a.mp4"}, {"duration": 10}]
assert estimate_total_duration(segs) == 10.0
def test_invalid_duration_skipped(self):
segs = [
{"duration": 10},
{"duration": "abc"},
{"duration": 20},
]
assert estimate_total_duration(segs) == 30.0
"""无效时长跳过."""
segs = [{"duration": 10}, {"duration": "abc"}, {"duration": 20}]
assert estimate_total_duration(segs) == pytest.approx(30.0)
def test_string_duration(self):
segs = [{"duration": "15.5"}]
assert estimate_total_duration(segs) == 15.5
def test_negative_duration(self):
segs = [{"duration": -5}]
assert estimate_total_duration(segs) == -5.0
# ── count_valid_segments ────────────────────────────────────────────────────
def test_missing_duration(self):
"""缺 duration 字段."""
segs = [{}, {"duration": 10}]
assert estimate_total_duration(segs) == pytest.approx(10.0)
class TestCountValidSegments:
"""有效段统计测试."""
def test_all_valid(self):
segs = [
{"video_path": "/a.mp4"},
{"video_path": "/b.mp4"},
]
"""全部有效."""
segs = [{"video_path": "/a.mp4"}, {"video_path": "/b.mp4"}]
assert count_valid_segments(segs) == 2
def test_some_invalid(self):
segs = [
{"video_path": "/a.mp4"},
{"video_path": ""},
{"video_path": "/c.mp4"},
]
assert count_valid_segments(segs) == 2
def test_none_valid(self):
segs = [
{"video_path": ""},
{"other_field": "x"},
]
assert count_valid_segments(segs) == 0
"""部分无效."""
segs = [{"video_path": "/a.mp4"}, {}, {"video_path": ""}]
assert count_valid_segments(segs) == 1
def test_empty_list(self):
"""空列表."""
assert count_valid_segments([]) == 0
File diff suppressed because it is too large Load Diff
-453
View File
@@ -1,453 +0,0 @@
"""shared.ai_service 单元测试.
主要测试纯逻辑部分:_parse_recommend_response / _fallback_recommend_clips / _call_ai_cover_service.
"""
from __future__ import annotations
import json
from unittest.mock import patch
import pytest
from shared.ai_service import (
_call_ai_cover_service,
_fallback_recommend_clips,
_parse_recommend_response,
)
# ── _parse_recommend_response 测试 ────────────────────────────────────────
class TestParseRecommendResponseBasic:
"""基础解析测试."""
def test_parse_valid_json(self):
content = json.dumps(
{
"clips": [
{
"clip_type": "intro",
"order": 0,
"text_content": "开场",
"duration": 3.0,
"transition_effect": "fade",
"asset_id": "asset1",
"start_time": 0.0,
"config": {},
},
{
"clip_type": "outro",
"order": 1,
"text_content": "结尾",
"duration": 2.0,
"transition_effect": "fade",
"asset_id": "",
"start_time": 0.0,
"config": {},
},
],
"title": "测试视频",
"confidence": 0.85,
}
)
result = _parse_recommend_response(content, ["asset1"], 30.0)
assert result is not None
assert len(result["clips"]) == 2
assert result["confidence"] == 0.85
assert result["total_duration"] == 5.0
assert result["config"]["title"]["text"] == "测试视频"
assert result["config"]["title"]["ai_auto"] is True
def test_parse_none_returns_none(self):
result = _parse_recommend_response(None, ["a1"], 30.0) # type: ignore[arg-type]
assert result is None
def test_parse_empty_string_returns_none(self):
result = _parse_recommend_response("", ["a1"], 30.0)
assert result is None
def test_parse_whitespace_only_returns_none(self):
result = _parse_recommend_response(" ", ["a1"], 30.0)
assert result is None
def test_parse_invalid_json_returns_none(self):
result = _parse_recommend_response("not json", ["a1"], 30.0)
assert result is None
def test_parse_non_dict_json_returns_none(self):
result = _parse_recommend_response("[1, 2, 3]", ["a1"], 30.0)
assert result is None
class TestParseRecommendResponseClips:
"""clips 解析测试."""
def test_parse_no_clips_returns_none(self):
content = json.dumps({"title": "test", "clips": []})
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is None
def test_parse_clips_not_list_returns_none(self):
content = json.dumps({"clips": "not a list"})
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is None
def test_parse_clips_sorted_by_order(self):
content = json.dumps(
{
"clips": [
{"clip_type": "outro", "order": 2, "duration": 2, "asset_id": "a1"},
{"clip_type": "intro", "order": 0, "duration": 3, "asset_id": "a1"},
{"clip_type": "showcase", "order": 1, "duration": 5, "asset_id": "a1"},
],
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert len(result["clips"]) == 3
assert result["clips"][0]["clip_type"] == "intro"
assert result["clips"][1]["clip_type"] == "showcase"
assert result["clips"][2]["clip_type"] == "outro"
def test_parse_clips_renumbered_continuously(self):
content = json.dumps(
{
"clips": [
{"clip_type": "intro", "order": 10, "duration": 2, "asset_id": "a1"},
{"clip_type": "outro", "order": 20, "duration": 2, "asset_id": "a1"},
],
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert result["clips"][0]["order"] == 0
assert result["clips"][1]["order"] == 1
def test_parse_skips_invalid_clip_dicts(self):
content = json.dumps(
{
"clips": [
{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"},
"not a dict",
{"clip_type": "outro", "order": 2, "duration": 2, "asset_id": "a1"},
],
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert len(result["clips"]) == 2
class TestParseRecommendResponseFields:
"""各字段解析与边界测试."""
def test_parse_duration_clamped_min(self):
content = json.dumps(
{
"clips": [
{"clip_type": "intro", "order": 0, "duration": 0.5, "asset_id": "a1"},
],
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert result["clips"][0]["duration"] == 1.0
def test_parse_duration_clamped_max(self):
content = json.dumps(
{
"clips": [
{"clip_type": "intro", "order": 0, "duration": 100, "asset_id": "a1"},
],
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert result["clips"][0]["duration"] == 30.0
def test_parse_start_time_clamped_min(self):
content = json.dumps(
{
"clips": [
{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1", "start_time": -5.0},
],
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert result["clips"][0]["start_time"] == 0.0
def test_parse_asset_id_not_in_list_empty(self):
content = json.dumps(
{
"clips": [
{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "unknown_asset"},
],
}
)
result = _parse_recommend_response(content, ["a1", "a2"], 30.0)
assert result is not None
assert result["clips"][0]["asset_id"] == ""
def test_parse_asset_id_in_list_kept(self):
content = json.dumps(
{
"clips": [
{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a2"},
],
}
)
result = _parse_recommend_response(content, ["a1", "a2"], 30.0)
assert result is not None
assert result["clips"][0]["asset_id"] == "a2"
def test_parse_default_values(self):
content = json.dumps(
{
"clips": [
{"order": 0},
],
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
clip = result["clips"][0]
assert clip["clip_type"] == "showcase"
assert clip["text_content"] == ""
assert clip["duration"] == 3.0
assert clip["transition_effect"] == "cut"
assert clip["asset_id"] == ""
assert clip["start_time"] == 0.0
assert clip["config"] == {}
class TestParseRecommendResponseMarkdown:
"""Markdown 代码块包裹的 JSON 测试."""
def test_parse_markdown_json(self):
content = (
"```json\n"
+ json.dumps(
{
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
"title": "md test",
}
)
+ "\n```"
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert len(result["clips"]) == 1
assert result["config"]["title"]["text"] == "md test"
def test_parse_backticks_no_language(self):
content = (
"```\n"
+ json.dumps(
{
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
}
)
+ "\n```"
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert len(result["clips"]) == 1
class TestParseRecommendResponseConfidence:
"""confidence 解析测试."""
def test_parse_confidence_normal(self):
content = json.dumps(
{
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
"confidence": 0.85,
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert result["confidence"] == 0.85
def test_parse_confidence_clamped_min(self):
content = json.dumps(
{
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
"confidence": -0.5,
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert result["confidence"] == 0.0
def test_parse_confidence_clamped_max(self):
content = json.dumps(
{
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
"confidence": 1.5,
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert result["confidence"] == 1.0
def test_parse_confidence_default(self):
content = json.dumps(
{
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert result["confidence"] == 0.7
class TestParseRecommendResponseConfig:
"""config 生成测试."""
def test_parse_no_title_no_ai_auto(self):
content = json.dumps(
{
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
# 没有 title 时,config 的 title.text 保持默认(DEFAULT_EDIT_PLAN_CONFIG 中的值)
assert "title" in result["config"]
def test_parse_config_is_deep_copy(self):
content = json.dumps(
{
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
"title": "test",
}
)
result1 = _parse_recommend_response(content, ["a1"], 30.0)
result2 = _parse_recommend_response(content, ["a1"], 30.0)
# 修改其中一个不影响另一个
result1["config"]["title"]["text"] = "modified"
assert result2["config"]["title"]["text"] != "modified"
class TestParseRecommendResponseTotalDuration:
"""total_duration 计算测试."""
def test_parse_total_duration_sum(self):
content = json.dumps(
{
"clips": [
{"clip_type": "intro", "order": 0, "duration": 3.5, "asset_id": "a1"},
{"clip_type": "showcase", "order": 1, "duration": 5.2, "asset_id": "a1"},
{"clip_type": "outro", "order": 2, "duration": 2.0, "asset_id": "a1"},
],
}
)
result = _parse_recommend_response(content, ["a1"], 30.0)
assert result is not None
assert result["total_duration"] == pytest.approx(10.7, abs=0.01)
# ── _fallback_recommend_clips 测试 ────────────────────────────────────────
class TestFallbackRecommendClips:
"""本地降级推荐方案测试."""
def test_fallback_returns_dict_with_clips(self):
with patch("shared.ai_service.time.sleep"):
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
assert "clips" in result
assert "config" in result
assert "total_duration" in result
assert "confidence" in result
def test_fallback_has_intro_and_outro(self):
with patch("shared.ai_service.time.sleep"):
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
clips = result["clips"]
assert clips[0]["clip_type"] == "intro"
assert clips[-1]["clip_type"] == "outro"
def test_fallback_showcase_count_matches_assets(self):
with patch("shared.ai_service.time.sleep"):
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
showcase_clips = [c for c in result["clips"] if c["clip_type"] == "showcase"]
assert len(showcase_clips) == 3
def test_fallback_no_assets_still_works(self):
with patch("shared.ai_service.time.sleep"):
result = _fallback_recommend_clips("plan1", "tmpl1", [], "one_take", 30.0)
assert len(result["clips"]) >= 2 # 至少有intro和outro
def test_fallback_intro_uses_first_asset(self):
with patch("shared.ai_service.time.sleep"):
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
assert result["clips"][0]["asset_id"] == "a1"
def test_fallback_outro_has_empty_asset(self):
with patch("shared.ai_service.time.sleep"):
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1"], "one_take", 30.0)
assert result["clips"][-1]["asset_id"] == ""
def test_fallback_confidence_in_range(self):
with patch("shared.ai_service.time.sleep"):
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1"], "one_take", 30.0)
assert 0.75 <= result["confidence"] <= 0.95
def test_fallback_title_contains_asset_count(self):
with patch("shared.ai_service.time.sleep"):
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
assert "3" in result["config"]["title"]["text"]
assert result["config"]["title"]["ai_auto"] is True
def test_fallback_total_duration_matches(self):
with patch("shared.ai_service.time.sleep"):
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
total = sum(c["duration"] for c in result["clips"])
assert result["total_duration"] == round(total, 1)
def test_fallback_orders_are_sequential(self):
with patch("shared.ai_service.time.sleep"):
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
orders = [c["order"] for c in result["clips"]]
assert orders == list(range(len(result["clips"])))
# ── _call_ai_cover_service 测试 ───────────────────────────────────────────
class TestAiCoverService:
"""AI封面生成服务测试."""
def test_cover_type_upload(self):
with patch("shared.ai_service.time.sleep"):
result = _call_ai_cover_service("plan1", ["a1"], "upload")
assert result["type"] == "upload"
assert result["image_url"] == ""
def test_cover_type_manual_with_frame_time(self):
with patch("shared.ai_service.time.sleep"):
result = _call_ai_cover_service("plan1", ["a1"], "manual", frame_time=5.5)
assert result["type"] == "manual"
assert result["frame_time"] == 5.5
assert "5.5" in result["image_url"]
def test_cover_type_ai_frame(self):
with patch("shared.ai_service.time.sleep"):
with patch("shared.ai_service.random.uniform", side_effect=[5.0, 0.9]):
result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
assert result["type"] == "ai_frame"
assert result["frame_time"] == 5.0
assert result["confidence"] == 0.9
assert "plan1" in result["image_url"]
def test_cover_type_ai_regenerate(self):
with patch("shared.ai_service.time.sleep"):
result = _call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
assert result["type"] == "ai_frame"
def test_cover_frame_time_in_range(self):
with patch("shared.ai_service.time.sleep"):
result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
assert 1.0 <= result["frame_time"] <= 10.0
File diff suppressed because it is too large Load Diff