Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 584a1059b8 | |||
| 92d5b3f26c | |||
| eb73eeab69 | |||
| 2325ffbc57 | |||
| faeed6f014 | |||
| 6dc388a794 | |||
| 7ca5b3732f | |||
| 7cdf56a1ac |
@@ -0,0 +1,4 @@
|
||||
export { useBatchDelete } from "./useBatchDelete"
|
||||
export { useBatchTag } from "./useBatchTag"
|
||||
export { useBatchClassify } from "./useBatchClassify"
|
||||
export { useBatchMark } from "./useBatchMark"
|
||||
+58
@@ -0,0 +1,58 @@
|
||||
import { useState, useCallback } from "react"
|
||||
import { message } from "antd"
|
||||
import { batchClassifyAssets, type BatchOperationResult } from "@/api/assets"
|
||||
|
||||
interface UseBatchClassifyOptions {
|
||||
selectedIds: Set<string>
|
||||
queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
|
||||
showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
|
||||
}
|
||||
|
||||
export const useBatchClassify = ({
|
||||
selectedIds,
|
||||
queryClient,
|
||||
showResult,
|
||||
}: UseBatchClassifyOptions) => {
|
||||
const [classifyModalOpen, setClassifyModalOpen] = useState(false)
|
||||
const [batchCategory, setBatchCategory] = useState("")
|
||||
const [batchLoading, setBatchLoading] = useState(false)
|
||||
|
||||
const handleBatchClassify = useCallback(async () => {
|
||||
if (!batchCategory) {
|
||||
message.warning("请选择分类")
|
||||
return
|
||||
}
|
||||
const ids = Array.from(selectedIds)
|
||||
setBatchLoading(true)
|
||||
try {
|
||||
const result = await batchClassifyAssets({
|
||||
asset_ids: ids,
|
||||
category: batchCategory,
|
||||
})
|
||||
queryClient.invalidateQueries({ queryKey: ["assets"] })
|
||||
showResult(result, "批量改分类")
|
||||
setClassifyModalOpen(false)
|
||||
setBatchCategory("")
|
||||
if (result.failure_count === 0) {
|
||||
message.success(`成功将 ${result.success_count} 个素材改为「${batchCategory}」`)
|
||||
} else {
|
||||
message.warning(
|
||||
`改分类完成:成功 ${result.success_count} 个,失败 ${result.failure_count} 个`,
|
||||
)
|
||||
}
|
||||
} catch {
|
||||
message.error("批量改分类失败,请重试")
|
||||
} finally {
|
||||
setBatchLoading(false)
|
||||
}
|
||||
}, [batchCategory, selectedIds, queryClient, showResult])
|
||||
|
||||
return {
|
||||
classifyModalOpen,
|
||||
setClassifyModalOpen,
|
||||
batchCategory,
|
||||
setBatchCategory,
|
||||
batchLoading,
|
||||
handleBatchClassify,
|
||||
}
|
||||
}
|
||||
+40
@@ -0,0 +1,40 @@
|
||||
import { useState, useCallback } from "react"
|
||||
import { message } from "antd"
|
||||
import { batchDeleteAssets, type BatchOperationResult } from "@/api/assets"
|
||||
|
||||
interface UseBatchDeleteOptions {
|
||||
selectedIds: Set<string>
|
||||
invalidateAssets: () => void
|
||||
showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
|
||||
}
|
||||
|
||||
export const useBatchDelete = ({
|
||||
selectedIds,
|
||||
invalidateAssets,
|
||||
showResult,
|
||||
}: UseBatchDeleteOptions) => {
|
||||
const [batchLoading, setBatchLoading] = useState(false)
|
||||
|
||||
const handleBatchDelete = useCallback(async () => {
|
||||
const ids = Array.from(selectedIds)
|
||||
setBatchLoading(true)
|
||||
try {
|
||||
const result = await batchDeleteAssets(ids)
|
||||
invalidateAssets()
|
||||
showResult(result, "批量删除")
|
||||
if (result.failure_count === 0) {
|
||||
message.success(`成功删除 ${result.success_count} 个素材`)
|
||||
} else {
|
||||
message.warning(
|
||||
`删除完成:成功 ${result.success_count} 个,失败 ${result.failure_count} 个`,
|
||||
)
|
||||
}
|
||||
} catch {
|
||||
message.error("批量删除失败,请重试")
|
||||
} finally {
|
||||
setBatchLoading(false)
|
||||
}
|
||||
}, [selectedIds, invalidateAssets, showResult])
|
||||
|
||||
return { batchLoading, handleBatchDelete }
|
||||
}
|
||||
+53
@@ -0,0 +1,53 @@
|
||||
import { useState, useCallback } from "react"
|
||||
import { message } from "antd"
|
||||
import { batchMarkAssets, type BatchOperationResult } from "@/api/assets"
|
||||
import type { SmartViewType } from "../../../components/BatchMarkModal"
|
||||
import { SMART_VIEW_LABELS } from "../constants"
|
||||
|
||||
interface UseBatchMarkOptions {
|
||||
selectedIds: Set<string>
|
||||
queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
|
||||
showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
|
||||
}
|
||||
|
||||
export const useBatchMark = ({ selectedIds, queryClient, showResult }: UseBatchMarkOptions) => {
|
||||
const [markModalOpen, setMarkModalOpen] = useState(false)
|
||||
const [batchSmartView, setBatchSmartView] = useState<SmartViewType>("recommended")
|
||||
const [batchLoading, setBatchLoading] = useState(false)
|
||||
|
||||
const handleBatchMark = useCallback(async () => {
|
||||
const ids = Array.from(selectedIds)
|
||||
setBatchLoading(true)
|
||||
try {
|
||||
const result = await batchMarkAssets({
|
||||
asset_ids: ids,
|
||||
smart_view: batchSmartView,
|
||||
})
|
||||
queryClient.invalidateQueries({ queryKey: ["assets"] })
|
||||
showResult(result, "批量智能标记")
|
||||
setMarkModalOpen(false)
|
||||
if (result.failure_count === 0) {
|
||||
message.success(
|
||||
`成功将 ${result.success_count} 个素材标记为「${SMART_VIEW_LABELS[batchSmartView]}」`,
|
||||
)
|
||||
} else {
|
||||
message.warning(
|
||||
`智能标记完成:成功 ${result.success_count} 个,失败 ${result.failure_count} 个`,
|
||||
)
|
||||
}
|
||||
} catch {
|
||||
message.error("批量智能标记失败,请重试")
|
||||
} finally {
|
||||
setBatchLoading(false)
|
||||
}
|
||||
}, [batchSmartView, selectedIds, queryClient, showResult])
|
||||
|
||||
return {
|
||||
markModalOpen,
|
||||
setMarkModalOpen,
|
||||
batchSmartView,
|
||||
setBatchSmartView,
|
||||
batchLoading,
|
||||
handleBatchMark,
|
||||
}
|
||||
}
|
||||
+86
@@ -0,0 +1,86 @@
|
||||
import { useState, useCallback } from "react"
|
||||
import { message } from "antd"
|
||||
import { batchTagAssets, type BatchOperationResult } from "@/api/assets"
|
||||
|
||||
interface UseBatchTagOptions {
|
||||
selectedIds: Set<string>
|
||||
queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
|
||||
showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
|
||||
}
|
||||
|
||||
export const useBatchTag = ({ selectedIds, queryClient, showResult }: UseBatchTagOptions) => {
|
||||
const [tagModalOpen, setTagModalOpen] = useState(false)
|
||||
const [batchTagInput, setBatchTagInput] = useState("")
|
||||
const [batchTags, setBatchTags] = useState<string[]>([])
|
||||
const [tagMode, setTagMode] = useState<"add" | "replace">("add")
|
||||
const [batchLoading, setBatchLoading] = useState(false)
|
||||
|
||||
const handleBatchTag = useCallback(async () => {
|
||||
if (batchTags.length === 0) {
|
||||
message.warning("请至少输入一个标签")
|
||||
return
|
||||
}
|
||||
const ids = Array.from(selectedIds)
|
||||
setBatchLoading(true)
|
||||
try {
|
||||
const result = await batchTagAssets({
|
||||
asset_ids: ids,
|
||||
tags: batchTags,
|
||||
mode: tagMode,
|
||||
})
|
||||
queryClient.invalidateQueries({ queryKey: ["assets"] })
|
||||
showResult(result, "批量打标签")
|
||||
setTagModalOpen(false)
|
||||
setBatchTags([])
|
||||
setBatchTagInput("")
|
||||
setTagMode("add")
|
||||
if (result.failure_count === 0) {
|
||||
message.success(`成功为 ${result.success_count} 个素材打标签`)
|
||||
} else {
|
||||
message.warning(
|
||||
`打标签完成:成功 ${result.success_count} 个,失败 ${result.failure_count} 个`,
|
||||
)
|
||||
}
|
||||
} catch {
|
||||
message.error("批量打标签失败,请重试")
|
||||
} finally {
|
||||
setBatchLoading(false)
|
||||
}
|
||||
}, [batchTags, selectedIds, tagMode, queryClient, showResult])
|
||||
|
||||
const handleTagInputKeyDown = useCallback(
|
||||
(e: React.KeyboardEvent) => {
|
||||
if (e.key === "Enter" && batchTagInput.trim()) {
|
||||
e.preventDefault()
|
||||
const tag = batchTagInput.trim()
|
||||
if (!batchTags.includes(tag)) {
|
||||
setBatchTags([...batchTags, tag])
|
||||
}
|
||||
setBatchTagInput("")
|
||||
}
|
||||
},
|
||||
[batchTagInput, batchTags],
|
||||
)
|
||||
|
||||
const removeBatchTag = useCallback(
|
||||
(tag: string) => {
|
||||
setBatchTags(batchTags.filter((t) => t !== tag))
|
||||
},
|
||||
[batchTags],
|
||||
)
|
||||
|
||||
return {
|
||||
tagModalOpen,
|
||||
setTagModalOpen,
|
||||
batchTagInput,
|
||||
setBatchTagInput,
|
||||
batchTags,
|
||||
setBatchTags,
|
||||
tagMode,
|
||||
setTagMode,
|
||||
batchLoading,
|
||||
handleBatchTag,
|
||||
handleTagInputKeyDown,
|
||||
removeBatchTag,
|
||||
}
|
||||
}
|
||||
@@ -1,238 +0,0 @@
|
||||
import { useState, useCallback } from "react"
|
||||
import { message } from "antd"
|
||||
import {
|
||||
batchDeleteAssets,
|
||||
batchTagAssets,
|
||||
batchClassifyAssets,
|
||||
batchMarkAssets,
|
||||
type BatchOperationResult,
|
||||
} from "@/api/assets"
|
||||
import type { SmartViewType } from "../../components/BatchMarkModal"
|
||||
import { SMART_VIEW_LABELS } from "./constants"
|
||||
|
||||
/* ── 批量删除 ── */
|
||||
interface UseBatchDeleteOptions {
|
||||
selectedIds: Set<string>
|
||||
invalidateAssets: () => void
|
||||
showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
|
||||
}
|
||||
|
||||
export const useBatchDelete = ({
|
||||
selectedIds,
|
||||
invalidateAssets,
|
||||
showResult,
|
||||
}: UseBatchDeleteOptions) => {
|
||||
const [batchLoading, setBatchLoading] = useState(false)
|
||||
|
||||
const handleBatchDelete = useCallback(async () => {
|
||||
const ids = Array.from(selectedIds)
|
||||
setBatchLoading(true)
|
||||
try {
|
||||
const result = await batchDeleteAssets(ids)
|
||||
invalidateAssets()
|
||||
showResult(result, "批量删除")
|
||||
if (result.failure_count === 0) {
|
||||
message.success(`成功删除 ${result.success_count} 个素材`)
|
||||
} else {
|
||||
message.warning(
|
||||
`删除完成:成功 ${result.success_count} 个,失败 ${result.failure_count} 个`,
|
||||
)
|
||||
}
|
||||
} catch {
|
||||
message.error("批量删除失败,请重试")
|
||||
} finally {
|
||||
setBatchLoading(false)
|
||||
}
|
||||
}, [selectedIds, invalidateAssets, showResult])
|
||||
|
||||
return { batchLoading, handleBatchDelete }
|
||||
}
|
||||
|
||||
/* ── 批量打标签 ── */
|
||||
interface UseBatchTagOptions {
|
||||
selectedIds: Set<string>
|
||||
queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
|
||||
showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
|
||||
}
|
||||
|
||||
export const useBatchTag = ({ selectedIds, queryClient, showResult }: UseBatchTagOptions) => {
|
||||
const [tagModalOpen, setTagModalOpen] = useState(false)
|
||||
const [batchTagInput, setBatchTagInput] = useState("")
|
||||
const [batchTags, setBatchTags] = useState<string[]>([])
|
||||
const [tagMode, setTagMode] = useState<"add" | "replace">("add")
|
||||
const [batchLoading, setBatchLoading] = useState(false)
|
||||
|
||||
const handleBatchTag = useCallback(async () => {
|
||||
if (batchTags.length === 0) {
|
||||
message.warning("请至少输入一个标签")
|
||||
return
|
||||
}
|
||||
const ids = Array.from(selectedIds)
|
||||
setBatchLoading(true)
|
||||
try {
|
||||
const result = await batchTagAssets({
|
||||
asset_ids: ids,
|
||||
tags: batchTags,
|
||||
mode: tagMode,
|
||||
})
|
||||
queryClient.invalidateQueries({ queryKey: ["assets"] })
|
||||
showResult(result, "批量打标签")
|
||||
setTagModalOpen(false)
|
||||
setBatchTags([])
|
||||
setBatchTagInput("")
|
||||
setTagMode("add")
|
||||
if (result.failure_count === 0) {
|
||||
message.success(`成功为 ${result.success_count} 个素材打标签`)
|
||||
} else {
|
||||
message.warning(
|
||||
`打标签完成:成功 ${result.success_count} 个,失败 ${result.failure_count} 个`,
|
||||
)
|
||||
}
|
||||
} catch {
|
||||
message.error("批量打标签失败,请重试")
|
||||
} finally {
|
||||
setBatchLoading(false)
|
||||
}
|
||||
}, [batchTags, selectedIds, tagMode, queryClient, showResult])
|
||||
|
||||
const handleTagInputKeyDown = useCallback(
|
||||
(e: React.KeyboardEvent) => {
|
||||
if (e.key === "Enter" && batchTagInput.trim()) {
|
||||
e.preventDefault()
|
||||
const tag = batchTagInput.trim()
|
||||
if (!batchTags.includes(tag)) {
|
||||
setBatchTags([...batchTags, tag])
|
||||
}
|
||||
setBatchTagInput("")
|
||||
}
|
||||
},
|
||||
[batchTagInput, batchTags],
|
||||
)
|
||||
|
||||
const removeBatchTag = useCallback(
|
||||
(tag: string) => {
|
||||
setBatchTags(batchTags.filter((t) => t !== tag))
|
||||
},
|
||||
[batchTags],
|
||||
)
|
||||
|
||||
return {
|
||||
tagModalOpen,
|
||||
setTagModalOpen,
|
||||
batchTagInput,
|
||||
setBatchTagInput,
|
||||
batchTags,
|
||||
setBatchTags,
|
||||
tagMode,
|
||||
setTagMode,
|
||||
batchLoading,
|
||||
handleBatchTag,
|
||||
handleTagInputKeyDown,
|
||||
removeBatchTag,
|
||||
}
|
||||
}
|
||||
|
||||
/* ── 批量改分类 ── */
|
||||
interface UseBatchClassifyOptions {
|
||||
selectedIds: Set<string>
|
||||
queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
|
||||
showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
|
||||
}
|
||||
|
||||
export const useBatchClassify = ({
|
||||
selectedIds,
|
||||
queryClient,
|
||||
showResult,
|
||||
}: UseBatchClassifyOptions) => {
|
||||
const [classifyModalOpen, setClassifyModalOpen] = useState(false)
|
||||
const [batchCategory, setBatchCategory] = useState("")
|
||||
const [batchLoading, setBatchLoading] = useState(false)
|
||||
|
||||
const handleBatchClassify = useCallback(async () => {
|
||||
if (!batchCategory) {
|
||||
message.warning("请选择分类")
|
||||
return
|
||||
}
|
||||
const ids = Array.from(selectedIds)
|
||||
setBatchLoading(true)
|
||||
try {
|
||||
const result = await batchClassifyAssets({
|
||||
asset_ids: ids,
|
||||
category: batchCategory,
|
||||
})
|
||||
queryClient.invalidateQueries({ queryKey: ["assets"] })
|
||||
showResult(result, "批量改分类")
|
||||
setClassifyModalOpen(false)
|
||||
setBatchCategory("")
|
||||
if (result.failure_count === 0) {
|
||||
message.success(`成功将 ${result.success_count} 个素材改为「${batchCategory}」`)
|
||||
} else {
|
||||
message.warning(
|
||||
`改分类完成:成功 ${result.success_count} 个,失败 ${result.failure_count} 个`,
|
||||
)
|
||||
}
|
||||
} catch {
|
||||
message.error("批量改分类失败,请重试")
|
||||
} finally {
|
||||
setBatchLoading(false)
|
||||
}
|
||||
}, [batchCategory, selectedIds, queryClient, showResult])
|
||||
|
||||
return {
|
||||
classifyModalOpen,
|
||||
setClassifyModalOpen,
|
||||
batchCategory,
|
||||
setBatchCategory,
|
||||
batchLoading,
|
||||
handleBatchClassify,
|
||||
}
|
||||
}
|
||||
|
||||
/* ── 批量智能标记 ── */
|
||||
interface UseBatchMarkOptions {
|
||||
selectedIds: Set<string>
|
||||
queryClient: ReturnType<typeof import("@tanstack/react-query").useQueryClient>
|
||||
showResult: (result: BatchOperationResult, title: string, clear?: boolean) => void
|
||||
}
|
||||
|
||||
export const useBatchMark = ({ selectedIds, queryClient, showResult }: UseBatchMarkOptions) => {
|
||||
const [markModalOpen, setMarkModalOpen] = useState(false)
|
||||
const [batchSmartView, setBatchSmartView] = useState<SmartViewType>("recommended")
|
||||
const [batchLoading, setBatchLoading] = useState(false)
|
||||
|
||||
const handleBatchMark = useCallback(async () => {
|
||||
const ids = Array.from(selectedIds)
|
||||
setBatchLoading(true)
|
||||
try {
|
||||
const result = await batchMarkAssets({
|
||||
asset_ids: ids,
|
||||
smart_view: batchSmartView,
|
||||
})
|
||||
queryClient.invalidateQueries({ queryKey: ["assets"] })
|
||||
showResult(result, "批量智能标记")
|
||||
setMarkModalOpen(false)
|
||||
if (result.failure_count === 0) {
|
||||
message.success(
|
||||
`成功将 ${result.success_count} 个素材标记为「${SMART_VIEW_LABELS[batchSmartView]}」`,
|
||||
)
|
||||
} else {
|
||||
message.warning(
|
||||
`智能标记完成:成功 ${result.success_count} 个,失败 ${result.failure_count} 个`,
|
||||
)
|
||||
}
|
||||
} catch {
|
||||
message.error("批量智能标记失败,请重试")
|
||||
} finally {
|
||||
setBatchLoading(false)
|
||||
}
|
||||
}, [batchSmartView, selectedIds, queryClient, showResult])
|
||||
|
||||
return {
|
||||
markModalOpen,
|
||||
setMarkModalOpen,
|
||||
batchSmartView,
|
||||
setBatchSmartView,
|
||||
batchLoading,
|
||||
handleBatchMark,
|
||||
}
|
||||
}
|
||||
@@ -12,7 +12,7 @@ import {
|
||||
useBatchTag,
|
||||
useBatchClassify,
|
||||
useBatchMark,
|
||||
} from "./asset-operations/batchOperations"
|
||||
} from "./asset-operations/batch-operations"
|
||||
import type { BatchOperationResult } from "@/api/assets"
|
||||
import type { SmartViewType } from "../components/BatchMarkModal"
|
||||
|
||||
|
||||
Regular → Executable
+14
-58
@@ -1,18 +1,16 @@
|
||||
/**
|
||||
* 字幕样式配置面板 — 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
|
||||
@@ -40,7 +38,6 @@ 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>
|
||||
@@ -55,26 +52,11 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
|
||||
|
||||
{config.enabled && (
|
||||
<>
|
||||
{/* ── 模式切换 ── */}
|
||||
<div className="sub-field">
|
||||
<label className="sub-label">字幕来源</label>
|
||||
<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>
|
||||
<SubtitleModeSwitch mode={config.mode} onModeChange={(mode) => update({ mode })} />
|
||||
</div>
|
||||
|
||||
{/* ── ASR 语言(仅 ASR 模式) ── */}
|
||||
{config.mode === "asr" && (
|
||||
<div className="sub-field">
|
||||
<label className="sub-label">识别语言</label>
|
||||
@@ -88,7 +70,6 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* ── 字体大小 ── */}
|
||||
<div className="sub-field">
|
||||
<label className="sub-label">
|
||||
字体大小 <span className="sub-value">{config.fontSize}px</span>
|
||||
@@ -101,7 +82,6 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
|
||||
/>
|
||||
</div>
|
||||
|
||||
{/* ── 字体颜色 ── */}
|
||||
<div className="sub-field">
|
||||
<label className="sub-label">字体颜色</label>
|
||||
<div className="sub-color-row">
|
||||
@@ -113,7 +93,6 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* ── 字体 ── */}
|
||||
<div className="sub-field">
|
||||
<label className="sub-label">字体</label>
|
||||
<Select
|
||||
@@ -125,46 +104,24 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
|
||||
/>
|
||||
</div>
|
||||
|
||||
{/* ── 字幕位置 ── */}
|
||||
<div className="sub-field">
|
||||
<label className="sub-label">位置</label>
|
||||
<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>
|
||||
<SubtitlePositionSelector
|
||||
position={config.position}
|
||||
onPositionChange={(position) => update({ position })}
|
||||
/>
|
||||
</div>
|
||||
|
||||
{/* ── 描边 / 阴影 ── */}
|
||||
<div className="sub-field">
|
||||
<label className="sub-label">效果</label>
|
||||
<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>
|
||||
<SubtitleEffectButtons
|
||||
stroke={config.stroke}
|
||||
shadow={config.shadow}
|
||||
onStrokeChange={(stroke) => update({ stroke })}
|
||||
onShadowChange={(shadow) => update({ shadow })}
|
||||
/>
|
||||
</div>
|
||||
|
||||
{/* ── 动画 ── */}
|
||||
<div className="sub-field">
|
||||
<label className="sub-label">动画效果</label>
|
||||
<Select
|
||||
@@ -179,7 +136,6 @@ const SubtitleStylePanel: React.FC<SubtitleStylePanelProps> = ({
|
||||
/>
|
||||
</div>
|
||||
|
||||
{/* ── 预览 ── */}
|
||||
<SubtitlePreview config={config} />
|
||||
</>
|
||||
)}
|
||||
|
||||
+32
@@ -0,0 +1,32 @@
|
||||
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>
|
||||
)
|
||||
}
|
||||
+25
@@ -0,0 +1,25 @@
|
||||
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>
|
||||
)
|
||||
}
|
||||
Executable
+27
@@ -0,0 +1,27 @@
|
||||
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>
|
||||
)
|
||||
}
|
||||
Executable → Regular
+123
-5
@@ -1,8 +1,11 @@
|
||||
import React from "react"
|
||||
import { Table } from "antd"
|
||||
import type { TaskItem } from "@/api/tasks"
|
||||
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 { TaskErrorDetail } from "./TaskErrorDetail"
|
||||
import { useTaskTableColumns, TaskEmptyState } from "./task-table"
|
||||
|
||||
interface TaskTableProps {
|
||||
dataSource: TaskItem[]
|
||||
@@ -21,6 +24,7 @@ interface TaskTableProps {
|
||||
|
||||
/**
|
||||
* 任务列表表格
|
||||
* 含列定义、分页、展开行
|
||||
*/
|
||||
export const TaskTable: React.FC<TaskTableProps> = ({
|
||||
dataSource,
|
||||
@@ -36,7 +40,116 @@ export const TaskTable: React.FC<TaskTableProps> = ({
|
||||
onRetry,
|
||||
onViewDetail,
|
||||
}) => {
|
||||
const columns = useTaskTableColumns({ retryLoading, 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>
|
||||
},
|
||||
},
|
||||
]
|
||||
|
||||
return (
|
||||
<Table
|
||||
@@ -65,7 +178,12 @@ export const TaskTable: React.FC<TaskTableProps> = ({
|
||||
scroll={{ x: 800 }}
|
||||
className="task-table"
|
||||
locale={{
|
||||
emptyText: <TaskEmptyState />,
|
||||
emptyText: (
|
||||
<div className="task-empty">
|
||||
<ClockCircleOutlined />
|
||||
<p>暂无任务记录</p>
|
||||
</div>
|
||||
),
|
||||
}}
|
||||
/>
|
||||
)
|
||||
|
||||
@@ -1,9 +0,0 @@
|
||||
import React from "react"
|
||||
import { ClockCircleOutlined } from "@ant-design/icons"
|
||||
|
||||
export const TaskEmptyState: React.FC = () => (
|
||||
<div className="task-empty">
|
||||
<ClockCircleOutlined />
|
||||
<p>暂无任务记录</p>
|
||||
</div>
|
||||
)
|
||||
@@ -1,2 +0,0 @@
|
||||
export { useTaskTableColumns } from "./useTaskTableColumns"
|
||||
export { TaskEmptyState } from "./TaskEmptyState"
|
||||
@@ -1,128 +0,0 @@
|
||||
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 } from "react"
|
||||
import { useRef, useCallback, useEffect } from "react"
|
||||
|
||||
interface UseRowProgressOptions {
|
||||
duration: number
|
||||
@@ -7,6 +7,22 @@ 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>) => {
|
||||
@@ -16,6 +32,7 @@ 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)
|
||||
}
|
||||
@@ -24,15 +41,26 @@ export function useRowProgress({ duration, onSeek }: UseRowProgressOptions) {
|
||||
|
||||
const handleMove = (ev: MouseEvent) => doSeek(ev)
|
||||
const handleUp = () => {
|
||||
document.removeEventListener("mousemove", handleMove)
|
||||
document.removeEventListener("mouseup", handleUp)
|
||||
cleanupListeners()
|
||||
}
|
||||
|
||||
// 先清理旧的,再添加新的
|
||||
cleanupListeners()
|
||||
listenersRef.current.move = handleMove
|
||||
listenersRef.current.up = handleUp
|
||||
|
||||
document.addEventListener("mousemove", handleMove)
|
||||
document.addEventListener("mouseup", handleUp)
|
||||
},
|
||||
[duration, onSeek],
|
||||
[duration, onSeek, cleanupListeners],
|
||||
)
|
||||
|
||||
// 组件卸载时清理事件监听器
|
||||
useEffect(() => {
|
||||
return () => {
|
||||
cleanupListeners()
|
||||
}
|
||||
}, [cleanupListeners])
|
||||
|
||||
return { progressRef, handleMouseDown }
|
||||
}
|
||||
|
||||
@@ -36,6 +36,11 @@ import "@/pages/assets/hooks/useLibraryManagement"
|
||||
import "@/pages/assets/hooks/useAssetUpload"
|
||||
import "@/pages/assets/hooks/useAssetSelection"
|
||||
import "@/pages/assets/hooks/useAssetOperations"
|
||||
import "@/pages/assets/hooks/asset-operations/batch-operations"
|
||||
import "@/pages/assets/hooks/asset-operations/batch-operations/useBatchDelete"
|
||||
import "@/pages/assets/hooks/asset-operations/batch-operations/useBatchTag"
|
||||
import "@/pages/assets/hooks/asset-operations/batch-operations/useBatchClassify"
|
||||
import "@/pages/assets/hooks/asset-operations/batch-operations/useBatchMark"
|
||||
|
||||
describe("AssetLibrary module smoke test", () => {
|
||||
it("should load all asset modules", () => {
|
||||
|
||||
@@ -65,6 +65,9 @@ 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"
|
||||
|
||||
@@ -1,17 +0,0 @@
|
||||
/**
|
||||
* 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,6 +20,8 @@ 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"
|
||||
|
||||
@@ -569,7 +569,9 @@ class CosyVoiceService:
|
||||
清洗后的 prefix
|
||||
"""
|
||||
# 只保留字母和数字
|
||||
cleaned = "".join(c for c in name if c.isalnum())
|
||||
import re
|
||||
|
||||
cleaned = re.sub(r"[^a-zA-Z0-9]", "", name)
|
||||
# 最多10字符
|
||||
cleaned = cleaned[:10]
|
||||
# 如果清洗后为空,用默认值
|
||||
|
||||
@@ -301,7 +301,7 @@ def main():
|
||||
pr_num = pr["number"]
|
||||
pr_title = pr["title"]
|
||||
head_sha = pr["head"]["sha"]
|
||||
base_ref = pr.get("base", {}).get("re", "")
|
||||
base_ref = pr.get("base", {}).get("ref", "")
|
||||
|
||||
# 跳过draft
|
||||
if pr.get("draft"):
|
||||
|
||||
+496
@@ -0,0 +1,496 @@
|
||||
"""模板片段转换器单测.
|
||||
|
||||
纯函数模块,覆盖:枚举安全解析、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")
|
||||
+224
-209
@@ -1,18 +1,14 @@
|
||||
"""路径安全校验工具单元测试 — 路径遍历防护."""
|
||||
|
||||
from __future__ import annotations
|
||||
"""path_security 单元测试."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker"))
|
||||
import pytest
|
||||
|
||||
from video_processing.path_security import ( # noqa: E402
|
||||
from apps.worker.video_processing.path_security import (
|
||||
LOCAL_SCHEMA_PREFIX,
|
||||
MAX_PATH_LENGTH,
|
||||
PathSecurityError,
|
||||
get_allowed_local_dirs,
|
||||
is_in_allowed_dirs,
|
||||
is_path_safe,
|
||||
safe_resolve_path,
|
||||
@@ -21,223 +17,242 @@ from video_processing.path_security import ( # noqa: E402
|
||||
)
|
||||
|
||||
|
||||
class TestSafeResolvePath(unittest.TestCase):
|
||||
"""安全路径解析测试."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
|
||||
def tearDown(self):
|
||||
import shutil
|
||||
|
||||
shutil.rmtree(self.tmpdir, ignore_errors=True)
|
||||
|
||||
# ── 正常路径 ─────────────────────────────────────────────────────────
|
||||
|
||||
def test_simple_relative_path(self):
|
||||
"""简单相对路径应该正常解析."""
|
||||
result = safe_resolve_path("test.mp4", self.tmpdir)
|
||||
self.assertEqual(result.name, "test.mp4")
|
||||
self.assertTrue(str(result).startswith(self.tmpdir))
|
||||
|
||||
def test_subdirectory_path(self):
|
||||
"""子目录路径应该正常解析."""
|
||||
result = safe_resolve_path("sub/dir/file.mp4", self.tmpdir)
|
||||
self.assertTrue(str(result).startswith(self.tmpdir))
|
||||
self.assertIn("sub/dir/file.mp4", str(result).replace("\\", "/"))
|
||||
|
||||
def test_dot_slash_path(self):
|
||||
"""./ 开头的路径应该正常解析."""
|
||||
result = safe_resolve_path("./test.mp4", self.tmpdir)
|
||||
self.assertEqual(result.name, "test.mp4")
|
||||
|
||||
# ── 路径遍历防护 ─────────────────────────────────────────────────────
|
||||
|
||||
def test_parent_traversal_rejected(self):
|
||||
"""../ 路径遍历应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path("../etc/passwd", self.tmpdir)
|
||||
|
||||
def test_multiple_parent_traversal_rejected(self):
|
||||
"""多级 ../ 遍历应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path("../../etc/passwd", self.tmpdir)
|
||||
|
||||
def test_mixed_traversal_rejected(self):
|
||||
"""混合路径遍历应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path("./sub/../../etc/shadow", self.tmpdir)
|
||||
|
||||
def test_absolute_path_rejected(self):
|
||||
"""绝对路径(超出基目录)应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path("/etc/passwd", self.tmpdir)
|
||||
|
||||
# ── 空字节注入 ───────────────────────────────────────────────────────
|
||||
|
||||
def test_null_byte_rejected(self):
|
||||
"""空字节注入应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path("test\x00.mp4", self.tmpdir)
|
||||
|
||||
# ── 空路径 ──────────────────────────────────────────────────────────
|
||||
|
||||
def test_empty_path_rejected(self):
|
||||
"""空路径应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path("", self.tmpdir)
|
||||
|
||||
def test_none_path_rejected(self):
|
||||
"""None 路径应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path(None, self.tmpdir) # type: ignore
|
||||
|
||||
def test_whitespace_path_rejected(self):
|
||||
"""空白路径应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path(" ", self.tmpdir)
|
||||
|
||||
# ── 路径长度 ────────────────────────────────────────────────────────
|
||||
|
||||
def test_too_long_path_rejected(self):
|
||||
"""超长路径应该被拒绝."""
|
||||
long_path = "a" * 5000 + ".mp4"
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path(long_path, self.tmpdir)
|
||||
|
||||
# ── 系统路径防护 ─────────────────────────────────────────────────────
|
||||
|
||||
def test_proc_path_rejected_when_absolute(self):
|
||||
"""/proc/ 路径在绝对路径模式下应该被拒绝(因为超出基目录)."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path("/proc/self/environ", self.tmpdir)
|
||||
|
||||
# ── 扩展名校验 ───────────────────────────────────────────────────────
|
||||
|
||||
def test_extension_whitelist_pass(self):
|
||||
"""白名单内的扩展名应该通过."""
|
||||
result = safe_resolve_path(
|
||||
"test.mp4",
|
||||
self.tmpdir,
|
||||
allowed_extensions={".mp4", ".mov"},
|
||||
)
|
||||
self.assertEqual(result.suffix.lower(), ".mp4")
|
||||
|
||||
def test_extension_whitelist_reject(self):
|
||||
"""白名单外的扩展名应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
safe_resolve_path(
|
||||
"test.exe",
|
||||
self.tmpdir,
|
||||
allowed_extensions={".mp4", ".mov"},
|
||||
)
|
||||
@pytest.fixture
|
||||
def base_dir():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
# 创建一个子文件用于测试
|
||||
with open(os.path.join(tmpdir, "test.mp4"), "w") as f:
|
||||
f.write("test")
|
||||
subdir = os.path.join(tmpdir, "subdir")
|
||||
os.makedirs(subdir)
|
||||
with open(os.path.join(subdir, "audio.mp3"), "w") as f:
|
||||
f.write("test")
|
||||
yield tmpdir
|
||||
|
||||
|
||||
class TestLocalSchemaPath(unittest.TestCase):
|
||||
"""local:// schema 路径测试."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
|
||||
def tearDown(self):
|
||||
import shutil
|
||||
|
||||
shutil.rmtree(self.tmpdir, ignore_errors=True)
|
||||
|
||||
def test_valid_local_schema(self):
|
||||
"""有效的 local:// 相对路径应该通过."""
|
||||
# 创建测试文件
|
||||
test_file = Path(self.tmpdir) / "test.mp4"
|
||||
test_file.touch()
|
||||
|
||||
result = validate_local_schema_path("local://test.mp4", self.tmpdir)
|
||||
self.assertTrue(result.exists())
|
||||
|
||||
def test_local_schema_absolute_rejected(self):
|
||||
"""local:// + 绝对路径应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
validate_local_schema_path("local:///etc/passwd", self.tmpdir)
|
||||
|
||||
def test_local_schema_traversal_rejected(self):
|
||||
"""local:// + 路径遍历应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
validate_local_schema_path("local://../etc/passwd", self.tmpdir)
|
||||
|
||||
def test_non_local_schema_rejected(self):
|
||||
"""非 local:// 开头的路径应该被拒绝."""
|
||||
with self.assertRaises(PathSecurityError):
|
||||
validate_local_schema_path("http://example.com/test", self.tmpdir)
|
||||
# ── safe_resolve_path ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSanitizeFilename(unittest.TestCase):
|
||||
"""文件名清理测试."""
|
||||
class TestSafeResolvePath:
|
||||
def test_none_path_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="不能为空"):
|
||||
safe_resolve_path(None, base_dir)
|
||||
|
||||
def test_empty_string_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="不能为空"):
|
||||
safe_resolve_path("", base_dir)
|
||||
|
||||
def test_whitespace_path_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="不能为空"):
|
||||
safe_resolve_path(" ", base_dir)
|
||||
|
||||
def test_too_long_path_raises(self, base_dir):
|
||||
long_path = "a" * (MAX_PATH_LENGTH + 1)
|
||||
with pytest.raises(PathSecurityError, match="路径过长"):
|
||||
safe_resolve_path(long_path, base_dir)
|
||||
|
||||
def test_null_byte_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="空字节"):
|
||||
safe_resolve_path("file\x00.mp4", base_dir)
|
||||
|
||||
def test_relative_path_within_base(self, base_dir):
|
||||
result = safe_resolve_path("test.mp4", base_dir)
|
||||
assert result.name == "test.mp4"
|
||||
assert str(result).startswith(str(os.path.realpath(base_dir)))
|
||||
|
||||
def test_subdirectory_path(self, base_dir):
|
||||
result = safe_resolve_path("subdir/audio.mp3", base_dir)
|
||||
assert result.name == "audio.mp3"
|
||||
assert "subdir" in str(result)
|
||||
|
||||
def test_parent_traversal_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="路径遍历"):
|
||||
safe_resolve_path("../etc/passwd", base_dir)
|
||||
|
||||
def test_nested_parent_traversal_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="路径遍历"):
|
||||
safe_resolve_path("subdir/../../etc/passwd", base_dir)
|
||||
|
||||
def test_absolute_path_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="绝对路径"):
|
||||
safe_resolve_path("/etc/passwd", base_dir)
|
||||
|
||||
def test_absolute_path_with_allow_outside(self, base_dir):
|
||||
# allow_outside=True 时允许绝对路径(但会被危险路径模式检查)
|
||||
with pytest.raises(PathSecurityError, match="系统路径"):
|
||||
safe_resolve_path("/etc/passwd", base_dir, allow_outside=True)
|
||||
|
||||
def test_local_schema_relative(self, base_dir):
|
||||
result = safe_resolve_path("local://test.mp4", base_dir)
|
||||
assert result.name == "test.mp4"
|
||||
assert str(result).startswith(str(os.path.realpath(base_dir)))
|
||||
|
||||
def test_local_schema_absolute_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="绝对路径"):
|
||||
safe_resolve_path("local:///etc/passwd", base_dir)
|
||||
|
||||
def test_local_schema_traversal_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="路径遍历"):
|
||||
safe_resolve_path("local://../secret", base_dir)
|
||||
|
||||
def test_invalid_base_dir_raises(self):
|
||||
with pytest.raises(PathSecurityError, match="基路径"):
|
||||
safe_resolve_path("file.txt", "/nonexistent/dir")
|
||||
|
||||
def test_allowed_extensions_valid(self, base_dir):
|
||||
result = safe_resolve_path("test.mp4", base_dir, allowed_extensions={".mp4"})
|
||||
assert result.suffix.lower() == ".mp4"
|
||||
|
||||
def test_allowed_extensions_invalid_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="文件类型"):
|
||||
safe_resolve_path("test.mp4", base_dir, allowed_extensions={".mp3"})
|
||||
|
||||
def test_no_extension_restriction(self, base_dir):
|
||||
# allowed_extensions=None 时不检查
|
||||
result = safe_resolve_path("test.mp4", base_dir, allowed_extensions=None)
|
||||
assert result is not None
|
||||
|
||||
def test_path_object_input(self, base_dir):
|
||||
from pathlib import Path
|
||||
|
||||
result = safe_resolve_path(Path("test.mp4"), base_dir)
|
||||
assert result.name == "test.mp4"
|
||||
|
||||
def test_path_object_base_dir(self, base_dir):
|
||||
from pathlib import Path
|
||||
|
||||
result = safe_resolve_path("test.mp4", Path(base_dir))
|
||||
assert result.name == "test.mp4"
|
||||
|
||||
|
||||
# ── is_path_safe ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestIsPathSafe:
|
||||
def test_safe_path_returns_true(self, base_dir):
|
||||
assert is_path_safe("test.mp4", base_dir) is True
|
||||
|
||||
def test_unsafe_path_returns_false(self, base_dir):
|
||||
assert is_path_safe("../etc/passwd", base_dir) is False
|
||||
|
||||
def test_none_returns_false(self, base_dir):
|
||||
assert is_path_safe(None, base_dir) is False
|
||||
|
||||
|
||||
# ── validate_local_schema_path ──────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestValidateLocalSchemaPath:
|
||||
def test_valid_local_path(self, base_dir):
|
||||
result = validate_local_schema_path("local://test.mp4", base_dir)
|
||||
assert result.name == "test.mp4"
|
||||
|
||||
def test_missing_prefix_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError, match="开头"):
|
||||
validate_local_schema_path("test.mp4", base_dir)
|
||||
|
||||
def test_traversal_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError):
|
||||
validate_local_schema_path("local://../secret", base_dir)
|
||||
|
||||
def test_absolute_path_raises(self, base_dir):
|
||||
with pytest.raises(PathSecurityError):
|
||||
validate_local_schema_path("local:///etc/passwd", base_dir)
|
||||
|
||||
|
||||
# ── sanitize_filename ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSanitizeFilename:
|
||||
def test_normal_filename(self):
|
||||
"""正常文件名应该保持不变."""
|
||||
self.assertEqual(sanitize_filename("video.mp4"), "video.mp4")
|
||||
assert sanitize_filename("hello.mp4") == "hello.mp4"
|
||||
|
||||
def test_path_separators_removed(self):
|
||||
"""路径分隔符应该被替换."""
|
||||
self.assertNotIn("/", sanitize_filename("../path/to/file.mp4"))
|
||||
self.assertNotIn("\\", sanitize_filename("..\\path\\file.mp4"))
|
||||
def test_empty_returns_unnamed(self):
|
||||
assert sanitize_filename("") == "unnamed"
|
||||
|
||||
def test_leading_dots_removed(self):
|
||||
"""开头的点应该被移除."""
|
||||
result = sanitize_filename(".hidden")
|
||||
self.assertFalse(result.startswith("."))
|
||||
self.assertEqual(result, "hidden")
|
||||
def test_none_default(self):
|
||||
# 空字符串会返回unnamed
|
||||
assert sanitize_filename("") == "unnamed"
|
||||
|
||||
def test_multiple_leading_dots_removed(self):
|
||||
"""多个开头的点应该全部被移除."""
|
||||
result = sanitize_filename("...hidden")
|
||||
self.assertFalse(result.startswith("."))
|
||||
def test_removes_path_separators(self):
|
||||
assert "/" not in sanitize_filename("path/to/file.mp4")
|
||||
assert "\\" not in sanitize_filename("path\\to\\file.mp4")
|
||||
|
||||
def test_empty_filename_default(self):
|
||||
"""空文件名应该返回 unnamed."""
|
||||
self.assertEqual(sanitize_filename(""), "unnamed")
|
||||
def test_removes_control_characters(self):
|
||||
result = sanitize_filename("file\x01\x02name.mp4")
|
||||
assert "\x01" not in result
|
||||
assert "\x02" not in result
|
||||
|
||||
def test_special_chars_removed(self):
|
||||
"""特殊字符应该被替换."""
|
||||
result = sanitize_filename('file<name>:"test|?*.mp4')
|
||||
self.assertNotIn("<", result)
|
||||
self.assertNotIn(">", result)
|
||||
self.assertNotIn(":", result)
|
||||
self.assertNotIn('"', result)
|
||||
self.assertNotIn("|", result)
|
||||
self.assertNotIn("?", result)
|
||||
self.assertNotIn("*", result)
|
||||
def test_removes_dangerous_chars(self):
|
||||
result = sanitize_filename("file<name>.mp4")
|
||||
assert "<" not in result
|
||||
assert ">" not in result
|
||||
|
||||
def test_chinese_filename_preserved(self):
|
||||
"""中文文件名应该保留."""
|
||||
result = sanitize_filename("视频素材.mp4")
|
||||
self.assertIn("视频素材", result)
|
||||
def test_removes_leading_dots(self):
|
||||
assert not sanitize_filename(".hidden").startswith(".")
|
||||
assert not sanitize_filename("..hidden").startswith(".")
|
||||
|
||||
def test_chinese_characters_preserved(self):
|
||||
result = sanitize_filename("视频文件.mp4")
|
||||
assert "视频文件" in result
|
||||
|
||||
def test_long_filename_truncated(self):
|
||||
"""超长文件名应该被截断."""
|
||||
long_name = "a" * 300 + ".mp4"
|
||||
result = sanitize_filename(long_name)
|
||||
self.assertLessEqual(len(result), 255)
|
||||
self.assertTrue(result.endswith(".mp4"))
|
||||
assert len(result) <= 255
|
||||
assert result.endswith(".mp4")
|
||||
|
||||
def test_spaces_preserved(self):
|
||||
result = sanitize_filename("my file.mp4")
|
||||
assert "my file.mp4" == result
|
||||
|
||||
def test_underscores_hyphens_preserved(self):
|
||||
result = sanitize_filename("my_file-name.mp4")
|
||||
assert result == "my_file-name.mp4"
|
||||
|
||||
def test_all_dots_returns_unnamed(self):
|
||||
assert sanitize_filename("...") == "unnamed"
|
||||
|
||||
|
||||
class TestAllowedDirs(unittest.TestCase):
|
||||
"""允许目录配置测试."""
|
||||
|
||||
def test_get_allowed_dirs_returns_list(self):
|
||||
"""get_allowed_local_dirs 应该返回列表."""
|
||||
dirs = get_allowed_local_dirs()
|
||||
self.assertIsInstance(dirs, list)
|
||||
|
||||
def test_is_in_allowed_dirs_tmp(self):
|
||||
"""/tmp 应该在默认允许目录内."""
|
||||
self.assertTrue(is_in_allowed_dirs("/tmp/test.mp4"))
|
||||
|
||||
def test_is_path_safe_convenience(self):
|
||||
"""is_path_safe 便捷函数应该正常工作."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
self.assertTrue(is_path_safe("test.mp4", tmpdir))
|
||||
self.assertFalse(is_path_safe("../etc/passwd", tmpdir))
|
||||
# ── is_in_allowed_dirs ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
class TestIsInAllowedDirs:
|
||||
def test_path_in_allowed_dir(self, base_dir):
|
||||
filepath = os.path.join(base_dir, "test.mp4")
|
||||
from pathlib import Path
|
||||
|
||||
assert is_in_allowed_dirs(filepath, [Path(base_dir)]) is True
|
||||
|
||||
def test_path_not_in_allowed_dir(self, base_dir):
|
||||
from pathlib import Path
|
||||
|
||||
assert is_in_allowed_dirs("/etc/passwd", [Path(base_dir)]) is False
|
||||
|
||||
def test_subdirectory_in_allowed(self, base_dir):
|
||||
from pathlib import Path
|
||||
|
||||
sub = os.path.join(base_dir, "subdir", "audio.mp3")
|
||||
assert is_in_allowed_dirs(sub, [Path(base_dir)]) is True
|
||||
|
||||
def test_none_allowed_dirs_uses_default(self):
|
||||
# None 使用默认配置(包含 /tmp)
|
||||
result = is_in_allowed_dirs("/tmp/test.mp4")
|
||||
assert isinstance(result, bool)
|
||||
|
||||
def test_allowed_dirs_list_is_empty(self):
|
||||
from pathlib import Path
|
||||
|
||||
assert is_in_allowed_dirs("/tmp/test", []) is False
|
||||
|
||||
|
||||
# ── PathSecurityError class ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestPathSecurityError:
|
||||
def test_is_value_error(self):
|
||||
assert issubclass(PathSecurityError, ValueError)
|
||||
|
||||
def test_message_preserved(self):
|
||||
err = PathSecurityError("test message")
|
||||
assert str(err) == "test message"
|
||||
|
||||
Reference in New Issue
Block a user