test(unit): 第86波 - worker层VoiceExtractor/CoverGenerator/VideoProcessor纯逻辑单测 (+63)

- VoiceExtractor: 命令构建 + 滤镜顺序 + 边界用例 + 异常传递 (19个) - CoverGenerator: 时间钳制 + 智能选帧均匀分布 + 文件大小选最佳 (23个) - VideoProcessor: 数据类 + fps解析 + concat文件格式 + 缩略图参数 (21个) - 全部通过mock外部依赖(ffmpeg/run_ffmpeg)实现，纯逻辑验证，无需真实FFmpeg
test(wave85): exceptions+bgm_utils+subtitle深度补充 +59 (#898 )
2026-07-26 00:26:06 +08:00 · 2026-07-26 00:13:44 +08:00 · 2026-07-26 00:13:44 +08:00 · 2026-07-25 23:55:16 +08:00
12 changed files with 2199 additions and 456 deletions
@@ -3,29 +3,11 @@
 * 两栏布局：左侧视频库列表（260px）+ 右侧素材网格
 * 使用 useQuery 对接后端真实 API（api/assets.ts）
 */
-import React, { useMemo, useState } from "react"
-import { Upload, message } from "antd"
+import React, { useState } from "react"
+import { Upload } from "antd"
 import { InboxOutlined, PictureOutlined, ExclamationCircleOutlined } from "@ant-design/icons"
-import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"
-import {
-  getAssetLibraries,
-  createAssetLibrary,
-  deleteAssetLibrary,
-  getAssets,
-  deleteAsset,
-  uploadAssetDirect,
-  getAssetDiagnosis,
-  batchDeleteAssets,
-  batchTagAssets,
-  batchClassifyAssets,
-  batchMarkAssets,
-  type AssetLibraryItem,
-  type AssetItem as ApiAssetItem,
-  type BatchOperationResult,
-} from "@/api/assets"
 import { Button } from "@/components/ui"
-import { type AssetItem, type AssetKind, mapLibrary, mapAsset } from "@/pages/assets/types"
-import { MAX_FILE_SIZE, LARGE_FILE_THRESHOLD } from "@/pages/assets/constants"
+import type { AssetItem } from "@/pages/assets/types"
 import AssetCard from "@/pages/assets/components/AssetCard"
 import { SkeletonCard } from "@/pages/assets/components/AssetSkeleton"
 import LibrarySidebar from "@/pages/assets/components/LibrarySidebar"
@@ -36,435 +18,102 @@ import PlayModal from "@/pages/assets/components/PlayModal"
 import BatchTagModal from "@/pages/assets/components/BatchTagModal"
 import BatchClassifyModal from "@/pages/assets/components/BatchClassifyModal"
 import BatchMarkModal from "@/pages/assets/components/BatchMarkModal"
-import type { SmartViewType } from "@/pages/assets/components/BatchMarkModal"
 import ResultDrawer from "@/pages/assets/components/ResultDrawer"
 import UploadProgressModal from "@/pages/assets/components/UploadProgressModal"
+import { useAssetsData } from "@/pages/assets/hooks/useAssetsData"
+import { useLibraryManagement } from "@/pages/assets/hooks/useLibraryManagement"
+import { useAssetUpload } from "@/pages/assets/hooks/useAssetUpload"
+import { useAssetSelection } from "@/pages/assets/hooks/useAssetSelection"
+import { useAssetOperations } from "@/pages/assets/hooks/useAssetOperations"
 import "./assets.css"

 /* ============================================================
 * 主组件
 * ============================================================ */
 const AssetLibrary: React.FC = () => {
-  const queryClient = useQueryClient()
-
-  /* ── 获取视频库列表 ── */
-  const { data: apiLibraries = [], isLoading: libLoading } = useQuery<AssetLibraryItem[], Error>({
-    queryKey: ["asset-libraries"],
-    queryFn: getAssetLibraries,
-    staleTime: 60_000,
-  })
-
-  const libraries = useMemo(
-    () =>
-      (Array.isArray(apiLibraries) ? apiLibraries : [])
-        .map(mapLibrary)
-        .filter((lib) => lib.kind === "video"),
-    [apiLibraries],
-  )
-
-  /* ── 当前选中的视频库 ── */
-  const [activeLibId, setActiveLibId] = useState<string>("")
-
-  // 当库列表加载完成后，自动选中第一个
-  const effectiveLibId = activeLibId || libraries[0]?.id || ""
-
-  /* ── 获取当前库的素材列表 ── */
+  /* ── 数据查询与筛选 ── */
  const {
-    data: apiAssets = { items: [], total: 0 },
-    isLoading: assetsLoading,
-    isError: assetsError,
-    error: assetsErrorObj,
-    refetch: refetchAssets,
-  } = useQuery<{ items: ApiAssetItem[]; total: number }, Error>({
-    queryKey: ["assets", effectiveLibId],
-    queryFn: () =>
-      getAssets(effectiveLibId, {
-        // 拉取所有非删除状态的素材，让用户上传后立刻能看到"处理中"的素材
-        status: "ready,uploading,ingesting,processing,pending,error,failed",
-      }),
-    enabled: !!effectiveLibId,
-    staleTime: 30_000,
+    libraries,
+    libLoading,
+    activeLibId,
+    setActiveLibId,
+    effectiveLibId,
+    assetsLoading,
+    assetsError,
+    assetsErrorObj,
+    refetchAssets,
+    searchText,
+    setSearchText,
+    filterType,
+    setFilterType,
+    filterTime,
+    setFilterTime,
+    filteredAssets,
+  } = useAssetsData()
+
+  /* ── 视频库管理 ── */
+  const {
+    createModalOpen,
+    setCreateModalOpen,
+    newLibName,
+    setNewLibName,
+    newLibKind,
+    setNewLibKind,
+    isCreating,
+    handleCreateLibrary,
+    handleDeleteLibrary,
+  } = useLibraryManagement({
+    libraries,
+    activeLibId,
+    setActiveLibId,
+    effectiveLibId,
  })

-  const assets = useMemo(
-    () => (Array.isArray(apiAssets?.items) ? apiAssets.items : []).map(mapAsset),
-    [apiAssets],
-  )
+  /* ── 上传 ── */
+  const { uploading, uploadProgress, handleUpload } = useAssetUpload({ effectiveLibId })

-  /* ── Mutations ── */
-  const createLibMutation = useMutation({
-    mutationFn: createAssetLibrary,
-    onSuccess: () => {
-      queryClient.invalidateQueries({ queryKey: ["asset-libraries"] })
-      message.success("视频库创建成功")
-    },
-    onError: () => {
-      message.error("创建视频库失败")
-    },
+  /* ── 选中态管理 ── */
+  const { selectedIds, setSelectedIds, toggleSelect, selectAll, deselectAll } = useAssetSelection({
+    filteredAssets,
  })

-  const deleteLibMutation = useMutation({
-    mutationFn: deleteAssetLibrary,
-    onSuccess: () => {
-      queryClient.invalidateQueries({ queryKey: ["asset-libraries"] })
-      message.success("视频库已删除")
-    },
-    onError: () => {
-      message.error("删除视频库失败")
-    },
-  })
+  /* ── 素材操作 ── */
+  const {
+    diagnosingId,
+    handleDiagnose,
+    handleSingleDelete,
+    batchLoading,
+    tagModalOpen,
+    setTagModalOpen,
+    batchTagInput,
+    setBatchTagInput,
+    batchTags,
+    setBatchTags,
+    tagMode,
+    setTagMode,
+    handleBatchTag,
+    handleTagInputKeyDown,
+    removeBatchTag,
+    classifyModalOpen,
+    setClassifyModalOpen,
+    batchCategory,
+    setBatchCategory,
+    handleBatchClassify,
+    markModalOpen,
+    setMarkModalOpen,
+    batchSmartView,
+    setBatchSmartView,
+    handleBatchMark,
+    handleBatchDelete,
+    resultDrawerOpen,
+    operationResult,
+    operationTitle,
+    handleResultDrawerClose,
+  } = useAssetOperations({ selectedIds, setSelectedIds })

-  /* 状态 */
-  const [selectedIds, setSelectedIds] = useState<Set<string>>(new Set())
-
-  // 大文件直传由 handleUpload 直接调用 uploadAssetDirect 处理
-
-  /* 筛选 */
-  const [searchText, setSearchText] = useState("")
-  const [filterType, setFilterType] = useState<string>("all")
-  const [filterTime, setFilterTime] = useState<string>("all")
-
-  /* 上传 */
-  const [uploading, setUploading] = useState(false)
-  const [uploadProgress, setUploadProgress] = useState(0)
-
-  /* 新建视频库 */
-  const [createModalOpen, setCreateModalOpen] = useState(false)
-  const [newLibName, setNewLibName] = useState("")
-  const [newLibKind, setNewLibKind] = useState<AssetKind>("video")
-
-  /* 视频播放 */
+  /* ── 视频播放 ── */
  const [playingAsset, setPlayingAsset] = useState<AssetItem | null>(null)

-  /* 诊断中状态 — 记录正在诊断的素材 ID */
-  const [diagnosingId, setDiagnosingId] = useState<string | null>(null)
-
-  /* ── 批量操作弹窗状态 ── */
-  const [tagModalOpen, setTagModalOpen] = useState(false)
-  const [classifyModalOpen, setClassifyModalOpen] = useState(false)
-  const [markModalOpen, setMarkModalOpen] = useState(false)
-  const [resultDrawerOpen, setResultDrawerOpen] = useState(false)
-
-  /* 批量打标签 */
-  const [batchTagInput, setBatchTagInput] = useState("")
-  const [batchTags, setBatchTags] = useState<string[]>([])
-  const [tagMode, setTagMode] = useState<"add" | "replace">("add")
-
-  /* 批量改分类 */
-  const [batchCategory, setBatchCategory] = useState("")
-
-  /* 批量智能标记 */
-  const [batchSmartView, setBatchSmartView] = useState<SmartViewType>("recommended")
-
-  /* 操作结果 */
-  const [operationResult, setOperationResult] = useState<BatchOperationResult | null>(null)
-  const [operationTitle, setOperationTitle] = useState("")
-
-  /* 批量操作 loading */
-  const [batchLoading, setBatchLoading] = useState(false)
-
-  /* 派生数据 */
-  const filteredAssets = useMemo(() => {
-    let list = assets
-
-    /* 按视频库类型过滤（如果筛选类型不是 all） */
-    if (filterType !== "all") {
-      list = list.filter((a) => a.kind === filterType)
-    }
-
-    /* 按时间筛选 */
-    if (filterTime !== "all") {
-      const now = new Date()
-      list = list.filter((a) => {
-        const d = new Date(a.createdAt)
-        const diffDays = (now.getTime() - d.getTime()) / (1000 * 60 * 60 * 24)
-        if (filterTime === "today") return diffDays < 1
-        if (filterTime === "week") return diffDays < 7
-        if (filterTime === "month") return diffDays < 30
-        return true
-      })
-    }
-
-    /* 搜索 */
-    if (searchText.trim()) {
-      const q = searchText.trim().toLowerCase()
-      list = list.filter((a) => a.name.toLowerCase().includes(q))
-    }
-
-    return list
-  }, [assets, filterType, filterTime, searchText])
-
-  /* 选择操作 */
-  const toggleSelect = (id: string) => {
-    setSelectedIds((prev) => {
-      const next = new Set(prev)
-      if (next.has(id)) next.delete(id)
-      else next.add(id)
-      return next
-    })
-  }
-
-  const selectAll = () => {
-    setSelectedIds(new Set(filteredAssets.map((a) => a.id)))
-  }
-
-  const deselectAll = () => {
-    setSelectedIds(new Set())
-  }
-
-  /* 上传 — 调用真实 API */
-  const handleUpload = async (file: File) => {
-    if (file.size > MAX_FILE_SIZE) {
-      message.error(`文件 "${file.name}" 超过 2GB 限制`)
-      return
-    }
-    if (!effectiveLibId) {
-      message.warning("请先选择或创建一个视频库")
-      return
-    }
-
-    setUploading(true)
-    setUploadProgress(0)
-    try {
-      if (file.size > LARGE_FILE_THRESHOLD) {
-        message.info(`大文件 "${file.name}" 将使用直传上传`)
-      }
-      await uploadAssetDirect({
-        file,
-        library_id: effectiveLibId,
-        onProgress: (pct) => setUploadProgress(pct),
-      })
-      message.success(`"${file.name}" 上传成功`)
-      queryClient.invalidateQueries({ queryKey: ["assets"] })
-      queryClient.invalidateQueries({ queryKey: ["asset-libraries"] })
-    } catch (err: unknown) {
-      const detail = err instanceof Error ? err.message : ""
-      console.error("[handleUpload] 上传失败:", err)
-      message.error(`"${file.name}" 上传失败${detail ? `：${detail}` : ""}`)
-      // 错误时延迟关闭弹窗，让用户能看到错误提示
-      await new Promise((r) => setTimeout(r, 1500))
-    } finally {
-      setUploading(false)
-      setUploadProgress(0)
-    }
-  }
-
-  /* 新建视频库 */
-  const handleCreateLibrary = async () => {
-    if (!newLibName.trim()) {
-      message.warning("请输入视频库名称")
-      return
-    }
-    try {
-      const newLib = await createLibMutation.mutateAsync({
-        name: newLibName.trim(),
-        kind: newLibKind,
-      })
-      setActiveLibId(newLib.id)
-      setCreateModalOpen(false)
-      setNewLibName("")
-      setNewLibKind("video")
-    } catch {
-      // error handled in mutation
-    }
-  }
-
-  /* 删除视频库 */
-  const handleDeleteLibrary = async (id: string) => {
-    try {
-      await deleteLibMutation.mutateAsync(id)
-      if (effectiveLibId === id) {
-        const remaining = libraries.filter((l) => l.id !== id)
-        if (remaining.length > 0) setActiveLibId(remaining[0].id)
-        else setActiveLibId("")
-      }
-    } catch {
-      // error handled in mutation
-    }
-  }
-
-  /* 诊断 — 调用真实 API，带 loading 状态 */
-  const handleDiagnose = async (asset: AssetItem) => {
-    setDiagnosingId(asset.id)
-    try {
-      const result = await getAssetDiagnosis(asset.id)
-      const score = result.readiness_score ?? "-"
-      message.success(`"${asset.name}" 诊断完成，就绪分：${score}`)
-      queryClient.invalidateQueries({ queryKey: ["assets"] })
-    } catch {
-      message.error(`"${asset.name}" 诊断失败`)
-    } finally {
-      setDiagnosingId(null)
-    }
-  }
-
-  /* 单个素材删除 */
-  const handleSingleDelete = async (assetId: string) => {
-    try {
-      await deleteAsset(assetId)
-      queryClient.invalidateQueries({ queryKey: ["assets"] })
-      queryClient.invalidateQueries({ queryKey: ["asset-libraries"] })
-      // 从选中集合中移除
-      setSelectedIds((prev) => {
-        const next = new Set(prev)
-        next.delete(assetId)
-        return next
-      })
-      message.success("素材已删除")
-    } catch {
-      message.error("删除失败，请重试")
-    }
-  }
-
-  /* 批量删除 */
-  const handleBatchDelete = async () => {
-    const ids = Array.from(selectedIds)
-    setBatchLoading(true)
-    try {
-      const result = await batchDeleteAssets(ids)
-      setOperationResult(result)
-      setOperationTitle("批量删除")
-      setResultDrawerOpen(true)
-      queryClient.invalidateQueries({ queryKey: ["assets"] })
-      queryClient.invalidateQueries({ queryKey: ["asset-libraries"] })
-      setSelectedIds(new Set())
-      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)
-    }
-  }
-
-  /* 批量打标签 */
-  const handleBatchTag = 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,
-      })
-      setOperationResult(result)
-      setOperationTitle("批量打标签")
-      setResultDrawerOpen(true)
-      setTagModalOpen(false)
-      setBatchTags([])
-      setBatchTagInput("")
-      setTagMode("add")
-      queryClient.invalidateQueries({ queryKey: ["assets"] })
-      setSelectedIds(new Set())
-      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)
-    }
-  }
-
-  /* 批量改分类 */
-  const handleBatchClassify = async () => {
-    if (!batchCategory) {
-      message.warning("请选择分类")
-      return
-    }
-    const ids = Array.from(selectedIds)
-    setBatchLoading(true)
-    try {
-      const result = await batchClassifyAssets({
-        asset_ids: ids,
-        category: batchCategory,
-      })
-      setOperationResult(result)
-      setOperationTitle("批量改分类")
-      setResultDrawerOpen(true)
-      setClassifyModalOpen(false)
-      setBatchCategory("")
-      queryClient.invalidateQueries({ queryKey: ["assets"] })
-      setSelectedIds(new Set())
-      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)
-    }
-  }
-
-  /* 批量智能标记 */
-  const handleBatchMark = async () => {
-    const ids = Array.from(selectedIds)
-    setBatchLoading(true)
-    try {
-      const result = await batchMarkAssets({
-        asset_ids: ids,
-        smart_view: batchSmartView,
-      })
-      setOperationResult(result)
-      setOperationTitle("批量智能标记")
-      setResultDrawerOpen(true)
-      setMarkModalOpen(false)
-      queryClient.invalidateQueries({ queryKey: ["assets"] })
-      setSelectedIds(new Set())
-      const labelMap: Record<SmartViewType, string> = {
-        recommended: "推荐",
-        caution: "慎用",
-        high_risk: "高风险",
-      }
-      if (result.failure_count === 0) {
-        message.success(
-          `成功将 ${result.success_count} 个素材标记为「${labelMap[batchSmartView]}」`,
-        )
-      } else {
-        message.warning(
-          `智能标记完成：成功 ${result.success_count} 个，失败 ${result.failure_count} 个`,
-        )
-      }
-    } catch {
-      message.error("批量智能标记失败，请重试")
-    } finally {
-      setBatchLoading(false)
-    }
-  }
-
-  /* 标签输入处理 */
-  const handleTagInputKeyDown = (e: React.KeyboardEvent) => {
-    if (e.key === "Enter" && batchTagInput.trim()) {
-      e.preventDefault()
-      const tag = batchTagInput.trim()
-      if (!batchTags.includes(tag)) {
-        setBatchTags([...batchTags, tag])
-      }
-      setBatchTagInput("")
-    }
-  }
-
-  const removeBatchTag = (tag: string) => {
-    setBatchTags(batchTags.filter((t) => t !== tag))
-  }
-
  // ── Loading 状态 ──
  if (libLoading) {
    return (
@@ -499,8 +148,6 @@ const AssetLibrary: React.FC = () => {
          {/* 上传区域 */}
          <Upload.Dragger
            beforeUpload={(file) => {
-              // 同步返回 false 阻止 antd 默认上传行为
-              // 异步上传由 handleUpload 处理
              handleUpload(file as File)
              return false
            }}
@@ -595,7 +242,7 @@ const AssetLibrary: React.FC = () => {
        onNameChange={setNewLibName}
        kind={newLibKind}
        onKindChange={setNewLibKind}
-        confirmLoading={createLibMutation.isPending}
+        confirmLoading={isCreating}
      />

      {/* ─── 视频/音频播放弹窗 ─── */}
@@ -651,10 +298,7 @@ const AssetLibrary: React.FC = () => {
        open={resultDrawerOpen}
        title={operationTitle}
        result={operationResult}
-        onClose={() => {
-          setResultDrawerOpen(false)
-          setOperationResult(null)
-        }}
+        onClose={handleResultDrawerClose}
      />
    </div>
  )
@@ -0,0 +1,301 @@
+import { useState, useCallback } from "react"
+import { useQueryClient } from "@tanstack/react-query"
+import { message } from "antd"
+import {
+  deleteAsset,
+  getAssetDiagnosis,
+  batchDeleteAssets,
+  batchTagAssets,
+  batchClassifyAssets,
+  batchMarkAssets,
+  type BatchOperationResult,
+} from "@/api/assets"
+import type { AssetItem } from "../types"
+import type { SmartViewType } from "../components/BatchMarkModal"
+
+/**
+ * 素材操作 Hook
+ * 封装素材的诊断、删除、批量打标签、批量改分类、批量智能标记等操作，
+ * 以及相关弹窗和结果展示的状态管理
+ */
+interface UseAssetOperationsProps {
+  selectedIds: Set<string>
+  setSelectedIds: (ids: Set<string>) => void
+}
+
+export function useAssetOperations({ selectedIds, setSelectedIds }: UseAssetOperationsProps) {
+  const queryClient = useQueryClient()
+
+  /* ── 诊断状态 ── */
+  const [diagnosingId, setDiagnosingId] = useState<string | null>(null)
+
+  /* ── 批量操作弹窗状态 ── */
+  const [tagModalOpen, setTagModalOpen] = useState(false)
+  const [classifyModalOpen, setClassifyModalOpen] = useState(false)
+  const [markModalOpen, setMarkModalOpen] = useState(false)
+  const [resultDrawerOpen, setResultDrawerOpen] = useState(false)
+
+  /* ── 批量打标签表单 ── */
+  const [batchTagInput, setBatchTagInput] = useState("")
+  const [batchTags, setBatchTags] = useState<string[]>([])
+  const [tagMode, setTagMode] = useState<"add" | "replace">("add")
+
+  /* ── 批量改分类表单 ── */
+  const [batchCategory, setBatchCategory] = useState("")
+
+  /* ── 批量智能标记表单 ── */
+  const [batchSmartView, setBatchSmartView] = useState<SmartViewType>("recommended")
+
+  /* ── 操作结果 ── */
+  const [operationResult, setOperationResult] = useState<BatchOperationResult | null>(null)
+  const [operationTitle, setOperationTitle] = useState("")
+
+  /* ── 批量操作 loading ── */
+  const [batchLoading, setBatchLoading] = useState(false)
+
+  /* ── 刷新数据辅助函数 ── */
+  const invalidateAssets = useCallback(() => {
+    queryClient.invalidateQueries({ queryKey: ["assets"] })
+    queryClient.invalidateQueries({ queryKey: ["asset-libraries"] })
+  }, [queryClient])
+
+  /* ── 诊断 ── */
+  const handleDiagnose = useCallback(
+    async (asset: AssetItem) => {
+      setDiagnosingId(asset.id)
+      try {
+        const result = await getAssetDiagnosis(asset.id)
+        const score = result.readiness_score ?? "-"
+        message.success(`"${asset.name}" 诊断完成，就绪分：${score}`)
+        queryClient.invalidateQueries({ queryKey: ["assets"] })
+      } catch {
+        message.error(`"${asset.name}" 诊断失败`)
+      } finally {
+        setDiagnosingId(null)
+      }
+    },
+    [queryClient],
+  )
+
+  /* ── 单个素材删除 ── */
+  const handleSingleDelete = useCallback(
+    async (assetId: string) => {
+      try {
+        await deleteAsset(assetId)
+        invalidateAssets()
+        // 从选中集合中移除
+        setSelectedIds(
+          (() => {
+            const next = new Set(selectedIds)
+            next.delete(assetId)
+            return next
+          })(),
+        )
+        message.success("素材已删除")
+      } catch {
+        message.error("删除失败，请重试")
+      }
+    },
+    [invalidateAssets, selectedIds, setSelectedIds],
+  )
+
+  /* ── 显示操作结果 ── */
+  const showOperationResult = useCallback(
+    (result: BatchOperationResult, title: string, clearSelection = true) => {
+      setOperationResult(result)
+      setOperationTitle(title)
+      setResultDrawerOpen(true)
+      if (clearSelection) setSelectedIds(new Set())
+    },
+    [setSelectedIds],
+  )
+
+  /* ── 批量删除 ── */
+  const handleBatchDelete = useCallback(async () => {
+    const ids = Array.from(selectedIds)
+    setBatchLoading(true)
+    try {
+      const result = await batchDeleteAssets(ids)
+      invalidateAssets()
+      showOperationResult(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, showOperationResult])
+
+  /* ── 批量打标签 ── */
+  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"] })
+      showOperationResult(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, showOperationResult])
+
+  /* ── 标签输入处理 ── */
+  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],
+  )
+
+  /* ── 批量改分类 ── */
+  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"] })
+      showOperationResult(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, showOperationResult])
+
+  /* ── 批量智能标记 ── */
+  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"] })
+      showOperationResult(result, "批量智能标记")
+      setMarkModalOpen(false)
+      const labelMap: Record<SmartViewType, string> = {
+        recommended: "推荐",
+        caution: "慎用",
+        high_risk: "高风险",
+      }
+      if (result.failure_count === 0) {
+        message.success(
+          `成功将 ${result.success_count} 个素材标记为「${labelMap[batchSmartView]}」`,
+        )
+      } else {
+        message.warning(
+          `智能标记完成：成功 ${result.success_count} 个，失败 ${result.failure_count} 个`,
+        )
+      }
+    } catch {
+      message.error("批量智能标记失败，请重试")
+    } finally {
+      setBatchLoading(false)
+    }
+  }, [batchSmartView, selectedIds, queryClient, showOperationResult])
+
+  /* ── 关闭结果 Drawer ── */
+  const handleResultDrawerClose = useCallback(() => {
+    setResultDrawerOpen(false)
+    setOperationResult(null)
+  }, [])
+
+  return {
+    // 诊断
+    diagnosingId,
+    handleDiagnose,
+    // 单个操作
+    handleSingleDelete,
+    // 批量操作 loading
+    batchLoading,
+    // 批量打标签
+    tagModalOpen,
+    setTagModalOpen,
+    batchTagInput,
+    setBatchTagInput,
+    batchTags,
+    setBatchTags,
+    tagMode,
+    setTagMode,
+    handleBatchTag,
+    handleTagInputKeyDown,
+    removeBatchTag,
+    // 批量改分类
+    classifyModalOpen,
+    setClassifyModalOpen,
+    batchCategory,
+    setBatchCategory,
+    handleBatchClassify,
+    // 批量智能标记
+    markModalOpen,
+    setMarkModalOpen,
+    batchSmartView,
+    setBatchSmartView,
+    handleBatchMark,
+    // 批量删除
+    handleBatchDelete,
+    // 操作结果
+    resultDrawerOpen,
+    operationResult,
+    operationTitle,
+    handleResultDrawerClose,
+  }
+}
@@ -0,0 +1,40 @@
+import { useState, useCallback } from "react"
+import type { AssetItem } from "../types"
+
+/**
+ * 素材选中态管理 Hook
+ * 封装单选、全选、取消全选等选中逻辑
+ */
+interface UseAssetSelectionProps {
+  filteredAssets: AssetItem[]
+}
+
+export function useAssetSelection({ filteredAssets }: UseAssetSelectionProps) {
+  const [selectedIds, setSelectedIds] = useState<Set<string>>(new Set())
+
+  const toggleSelect = useCallback((id: string) => {
+    setSelectedIds((prev) => {
+      const next = new Set(prev)
+      if (next.has(id)) next.delete(id)
+      else next.add(id)
+      return next
+    })
+  }, [])
+
+  const selectAll = useCallback(() => {
+    setSelectedIds(new Set(filteredAssets.map((a) => a.id)))
+  }, [filteredAssets])
+
+  const deselectAll = useCallback(() => {
+    setSelectedIds(new Set())
+  }, [])
+
+  return {
+    selectedIds,
+    setSelectedIds,
+    toggleSelect,
+    selectAll,
+    deselectAll,
+    selectedCount: selectedIds.size,
+  }
+}
@@ -0,0 +1,65 @@
+import { useState, useCallback } from "react"
+import { useQueryClient } from "@tanstack/react-query"
+import { message } from "antd"
+import { uploadAssetDirect } from "@/api/assets"
+import { MAX_FILE_SIZE, LARGE_FILE_THRESHOLD } from "../constants"
+
+/**
+ * 素材上传 Hook
+ * 封装上传状态、进度管理和上传逻辑
+ */
+interface UseAssetUploadProps {
+  effectiveLibId: string
+}
+
+export function useAssetUpload({ effectiveLibId }: UseAssetUploadProps) {
+  const queryClient = useQueryClient()
+
+  const [uploading, setUploading] = useState(false)
+  const [uploadProgress, setUploadProgress] = useState(0)
+
+  const handleUpload = useCallback(
+    async (file: File) => {
+      if (file.size > MAX_FILE_SIZE) {
+        message.error(`文件 "${file.name}" 超过 2GB 限制`)
+        return
+      }
+      if (!effectiveLibId) {
+        message.warning("请先选择或创建一个视频库")
+        return
+      }
+
+      setUploading(true)
+      setUploadProgress(0)
+      try {
+        if (file.size > LARGE_FILE_THRESHOLD) {
+          message.info(`大文件 "${file.name}" 将使用直传上传`)
+        }
+        await uploadAssetDirect({
+          file,
+          library_id: effectiveLibId,
+          onProgress: (pct) => setUploadProgress(pct),
+        })
+        message.success(`"${file.name}" 上传成功`)
+        queryClient.invalidateQueries({ queryKey: ["assets"] })
+        queryClient.invalidateQueries({ queryKey: ["asset-libraries"] })
+      } catch (err: unknown) {
+        const detail = err instanceof Error ? err.message : ""
+        console.error("[handleUpload] 上传失败:", err)
+        message.error(`"${file.name}" 上传失败${detail ? `：${detail}` : ""}`)
+        // 错误时延迟关闭弹窗，让用户能看到错误提示
+        await new Promise((r) => setTimeout(r, 1500))
+      } finally {
+        setUploading(false)
+        setUploadProgress(0)
+      }
+    },
+    [effectiveLibId, queryClient],
+  )
+
+  return {
+    uploading,
+    uploadProgress,
+    handleUpload,
+  }
+}
@@ -0,0 +1,118 @@
+import { useState, useMemo } from "react"
+import { useQuery } from "@tanstack/react-query"
+import {
+  getAssetLibraries,
+  getAssets,
+  type AssetLibraryItem,
+  type AssetItem as ApiAssetItem,
+} from "@/api/assets"
+import { mapLibrary, mapAsset, type AssetItem, type LibraryItem } from "../types"
+
+/**
+ * 素材库数据 Hook
+ * 封装视频库列表、素材列表的数据查询，以及筛选、搜索状态管理
+ */
+export function useAssetsData() {
+  /* ── 视频库列表查询 ── */
+  const { data: apiLibraries = [], isLoading: libLoading } = useQuery<AssetLibraryItem[], Error>({
+    queryKey: ["asset-libraries"],
+    queryFn: getAssetLibraries,
+    staleTime: 60_000,
+  })
+
+  const libraries: LibraryItem[] = useMemo(
+    () =>
+      (Array.isArray(apiLibraries) ? apiLibraries : [])
+        .map(mapLibrary)
+        .filter((lib) => lib.kind === "video"),
+    [apiLibraries],
+  )
+
+  /* ── 当前选中的视频库 ── */
+  const [activeLibId, setActiveLibId] = useState<string>("")
+
+  // 当库列表加载完成后，自动选中第一个
+  const effectiveLibId = activeLibId || libraries[0]?.id || ""
+
+  /* ── 当前库的素材列表查询 ── */
+  const {
+    data: apiAssets = { items: [], total: 0 },
+    isLoading: assetsLoading,
+    isError: assetsError,
+    error: assetsErrorObj,
+    refetch: refetchAssets,
+  } = useQuery<{ items: ApiAssetItem[]; total: number }, Error>({
+    queryKey: ["assets", effectiveLibId],
+    queryFn: () =>
+      getAssets(effectiveLibId, {
+        // 拉取所有非删除状态的素材，让用户上传后立刻能看到"处理中"的素材
+        status: "ready,uploading,ingesting,processing,pending,error,failed",
+      }),
+    enabled: !!effectiveLibId,
+    staleTime: 30_000,
+  })
+
+  const assets: AssetItem[] = useMemo(
+    () => (Array.isArray(apiAssets?.items) ? apiAssets.items : []).map(mapAsset),
+    [apiAssets],
+  )
+
+  /* ── 筛选状态 ── */
+  const [searchText, setSearchText] = useState("")
+  const [filterType, setFilterType] = useState<string>("all")
+  const [filterTime, setFilterTime] = useState<string>("all")
+
+  /* ── 筛选后的素材列表 ── */
+  const filteredAssets = useMemo(() => {
+    let list = assets
+
+    /* 按素材类型过滤 */
+    if (filterType !== "all") {
+      list = list.filter((a) => a.kind === filterType)
+    }
+
+    /* 按时间筛选 */
+    if (filterTime !== "all") {
+      const now = new Date()
+      list = list.filter((a) => {
+        const d = new Date(a.createdAt)
+        const diffDays = (now.getTime() - d.getTime()) / (1000 * 60 * 60 * 24)
+        if (filterTime === "today") return diffDays < 1
+        if (filterTime === "week") return diffDays < 7
+        if (filterTime === "month") return diffDays < 30
+        return true
+      })
+    }
+
+    /* 搜索 */
+    if (searchText.trim()) {
+      const q = searchText.trim().toLowerCase()
+      list = list.filter((a) => a.name.toLowerCase().includes(q))
+    }
+
+    return list
+  }, [assets, filterType, filterTime, searchText])
+
+  return {
+    // 视频库
+    libraries,
+    libLoading,
+    activeLibId,
+    setActiveLibId,
+    effectiveLibId,
+    // 素材列表
+    assets,
+    assetsLoading,
+    assetsError,
+    assetsErrorObj,
+    refetchAssets,
+    // 筛选
+    searchText,
+    setSearchText,
+    filterType,
+    setFilterType,
+    filterTime,
+    setFilterTime,
+    filteredAssets,
+  }
+}
@@ -0,0 +1,106 @@
+import { useState, useCallback } from "react"
+import { useMutation, useQueryClient } from "@tanstack/react-query"
+import { message } from "antd"
+import { createAssetLibrary, deleteAssetLibrary } from "@/api/assets"
+import type { AssetKind, LibraryItem } from "../types"
+
+/**
+ * 视频库管理 Hook
+ * 封装视频库的创建、删除操作，以及新建弹窗的表单状态
+ */
+interface UseLibraryManagementProps {
+  libraries: LibraryItem[]
+  activeLibId: string
+  setActiveLibId: (id: string) => void
+  effectiveLibId: string
+}
+
+export function useLibraryManagement({
+  libraries,
+  setActiveLibId,
+  effectiveLibId,
+}: UseLibraryManagementProps) {
+  const queryClient = useQueryClient()
+
+  /* ── 状态 ── */
+  const [createModalOpen, setCreateModalOpen] = useState(false)
+  const [newLibName, setNewLibName] = useState("")
+  const [newLibKind, setNewLibKind] = useState<AssetKind>("video")
+
+  /* ── Mutations ── */
+  const createLibMutation = useMutation({
+    mutationFn: createAssetLibrary,
+    onSuccess: () => {
+      queryClient.invalidateQueries({ queryKey: ["asset-libraries"] })
+      message.success("视频库创建成功")
+    },
+    onError: () => {
+      message.error("创建视频库失败")
+    },
+  })
+
+  const deleteLibMutation = useMutation({
+    mutationFn: deleteAssetLibrary,
+    onSuccess: () => {
+      queryClient.invalidateQueries({ queryKey: ["asset-libraries"] })
+      message.success("视频库已删除")
+    },
+    onError: () => {
+      message.error("删除视频库失败")
+    },
+  })
+
+  /* ── 新建视频库 ── */
+  const handleCreateLibrary = useCallback(async () => {
+    if (!newLibName.trim()) {
+      message.warning("请输入视频库名称")
+      return
+    }
+    try {
+      const newLib = await createLibMutation.mutateAsync({
+        name: newLibName.trim(),
+        kind: newLibKind,
+      })
+      setActiveLibId(newLib.id)
+      setCreateModalOpen(false)
+      setNewLibName("")
+      setNewLibKind("video")
+    } catch {
+      // error handled in mutation
+    }
+  }, [newLibName, newLibKind, createLibMutation, setActiveLibId])
+
+  /* ── 删除视频库 ── */
+  const handleDeleteLibrary = useCallback(
+    async (id: string) => {
+      try {
+        await deleteLibMutation.mutateAsync(id)
+        if (effectiveLibId === id) {
+          const remaining = libraries.filter((l) => l.id !== id)
+          if (remaining.length > 0) setActiveLibId(remaining[0].id)
+          else setActiveLibId("")
+        }
+      } catch {
+        // error handled in mutation
+      }
+    },
+    [deleteLibMutation, effectiveLibId, libraries, setActiveLibId],
+  )
+
+  return {
+    // 弹窗状态
+    createModalOpen,
+    setCreateModalOpen,
+    // 表单状态
+    newLibName,
+    setNewLibName,
+    newLibKind,
+    setNewLibKind,
+    // Mutations
+    isCreating: createLibMutation.isPending,
+    isDeleting: deleteLibMutation.isPending,
+    // Handlers
+    handleCreateLibrary,
+    handleDeleteLibrary,
+  }
+}
@@ -1,13 +1,27 @@
 /**
 * AssetLibrary 模块 smoke test
 * 建立完整依赖链，确保 vitest related 模式能匹配到
- * assets 目录下所有文件的改动（包括子组件和工具函数）
+ * assets 目录下所有文件的改动（包括子组件、Hook 和工具函数）
 */
 import { describe, it, expect } from "vitest"

 // 主组件
 import "@/pages/assets/AssetLibrary"

+// 子组件
+import "@/pages/assets/components/AssetCard"
+import "@/pages/assets/components/AssetFilterBar"
+import "@/pages/assets/components/AssetSkeleton"
+import "@/pages/assets/components/BatchClassifyModal"
+import "@/pages/assets/components/BatchMarkModal"
+import "@/pages/assets/components/BatchOperationBar"
+import "@/pages/assets/components/BatchTagModal"
+import "@/pages/assets/components/CreateLibraryModal"
+import "@/pages/assets/components/LibrarySidebar"
+import "@/pages/assets/components/PlayModal"
+import "@/pages/assets/components/ResultDrawer"
+import "@/pages/assets/components/UploadProgressModal"
+
 // 类型与常量
 import "@/pages/assets/types"
 import "@/pages/assets/constants"
@@ -15,21 +29,13 @@ import "@/pages/assets/constants"
 // 工具函数
 import "@/pages/assets/utils/format"
 import "@/pages/assets/utils/asset"
-import "@/pages/assets/utils/kindIcon"

-// UI 组件
-import "@/pages/assets/components/AssetCard"
-import "@/pages/assets/components/AssetSkeleton"
-import "@/pages/assets/components/LibrarySidebar"
-import "@/pages/assets/components/AssetFilterBar"
-import "@/pages/assets/components/BatchOperationBar"
-import "@/pages/assets/components/CreateLibraryModal"
-import "@/pages/assets/components/PlayModal"
-import "@/pages/assets/components/BatchTagModal"
-import "@/pages/assets/components/BatchClassifyModal"
-import "@/pages/assets/components/BatchMarkModal"
-import "@/pages/assets/components/ResultDrawer"
-import "@/pages/assets/components/UploadProgressModal"
+// Hooks
+import "@/pages/assets/hooks/useAssetsData"
+import "@/pages/assets/hooks/useLibraryManagement"
+import "@/pages/assets/hooks/useAssetUpload"
+import "@/pages/assets/hooks/useAssetSelection"
+import "@/pages/assets/hooks/useAssetOperations"

 describe("AssetLibrary module smoke test", () => {
  it("should load all asset modules", () => {
@@ -206,4 +206,167 @@ class TestClassificationJobCreate:

    def test_create_id_is_hex(self):
        job = ClassificationJob.create(project_id="p", asset_id="a")
-        int(job.id, 16)  # 不抛异常就是合法 hex
+        assert job.created_at.tzinfo is not None
+        assert job.updated_at.tzinfo is not None
+
+
+class TestClassificationJobState:
+    """ClassificationJob 状态操作测试"""
+
+    def test_set_processing(self):
+        job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
+        job.status = ClassificationJobStatus.PROCESSING
+        assert job.status == ClassificationJobStatus.PROCESSING
+
+    def test_set_completed_with_result(self):
+        job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
+        job.status = ClassificationJobStatus.COMPLETED
+        job.classification = AssetClassification.SCENIC
+        job.confidence = 0.95
+        assert job.status == ClassificationJobStatus.COMPLETED
+        assert job.classification == "scenic"
+        assert job.confidence == pytest.approx(0.95)
+
+    def test_set_failed_with_error(self):
+        job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
+        job.status = ClassificationJobStatus.FAILED
+        job.error_message = "model timeout"
+        assert job.status == ClassificationJobStatus.FAILED
+        assert job.error_message == "model timeout"
+
+    def test_confidence_range_zero(self):
+        job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
+        job.confidence = 0.0
+        assert job.confidence == 0.0
+
+    def test_confidence_range_one(self):
+        job = ClassificationJob.create(project_id="proj-1", asset_id="asset-1")
+        job.confidence = 1.0
+        assert job.confidence == 1.0
+
+
+class TestClassificationJobStatusMissing:
+    """ClassificationJobStatus._missing_ 兼容行为测试"""
+
+    def test_done_maps_to_completed(self):
+        assert ClassificationJobStatus("done") == ClassificationJobStatus.COMPLETED
+
+    def test_success_maps_to_completed(self):
+        assert ClassificationJobStatus("success") == ClassificationJobStatus.COMPLETED
+
+    def test_finished_maps_to_completed(self):
+        assert ClassificationJobStatus("finished") == ClassificationJobStatus.COMPLETED
+
+    def test_complete_maps_to_completed(self):
+        assert ClassificationJobStatus("complete") == ClassificationJobStatus.COMPLETED
+
+    def test_fail_maps_to_failed(self):
+        assert ClassificationJobStatus("fail") == ClassificationJobStatus.FAILED
+
+    def test_error_maps_to_failed(self):
+        assert ClassificationJobStatus("error") == ClassificationJobStatus.FAILED
+
+    def test_err_maps_to_failed(self):
+        assert ClassificationJobStatus("err") == ClassificationJobStatus.FAILED
+
+    def test_unknown_maps_to_pending(self):
+        assert ClassificationJobStatus("unknown_status") == ClassificationJobStatus.PENDING
+
+    def test_case_insensitive_mapping(self):
+        assert ClassificationJobStatus("DONE") == ClassificationJobStatus.COMPLETED
+        assert ClassificationJobStatus("Done") == ClassificationJobStatus.COMPLETED
+
+    def test_whitespace_stripped(self):
+        assert ClassificationJobStatus("  done  ") == ClassificationJobStatus.COMPLETED
+
+
+class TestClassificationJobExtended:
+    """ClassificationJob 深度补充测试"""
+
+    def test_id_is_hex(self):
+        job = ClassificationJob.create(project_id="p", asset_id="a")
+        int(job.id, 16)
+
+    def test_empty_classification(self):
+        job = ClassificationJob.create(project_id="p", asset_id="a")
+        assert job.classification == ""
+
+    def test_zero_confidence(self):
+        job = ClassificationJob.create(project_id="p", asset_id="a")
+        assert job.confidence == 0.0
+
+    def test_high_confidence(self):
+        job = ClassificationJob.create(project_id="p", asset_id="a")
+        job.confidence = 0.99
+        assert job.confidence == pytest.approx(0.99)
+
+    def test_negative_confidence(self):
+        job = ClassificationJob.create(project_id="p", asset_id="a")
+        job.confidence = -0.1
+        assert job.confidence == pytest.approx(-0.1)
+
+    def test_confidence_over_one(self):
+        job = ClassificationJob.create(project_id="p", asset_id="a")
+        job.confidence = 1.5
+        assert job.confidence == pytest.approx(1.5)
+
+    def test_empty_error_message(self):
+        job = ClassificationJob.create(project_id="p", asset_id="a")
+        assert job.error_message == ""
+
+    def test_long_error_message(self):
+        job = ClassificationJob.create(project_id="p", asset_id="a")
+        long_msg = "error" * 100
+        job.error_message = long_msg
+        assert job.error_message == long_msg
+        assert len(job.error_message) == 500
+
+    def test_status_with_string_assignment(self):
+        job = ClassificationJob.create(project_id="p", asset_id="a")
+        job.status = "processing"
+        assert job.status == ClassificationJobStatus.PROCESSING
+
+
+class TestAssetLibraryKindExtended:
+    """AssetLibraryKind 深度补充测试"""
+
+    def test_image_value(self):
+        assert AssetLibraryKind.IMAGE == "image"
+
+    def test_all_three_kinds(self):
+        assert len(AssetLibraryKind) == 3
+
+    def test_is_string_enum(self):
+        assert isinstance(AssetLibraryKind.VIDEO, str)
+
+    def test_from_string(self):
+        assert AssetLibraryKind("video") == AssetLibraryKind.VIDEO
+
+
+class TestIngestJobStatusExtended:
+    """IngestJobStatus 深度补充测试"""
+
+    def test_is_string_enum(self):
+        assert isinstance(IngestJobStatus.PENDING, str)
+
+    def test_total_count(self):
+        assert len(IngestJobStatus) == 4
+
+    def test_from_string(self):
+        assert IngestJobStatus("pending") == IngestJobStatus.PENDING
+
+
+class TestAssetClassificationExtended:
+    """AssetClassification 深度补充测试"""
+
+    def test_total_count(self):
+        assert len(AssetClassification) == 9
+
+    def test_is_string_enum(self):
+        assert isinstance(AssetClassification.SCENIC, str)
+
+    def test_from_string(self):
+        assert AssetClassification("scenic") == AssetClassification.SCENIC
+
+    def test_other_category(self):
+        assert AssetClassification.OTHER == "other"
@@ -0,0 +1,539 @@
+"""CoverGenerator 纯逻辑单测 — 时间钳制 + 智能选帧算法.
+
+通过 mock run_ffmpeg 和 probe_video_info 验证纯逻辑部分，
+不实际执行 FFmpeg，确保测试轻量快速。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from video_processing.cover_generator import (
+    CoverGenerator,
+    DEFAULT_COVER_HEIGHT,
+    DEFAULT_COVER_QUALITY,
+    DEFAULT_COVER_TIME,
+    DEFAULT_COVER_WIDTH,
+    SMART_COVER_FRAME_COUNT,
+)
+
+
+class TestCoverGeneratorConstants:
+    """常量默认值测试."""
+
+    def test_default_cover_time(self):
+        """默认抽帧时间为 1.0 秒."""
+        assert DEFAULT_COVER_TIME == 1.0
+
+    def test_default_dimensions(self):
+        """默认封面尺寸 1080x1920 (竖屏)."""
+        assert DEFAULT_COVER_WIDTH == 1080
+        assert DEFAULT_COVER_HEIGHT == 1920
+
+    def test_default_quality(self):
+        """默认质量为 5 (JPEG q:v, 越小越好)."""
+        assert DEFAULT_COVER_QUALITY == 5
+
+    def test_smart_cover_frame_count(self):
+        """智能封面默认抽 3 帧."""
+        assert SMART_COVER_FRAME_COUNT == 3
+
+
+class TestExtractFrameCommand:
+    """extract_frame 命令构建测试."""
+
+    def _probe_video_info_mock(self, duration=10.0):
+        """创建 probe_video_info 的 mock."""
+        return {"duration": duration, "width": 1920, "height": 1080, "fps": 25.0}
+
+    def test_default_params_command(self, tmp_path):
+        """默认参数下 FFmpeg 命令正确."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+            # 让 output_path 在 run_ffmpeg 后存在
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            result = CoverGenerator.extract_frame(str(video_file), str(output_file))
+
+        assert result == Path(output_file)
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+
+        # 基本结构
+        assert cmd[0].endswith("ffmpeg") or "ffmpeg" in cmd[0]
+        assert "-y" in cmd
+        assert "-vframes" in cmd
+        assert cmd[cmd.index("-vframes") + 1] == "1"
+        assert "-f" in cmd
+        assert "mjpeg" in cmd[cmd.index("-f") + 1]
+
+        # 时间点
+        ss_idx = cmd.index("-ss")
+        assert float(cmd[ss_idx + 1]) == pytest.approx(DEFAULT_COVER_TIME, abs=0.001)
+
+        # 输入文件
+        i_idx = cmd.index("-i")
+        assert cmd[i_idx + 1] == str(video_file)
+
+        # 输出文件
+        assert cmd[-1] == str(output_file)
+
+        # scale + crop 滤镜
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "scale=" in vf_value
+        assert "crop=" in vf_value
+        assert "force_original_aspect_ratio=increase" in vf_value
+
+    def test_custom_time(self, tmp_path):
+        """自定义抽帧时间点."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(duration=30.0),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=5.5)
+
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        assert float(cmd[ss_idx + 1]) == pytest.approx(5.5, abs=0.001)
+
+    def test_custom_dimensions(self, tmp_path):
+        """自定义输出尺寸."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), width=1920, height=1080)
+
+        cmd = mock_run.call_args[0][0]
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "scale=1920:1080:" in vf_value
+        assert "crop=1920:1080" in vf_value
+
+    def test_custom_quality(self, tmp_path):
+        """自定义 JPEG 质量."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), quality=2)
+
+        cmd = mock_run.call_args[0][0]
+        q_idx = cmd.index("-q:v")
+        assert cmd[q_idx + 1] == "2"
+
+    def test_time_exceeds_duration_clamps_to_midpoint(self, tmp_path):
+        """抽帧时间超过视频时长时，钳制到中间帧."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(duration=5.0),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=10.0)
+
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        # 钳制到 duration/2 = 2.5
+        assert float(cmd[ss_idx + 1]) == pytest.approx(2.5, abs=0.001)
+
+    def test_negative_time_clamps_to_zero(self, tmp_path):
+        """负时间钳制到 0."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(duration=10.0),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=-2.0)
+
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        assert float(cmd[ss_idx + 1]) == pytest.approx(0.0, abs=0.001)
+
+    def test_time_equals_duration_clamps_to_midpoint(self, tmp_path):
+        """时间点等于时长时钳制到中间帧."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(duration=10.0),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=10.0)
+
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        assert float(cmd[ss_idx + 1]) == pytest.approx(5.0, abs=0.001)
+
+    def test_zero_duration_video(self, tmp_path):
+        """视频时长为 0 时的行为（不钳制，用原始时间）."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(duration=0.0),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=0.5)
+
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        assert float(cmd[ss_idx + 1]) == pytest.approx(0.5, abs=0.001)
+
+    def test_video_not_found_raises(self, tmp_path):
+        """视频文件不存在时抛出 FileNotFoundError."""
+        output_file = tmp_path / "cover.jpg"
+
+        with pytest.raises(FileNotFoundError):
+            CoverGenerator.extract_frame(str(tmp_path / "nonexistent.mp4"), str(output_file))
+
+    def test_output_creates_parent_dir(self, tmp_path):
+        """输出目录不存在时自动创建."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        out_dir = tmp_path / "deep" / "nested"
+        output_file = out_dir / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file))
+
+        assert out_dir.exists()
+        assert out_dir.is_dir()
+
+    def test_ffmpeg_failure_propagates(self, tmp_path):
+        """FFmpeg 失败时异常向上传递."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(),
+            ),
+            patch(
+                "video_processing.cover_generator.run_ffmpeg",
+                side_effect=RuntimeError("FFmpeg error"),
+            ),
+        ):
+            with pytest.raises(RuntimeError, match="FFmpeg error"):
+                CoverGenerator.extract_frame(str(video_file), str(output_file))
+
+
+class TestSmartCoverTimePoints:
+    """智能封面时间点计算测试."""
+
+    def test_single_frame_falls_back_to_default(self, tmp_path):
+        """只有 1 帧时退化为普通抽帧（取 DEFAULT_COVER_TIME 和 midpoint 中较小值）."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value={"duration": 20.0},
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            # frame_count=1 时退化为普通抽帧
+            CoverGenerator.extract_smart_cover(str(video_file), str(output_file), frame_count=1)
+
+        # 只调用一次（退化路径）
+        assert mock_run.call_count == 1
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        # min(DEFAULT_COVER_TIME=1.0, duration/2=10.0) = 1.0
+        assert float(cmd[ss_idx + 1]) == pytest.approx(1.0, abs=0.001)
+
+    def test_zero_duration_falls_back(self, tmp_path):
+        """视频时长为 0 时退化为普通抽帧."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value={"duration": 0.0},
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
+
+        # 只调用一次（退化路径）
+        assert mock_run.call_count == 1
+
+    def test_three_frames_uniform_distribution(self, tmp_path):
+        """3 帧均匀分布在 5%~95% 区间."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        call_times = []
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value={"duration": 100.0},
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                # 记录抽帧时间
+                ss_idx = cmd.index("-ss")
+                call_times.append(float(cmd[ss_idx + 1]))
+                # 在输出路径写文件
+                output_arg = cmd[-1]
+                Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
+                # 不同文件大小，让第三帧"最清晰"
+                idx = len(call_times) - 1
+                size = 1000 * (idx + 1)  # 递增的文件大小
+                Path(output_arg).write_bytes(b"x" * size)
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
+
+        # 3 帧：5%、50%、95%
+        assert len(call_times) == 3
+        assert call_times[0] == pytest.approx(5.0, abs=0.1)  # 5%
+        assert call_times[1] == pytest.approx(50.0, abs=0.1)  # 50%
+        assert call_times[2] == pytest.approx(95.0, abs=0.1)  # 95%
+
+    def test_five_frames_distribution(self, tmp_path):
+        """5 帧均匀分布."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        call_times = []
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value={"duration": 100.0},
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                ss_idx = cmd.index("-ss")
+                call_times.append(float(cmd[ss_idx + 1]))
+                output_arg = cmd[-1]
+                Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
+                idx = len(call_times) - 1
+                Path(output_arg).write_bytes(b"x" * (1000 * (idx + 1)))
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_smart_cover(str(video_file), str(output_file), frame_count=5)
+
+        assert len(call_times) == 5
+        # step = (95-5) / (5-1) = 22.5
+        # times: 5, 27.5, 50, 72.5, 95
+        assert call_times[0] == pytest.approx(5.0, abs=0.1)
+        assert call_times[1] == pytest.approx(27.5, abs=0.1)
+        assert call_times[2] == pytest.approx(50.0, abs=0.1)
+        assert call_times[3] == pytest.approx(72.5, abs=0.1)
+        assert call_times[4] == pytest.approx(95.0, abs=0.1)
+
+    def test_selects_largest_file_as_best(self, tmp_path):
+        """选择文件最大的帧作为最佳封面（清晰度近似）."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        sizes = [5000, 15000, 8000]  # 第二帧最大
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value={"duration": 100.0},
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+            call_idx = [0]
+
+            def fake_run(cmd):
+                output_arg = cmd[-1]
+                Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
+                idx = call_idx[0]
+                Path(output_arg).write_bytes(b"x" * sizes[idx])
+                call_idx[0] += 1
+
+            mock_run.side_effect = fake_run
+            result = CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
+
+        # 第二帧（索引1）应该是最佳
+        assert result == output_file
+        # 输出文件大小应等于第二帧大小
+        assert output_file.stat().st_size == 15000
+
+
+class TestProcessCustomCover:
+    """自定义封面处理测试."""
+
+    def test_custom_cover_resize_command(self, tmp_path):
+        """自定义封面调整尺寸命令正确."""
+        input_file = tmp_path / "upload.jpg"
+        input_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with patch("video_processing.cover_generator.run_ffmpeg") as mock_run:
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.process_custom_cover(str(input_file), str(output_file))
+
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+
+        assert "-i" in cmd
+        assert cmd[cmd.index("-i") + 1] == str(input_file)
+        assert cmd[-1] == str(output_file)
+
+        # scale + crop
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "scale=" in vf_value
+        assert "crop=" in vf_value
+
+    def test_custom_cover_not_found_raises(self, tmp_path):
+        """自定义封面文件不存在时抛出 FileNotFoundError."""
+        output_file = tmp_path / "cover.jpg"
+
+        with pytest.raises(FileNotFoundError):
+            CoverGenerator.process_custom_cover(str(tmp_path / "nonexistent.jpg"), str(output_file))
+
+    def test_custom_cover_custom_dimensions(self, tmp_path):
+        """自定义封面自定义输出尺寸."""
+        input_file = tmp_path / "upload.jpg"
+        input_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with patch("video_processing.cover_generator.run_ffmpeg") as mock_run:
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.process_custom_cover(str(input_file), str(output_file), width=800, height=600)
+
+        cmd = mock_run.call_args[0][0]
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "scale=800:600:" in vf_value
+        assert "crop=800:600" in vf_value
@@ -202,3 +202,139 @@ class TestTtsConfigClamp:
        config = TtsConfig.parse(data)
        assert isinstance(config.volume, float)
        assert config.volume == 1.0
+
+
+class TestTtsConfigTextEdge:
+    """文本字段边界测试."""
+
+    def test_long_text_preserved(self):
+        long_text = "配音文本" * 500
+        data = {"enabled": True, "voice_id": "v1", "text": long_text}
+        config = TtsConfig.parse(data)
+        assert config.text == long_text
+        assert len(config.text) == 2000
+
+    def test_unicode_text_preserved(self):
+        data = {"enabled": True, "voice_id": "v1", "text": "こんにちは世界🎵"}
+        config = TtsConfig.parse(data)
+        assert config.text == "こんにちは世界🎵"
+
+    def test_special_chars_text_preserved(self):
+        data = {"enabled": True, "voice_id": "v1", "text": "line1\nline2\t tab <>&\"'"}
+        config = TtsConfig.parse(data)
+        assert config.text == "line1\nline2\t tab <>&\"'"
+
+    def test_empty_text_ok(self):
+        data = {"enabled": True, "voice_id": "v1", "text": ""}
+        config = TtsConfig.parse(data)
+        assert config.text == ""
+
+    def test_text_none_fallback(self):
+        data = {"enabled": True, "voice_id": "v1", "text": None}
+        config = TtsConfig.parse(data)
+        assert config.text == ""
+
+
+class TestTtsConfigVoiceIdEdge:
+    """voice_id 边界测试."""
+
+    def test_very_long_voice_id_preserved(self):
+        long_id = "voice_" + "x" * 200
+        data = {"enabled": True, "voice_id": long_id}
+        config = TtsConfig.parse(data)
+        assert config.voice_id == long_id
+
+    def test_voice_id_empty_string_ok(self):
+        data = {"enabled": True, "voice_id": ""}
+        config = TtsConfig.parse(data)
+        assert config.voice_id == ""
+
+    def test_voice_id_unicode_ok(self):
+        data = {"enabled": True, "voice_id": "音色_测试_001"}
+        config = TtsConfig.parse(data)
+        assert config.voice_id == "音色_测试_001"
+
+
+class TestTtsConfigClampEdge:
+    """钳制边界附近值测试."""
+
+    def test_speed_just_below_min_clamped(self):
+        data = {"enabled": True, "speed": 0.499}
+        config = TtsConfig.parse(data)
+        assert config.speed == 0.5
+
+    def test_speed_just_above_max_clamped(self):
+        data = {"enabled": True, "speed": 2.001}
+        config = TtsConfig.parse(data)
+        assert config.speed == 2.0
+
+    def test_pitch_just_below_min_clamped(self):
+        data = {"enabled": True, "pitch": -12.1}
+        config = TtsConfig.parse(data)
+        assert config.pitch == -12
+
+    def test_pitch_just_above_max_clamped(self):
+        data = {"enabled": True, "pitch": 12.1}
+        config = TtsConfig.parse(data)
+        assert config.pitch == 12
+
+    def test_volume_just_below_min_clamped(self):
+        data = {"enabled": True, "volume": -0.001}
+        config = TtsConfig.parse(data)
+        assert config.volume == 0.0
+
+    def test_volume_just_above_max_clamped(self):
+        data = {"enabled": True, "volume": 1.001}
+        config = TtsConfig.parse(data)
+        assert config.volume == 1.0
+
+    def test_direct_construct_clamp_speed(self):
+        config = TtsConfig(enabled=True, speed=0.1)
+        config._clamp()
+        assert config.speed == 0.5
+
+    def test_direct_construct_clamp_pitch_volume(self):
+        config = TtsConfig(enabled=True, pitch=-20, volume=2.0)
+        config._clamp()
+        assert config.pitch == -12
+        assert config.volume == 1.0
+
+
+class TestTtsConfigAlignOverlapEdge:
+    """对齐与叠加模式边界."""
+
+    def test_align_mode_empty_string_fallback(self):
+        data = {"enabled": True, "align_mode": ""}
+        config = TtsConfig.parse(data)
+        assert config.align_mode == "full"
+
+    def test_overlap_mode_empty_string_fallback(self):
+        data = {"enabled": True, "overlap_mode": ""}
+        config = TtsConfig.parse(data)
+        assert config.overlap_mode == "replace"
+
+    def test_align_mode_case_sensitive(self):
+        data = {"enabled": True, "align_mode": "SUBTITLE"}
+        config = TtsConfig.parse(data)
+        assert config.align_mode == "full"
+
+
+class TestTtsConfigEquality:
+    """相等性与独立性测试."""
+
+    def test_same_config_equal(self):
+        c1 = TtsConfig(enabled=True, voice_id="v1", speed=1.5)
+        c2 = TtsConfig(enabled=True, voice_id="v1", speed=1.5)
+        assert c1 == c2
+
+    def test_different_config_not_equal(self):
+        c1 = TtsConfig(enabled=True, voice_id="v1")
+        c2 = TtsConfig(enabled=True, voice_id="v2")
+        assert c1 != c2
+
+    def test_modify_one_does_not_affect_other(self):
+        c1 = TtsConfig(enabled=True, voice_id="v1")
+        c2 = TtsConfig(enabled=True, voice_id="v1")
+        c2.speed = 2.0
+        assert c1.speed == 1.0
+        assert c1 != c2
@@ -0,0 +1,365 @@
+"""VideoProcessor 纯逻辑单测 — 数据类 + 输入校验 + 解析逻辑.
+
+通过 mock ffmpeg-python 库验证纯逻辑部分，
+不实际执行 FFmpeg，确保测试轻量快速。
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import fields
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from video_processing.processor import VideoProcessor, VideoResult
+
+
+class TestVideoResultDataclass:
+    """VideoResult 数据类测试."""
+
+    def test_all_fields_exist(self):
+        """所有字段都存在."""
+        field_names = {f.name for f in fields(VideoResult)}
+        expected = {
+            "output_path",
+            "thumbnail_path",
+            "duration",
+            "width",
+            "height",
+            "fps",
+            "file_size",
+        }
+        assert expected.issubset(field_names)
+
+    def test_default_construction(self):
+        """正常构造 VideoResult."""
+        result = VideoResult(
+            output_path="/tmp/out.mp4",
+            thumbnail_path="/tmp/out.jpg",
+            duration=10.5,
+            width=1920,
+            height=1080,
+            fps=25.0,
+            file_size=1024000,
+        )
+        assert result.output_path == "/tmp/out.mp4"
+        assert result.thumbnail_path == "/tmp/out.jpg"
+        assert result.duration == 10.5
+        assert result.width == 1920
+        assert result.height == 1080
+        assert result.fps == 25.0
+        assert result.file_size == 1024000
+
+    def test_zero_values(self):
+        """零值/边界值构造."""
+        result = VideoResult(
+            output_path="",
+            thumbnail_path="",
+            duration=0.0,
+            width=0,
+            height=0,
+            fps=0.0,
+            file_size=0,
+        )
+        assert result.duration == 0.0
+        assert result.file_size == 0
+
+
+class TestVideoProcessorInit:
+    """VideoProcessor 初始化测试."""
+
+    def test_default_temp_dir(self):
+        """默认使用系统临时目录."""
+        import tempfile
+
+        vp = VideoProcessor()
+        assert vp.temp_dir == tempfile.gettempdir()
+
+    def test_custom_temp_dir(self):
+        """自定义临时目录."""
+        vp = VideoProcessor(temp_dir="/my/temp")
+        assert vp.temp_dir == "/my/temp"
+
+
+class TestVideoProcessorConcatenateValidation:
+    """concatenate_videos 输入校验测试."""
+
+    def test_empty_input_raises(self):
+        """空输入列表抛出 ValueError."""
+        vp = VideoProcessor()
+        with pytest.raises(ValueError, match="cannot be empty"):
+            vp.concatenate_videos([], "/tmp/output.mp4")
+
+    def test_none_input_raises(self):
+        """None 输入抛出异常."""
+        vp = VideoProcessor()
+        with pytest.raises((ValueError, TypeError)):
+            vp.concatenate_videos(None, "/tmp/output.mp4")  # type: ignore[arg-type]
+
+
+class TestVideoProcessorGetVideoInfoParsing:
+    """get_video_info 解析逻辑测试（mock ffmpeg.probe）."""
+
+    def _mock_probe(self, streams=None, fmt=None):
+        """创建 ffmpeg.probe 的 mock 返回值."""
+        return {
+            "streams": streams
+            or [{"codec_type": "video", "width": 1920, "height": 1080, "r_frame_rate": "25/1", "codec_name": "h264"}],
+            "format": fmt or {"duration": "10.5", "bit_rate": "5000000"},
+        }
+
+    def test_basic_info_parsing(self):
+        """基本视频信息解析正确."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe()
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["duration"] == 10.5
+        assert info["width"] == 1920
+        assert info["height"] == 1080
+        assert info["fps"] == 25.0
+        assert info["codec"] == "h264"
+        assert info["bitrate"] == 5000000
+
+    def test_fps_fraction_parsing(self):
+        """分数帧率解析（如 30000/1001 = 29.97）."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe(
+            streams=[
+                {
+                    "codec_type": "video",
+                    "width": 1920,
+                    "height": 1080,
+                    "r_frame_rate": "30000/1001",
+                    "codec_name": "h264",
+                }
+            ]
+        )
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["fps"] == pytest.approx(29.97, abs=0.01)
+
+    def test_fps_integer_string(self):
+        """整数字符串帧率（如 "60"）."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe(
+            streams=[{"codec_type": "video", "width": 1920, "height": 1080, "r_frame_rate": "60", "codec_name": "h264"}]
+        )
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["fps"] == 60.0
+
+    def test_missing_r_frame_rate(self):
+        """缺少 r_frame_rate 时使用默认值."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe(
+            streams=[{"codec_type": "video", "width": 1920, "height": 1080, "codec_name": "h264"}]
+        )
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["fps"] == 25.0
+
+    def test_no_video_stream(self):
+        """没有视频流时的行为."""
+        vp = VideoProcessor()
+        probe_data = {
+            "streams": [{"codec_type": "audio", "codec_name": "aac"}],
+            "format": {"duration": "10.0", "bit_rate": "128000"},
+        }
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            with pytest.raises(StopIteration):
+                vp.get_video_info("/tmp/test.mp4")
+
+    def test_float_duration(self):
+        """浮点时长解析."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe(fmt={"duration": "123.456", "bit_rate": "0"})
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["duration"] == pytest.approx(123.456, abs=0.001)
+
+    def test_bitrate_zero(self):
+        """码率为 0 时."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe(fmt={"duration": "10.0", "bit_rate": "0"})
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["bitrate"] == 0
+
+    def test_ffmpeg_probe_error_raises(self):
+        """ffmpeg.probe 失败时抛出 RuntimeError."""
+        vp = VideoProcessor()
+
+        import ffmpeg
+
+        with patch(
+            "video_processing.processor.ffmpeg.probe",
+            side_effect=ffmpeg.Error([], b"", b"No such file"),
+        ):
+            with pytest.raises(RuntimeError, match="probe error"):
+                vp.get_video_info("/tmp/nonexistent.mp4")
+
+
+class TestVideoProcessorGenerateThumbnail:
+    """generate_thumbnail 测试."""
+
+    def _build_mock_chain(self):
+        """构建 ffmpeg.input → .output → .overwrite_output → .run 调用链."""
+        mock_input_node = MagicMock()
+        mock_output_node = MagicMock()
+        mock_overwrite_node = MagicMock()
+        mock_input_node.output.return_value = mock_output_node
+        mock_output_node.overwrite_output.return_value = mock_overwrite_node
+        return mock_input_node, mock_output_node, mock_overwrite_node
+
+    def test_default_output_path(self):
+        """默认输出路径为视频路径 + _thumb.jpg."""
+        vp = VideoProcessor()
+        mock_input_node, _mock_output, mock_overwrite = self._build_mock_chain()
+
+        with patch("video_processing.processor.ffmpeg.input", return_value=mock_input_node) as mock_ff_input:
+            result = vp.generate_thumbnail("/tmp/video.mp4")
+
+        assert result == "/tmp/video_thumb.jpg"
+        mock_ff_input.assert_called_once_with("/tmp/video.mp4", ss=1.0)
+        mock_input_node.output.assert_called_once()
+        # 验证输出路径和参数
+        output_args = mock_input_node.output.call_args
+        assert output_args[0][0] == "/tmp/video_thumb.jpg"
+        assert output_args[1].get("vframes") == 1
+        assert output_args[1].get("format") == "image2"
+        assert output_args[1].get("vcodec") == "mjpeg"
+
+    def test_custom_output_path(self):
+        """自定义输出路径."""
+        vp = VideoProcessor()
+        mock_input_node, _mock_output, mock_overwrite = self._build_mock_chain()
+
+        with patch("video_processing.processor.ffmpeg.input", return_value=mock_input_node) as mock_ff_input:
+            result = vp.generate_thumbnail("/tmp/video.mp4", output_path="/custom/thumb.jpg")
+
+        assert result == "/custom/thumb.jpg"
+
+    def test_custom_timestamp(self):
+        """自定义截图时间点."""
+        vp = VideoProcessor()
+        mock_input_node, _mock_output, mock_overwrite = self._build_mock_chain()
+
+        with patch("video_processing.processor.ffmpeg.input", return_value=mock_input_node) as mock_ff_input:
+            vp.generate_thumbnail("/tmp/video.mp4", timestamp=3.5)
+
+        # 验证 ss 参数
+        mock_ff_input.assert_called_once_with("/tmp/video.mp4", ss=3.5)
+
+    def test_ffmpeg_error_raises_runtime(self):
+        """FFmpeg 失败时抛出 RuntimeError."""
+        vp = VideoProcessor()
+
+        import ffmpeg
+
+        mock_input_node, mock_output, mock_overwrite = self._build_mock_chain()
+        mock_overwrite.run.side_effect = ffmpeg.Error([], b"", b"Output file #0 does not contain any stream")
+
+        with patch("video_processing.processor.ffmpeg.input", return_value=mock_input_node):
+            with pytest.raises(RuntimeError, match="thumbnail error"):
+                vp.generate_thumbnail("/tmp/video.mp4")
+
+
+class TestVideoProcessorConcatFileFormat:
+    """concat 临时文件格式验证."""
+
+    def test_concat_file_format(self, tmp_path):
+        """concat 临时文件格式符合 FFmpeg concat demuxer 规范."""
+        import os
+
+        vp = VideoProcessor(temp_dir=str(tmp_path))
+
+        written_content = {}
+
+        def fake_input(path, *args, **kwargs):
+            mock_node = MagicMock()
+            mock_output = MagicMock()
+            mock_overwrite = MagicMock()
+            mock_node.output.return_value = mock_output
+            mock_output.overwrite_output.return_value = mock_overwrite
+
+            if kwargs.get("format") == "concat":
+                # 读取 concat 文件内容
+                with open(path) as f:
+                    written_content["concat"] = f.read()
+            return mock_node
+
+        mock_probe = MagicMock(
+            return_value={
+                "streams": [{"codec_type": "video", "width": 1920, "height": 1080, "r_frame_rate": "25/1"}],
+                "format": {"duration": "5.0", "bit_rate": "1000000"},
+            }
+        )
+
+        with (
+            patch("video_processing.processor.ffmpeg.input", side_effect=fake_input),
+            patch("video_processing.processor.ffmpeg.probe", mock_probe),
+            patch("video_processing.processor.os.path.getsize", return_value=1024),
+        ):
+            with patch.object(VideoProcessor, "generate_thumbnail", return_value="/tmp/thumb.jpg"):
+                vp.concatenate_videos(
+                    ["/tmp/a.mp4", "/tmp/b.mp4", "/tmp/c.mp4"],
+                    str(tmp_path / "output.mp4"),
+                )
+
+        # 验证 concat 文件格式
+        assert "concat" in written_content
+        lines = written_content["concat"].strip().split("\n")
+        assert len(lines) == 3
+        assert lines[0].startswith("file '")
+        assert "a.mp4'" in lines[0]
+        assert "b.mp4'" in lines[1]
+        assert "c.mp4'" in lines[2]
+        # 使用绝对路径
+        first_path = lines[0].replace("file '", "").rstrip("'")
+        assert os.path.isabs(first_path)
+
+    def test_concat_creates_output_directory(self, tmp_path):
+        """输出目录不存在时自动创建."""
+        vp = VideoProcessor(temp_dir=str(tmp_path))
+
+        out_dir = tmp_path / "deep" / "output"
+        out_file = out_dir / "result.mp4"
+
+        mock_node = MagicMock()
+        mock_output = MagicMock()
+        mock_overwrite = MagicMock()
+        mock_node.output.return_value = mock_output
+        mock_output.overwrite_output.return_value = mock_overwrite
+
+        mock_probe = MagicMock(
+            return_value={
+                "streams": [{"codec_type": "video", "width": 1920, "height": 1080, "r_frame_rate": "25/1"}],
+                "format": {"duration": "5.0", "bit_rate": "1000000"},
+            }
+        )
+
+        with (
+            patch("video_processing.processor.ffmpeg.input", return_value=mock_node),
+            patch("video_processing.processor.ffmpeg.probe", mock_probe),
+            patch("video_processing.processor.os.path.getsize", return_value=1024),
+        ):
+            with patch.object(VideoProcessor, "generate_thumbnail", return_value=str(out_dir / "thumb.jpg")):
+                vp.concatenate_videos(["/tmp/a.mp4"], str(out_file))
+
+        assert out_dir.exists()
+        assert out_dir.is_dir()
@@ -0,0 +1,260 @@
+"""VoiceExtractor 纯逻辑单测 — 命令构建 + 边界用例.
+
+通过 mock run_ffmpeg 验证 FFmpeg 命令参数是否正确，
+不实际执行 FFmpeg，确保测试轻量快速。
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from worker_app.tasks.voice_extraction import VoiceExtractor
+
+
+class TestVoiceExtractorExtractVoiceCommand:
+    """extract_voice 命令构建测试."""
+
+    def test_default_params_correct_command(self):
+        """默认参数下 FFmpeg 命令正确."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            result = extractor.extract_voice("/tmp/input.mp4", "/tmp/output.mp3")
+
+        assert result == "/tmp/output.mp3"
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+
+        # 基本结构验证
+        assert cmd[0] == "ffmpeg"
+        assert "-y" in cmd
+        assert cmd[cmd.index("-i") + 1] == "/tmp/input.mp4"
+        assert "-vn" in cmd  # 无视频流
+        assert cmd[-1] == "/tmp/output.mp3"
+
+        # 音频滤镜验证
+        af_idx = cmd.index("-af")
+        af_value = cmd[af_idx + 1]
+        assert "highpass=f=200" in af_value
+        assert "afftdn=bn=20" in af_value
+        assert "bandpass=f=300:width_type=h:width=3000" in af_value
+        assert "loudnorm" in af_value
+
+        # 编码验证
+        assert "libmp3lame" in cmd
+        assert "-q:a" in cmd
+        assert cmd[cmd.index("-q:a") + 1] == "2"
+
+    def test_custom_highpass(self):
+        """自定义 highpass 频率."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", highpass=500)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "highpass=f=500" in af_value
+
+    def test_custom_bandpass_freq(self):
+        """自定义 bandpass 中心频率."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", bandpass_freq=500)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "bandpass=f=500:" in af_value
+
+    def test_custom_bandpass_width(self):
+        """自定义 bandpass 宽度."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", bandpass_width=5000)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "width=5000" in af_value
+
+    def test_custom_noise_reduction(self):
+        """自定义降噪强度."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", noise_reduction=30)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "afftdn=bn=30" in af_value
+
+    def test_filter_order_is_correct(self):
+        """滤镜顺序：highpass → 降噪 → bandpass → loudnorm."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3")
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+
+        hp_pos = af_value.index("highpass")
+        dn_pos = af_value.index("afftdn")
+        bp_pos = af_value.index("bandpass")
+        ln_pos = af_value.index("loudnorm")
+
+        assert hp_pos < dn_pos < bp_pos < ln_pos
+
+    def test_creates_output_directory(self, tmp_path):
+        """输出目录不存在时自动创建."""
+        out_dir = tmp_path / "nested" / "deep"
+        out_file = out_dir / "voice.mp3"
+
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg"):
+            extractor.extract_voice("/tmp/in.mp4", str(out_file))
+
+        assert out_dir.exists()
+        assert out_dir.is_dir()
+
+    def test_returns_output_path(self):
+        """返回值为输出路径."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg"):
+            result = extractor.extract_voice("/tmp/in.mp4", "/tmp/voice.mp3")
+
+        assert result == "/tmp/voice.mp3"
+
+
+class TestVoiceExtractorExtractBackgroundCommand:
+    """extract_background 命令构建测试."""
+
+    def test_default_params_correct_command(self):
+        """默认参数下 FFmpeg 命令正确."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            result = extractor.extract_background("/tmp/input.mp4", "/tmp/output.mp3")
+
+        assert result == "/tmp/output.mp3"
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+
+        # 基本结构
+        assert cmd[0] == "ffmpeg"
+        assert "-y" in cmd
+        assert cmd[cmd.index("-i") + 1] == "/tmp/input.mp4"
+        assert "-vn" in cmd
+        assert cmd[-1] == "/tmp/output.mp3"
+
+        # 音频滤镜
+        af_idx = cmd.index("-af")
+        af_value = cmd[af_idx + 1]
+        assert "lowpass=f=200" in af_value
+        assert "loudnorm" in af_value
+
+        # 编码
+        assert "libmp3lame" in cmd
+
+    def test_custom_lowpass_freq(self):
+        """自定义 lowpass 频率."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_background("/tmp/in.mp4", "/tmp/out.mp3", lowpass=500)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "lowpass=f=500" in af_value
+
+    def test_filter_order_background(self):
+        """背景音滤镜顺序：lowpass → loudnorm."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_background("/tmp/in.mp4", "/tmp/out.mp3")
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+
+        lp_pos = af_value.index("lowpass")
+        ln_pos = af_value.index("loudnorm")
+        assert lp_pos < ln_pos
+
+    def test_background_creates_output_directory(self, tmp_path):
+        """背景音输出目录不存在时自动创建."""
+        out_dir = tmp_path / "bgm" / "tracks"
+        out_file = out_dir / "bg.mp3"
+
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg"):
+            extractor.extract_background("/tmp/in.mp4", str(out_file))
+
+        assert out_dir.exists()
+
+
+class TestVoiceExtractorEdgeCases:
+    """边界情况测试."""
+
+    def test_zero_highpass(self):
+        """highpass=0 时的行为（极端低值）."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", highpass=0)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "highpass=f=0" in af_value
+
+    def test_zero_bandpass_freq(self):
+        """bandpass_freq=0 时的极端情况."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", bandpass_freq=0)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "bandpass=f=0:" in af_value
+
+    def test_very_high_noise_reduction(self):
+        """极高降噪强度."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", noise_reduction=100)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "afftdn=bn=100" in af_value
+
+    def test_negative_lowpass_allowed(self):
+        """lowpass 负值（由调用方保证合法性，函数不做校验）."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_background("/tmp/in.mp4", "/tmp/out.mp3", lowpass=-10)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "lowpass=f=-10" in af_value
+
+    def test_run_ffmpeg_propagates_error(self):
+        """_run_ffmpeg 抛出异常时向上传递."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg", side_effect=RuntimeError("FFmpeg failed")):
+            with pytest.raises(RuntimeError, match="FFmpeg failed"):
+                extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3")
+
+    def test_voice_extractor_is_static_method(self):
+        """_run_ffmpeg 是静态方法，可在类上直接调用."""
+        # 验证 VoiceExtractor 可以直接实例化（无需参数）
+        extractor = VoiceExtractor()
+        assert extractor is not None
+
+    def test_multiple_extractions_same_instance(self):
+        """同一个实例可多次执行提取."""
+        extractor = VoiceExtractor()
+        call_count = 0
+
+        def fake_run(cmd):
+            nonlocal call_count
+            call_count += 1
+
+        with patch.object(VoiceExtractor, "_run_ffmpeg", side_effect=fake_run):
+            extractor.extract_voice("/tmp/a.mp4", "/tmp/a_voice.mp3")
+            extractor.extract_background("/tmp/a.mp4", "/tmp/a_bg.mp3")
+
+        assert call_count == 2
Author	SHA1	Message	Date
CI Bot	d548075b8d	test(unit): 第86波 - worker层VoiceExtractor/CoverGenerator/VideoProcessor纯逻辑单测 (+63) AI Code Review / AI Code Review (pull_request) Failing after 1s Details CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 21s Details CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m13s Details CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m15s Details CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m1s Details CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 42s Details CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 57s Details CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 2m13s Details Preview Deploy / Deploy Preview Environment (pull_request) Failing after 43s Details PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m2s Details CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m43s Details PR Automation / Auto Approve on CI Green (pull_request) Successful in 4m3s Details CI/CD Pipeline / Unit Tests (pull_request) Failing after 2m33s Details CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m6s Details ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled Details Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 18s Details CI/CD Pipeline / Canary Release to Production (pull_request) Failing after 1262h50m25s Details CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1262h50m26s Details CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 1262h50m28s Details CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1262h50m29s Details CI/CD Pipeline / Deploy Production (pull_request) Failing after 1262h50m33s Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 1262h50m48s Details CI/CD Pipeline / Build Production API Image (pull_request) Failing after 1262h50m52s Details CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 1262h50m54s Details CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 1262h50m50s Details CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 1262h54m55s Details CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 1262h54m59s Details CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 1263h23m26s Details CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1263h23m7s Details CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 1263h27m33s Details - VoiceExtractor: 命令构建 + 滤镜顺序 + 边界用例 + 异常传递 (19个) - CoverGenerator: 时间钳制 + 智能选帧均匀分布 + 文件大小选最佳 (23个) - VideoProcessor: 数据类 + fps解析 + concat文件格式 + 缩略图参数 (21个) - 全部通过mock外部依赖(ffmpeg/run_ffmpeg)实现，纯逻辑验证，无需真实FFmpeg	2026-07-26 00:26:06 +08:00
xiaoxia	f506048240	test(wave85): exceptions+bgm_utils+subtitle深度补充 +59 (#898 ) CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m47s Details CI/CD Pipeline / Validate - Code Quality (push) Failing after 3m8s Details CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m19s Details CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m18s Details CI/CD Pipeline / Unit Tests (push) Failing after 4m49s Details CI/CD Pipeline / Integration Tests (push) Successful in 2m21s Details CI/CD Pipeline / Frontend Lint (push) Successful in 52s Details CI/CD Pipeline / Frontend Unit Tests (push) Successful in 2m42s Details CI/CD Pipeline / Build Staging API Image (push) Successful in 11m24s Details CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m44s Details CI/CD Pipeline / Build Staging Worker Image (push) Successful in 3m15s Details CI/CD Pipeline / ACR Image Cleanup (push) Failing after 32s Details CI/CD Pipeline / Staging E2E Tests (push) Successful in 3m9s Details CI/CD Pipeline / Staging API Integration Tests (push) Successful in 5m3s Details CI/CD Pipeline / Canary Release to Production (push) Failing after 1262h48m31s Details CI/CD Pipeline / Production Browser E2E (push) Failing after 1263h9m53s Details CI/CD Pipeline / Build Production API Image (push) Failing after 1263h13m3s Details CI/CD Pipeline / PR Build Worker Image (push) Failing after 1263h13m59s Details CI/CD Pipeline / Deploy Production (push) Failing after 1263h10m40s Details CI/CD Pipeline / PR Build API Image (push) Failing after 1263h14m3s Details CI/CD Pipeline / Build Production Worker Image (push) Failing after 1263h12m59s Details CI/CD Pipeline / Check if frontend-only change (push) Failing after 1263h14m30s Details CI/CD Pipeline / Build Production Web Image (push) Failing after 1263h45m37s Details CI/CD Pipeline / PR Build Web Image (push) Failing after 1263h46m37s Details	2026-07-26 00:13:44 +08:00
xiaoxia	69f2846c95	test(wave84): classification + tts_config 单测深度补充 +52 (#893 )	2026-07-26 00:13:44 +08:00
xiaoxia	6437b96e54	refactor(assets): Phase 3 - extract business hooks (#899 ) CI/CD Pipeline / Validate - Code Quality (push) Failing after 3m17s Details CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m8s Details CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 1m20s Details CI/CD Pipeline / Unit Tests (push) Failing after 5m11s Details CI/CD Pipeline / Integration Tests (push) Successful in 3m7s Details CI/CD Pipeline / Frontend Lint (push) Successful in 55s Details CI/CD Pipeline / Frontend Unit Tests (push) Successful in 2m8s Details CI/CD Pipeline / Build Staging API Image (push) Successful in 13m32s Details CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m27s Details CI/CD Pipeline / Build Staging Worker Image (push) Successful in 5m42s Details CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m33s Details CI/CD Pipeline / Staging E2E Tests (push) Successful in 8m18s Details CI/CD Pipeline / Staging API Integration Tests (push) Successful in 11m0s Details CI/CD Pipeline / ACR Image Cleanup (push) Failing after 28s Details CI/CD Pipeline / Canary Release to Production (push) Failing after 1263h2m30s Details CI/CD Pipeline / Production Browser E2E (push) Failing after 1263h24m8s Details CI/CD Pipeline / Deploy Production (push) Failing after 1263h24m14s Details CI/CD Pipeline / Build Production Worker Image (push) Failing after 1263h25m30s Details CI/CD Pipeline / Build Production Web Image (push) Failing after 1263h25m32s Details CI/CD Pipeline / Build Production API Image (push) Failing after 1263h25m34s Details CI/CD Pipeline / PR Build Worker Image (push) Failing after 1263h29m42s Details CI/CD Pipeline / Check if frontend-only change (push) Failing after 1263h30m8s Details CI/CD Pipeline / PR Build Web Image (push) Failing after 1263h29m44s Details CI/CD Pipeline / PR Build API Image (push) Failing after 1264h2m22s Details Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>	2026-07-25 23:55:16 +08:00