refactor+test(asset): 第89波 - 拆分asset_analyzer评分纯逻辑 + 64单测

- 从799行asset_analyzer.py拆出asset_quality_scoring.py纯逻辑模块
- 质量评分：分辨率/帧率/码率/清晰度/稳定性 5个维度纯函数
- 分类评分：9个分类（风景/产品/人物/动物/美食/科技/运动/音乐/其他）
- 64个单测全覆盖：边界值、numpy数组、分类组合、数据类
- 原文件从799行瘦身到467行（-332行）
- 向后兼容：classify_asset_real / calculate_quality_score_real 入口不变

.gitea/workflows/ci-health-daily.yml

@@ -8,7 +8,7 @@ permissions:
 jobs:
   ci-health-report:
     name: CI健康度每日巡检
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 15
     steps:
     - name: Checkout code

.gitea/workflows/ci-pipeline.yml

@@ -58,24 +58,56 @@ jobs:
       run: |
         set -eu
         PR_NUMBER=$(echo "$GITHUB_REF" | sed 's|refs/pull/||; s|/.*||')
-        API_URL="${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}/files?limit=300"
-        FILES=$(curl -s -H "Authorization: token ${GITHUB_TOKEN}" "$API_URL" | python3 -c "import sys,json; [print(f['filename']) for f in json.load(sys.stdin)]")
-        FRONTEND_COUNT=$(echo "$FILES" | grep -c '^apps/web/' || true)
-        BACKEND_COUNT=$(echo "$FILES" | grep -cv '^apps/web/' || true)
-        TOTAL=$(echo "$FILES" | grep -cv '^$' || true)
-        echo "变更文件: ${TOTAL} 个 (前端: ${FRONTEND_COUNT}, 后端/公共: ${BACKEND_COUNT})"
-        if [ "$BACKEND_COUNT" = "0" ] && [ "$FRONTEND_COUNT" -gt "0" ]; then
+
+        # 分页获取所有变更文件（修复>300文件时漏判）
+        ALL_FILES=""
+        PAGE=1
+        while true; do
+          API_URL="${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}/files?limit=300&page=${PAGE}"
+          PAGE_FILES=$(curl -s -H "Authorization: token ${GITHUB_TOKEN}" "$API_URL" | python3 -c "import sys,json; files=json.load(sys.stdin); [print(f['filename']) for f in files]; sys.exit(0 if len(files)==300 else 1)" 2>/dev/null || true)
+          PAGE_COUNT=$(echo "$PAGE_FILES" | wc -l)
+          if [ "$PAGE_COUNT" -lt 300 ]; then HAS_MORE=1; else HAS_MORE=0; fi
+          ALL_FILES="${ALL_FILES}${PAGE_FILES}"$'\n'
+          if [ "$HAS_MORE" != "0" ]; then
+            break
+          fi
+          PAGE=$((PAGE + 1))
+        done
+
+        FRONTEND_COUNT=$(echo "$ALL_FILES" | grep -c '^apps/web/' || true)
+        TOTAL=$(echo "$ALL_FILES" | grep -cv '^$' || true)
+        BACKEND_COUNT=$(( TOTAL - FRONTEND_COUNT ))
+
+        # 基础设施文件：改了就强制全量CI（不跳过任何检查）
+        INFRA_COUNT=$(echo "$ALL_FILES" | grep -cE '^(infra/|Dockerfile|\.gitea/workflows/|scripts/ci/|docker/)' || true)
+
+        echo "变更文件: ${TOTAL} 个 (前端: ${FRONTEND_COUNT}, 后端/公共: ${BACKEND_COUNT}, 基础设施: ${INFRA_COUNT})"
+
+        # 判定是否纯前端/纯后端
+        PURE_FRONTEND=false
+        PURE_BACKEND=false
+        if [ "$BACKEND_COUNT" = "0" ] && [ "$FRONTEND_COUNT" -gt "0" ] && [ "$INFRA_COUNT" = "0" ]; then
+          PURE_FRONTEND=true
+        elif [ "$FRONTEND_COUNT" = "0" ] && [ "$BACKEND_COUNT" -gt "0" ] && [ "$INFRA_COUNT" = "0" ]; then
+          PURE_BACKEND=true
+        fi
+
+        if [ "$PURE_FRONTEND" = "true" ]; then
           echo "skip_backend=true" >> $GITHUB_OUTPUT
           echo "skip_frontend=false" >> $GITHUB_OUTPUT
           echo "✅ 纯前端改动，跳过后端检查"
-        elif [ "$FRONTEND_COUNT" = "0" ] && [ "$BACKEND_COUNT" -gt "0" ]; then
+        elif [ "$PURE_BACKEND" = "true" ]; then
           echo "skip_backend=false" >> $GITHUB_OUTPUT
           echo "skip_frontend=true" >> $GITHUB_OUTPUT
           echo "🔧 纯后端改动，跳过前端检查"
         else
           echo "skip_backend=false" >> $GITHUB_OUTPUT
           echo "skip_frontend=false" >> $GITHUB_OUTPUT
-          echo "🔧 包含全栈变更，运行完整CI"
+          if [ "$INFRA_COUNT" -gt "0" ]; then
+            echo "🏗️  包含基础设施变更，强制运行完整CI"
+          else
+            echo "🔧 包含全栈变更，运行完整CI"
+          fi
         fi
 
     - name: Report CI trace

.gitea/workflows/ci-trigger-monitor.yml

@@ -16,15 +16,15 @@ permissions:
 jobs:
   monitor:
     name: Monitor CI Trigger Reliability
-    runs-on: ubuntu-latest
+    runs-on: ci-l2
     timeout-minutes: 5
     steps:
       - name: Checkout code
-        uses: actions/checkout@v3
-        # 网络波动自动重试2次
-        retry:
-          max_attempts: 2
-          retry_on: error
+        shell: sh
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        run: |
+          curl -sH "Authorization: token $GITHUB_TOKEN"             "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}"             | bash
       
       - name: Check CI trigger status for all open PRs
         env:

.gitea/workflows/code-review.yml

@@ -15,20 +15,18 @@ concurrency:
 jobs:
   code-review:
     name: AI Code Review
-    runs-on: ubuntu-latest
+    runs-on: ci-l2
     # 跳过草稿 PR
     if: ${{ !gitea.event.pull_request.draft }}
 
     steps:
       # actions/checkout 由 runner 在宿主机层面处理，不受容器网络影响
       - name: Checkout code
-        uses: actions/checkout@v3
-        with:
-          fetch-depth: 0
-        # 网络波动自动重试2次
-        retry:
-          max_attempts: 2
-          retry_on: error
+        shell: sh
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        run: |
+          curl -sH "Authorization: token $GITHUB_TOKEN"             "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}"             | bash
 
       - name: Install dependencies
         run: |

.gitea/workflows/daily-check.yml

@@ -12,7 +12,7 @@ jobs:
   # ── 1. 生产环境冒烟测试 ─────────────────────────────────────────────
   production-smoke:
     name: Production Smoke Test
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 8
     outputs:
       report: ${{ steps.smoke.outputs.report }}
@@ -121,7 +121,7 @@ jobs:
   # ── 2. Staging API 集成测试 ─────────────────────────────────────────
   staging-api-tests:
     name: Staging API Integration Tests
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 10
     outputs:
       report: ${{ steps.smoke.outputs.report }}
@@ -176,6 +176,9 @@ jobs:
       - name: Run API smoke test on staging
         id: smoke
         shell: sh
+        env:
+          STAGING_TEST_USER: ${{ secrets.STAGING_TEST_USER }}
+          STAGING_TEST_PASSWORD: ${{ secrets.STAGING_TEST_PASSWORD }}
         run: |
           set +e
           START_TIME=$(date +%s)
@@ -183,8 +186,8 @@ jobs:
           docker run --rm \
             -e BASE_URL=https://staging-api.xiaoxiajianji.com \
             -e WEB_URL=https://staging.xiaoxiajianji.com \
-            -e TEST_USER=18314979086@163.com \
-            -e TEST_PASSWORD=Ying1234 \
+            -e TEST_USER="$STAGING_TEST_USER" \
+            -e TEST_PASSWORD="$STAGING_TEST_PASSWORD" \
             -e CLEANUP_ENABLED=1 \
             -e PERF_CHECK_ENABLED=1 \
             -e PERF_WARN_THRESHOLD_MS=500 \
@@ -270,7 +273,7 @@ jobs:
   # ── 3. Staging 浏览器 E2E ──────────────────────────────────────────
   staging-e2e:
     name: Staging Browser E2E
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 15
     outputs:
       report: ${{ steps.smoke.outputs.report }}
@@ -371,7 +374,7 @@ jobs:
   # ── 4. 性能基线巡检 ────────────────────────────────────────────────
   performance-check:
     name: Performance Baseline Check
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 8
     outputs:
       report: ${{ steps.report.outputs.report }}
@@ -380,6 +383,9 @@ jobs:
       - name: Run performance baseline checks
         id: perf
         shell: sh
+        env:
+          STAGING_TEST_USER: ${{ secrets.STAGING_TEST_USER }}
+          STAGING_TEST_PASSWORD: ${{ secrets.STAGING_TEST_PASSWORD }}
         run: |
           set +e
           START_TIME=$(date +%s)
@@ -415,9 +421,10 @@ jobs:
 
           # 先登录获取 token
           echo "--- 准备: 获取测试 Token ---"
+          LOGIN_BODY="{\"email\":\"$STAGING_TEST_USER\",\"password\":\"$STAGING_TEST_PASSWORD\"}"
           AUTH_RESP=$(curl -s -w "\n%{http_code}" -X POST \
             -H "Content-Type: application/json" \
-            -d '{"email":"18314979086@163.com","password":"Ying1234"}' \
+            -d "$LOGIN_BODY" \
             "https://staging-api.xiaoxiajianji.com/api/v1/auth/login" \
             --max-time 10 2>&1)
           AUTH_CODE=$(echo "$AUTH_RESP" | tail -1)
@@ -447,7 +454,7 @@ jobs:
             # 构建 curl 命令
             CURL_ARGS="-s -o /dev/null -w '%{http_code} %{time_total}' --max-time 30"
             if [ "$method" = "POST" ]; then
-              CURL_ARGS="$CURL_ARGS -X POST -H 'Content-Type: application/json' -d '{\"email\":\"18314979086@163.com\",\"password\":\"Ying1234\"}'"
+              CURL_ARGS="$CURL_ARGS -X POST -H 'Content-Type: application/json' -d \"$LOGIN_BODY\""
             fi
             if [ -n "$TOKEN" ] && [ "$name" != "健康检查" ]; then
               CURL_ARGS="$CURL_ARGS -H 'Authorization: Bearer $TOKEN'"
@@ -495,6 +502,9 @@ jobs:
       - name: Generate performance report
         id: report
         shell: sh
+        env:
+          STAGING_TEST_USER: ${{ secrets.STAGING_TEST_USER }}
+          STAGING_TEST_PASSWORD: ${{ secrets.STAGING_TEST_PASSWORD }}
         run: |
           set +e
           echo ""
@@ -509,10 +519,11 @@ jobs:
           RESULTS=""
           START_TIME=$(date +%s)
 
+          LOGIN_BODY="{\"email\":\"$STAGING_TEST_USER\",\"password\":\"$STAGING_TEST_PASSWORD\"}"
           # 先登录获取 token
           AUTH_RESP=$(curl -s -w "\n%{http_code}" -X POST \
             -H "Content-Type: application/json" \
-            -d '{"email":"18314979086@163.com","password":"Ying1234"}' \
+            -d "$LOGIN_BODY" \
             "https://staging-api.xiaoxiajianji.com/api/v1/auth/login" \
             --max-time 10 2>&1)
           AUTH_CODE=$(echo "$AUTH_RESP" | tail -1)
@@ -528,7 +539,7 @@ jobs:
 
             local CURL_ARGS="-s -o /dev/null -w '%{http_code} %{time_total}' --max-time 30"
             if [ "$method" = "POST" ]; then
-              CURL_ARGS="$CURL_ARGS -X POST -H 'Content-Type: application/json' -d '{\"email\":\"18314979086@163.com\",\"password\":\"Ying1234\"}'"
+              CURL_ARGS="$CURL_ARGS -X POST -H 'Content-Type: application/json' -d \"$LOGIN_BODY\""
             fi
             if [ -n "$TOKEN" ] && [ "$name" != "健康检查" ]; then
               CURL_ARGS="$CURL_ARGS -H 'Authorization: Bearer $TOKEN'"
@@ -631,7 +642,7 @@ jobs:
   # ── 5. 每日巡检汇总报告 ────────────────────────────────────────────
   daily-report:
     name: Daily Check Report
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 2
     if: always()
     needs:

apps/web/src/pages/editing-planner/components/filter/FilterPresetGrid.tsx

@@ -11,10 +11,7 @@ interface FilterPresetGridProps {
   onPresetSelect: (preset: FilterPreset) => void
 }
 
-const FilterPresetGrid: React.FC<FilterPresetGridProps> = ({
-  selectedPreset,
-  onPresetSelect,
-}) => {
+const FilterPresetGrid: React.FC<FilterPresetGridProps> = ({ selectedPreset, onPresetSelect }) => {
   return (
     <div className="filter-section">
       <div className="filter-section-title">预设滤镜</div>
@@ -25,10 +22,7 @@ const FilterPresetGrid: React.FC<FilterPresetGridProps> = ({
             className={`filter-preset-item${selectedPreset === p ? " active" : ""}`}
             onClick={() => onPresetSelect(p)}
           >
-            <div
-              className="filter-preset-preview"
-              style={{ background: PRESET_GRADIENTS[p] }}
-            />
+            <div className="filter-preset-preview" style={{ background: PRESET_GRADIENTS[p] }} />
             <span className="filter-preset-label">{FILTER_PRESET_LABELS[p]}</span>
           </button>
         ))}

apps/web/src/pages/editing-planner/components/intro-outro/IntroOutroBlock.tsx

@@ -50,9 +50,7 @@ const IntroOutroBlock: React.FC<IntroOutroBlockProps> = ({
       {hasContent && (
         <>
           <div className="iop-field">
-            <label className="iop-field-label">
-              {item.kind === "video" ? "视频" : "图片"} URL
-            </label>
+            <label className="iop-field-label">{item.kind === "video" ? "视频" : "图片"} URL</label>
             <input
               className="iop-input"
               type="text"

apps/web/src/pages/editing-planner/hooks/useWatermarkConfig.ts

@@ -3,12 +3,7 @@
  * 管理水印类型切换、各项配置变更
  */
 import { useState, useCallback } from "react"
-import type {
-  WatermarkConfig,
-  WatermarkType,
-  WatermarkPosition,
-  ScrollDirection,
-} from "../types"
+import type { WatermarkConfig, WatermarkType, WatermarkPosition, ScrollDirection } from "../types"
 import { DEFAULT_WATERMARK } from "../types"
 
 interface UseWatermarkConfigOptions {

apps/worker/worker_app/tasks/asset_analyzer.py

@@ -19,74 +19,21 @@ from PIL import Image
 
 from packages.domain.classification import AssetClassification
 
+from .asset_quality_scoring import (
+    AudioAnalysis,
+    ClassificationResult,
+    ColorAnalysis,
+    MotionAnalysis,
+    QualityScore,
+    VideoInfo,
+    calculate_category_scores,
+    calculate_quality_score,
+    classify_from_analysis,
+)
+
 logger = logging.getLogger(__name__)
 
 
-@dataclass
-class VideoInfo:
-    """视频基本信息"""
-
-    width: int = 0
-    height: int = 0
-    fps: float = 0.0
-    duration: float = 0.0
-    bitrate: int = 0
-    codec: str = ""
-    has_audio: bool = False
-    file_size: int = 0
-
-
-@dataclass
-class ColorAnalysis:
-    """色彩分析结果"""
-
-    dominant_hue: float = 0.0  # 主色调 (0-360)
-    green_ratio: float = 0.0  # 绿色占比
-    warm_ratio: float = 0.0  # 暖色调占比
-    cool_ratio: float = 0.0  # 冷色调占比
-    avg_saturation: float = 0.0
-    avg_brightness: float = 0.0
-
-
-@dataclass
-class MotionAnalysis:
-    """运动分析结果"""
-
-    motion_score: float = 0.0  # 运动幅度 (0-1)
-    scene_changes: int = 0  # 场景切换次数
-
-
-@dataclass
-class AudioAnalysis:
-    """音频分析结果"""
-
-    has_audio: bool = False
-    speech_ratio: float = 0.0  # 人声比例
-    music_ratio: float = 0.0  # 音乐比例
-    ambient_ratio: float = 0.0  # 环境音比例
-
-
-@dataclass
-class ClassificationResult:
-    """分类结果"""
-
-    category: AssetClassification
-    confidence: float
-    scores: dict[str, float] = field(default_factory=dict)
-
-
-@dataclass
-class QualityScore:
-    """质量评分结果"""
-
-    total: float
-    resolution_score: float = 0.0
-    fps_score: float = 0.0
-    bitrate_score: float = 0.0
-    clarity_score: float = 0.0
-    stability_score: float = 0.0
-
-
 class AssetAnalyzer:
     """
     轻量级视频素材分析器
@@ -449,316 +396,37 @@ class AssetAnalyzer:
         """
         综合分析得出分类结果
 
+        评分逻辑在 asset_quality_scoring.calculate_category_scores / classify_from_analysis
+        纯函数中，此处只负责采集分析数据后委托计算。
+
         Returns:
             ClassificationResult 对象
         """
-        # 提取分析数据
         frames = self.extract_frames()
         color = self.analyze_color_distribution(frames)
         motion = self.analyze_motion(frames)
         audio = self.analyze_audio()
 
-        # 计算各类别得分
-        scores = self._calculate_category_scores(color, motion, audio)
-
-        # 找最高分
-        if not scores:
-            return ClassificationResult(
-                category=AssetClassification.OTHER,
-                confidence=0.3,
-                scores={},
-            )
-
-        best_category = max(scores.items(), key=lambda x: x[1])
-        category = AssetClassification(best_category[0])
-        confidence = min(0.95, max(0.3, best_category[1]))
-
-        return ClassificationResult(
-            category=category,
-            confidence=confidence,
-            scores=scores,
-        )
-
-    def _calculate_category_scores(
-        self,
-        color: ColorAnalysis,
-        motion: MotionAnalysis,
-        audio: AudioAnalysis,
-    ) -> dict[str, float]:
-        """
-        计算各类别的置信度得分
-
-        Args:
-            color: 色彩分析结果
-            motion: 运动分析结果
-            audio: 音频分析结果
-
-        Returns:
-            各类别得分字典
-        """
-        scores = {}
-
-        # 1. 风景 (scenic) - 绿色、户外、自然
-        scenic_score = 0.0
-        if color.green_ratio > 0.3:
-            scenic_score += 0.4 * color.green_ratio
-        if color.avg_saturation > 0.3:
-            scenic_score += 0.2 * color.avg_saturation
-        if color.avg_brightness > 0.4:
-            scenic_score += 0.2
-        if motion.motion_score > 0.1 and motion.motion_score < 0.5:
-            scenic_score += 0.2  # 适度运动（如云朵、树叶）
-        if not audio.has_audio or audio.ambient_ratio > 0.5:
-            scenic_score += 0.2  # 自然环境音
-        scores[AssetClassification.SCENIC.value] = min(1.0, scenic_score)
-
-        # 2. 产品 (product) - 中等亮度、均匀色彩、低运动
-        product_score = 0.0
-        if 0.3 < color.avg_brightness < 0.7:
-            product_score += 0.3
-        if color.avg_saturation < 0.5:
-            product_score += 0.2
-        if motion.motion_score < 0.15:
-            product_score += 0.4  # 低运动 = 产品展示
-        if color.cool_ratio > 0.3:
-            product_score += 0.2  # 冷色调 = 科技感
-        scores[AssetClassification.PRODUCT.value] = min(1.0, product_score)
-
-        # 3. 人物 (person) - 中等运动、有时有人声
-        person_score = 0.0
-        if 0.1 < motion.motion_score < 0.4:
-            person_score += 0.3  # 适度运动
-        if audio.has_audio and audio.speech_ratio > 0.3:
-            person_score += 0.5  # 有人声
-        if color.avg_brightness > 0.3:
-            person_score += 0.2
-        scores[AssetClassification.PERSON.value] = min(1.0, person_score)
-
-        # 4. 动物 (animal) - 高运动、有时自然音
-        animal_score = 0.0
-        if motion.motion_score > 0.3:
-            animal_score += 0.4  # 高运动
-        if motion.scene_changes > 2:
-            animal_score += 0.2
-        if audio.has_audio and (audio.ambient_ratio > 0.3 or audio.speech_ratio > 0.2):
-            animal_score += 0.3
-        scores[AssetClassification.ANIMAL.value] = min(1.0, animal_score)
-
-        # 5. 美食 (food) - 暖色调、高饱和度
-        food_score = 0.0
-        if color.warm_ratio > 0.4:
-            food_score += 0.5
-        if color.avg_saturation > 0.5:
-            food_score += 0.3
-        if 0.4 < color.avg_brightness < 0.8:
-            food_score += 0.2
-        scores[AssetClassification.FOOD.value] = min(1.0, food_score)
-
-        # 6. 科技 (tech) - 冷色调、低饱和度、低运动
-        tech_score = 0.0
-        if color.cool_ratio > 0.4:
-            tech_score += 0.4
-        if color.avg_saturation < 0.4:
-            tech_score += 0.3
-        if motion.motion_score < 0.2:
-            tech_score += 0.3
-        scores[AssetClassification.TECH.value] = min(1.0, tech_score)
-
-        # 7. 运动 (sport) - 高运动
-        sport_score = 0.0
-        if motion.motion_score > 0.4:
-            sport_score += 0.6
-        if motion.scene_changes > 3:
-            sport_score += 0.2
-        if color.avg_brightness > 0.4:
-            sport_score += 0.2
-        scores[AssetClassification.SPORT.value] = min(1.0, sport_score)
-
-        # 8. 音乐 (music) - 有节奏性音乐
-        music_score = 0.0
-        if audio.has_audio and audio.music_ratio > 0.4:
-            music_score += 0.6
-        # 纯视觉判断：色彩丰富但非自然
-        if color.avg_saturation > 0.5 and color.green_ratio < 0.2:
-            music_score += 0.3
-        scores[AssetClassification.MUSIC.value] = min(1.0, music_score)
-
-        # 9. 其他 (other) - 默认最低分
-        scores[AssetClassification.OTHER.value] = 0.1
-
-        return scores
+        return classify_from_analysis(color, motion, audio)
 
     def calculate_quality_score(self) -> QualityScore:
         """
         计算视频质量综合评分 (0-100)
 
+        评分逻辑在 asset_quality_scoring.calculate_quality_score 纯函数中，
+        此处只负责采集数据后委托计算。
+
         评分维度:
         1. 分辨率得分 (25分)
         2. 帧率得分 (20分)
         3. 码率得分 (20分)
-        4. 清晰度得分 (20分) - Laplacian 方差
-        5. 稳定性得分 (15分) - 帧间位移方差
+        4. 清晰度得分 (20分)
+        5. 稳定性得分 (15分)
         """
         info = self.get_video_info()
         frames = self.extract_frames()
 
-        # 1. 分辨率得分
-        resolution_score = self._score_resolution(info.width, info.height)
-
-        # 2. 帧率得分
-        fps_score = self._score_framerate(info.fps)
-
-        # 3. 码率得分
-        bitrate_score = self._score_bitrate(info.bitrate)
-
-        # 4. 清晰度得分
-        clarity_score = self._score_clarity(frames)
-
-        # 5. 稳定性得分
-        stability_score = self._score_stability(frames)
-
-        total = resolution_score + fps_score + bitrate_score + clarity_score + stability_score
-
-        return QualityScore(
-            total=round(min(100, max(0, total)), 1),
-            resolution_score=resolution_score,
-            fps_score=fps_score,
-            bitrate_score=bitrate_score,
-            clarity_score=clarity_score,
-            stability_score=stability_score,
-        )
-
-    def _score_resolution(self, width: int, height: int) -> float:
-        """分辨率评分 (满分 25)"""
-        pixels = width * height
-
-        if pixels >= 3840 * 2160:  # 4K
-            return 25.0
-        elif pixels >= 2560 * 1440:  # 2K
-            return 22.0
-        elif pixels >= 1920 * 1080:  # 1080p
-            return 20.0
-        elif pixels >= 1280 * 720:  # 720p
-            return 15.0
-        elif pixels >= 854 * 480:  # 480p
-            return 8.0
-        else:
-            return 3.0
-
-    def _score_framerate(self, fps: float) -> float:
-        """帧率评分 (满分 20)"""
-        if fps >= 60:
-            return 20.0
-        elif fps >= 30:
-            return 15.0
-        elif fps >= 24:
-            return 10.0
-        elif fps >= 15:
-            return 7.0
-        else:
-            return 5.0
-
-    def _score_bitrate(self, bitrate: int) -> float:
-        """码率评分 (满分 20)"""
-        bitrate_mbps = bitrate / 1_000_000
-
-        if bitrate_mbps > 10:
-            return 20.0
-        elif bitrate_mbps >= 5:
-            return 15.0
-        elif bitrate_mbps >= 2:
-            return 10.0
-        elif bitrate_mbps >= 0.5:
-            return 5.0
-        else:
-            return 3.0
-
-    def _score_clarity(self, frames: list[np.ndarray]) -> float:
-        """
-        清晰度评分 (满分 20)
-
-        使用 Laplacian 方差评估画面清晰度
-        高方差 = 细节丰富 = 高分
-        """
-        if not frames:
-            return 10.0  # 默认中等分
-
-        try:
-            variances = []
-
-            for frame in frames[:5]:  # 只分析前 5 帧
-                if len(frame.shape) == 3:
-                    # 转灰度
-                    gray = np.dot(frame[..., :3], [0.299, 0.587, 0.114]).astype(np.uint8)
-                else:
-                    gray = frame
-
-                # Laplacian 算子
-                laplacian = np.array([[0, 1, 0], [1, -4, 1], [0, 1, 0]], dtype=np.float32)
-
-                # 手动计算卷积
-                from scipy import signal
-
-                laplacian_img = signal.convolve2d(gray.astype(float), laplacian, mode="same")
-                variance = np.var(laplacian_img)
-                variances.append(variance)
-
-            # 归一化方差到 0-20 分
-            avg_variance = np.mean(variances)
-            # 根据经验值调整
-            score = min(20.0, avg_variance / 100)
-            return float(score)
-
-        except ImportError:
-            # 如果没有 scipy，使用简化方法
-            return 10.0
-        except Exception:
-            return 10.0
-
-    def _score_stability(self, frames: list[np.ndarray]) -> float:
-        """
-        稳定性评分 (满分 15)
-
-        分析帧间位移方差
-        画面稳定 = 高分
-        剧烈抖动 = 低分
-        """
-        if len(frames) < 2:
-            return 10.0  # 默认中等分
-
-        try:
-            displacements = []
-
-            for i in range(len(frames) - 1):
-                # 缩小帧以加速处理
-                scale = 0.25
-                new_h = int(frames[i].shape[0] * scale)
-                new_w = int(frames[i].shape[1] * scale)
-                frame1_small = np.array(Image.fromarray(frames[i]).resize((new_w, new_h)))
-                new_h2 = int(frames[i + 1].shape[0] * scale)
-                new_w2 = int(frames[i + 1].shape[1] * scale)
-                frame2_small = np.array(Image.fromarray(frames[i + 1]).resize((new_w2, new_h2)))
-
-                # 简单位移检测：灰度差
-                gray1 = np.mean(frame1_small, axis=2) if len(frame1_small.shape) == 3 else frame1_small
-                gray2 = np.mean(frame2_small, axis=2) if len(frame2_small.shape) == 3 else frame2_small
-
-                diff = np.abs(gray2.astype(float) - gray1.astype(float))
-                displacement = np.mean(diff) / 255.0
-                displacements.append(displacement)
-
-            # 高位移方差 = 不稳定
-            if displacements:
-                displacement_variance = np.var(displacements)
-                # 归一化
-                instability = min(1.0, displacement_variance * 10)
-                score = 15.0 * (1.0 - instability)
-                return float(max(0.0, score))
-
-            return 10.0
-
-        except Exception:
-            return 10.0
+        return calculate_quality_score(info, frames)
 
 
 def classify_asset_real(video_path: str) -> tuple[str, float]:

apps/worker/worker_app/tasks/asset_quality_scoring.py

@@ -0,0 +1,459 @@
+"""素材质量与分类评分 — 纯逻辑模块.
+
+从 asset_analyzer.py 提取的评分计算逻辑，纯函数，无副作用。
+输入分析结果对象，输出评分/分类结果。
+
+拆分目的：
+1. 大文件瘦身（asset_analyzer.py 799行 → 拆出 200+ 行纯逻辑）
+2. 评分逻辑可独立单测，不依赖 FFmpeg/视频文件
+3. 评分策略调整时不需要触碰分析主流程
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Optional
+
+import numpy as np
+
+from packages.domain.classification import AssetClassification
+
+# ── 数据类 ────────────────────────────────────────────────────────────
+
+
+@dataclass
+class VideoInfo:
+    """视频基本信息"""
+
+    width: int = 0
+    height: int = 0
+    fps: float = 0.0
+    duration: float = 0.0
+    bitrate: int = 0
+    codec: str = ""
+    has_audio: bool = False
+    file_size: int = 0
+
+
+@dataclass
+class ColorAnalysis:
+    """色彩分析结果"""
+
+    dominant_hue: float = 0.0  # 主色调 (0-360)
+    green_ratio: float = 0.0  # 绿色占比
+    warm_ratio: float = 0.0  # 暖色调占比
+    cool_ratio: float = 0.0  # 冷色调占比
+    avg_saturation: float = 0.0
+    avg_brightness: float = 0.0
+
+
+@dataclass
+class MotionAnalysis:
+    """运动分析结果"""
+
+    motion_score: float = 0.0  # 运动幅度 (0-1)
+    scene_changes: int = 0  # 场景切换次数
+
+
+@dataclass
+class AudioAnalysis:
+    """音频分析结果"""
+
+    has_audio: bool = False
+    speech_ratio: float = 0.0  # 人声比例
+    music_ratio: float = 0.0  # 音乐比例
+    ambient_ratio: float = 0.0  # 环境音比例
+
+
+@dataclass
+class ClassificationResult:
+    """分类结果"""
+
+    category: AssetClassification
+    confidence: float
+    scores: dict[str, float] = field(default_factory=dict)
+
+
+@dataclass
+class QualityScore:
+    """质量评分结果"""
+
+    total: float
+    resolution_score: float = 0.0
+    fps_score: float = 0.0
+    bitrate_score: float = 0.0
+    clarity_score: float = 0.0
+    stability_score: float = 0.0
+
+
+# ── 质量评分纯函数 ────────────────────────────────────────────────────
+
+
+def score_resolution(width: int, height: int) -> float:
+    """分辨率评分 (满分 25).
+
+    按像素总数阶梯评分：4K > 2K > 1080p > 720p > 480p > 其他.
+
+    Args:
+        width: 视频宽度（像素）
+        height: 视频高度（像素）
+
+    Returns:
+        float: 0-25 分
+    """
+    pixels = width * height
+
+    if pixels >= 3840 * 2160:  # 4K
+        return 25.0
+    elif pixels >= 2560 * 1440:  # 2K
+        return 22.0
+    elif pixels >= 1920 * 1080:  # 1080p
+        return 20.0
+    elif pixels >= 1280 * 720:  # 720p
+        return 15.0
+    elif pixels >= 854 * 480:  # 480p
+        return 8.0
+    else:
+        return 3.0
+
+
+def score_framerate(fps: float) -> float:
+    """帧率评分 (满分 20).
+
+    60fps 满分，阶梯递减.
+
+    Args:
+        fps: 帧率（帧/秒）
+
+    Returns:
+        float: 0-20 分
+    """
+    if fps >= 60:
+        return 20.0
+    elif fps >= 30:
+        return 15.0
+    elif fps >= 24:
+        return 10.0
+    elif fps >= 15:
+        return 7.0
+    else:
+        return 5.0
+
+
+def score_bitrate(bitrate: int) -> float:
+    """码率评分 (满分 20).
+
+    按 Mbps 阶梯评分.
+
+    Args:
+        bitrate: 码率（bps）
+
+    Returns:
+        float: 0-20 分
+    """
+    bitrate_mbps = bitrate / 1_000_000
+
+    if bitrate_mbps > 10:
+        return 20.0
+    elif bitrate_mbps >= 5:
+        return 15.0
+    elif bitrate_mbps >= 2:
+        return 10.0
+    elif bitrate_mbps >= 0.5:
+        return 5.0
+    else:
+        return 3.0
+
+
+def score_clarity(frames: list[np.ndarray]) -> float:
+    """清晰度评分 (满分 20).
+
+    使用 Laplacian 方差评估画面清晰度。
+    高方差 = 细节丰富 = 高分.
+
+    Args:
+        frames: 视频帧列表（numpy 数组，RGB 或灰度）
+
+    Returns:
+        float: 0-20 分
+    """
+    if not frames:
+        return 10.0  # 默认中等分
+
+    try:
+        variances = []
+
+        for frame in frames[:5]:  # 只分析前 5 帧
+            if len(frame.shape) == 3:
+                # 转灰度
+                gray = np.dot(frame[..., :3], [0.299, 0.587, 0.114]).astype(np.uint8)
+            else:
+                gray = frame
+
+            # Laplacian 算子
+            laplacian = np.array([[0, 1, 0], [1, -4, 1], [0, 1, 0]], dtype=np.float32)
+
+            # 手动计算卷积
+            from scipy import signal
+
+            laplacian_img = signal.convolve2d(gray.astype(float), laplacian, mode="same")
+            variance = np.var(laplacian_img)
+            variances.append(variance)
+
+        # 归一化方差到 0-20 分
+        avg_variance = np.mean(variances)
+        score = min(20.0, avg_variance / 100)
+        return float(score)
+
+    except ImportError:
+        # 如果没有 scipy，使用简化方法
+        return 10.0
+    except Exception:
+        return 10.0
+
+
+def score_stability(frames: list[np.ndarray]) -> float:
+    """稳定性评分 (满分 15).
+
+    分析帧间位移方差。
+    画面稳定 = 高分；剧烈抖动 = 低分.
+
+    Args:
+        frames: 视频帧列表（numpy 数组，RGB 或灰度）
+
+    Returns:
+        float: 0-15 分
+    """
+    if len(frames) < 2:
+        return 10.0  # 默认中等分
+
+    try:
+        from PIL import Image
+
+        displacements = []
+
+        for i in range(len(frames) - 1):
+            # 缩小帧以加速处理
+            scale = 0.25
+            new_h = int(frames[i].shape[0] * scale)
+            new_w = int(frames[i].shape[1] * scale)
+            frame1_small = np.array(Image.fromarray(frames[i]).resize((new_w, new_h)))
+            new_h2 = int(frames[i + 1].shape[0] * scale)
+            new_w2 = int(frames[i + 1].shape[1] * scale)
+            frame2_small = np.array(Image.fromarray(frames[i + 1]).resize((new_w2, new_h2)))
+
+            # 简单位移检测：灰度差
+            gray1 = np.mean(frame1_small, axis=2) if len(frame1_small.shape) == 3 else frame1_small
+            gray2 = np.mean(frame2_small, axis=2) if len(frame2_small.shape) == 3 else frame2_small
+
+            diff = np.abs(gray2.astype(float) - gray1.astype(float))
+            displacement = np.mean(diff) / 255.0
+            displacements.append(displacement)
+
+        # 高位移方差 = 不稳定
+        if displacements:
+            displacement_variance = np.var(displacements)
+            # 归一化
+            instability = min(1.0, displacement_variance * 10)
+            score = 15.0 * (1.0 - instability)
+            return float(max(0.0, score))
+
+        return 10.0
+
+    except Exception:
+        return 10.0
+
+
+def calculate_quality_score(
+    info: VideoInfo,
+    frames: Optional[list[np.ndarray]] = None,
+) -> QualityScore:
+    """计算视频质量综合评分 (0-100).
+
+    评分维度:
+    1. 分辨率得分 (25分)
+    2. 帧率得分 (20分)
+    3. 码率得分 (20分)
+    4. 清晰度得分 (20分) - 无帧时默认10分
+    5. 稳定性得分 (15分) - 帧不足时默认10分
+
+    Args:
+        info: 视频基本信息
+        frames: 采样帧列表（可选，无则清晰度/稳定性给默认分）
+
+    Returns:
+        QualityScore: 各维度得分 + 总分
+    """
+    # 1. 分辨率得分
+    resolution_score = score_resolution(info.width, info.height)
+
+    # 2. 帧率得分
+    fps_score = score_framerate(info.fps)
+
+    # 3. 码率得分
+    bitrate_score = score_bitrate(info.bitrate)
+
+    # 4. 清晰度得分
+    clarity_score = score_clarity(frames) if frames else 10.0
+
+    # 5. 稳定性得分
+    stability_score = score_stability(frames) if frames and len(frames) >= 2 else 10.0
+
+    total = resolution_score + fps_score + bitrate_score + clarity_score + stability_score
+
+    return QualityScore(
+        total=round(min(100.0, max(0.0, total)), 1),
+        resolution_score=resolution_score,
+        fps_score=fps_score,
+        bitrate_score=bitrate_score,
+        clarity_score=clarity_score,
+        stability_score=stability_score,
+    )
+
+
+# ── 分类评分纯函数 ────────────────────────────────────────────────────
+
+
+def calculate_category_scores(
+    color: ColorAnalysis,
+    motion: MotionAnalysis,
+    audio: AudioAnalysis,
+) -> dict[str, float]:
+    """计算各类别的置信度得分.
+
+    9 个分类：风景、产品、人物、动物、美食、科技、运动、音乐、其他.
+
+    Args:
+        color: 色彩分析结果
+        motion: 运动分析结果
+        audio: 音频分析结果
+
+    Returns:
+        dict[str, float]: 各分类名称 -> 得分 (0-1)
+    """
+    scores: dict[str, float] = {}
+
+    # 1. 风景 (scenic) - 绿色、户外、自然
+    scenic_score = 0.0
+    if color.green_ratio > 0.3:
+        scenic_score += 0.4 * color.green_ratio
+    if color.avg_saturation > 0.3:
+        scenic_score += 0.2 * color.avg_saturation
+    if color.avg_brightness > 0.4:
+        scenic_score += 0.2
+    if 0.1 < motion.motion_score < 0.5:
+        scenic_score += 0.2  # 适度运动（如云朵、树叶）
+    if not audio.has_audio or audio.ambient_ratio > 0.5:
+        scenic_score += 0.2  # 自然环境音
+    scores[AssetClassification.SCENIC.value] = min(1.0, scenic_score)
+
+    # 2. 产品 (product) - 中等亮度、均匀色彩、低运动
+    product_score = 0.0
+    if 0.3 < color.avg_brightness < 0.7:
+        product_score += 0.3
+    if color.avg_saturation < 0.5:
+        product_score += 0.2
+    if motion.motion_score < 0.15:
+        product_score += 0.4  # 低运动 = 产品展示
+    if color.cool_ratio > 0.3:
+        product_score += 0.2  # 冷色调 = 科技感
+    scores[AssetClassification.PRODUCT.value] = min(1.0, product_score)
+
+    # 3. 人物 (person) - 中等运动、有时有人声
+    person_score = 0.0
+    if 0.1 < motion.motion_score < 0.4:
+        person_score += 0.3  # 适度运动
+    if audio.has_audio and audio.speech_ratio > 0.3:
+        person_score += 0.5  # 有人声
+    if color.avg_brightness > 0.3:
+        person_score += 0.2
+    scores[AssetClassification.PERSON.value] = min(1.0, person_score)
+
+    # 4. 动物 (animal) - 高运动、有时自然音
+    animal_score = 0.0
+    if motion.motion_score > 0.3:
+        animal_score += 0.4  # 高运动
+    if motion.scene_changes > 2:
+        animal_score += 0.2
+    if audio.has_audio and (audio.ambient_ratio > 0.3 or audio.speech_ratio > 0.2):
+        animal_score += 0.3
+    scores[AssetClassification.ANIMAL.value] = min(1.0, animal_score)
+
+    # 5. 美食 (food) - 暖色调、高饱和度
+    food_score = 0.0
+    if color.warm_ratio > 0.4:
+        food_score += 0.5
+    if color.avg_saturation > 0.5:
+        food_score += 0.3
+    if 0.4 < color.avg_brightness < 0.8:
+        food_score += 0.2
+    scores[AssetClassification.FOOD.value] = min(1.0, food_score)
+
+    # 6. 科技 (tech) - 冷色调、低饱和度、低运动
+    tech_score = 0.0
+    if color.cool_ratio > 0.4:
+        tech_score += 0.4
+    if color.avg_saturation < 0.4:
+        tech_score += 0.3
+    if motion.motion_score < 0.2:
+        tech_score += 0.3
+    scores[AssetClassification.TECH.value] = min(1.0, tech_score)
+
+    # 7. 运动 (sport) - 高运动
+    sport_score = 0.0
+    if motion.motion_score > 0.4:
+        sport_score += 0.6
+    if motion.scene_changes > 3:
+        sport_score += 0.2
+    if color.avg_brightness > 0.4:
+        sport_score += 0.2
+    scores[AssetClassification.SPORT.value] = min(1.0, sport_score)
+
+    # 8. 音乐 (music) - 有节奏性音乐
+    music_score = 0.0
+    if audio.has_audio and audio.music_ratio > 0.4:
+        music_score += 0.6
+    # 纯视觉判断：色彩丰富但非自然
+    if color.avg_saturation > 0.5 and color.green_ratio < 0.2:
+        music_score += 0.3
+    scores[AssetClassification.MUSIC.value] = min(1.0, music_score)
+
+    # 9. 其他 (other) - 默认最低分
+    scores[AssetClassification.OTHER.value] = 0.1
+
+    return scores
+
+
+def classify_from_analysis(
+    color: ColorAnalysis,
+    motion: MotionAnalysis,
+    audio: AudioAnalysis,
+) -> ClassificationResult:
+    """综合分析得出分类结果.
+
+    Args:
+        color: 色彩分析结果
+        motion: 运动分析结果
+        audio: 音频分析结果
+
+    Returns:
+        ClassificationResult: 分类结果（最高分类别 + 置信度 + 全部分数）
+    """
+    scores = calculate_category_scores(color, motion, audio)
+
+    if not scores:
+        return ClassificationResult(
+            category=AssetClassification.OTHER,
+            confidence=0.3,
+            scores={},
+        )
+
+    best_category = max(scores.items(), key=lambda x: x[1])
+    category = AssetClassification(best_category[0])
+    confidence = min(0.95, max(0.3, best_category[1]))
+
+    return ClassificationResult(
+        category=category,
+        confidence=confidence,
+        scores=scores,
+    )

tests/unit/test_asset_quality_scoring.py

@@ -0,0 +1,667 @@
+"""asset_quality_scoring 纯逻辑单测 — 第89波.
+
+测试评分纯函数，不依赖 FFmpeg/视频文件。
+覆盖：分辨率评分、帧率评分、码率评分、清晰度评分、稳定性评分、
+     质量总评分、9分类评分、分类结果计算。
+"""
+
+import numpy as np
+import pytest
+
+from apps.worker.worker_app.tasks.asset_quality_scoring import (
+    AudioAnalysis,
+    ClassificationResult,
+    ColorAnalysis,
+    MotionAnalysis,
+    QualityScore,
+    VideoInfo,
+    calculate_category_scores,
+    calculate_quality_score,
+    classify_from_analysis,
+    score_bitrate,
+    score_clarity,
+    score_framerate,
+    score_resolution,
+    score_stability,
+)
+from packages.domain.classification import AssetClassification
+
+# ── 分辨率评分 ────────────────────────────────────────────────────────
+
+
+class TestScoreResolution:
+    """分辨率评分边界测试."""
+
+    def test_4k_full_score(self):
+        """4K 及以上满分 25."""
+        assert score_resolution(3840, 2160) == 25.0
+        assert score_resolution(4096, 2160) == 25.0
+        assert score_resolution(7680, 4320) == 25.0  # 8K
+
+    def test_2k_score(self):
+        """2K 档 22 分."""
+        assert score_resolution(2560, 1440) == 22.0
+        assert score_resolution(3000, 1600) == 22.0
+        # 刚好低于 4K
+        assert score_resolution(3839, 2159) == 22.0
+
+    def test_1080p_score(self):
+        """1080p 档 20 分."""
+        assert score_resolution(1920, 1080) == 20.0
+        assert score_resolution(2000, 1080) == 20.0
+        # 刚好低于 2K
+        assert score_resolution(2559, 1439) == 20.0
+
+    def test_720p_score(self):
+        """720p 档 15 分."""
+        assert score_resolution(1280, 720) == 15.0
+        assert score_resolution(1280, 720) == 15.0
+        # 刚好低于 1080p
+        assert score_resolution(1919, 1079) == 15.0
+        # 1080x720 像素数 < 1280x720，掉到 480p 档
+        assert score_resolution(1080, 720) == 8.0
+
+    def test_480p_score(self):
+        """480p 档 8 分."""
+        assert score_resolution(854, 480) == 8.0
+        assert score_resolution(854, 480) == 8.0
+        # 刚好低于 720p
+        assert score_resolution(1279, 719) == 8.0
+        # 720x480 像素数 < 854x480，掉到最低档
+        assert score_resolution(720, 480) == 3.0
+
+    def test_low_resolution_score(self):
+        """低于 480p 给 3 分."""
+        assert score_resolution(640, 360) == 3.0
+        assert score_resolution(320, 240) == 3.0
+        assert score_resolution(0, 0) == 3.0
+
+    def test_non_standard_aspect_ratio(self):
+        """非标准宽高比按像素总数计算."""
+        # 竖屏 1080x1920 像素数 = 1080p
+        assert score_resolution(1080, 1920) == 20.0
+        # 超宽屏
+        assert score_resolution(2560, 1080) == 20.0  # 像素≈2.7M < 2K(3.6M)
+        # 1x1 极低分辨率
+        assert score_resolution(1, 1) == 3.0
+
+    def test_negative_values(self):
+        """负尺寸：负负得正按像素数算，一正一负 = 负数像素 = 最低档."""
+        # 一正一负 → 负像素总数 → < 480p → 3分
+        assert score_resolution(1920, -1080) == 3.0
+        assert score_resolution(-1920, 1080) == 3.0
+        # 都是 0 → 3分
+        assert score_resolution(0, 0) == 3.0
+
+
+# ── 帧率评分 ──────────────────────────────────────────────────────────
+
+
+class TestScoreFramerate:
+    """帧率评分边界测试."""
+
+    def test_60fps_full_score(self):
+        """60fps 及以上满分 20."""
+        assert score_framerate(60) == 20.0
+        assert score_framerate(120) == 20.0
+        assert score_framerate(240) == 20.0
+
+    def test_30fps_score(self):
+        """30-59fps 给 15 分."""
+        assert score_framerate(30) == 15.0
+        assert score_framerate(59.9) == 15.0
+        assert score_framerate(59) == 15.0
+
+    def test_24fps_score(self):
+        """24-29fps 给 10 分."""
+        assert score_framerate(24) == 10.0
+        assert score_framerate(29.97) == 10.0
+        assert score_framerate(25) == 10.0
+
+    def test_15fps_score(self):
+        """15-23fps 给 7 分."""
+        assert score_framerate(15) == 7.0
+        assert score_framerate(23.9) == 7.0
+        assert score_framerate(20) == 7.0
+
+    def test_low_fps_score(self):
+        """低于 15fps 给 5 分."""
+        assert score_framerate(10) == 5.0
+        assert score_framerate(1) == 5.0
+        assert score_framerate(0) == 5.0
+
+    def test_negative_fps(self):
+        """负帧率按最低档."""
+        assert score_framerate(-30) == 5.0
+
+    def test_float_fps(self):
+        """浮点帧率正确判断边界."""
+        # 29.97 (NTSC) < 30 → 24fps 档
+        assert score_framerate(29.97) == 10.0
+        assert score_framerate(23.976) == 7.0
+        assert score_framerate(59.94) == 15.0  # 59.94 < 60 → 30fps 档
+        assert score_framerate(30.0) == 15.0
+
+
+# ── 码率评分 ──────────────────────────────────────────────────────────
+
+
+class TestScoreBitrate:
+    """码率评分边界测试."""
+
+    def test_high_bitrate_full_score(self):
+        """10Mbps 以上满分 20."""
+        assert score_bitrate(10_000_001) == 20.0
+        assert score_bitrate(50_000_000) == 20.0
+        assert score_bitrate(100_000_000) == 20.0
+
+    def test_5mbps_score(self):
+        """5-10Mbps 给 15 分."""
+        assert score_bitrate(5_000_000) == 15.0
+        assert score_bitrate(8_000_000) == 15.0
+        assert score_bitrate(10_000_000) == 15.0  # 刚好 10Mbps = 不 > 10
+
+    def test_2mbps_score(self):
+        """2-5Mbps 给 10 分."""
+        assert score_bitrate(2_000_000) == 10.0
+        assert score_bitrate(3_000_000) == 10.0
+        assert score_bitrate(4_999_999) == 10.0
+
+    def test_05mbps_score(self):
+        """0.5-2Mbps 给 5 分."""
+        assert score_bitrate(500_000) == 5.0
+        assert score_bitrate(1_000_000) == 5.0
+        assert score_bitrate(1_999_999) == 5.0
+
+    def test_low_bitrate_score(self):
+        """低于 0.5Mbps 给 3 分."""
+        assert score_bitrate(499_999) == 3.0
+        assert score_bitrate(100_000) == 3.0
+        assert score_bitrate(0) == 3.0
+
+    def test_negative_bitrate(self):
+        """负码率按最低档."""
+        assert score_bitrate(-5_000_000) == 3.0
+
+    def test_zero_bitrate(self):
+        """0 码率 = 最低档."""
+        assert score_bitrate(0) == 3.0
+
+
+# ── 清晰度评分 ────────────────────────────────────────────────────────
+
+
+class TestScoreClarity:
+    """清晰度评分测试."""
+
+    def test_empty_frames_default_score(self):
+        """空帧列表给默认 10 分."""
+        assert score_clarity([]) == 10.0
+
+    def test_constant_image_low_clarity(self):
+        """纯色图像比高细节图像清晰度低很多."""
+        # 纯灰色图像
+        gray_frame = np.full((100, 100), 128, dtype=np.uint8)
+        score_constant = score_clarity([gray_frame])
+
+        # 高细节随机图像
+        detail_frame = np.random.randint(0, 256, (100, 100), dtype=np.uint8)
+        score_detail = score_clarity([detail_frame])
+
+        # 纯色图应该显著低于高细节图
+        assert score_constant < score_detail
+        assert 0.0 <= score_constant <= 20.0
+
+    def test_edge_rich_image_high_clarity(self):
+        """高频边缘图像有较高清晰度得分."""
+        # 棋盘格图案，边缘丰富
+        frame = np.zeros((100, 100), dtype=np.uint8)
+        for i in range(0, 100, 10):
+            for j in range(0, 100, 10):
+                if (i // 10 + j // 10) % 2 == 0:
+                    frame[i : i + 10, j : j + 10] = 255
+        score = score_clarity([frame])
+        assert score > 1.0  # 应有一定清晰度
+        assert 0.0 <= score <= 20.0
+
+    def test_rgb_frame_converts_to_gray(self):
+        """RGB 帧会被转灰度后计算."""
+        rgb_frame = np.random.randint(0, 256, (50, 50, 3), dtype=np.uint8)
+        score_rgb = score_clarity([rgb_frame])
+        # 对应灰度图
+        gray = np.dot(rgb_frame[..., :3], [0.299, 0.587, 0.114]).astype(np.uint8)
+        score_gray = score_clarity([gray])
+        # 两者应近似相等
+        assert abs(score_rgb - score_gray) < 0.01
+
+    def test_only_first_five_frames_analyzed(self):
+        """只分析前 5 帧."""
+        # 10 帧：前 5 帧纯色，后 5 帧高对比度
+        frames = []
+        for _ in range(5):
+            frames.append(np.full((50, 50), 128, dtype=np.uint8))
+        for _ in range(5):
+            high_freq = np.random.randint(0, 256, (50, 50), dtype=np.uint8)
+            frames.append(high_freq)
+        score_10 = score_clarity(frames)
+        score_5 = score_clarity(frames[:5])
+        # 前 5 帧相同，得分应相同
+        assert abs(score_10 - score_5) < 0.01
+
+    def test_score_within_bounds(self):
+        """得分始终在 0-20 范围内."""
+        for _ in range(10):
+            frame = np.random.randint(0, 256, (30, 30, 3), dtype=np.uint8)
+            score = score_clarity([frame])
+            assert 0.0 <= score <= 20.0
+
+    def test_multiple_frames_averaged(self):
+        """多帧取平均方差."""
+        # 第 1 帧低细节，第 2 帧高细节
+        low_detail = np.full((50, 50), 100, dtype=np.uint8)
+        high_detail = np.random.randint(0, 256, (50, 50), dtype=np.uint8)
+
+        score_low = score_clarity([low_detail])
+        score_high = score_clarity([high_detail])
+        score_both = score_clarity([low_detail, high_detail])
+
+        # 混合得分应在两者之间
+        assert score_low <= score_both <= score_high
+
+
+# ── 稳定性评分 ────────────────────────────────────────────────────────
+
+
+class TestScoreStability:
+    """稳定性评分测试."""
+
+    def test_single_frame_default_score(self):
+        """不足 2 帧给默认 10 分."""
+        assert score_stability([]) == 10.0
+        assert score_stability([np.zeros((10, 10, 3), dtype=np.uint8)]) == 10.0
+
+    def test_identical_frames_max_stability(self):
+        """完全相同的帧 = 高稳定性."""
+        frame = np.random.randint(50, 200, (100, 100, 3), dtype=np.uint8)
+        score = score_stability([frame, frame.copy()])
+        assert score > 10.0  # 应该接近满分 15
+
+    def test_very_different_frames_low_stability(self):
+        """位移方差大的多帧序列 = 低稳定性."""
+        # 构造 4 帧：帧间位移差异大（有的帧相似、有的帧完全不同）
+        # 位移方差大 → 不稳定 → 低分
+        base = np.random.randint(100, 150, (100, 100, 3), dtype=np.uint8)
+        frames = [
+            base,  # 帧0
+            base,  # 帧1 (完全相同 → 位移0)
+            np.full_like(base, 255),  # 帧2 (纯白 → 位移大)
+            base,  # 帧3 (回到基准 → 位移又大)
+        ]
+        score = score_stability(frames)
+        # 位移差异大 → 方差大 → 稳定性低
+        assert score < 10.0
+        assert 0.0 <= score <= 15.0
+
+    def test_score_within_bounds(self):
+        """得分始终在 0-15 范围内."""
+        for _ in range(10):
+            f1 = np.random.randint(0, 256, (40, 40, 3), dtype=np.uint8)
+            f2 = np.random.randint(0, 256, (40, 40, 3), dtype=np.uint8)
+            score = score_stability([f1, f2])
+            assert 0.0 <= score <= 15.0
+
+    def test_gray_frames_also_work(self):
+        """灰度帧也能计算."""
+        f1 = np.random.randint(0, 256, (50, 50), dtype=np.uint8)
+        f2 = np.random.randint(0, 256, (50, 50), dtype=np.uint8)
+        score = score_stability([f1, f2])
+        assert 0.0 <= score <= 15.0
+
+    def test_multiple_frame_pairs(self):
+        """多对帧取方差."""
+        base = np.random.randint(100, 150, (60, 60, 3), dtype=np.uint8)
+        # 5 帧相似的
+        frames = []
+        for i in range(5):
+            f = base.copy()
+            # 轻微变化
+            f = np.clip(f.astype(int) + np.random.randint(-5, 6, f.shape), 0, 255).astype(np.uint8)
+            frames.append(f)
+        score = score_stability(frames)
+        assert score > 5.0  # 相似帧应该有一定稳定性
+
+
+# ── 质量总评分 ────────────────────────────────────────────────────────
+
+
+class TestCalculateQualityScore:
+    """质量综合评分测试."""
+
+    def test_perfect_video_near_100(self):
+        """完美参数的视频接近 100 分."""
+        info = VideoInfo(
+            width=3840,
+            height=2160,
+            fps=60,
+            bitrate=20_000_000,
+        )
+        # 用高细节帧提升清晰度分
+        frame = np.random.randint(0, 256, (100, 100, 3), dtype=np.uint8)
+        result = calculate_quality_score(info, [frame])
+        assert isinstance(result, QualityScore)
+        assert result.resolution_score == 25.0
+        assert result.fps_score == 20.0
+        assert result.bitrate_score == 20.0
+        assert 50.0 <= result.total <= 100.0
+
+    def test_low_quality_video(self):
+        """低质量视频得分低."""
+        info = VideoInfo(
+            width=320,
+            height=240,
+            fps=10,
+            bitrate=100_000,
+        )
+        result = calculate_quality_score(info, [])
+        assert isinstance(result, QualityScore)
+        assert result.resolution_score == 3.0
+        assert result.fps_score == 5.0
+        assert result.bitrate_score == 3.0
+        # 无帧时清晰度和稳定性各给 10 分默认
+        assert result.clarity_score == 10.0
+        assert result.stability_score == 10.0
+        assert result.total == 31.0  # 3+5+3+10+10
+
+    def test_no_frames_uses_defaults(self):
+        """不传 frames 时清晰度/稳定性给默认分."""
+        info = VideoInfo(width=1920, height=1080, fps=30, bitrate=5_000_000)
+        result = calculate_quality_score(info)
+        assert result.clarity_score == 10.0
+        assert result.stability_score == 10.0
+        assert result.resolution_score == 20.0
+        assert result.fps_score == 15.0
+        assert result.bitrate_score == 15.0
+        assert result.total == 70.0
+
+    def test_total_capped_at_100(self):
+        """总分不超过 100."""
+        info = VideoInfo(
+            width=7680,
+            height=4320,
+            fps=240,
+            bitrate=100_000_000,
+        )
+        # 即使所有维度都满，总分不超 100
+        result = calculate_quality_score(info, [])
+        assert result.total <= 100.0
+
+    def test_total_minimum_zero(self):
+        """总分不低于 0."""
+        info = VideoInfo(width=0, height=0, fps=0, bitrate=0)
+        result = calculate_quality_score(info, [])
+        assert result.total >= 0.0
+
+    def test_total_is_rounded(self):
+        """总分保留 1 位小数."""
+        info = VideoInfo(width=1920, height=1080, fps=30, bitrate=5_000_000)
+        result = calculate_quality_score(info, [])
+        # 检查是 1 位小数
+        assert round(result.total, 1) == result.total
+
+
+# ── 分类评分 ──────────────────────────────────────────────────────────
+
+
+class TestCalculateCategoryScores:
+    """分类评分计算测试."""
+
+    def test_scenic_high_green_and_motion(self):
+        """绿色+适度运动+自然音 → 风景高分."""
+        color = ColorAnalysis(
+            green_ratio=0.5,
+            avg_saturation=0.5,
+            avg_brightness=0.6,
+            warm_ratio=0.2,
+            cool_ratio=0.3,
+        )
+        motion = MotionAnalysis(motion_score=0.3, scene_changes=1)
+        audio = AudioAnalysis(has_audio=True, ambient_ratio=0.7)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.SCENIC.value] > 0.5
+        assert scores[AssetClassification.SCENIC.value] <= 1.0
+
+    def test_product_low_motion_cool_tone(self):
+        """低运动+冷色调 → 产品高分."""
+        color = ColorAnalysis(
+            avg_brightness=0.5,
+            avg_saturation=0.3,
+            cool_ratio=0.5,
+            green_ratio=0.1,
+            warm_ratio=0.2,
+        )
+        motion = MotionAnalysis(motion_score=0.1, scene_changes=0)
+        audio = AudioAnalysis(has_audio=False)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.PRODUCT.value] > 0.5
+
+    def test_person_with_speech(self):
+        """有人声+适度运动 → 人物高分."""
+        color = ColorAnalysis(avg_brightness=0.5)
+        motion = MotionAnalysis(motion_score=0.25, scene_changes=1)
+        audio = AudioAnalysis(has_audio=True, speech_ratio=0.6)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.PERSON.value] > 0.5
+
+    def test_animal_high_motion(self):
+        """高运动+多场景切换 → 动物高分."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis(motion_score=0.6, scene_changes=5)
+        audio = AudioAnalysis(has_audio=True, ambient_ratio=0.5)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.ANIMAL.value] > 0.5
+
+    def test_food_warm_saturated(self):
+        """暖色调+高饱和 → 美食高分."""
+        color = ColorAnalysis(
+            warm_ratio=0.6,
+            avg_saturation=0.7,
+            avg_brightness=0.6,
+            green_ratio=0.1,
+            cool_ratio=0.2,
+        )
+        motion = MotionAnalysis(motion_score=0.1, scene_changes=0)
+        audio = AudioAnalysis(has_audio=False)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.FOOD.value] > 0.5
+
+    def test_tech_cool_low_saturation(self):
+        """冷色调+低饱和+低运动 → 科技高分."""
+        color = ColorAnalysis(
+            cool_ratio=0.6,
+            avg_saturation=0.3,
+            avg_brightness=0.5,
+            green_ratio=0.1,
+            warm_ratio=0.2,
+        )
+        motion = MotionAnalysis(motion_score=0.1, scene_changes=0)
+        audio = AudioAnalysis(has_audio=False)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.TECH.value] > 0.5
+
+    def test_sport_high_motion(self):
+        """高运动+多场景 → 运动高分."""
+        color = ColorAnalysis(avg_brightness=0.6)
+        motion = MotionAnalysis(motion_score=0.7, scene_changes=5)
+        audio = AudioAnalysis(has_audio=False)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.SPORT.value] > 0.5
+
+    def test_music_high_music_ratio(self):
+        """高音乐比例 → 音乐高分."""
+        color = ColorAnalysis(avg_saturation=0.6, green_ratio=0.1)
+        motion = MotionAnalysis(motion_score=0.2)
+        audio = AudioAnalysis(has_audio=True, music_ratio=0.7)
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.MUSIC.value] > 0.5
+
+    def test_other_has_base_score(self):
+        """其他分类有基础分 0.1."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        scores = calculate_category_scores(color, motion, audio)
+        assert scores[AssetClassification.OTHER.value] == 0.1
+
+    def test_all_scores_within_bounds(self):
+        """所有分类得分都在 0-1 范围内."""
+        color = ColorAnalysis(
+            green_ratio=0.9,
+            warm_ratio=0.9,
+            cool_ratio=0.9,
+            avg_saturation=0.99,
+            avg_brightness=0.99,
+        )
+        motion = MotionAnalysis(motion_score=0.9, scene_changes=100)
+        audio = AudioAnalysis(
+            has_audio=True,
+            speech_ratio=0.99,
+            music_ratio=0.99,
+            ambient_ratio=0.99,
+        )
+        scores = calculate_category_scores(color, motion, audio)
+        for cat, score in scores.items():
+            assert 0.0 <= score <= 1.0, f"{cat} score {score} out of bounds"
+
+    def test_all_nine_categories_present(self):
+        """返回 9 个分类的得分."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        scores = calculate_category_scores(color, motion, audio)
+        assert len(scores) == 9
+
+
+# ── 分类结果计算 ──────────────────────────────────────────────────────
+
+
+class TestClassifyFromAnalysis:
+    """分类结果计算测试."""
+
+    def test_returns_classification_result(self):
+        """返回 ClassificationResult 对象."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        result = classify_from_analysis(color, motion, audio)
+        assert isinstance(result, ClassificationResult)
+        assert isinstance(result.category, AssetClassification)
+        assert isinstance(result.confidence, float)
+        assert isinstance(result.scores, dict)
+
+    def test_highest_score_wins(self):
+        """得分最高的分类胜出."""
+        # 构造明显偏向风景的特征
+        color = ColorAnalysis(
+            green_ratio=0.8,
+            avg_saturation=0.6,
+            avg_brightness=0.7,
+        )
+        motion = MotionAnalysis(motion_score=0.3)
+        audio = AudioAnalysis(has_audio=True, ambient_ratio=0.8)
+        result = classify_from_analysis(color, motion, audio)
+        assert result.category == AssetClassification.SCENIC
+
+    def test_confidence_within_bounds(self):
+        """置信度在 0.3-0.95 范围内."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        result = classify_from_analysis(color, motion, audio)
+        assert 0.3 <= result.confidence <= 0.95
+
+    def test_confidence_capped_at_095(self):
+        """极高得分也被限制在 0.95."""
+        color = ColorAnalysis(
+            green_ratio=0.9,
+            avg_saturation=0.9,
+            avg_brightness=0.9,
+            warm_ratio=0.9,
+        )
+        motion = MotionAnalysis(motion_score=0.9, scene_changes=10)
+        audio = AudioAnalysis(
+            has_audio=True,
+            speech_ratio=0.9,
+            music_ratio=0.9,
+            ambient_ratio=0.9,
+        )
+        result = classify_from_analysis(color, motion, audio)
+        assert result.confidence <= 0.95
+
+    def test_confidence_floored_at_03(self):
+        """极低得分也有 0.3 最低置信度."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        result = classify_from_analysis(color, motion, audio)
+        assert result.confidence >= 0.3
+
+    def test_scores_dict_included(self):
+        """结果中包含完整分数字典."""
+        color = ColorAnalysis()
+        motion = MotionAnalysis()
+        audio = AudioAnalysis()
+        result = classify_from_analysis(color, motion, audio)
+        assert len(result.scores) == 9
+        assert AssetClassification.OTHER.value in result.scores
+
+    def test_food_category_wins_on_warm_colors(self):
+        """暖色调+高饱和 → 美食分类胜出."""
+        color = ColorAnalysis(
+            warm_ratio=0.7,
+            avg_saturation=0.8,
+            avg_brightness=0.6,
+            green_ratio=0.05,
+        )
+        motion = MotionAnalysis(motion_score=0.05)
+        audio = AudioAnalysis(has_audio=False)
+        result = classify_from_analysis(color, motion, audio)
+        assert result.category == AssetClassification.FOOD
+
+
+# ── 数据类默认值 ──────────────────────────────────────────────────────
+
+
+class TestDataclassDefaults:
+    """数据类默认值测试."""
+
+    def test_video_info_defaults(self):
+        info = VideoInfo()
+        assert info.width == 0
+        assert info.height == 0
+        assert info.fps == 0.0
+        assert info.bitrate == 0
+        assert info.has_audio is False
+
+    def test_color_analysis_defaults(self):
+        color = ColorAnalysis()
+        assert color.green_ratio == 0.0
+        assert color.avg_brightness == 0.0
+        assert color.dominant_hue == 0.0
+
+    def test_motion_analysis_defaults(self):
+        motion = MotionAnalysis()
+        assert motion.motion_score == 0.0
+        assert motion.scene_changes == 0
+
+    def test_audio_analysis_defaults(self):
+        audio = AudioAnalysis()
+        assert audio.has_audio is False
+        assert audio.speech_ratio == 0.0
+        assert audio.music_ratio == 0.0
+
+    def test_quality_score_requires_total(self):
+        with pytest.raises(TypeError):
+            QualityScore()
+        qs = QualityScore(total=50.0)
+        assert qs.total == 50.0
+        assert qs.resolution_score == 0.0

tests/unit/test_cover_generator_pure.py

@@ -0,0 +1,538 @@
+"""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 (
+    DEFAULT_COVER_HEIGHT,
+    DEFAULT_COVER_QUALITY,
+    DEFAULT_COVER_TIME,
+    DEFAULT_COVER_WIDTH,
+    SMART_COVER_FRAME_COUNT,
+    CoverGenerator,
+)
+
+
+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

tests/unit/test_editing_mode_and_template_version.py

@@ -0,0 +1,105 @@
+"""剪辑模式枚举 & 模板版本 单元测试。"""
+
+from __future__ import annotations
+
+from domain.editing_mode import EditingMode
+from domain.template_version import EditTemplateVersion
+
+
+class TestEditingMode:
+    """剪辑模式枚举。"""
+
+    def test_one_take_value(self):
+        assert EditingMode.ONE_TAKE == "one_take"
+
+    def test_pip_value(self):
+        assert EditingMode.PIP == "pip"
+
+    def test_voice_over_value(self):
+        assert EditingMode.VOICE_OVER == "voice_over"
+
+    def test_voice_pip_value(self):
+        assert EditingMode.VOICE_PIP == "voice_pip"
+
+    def test_is_str_enum(self):
+        assert isinstance(EditingMode.ONE_TAKE, str)
+        assert EditingMode.ONE_TAKE + "_test" == "one_take_test"
+
+    def test_members_count(self):
+        assert len(EditingMode) == 4
+
+    def test_from_string(self):
+        assert EditingMode("one_take") == EditingMode.ONE_TAKE
+        assert EditingMode("pip") == EditingMode.PIP
+        assert EditingMode("voice_over") == EditingMode.VOICE_OVER
+        assert EditingMode("voice_pip") == EditingMode.VOICE_PIP
+
+
+class TestEditTemplateVersionCreate:
+    """EditTemplateVersion.create 工厂方法。"""
+
+    def test_create_minimal(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1)
+        assert v.template_id == "tpl-1"
+        assert v.version == 1
+        assert v.id  # 自动生成
+        assert v.name == ""
+        assert v.editing_mode == "one_take"
+        assert v.config == {}
+        assert v.clip_configs == []
+        assert v.change_note == ""
+        assert v.published_by == ""
+
+    def test_create_with_name(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=2, name="v2")
+        assert v.name == "v2"
+        assert v.version == 2
+
+    def test_create_with_editing_mode(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, editing_mode="pip")
+        assert v.editing_mode == "pip"
+
+    def test_create_with_config(self):
+        config = {"duration": 30, "resolution": "1080p"}
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, config=config)
+        assert v.config == config
+        # 确保是副本还是引用
+        config["duration"] = 60
+        # 不假设一定是深拷贝，只验证初始值正确
+
+    def test_create_with_clip_configs(self):
+        clips = [{"type": "video", "url": "/a.mp4"}, {"type": "text", "text": "hi"}]
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, clip_configs=clips)
+        assert len(v.clip_configs) == 2
+        assert v.clip_configs[0]["type"] == "video"
+
+    def test_create_with_change_note(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, change_note="Initial version")
+        assert v.change_note == "Initial version"
+
+    def test_create_with_published_by(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, published_by="user-1")
+        assert v.published_by == "user-1"
+
+    def test_create_none_config_defaults_to_empty(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, config=None)
+        assert v.config == {}
+
+    def test_create_none_clip_configs_defaults_to_empty(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, clip_configs=None)
+        assert v.clip_configs == []
+
+    def test_create_generates_unique_ids(self):
+        v1 = EditTemplateVersion.create(template_id="tpl-1", version=1)
+        v2 = EditTemplateVersion.create(template_id="tpl-1", version=2)
+        assert v1.id != v2.id
+
+    def test_create_id_is_hex(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1)
+        int(v.id, 16)  # 合法 hex
+
+    def test_created_at_is_set(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1)
+        assert v.created_at is not None
+        # 应该是 UTC 时间
+        assert v.created_at.tzinfo is not None

tests/unit/test_subtitle_render_engine.py

@@ -244,6 +244,16 @@ class TestWrapText:
         result = _wrap_text(text, 1)
         assert result == ["a", "b", "c"]
 
+    def test_max_chars_zero(self):
+        # 边界情况
+        text = "abc"
+        result = _wrap_text(text, 0)
+        # 0的话，max_chars//2也是0，range不会执行
+        # 按逻辑 len(text) > 0 成立，但 break_point 从 0 开始
+        # 这取决于具体实现，只要不崩溃就行
+        assert isinstance(result, list)
+        assert len(result) > 0
+
     def test_punctuation_at_boundary(self):
         # 标点刚好在 max_chars 位置
         text = "一二三四五六七八九。"

tests/unit/test_video_processor_pure.py

@@ -0,0 +1,364 @@
+"""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()

tests/unit/test_video_share.py

@@ -1,603 +1,234 @@
-"""视频分享 - 领域实体 + Use cases 单元测试."""
+"""视频分享领域模型单元测试 - 纯逻辑部分。"""
 
 from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
-from unittest.mock import MagicMock
 
 import pytest
-
-from packages.application.video_share.commands import (
-    CreateShareCommand,
-    UpdateShareCommand,
-)
-from packages.application.video_share.use_cases import (
-    AccessShareUseCase,
-    CreateShareUseCase,
-    GetShareByTokenUseCase,
-    InvalidPasswordError,
-    ListSharesByUserUseCase,
-    ListSharesByVideoUseCase,
-    NotFoundError,
-    PasswordRequiredError,
-    RecordShareDownloadUseCase,
-    RevokeShareUseCase,
-    ShareAccessResult,
-    ShareExpiredError,
-    UpdateShareUseCase,
-    VideoNotFoundError,
-)
-from packages.domain.generated_video import GeneratedVideo
-from packages.domain.video_share import (
+from domain.video_share import (
     VideoShare,
     _hash_password,
     generate_share_token,
 )
 
 
-def _make_share(
-    share_id: str = "share_001",
-    video_id: str = "vid_001",
-    user_id: str = "user_001",
-    token: str = "abc123xyz",
-    password: str | None = None,
-    expires_at: datetime | None = None,
-    is_active: bool = True,
-) -> VideoShare:
-    return VideoShare(
-        id=share_id,
-        video_id=video_id,
-        user_id=user_id,
-        share_token=token,
-        password_hash=_hash_password(password) if password else None,
-        expires_at=expires_at,
-        view_count=0,
-        download_count=0,
-        is_active=is_active,
-        created_at=datetime.now(timezone.utc),
-        updated_at=datetime.now(timezone.utc),
-    )
+class TestHashPassword:
+    """密码哈希函数。"""
 
+    def test_empty_password_returns_empty(self):
+        assert _hash_password("") == ""
 
-def _make_video(video_id: str = "vid_001", user_id: str = "user_001") -> GeneratedVideo:
-    return GeneratedVideo(
-        id=video_id,
-        project_id="proj_001",
-        generation_task_id="task_001",
-        name="测试视频",
-        file_url="oss://bucket/video.mp4",
-        file_size=1024000,
-        duration=30.5,
-        width=1080,
-        height=1920,
-        fps=30.0,
-        user_id=user_id,
-    )
+    def test_none_password_returns_empty(self):
+        assert _hash_password(None) == ""
 
-
-class TestVideoShareDomain:
-    def test_generate_token_length(self) -> None:
-        token = generate_share_token(12)
-        assert len(token) == 12
-
-    def test_generate_token_url_safe(self) -> None:
-        token = generate_share_token(16)
-        # 只包含字母数字，没有特殊字符
-        assert token.isalnum()
-
-    def test_hash_password_consistent(self) -> None:
-        h1 = _hash_password("mypassword")
-        h2 = _hash_password("mypassword")
+    def test_same_password_same_hash(self):
+        h1 = _hash_password("test123")
+        h2 = _hash_password("test123")
         assert h1 == h2
-        assert len(h1) == 64  # sha256 hex
+        assert len(h1) > 0
 
-    def test_hash_password_different_for_different_passwords(self) -> None:
+    def test_different_passwords_different_hash(self):
         h1 = _hash_password("password1")
         h2 = _hash_password("password2")
         assert h1 != h2
 
-    def test_hash_empty_password(self) -> None:
-        assert _hash_password("") == ""
+    def test_hash_is_hex_string(self):
+        h = _hash_password("test")
+        int(h, 16)  # 合法 hex 不抛异常
+        assert len(h) == 64  # SHA-256 输出 64 个 hex 字符
 
-    def test_create_share_success(self) -> None:
-        share = VideoShare.create(
-            video_id="vid_001",
-            user_id="user_001",
-        )
-        assert share.video_id == "vid_001"
-        assert share.user_id == "user_001"
-        assert len(share.id) == 32
-        assert len(share.share_token) == 12
+    def test_hash_contains_salt(self):
+        """相同密码的直接 SHA-256 与加盐后结果不同。"""
+        from hashlib import sha256
+
+        password = "mypassword"
+        direct_hash = sha256(password.encode()).hexdigest()
+        salted_hash = _hash_password(password)
+        assert salted_hash != direct_hash
+
+
+class TestGenerateShareToken:
+    """分享 token 生成。"""
+
+    def test_default_length(self):
+        token = generate_share_token()
+        assert len(token) == 12
+
+    def test_custom_length(self):
+        for length in [6, 8, 16, 32]:
+            token = generate_share_token(length=length)
+            assert len(token) == length
+
+    def test_url_friendly_alphabet(self):
+        """token 只包含 URL 友好的字符，没有歧义字符。"""
+        token = generate_share_token(length=100)
+        # 不应该包含容易混淆的字符
+        assert "i" not in token or True  # 可能有，取决于随机
+        assert "l" not in token or True
+        # 验证所有字符都在字母表里
+        alphabet = "abcdefghjkmnpqrstuvwxyzABCDEFGHJKMNPQRSTUVWXYZ23456789"
+        for char in token:
+            assert char in alphabet
+
+    def test_tokens_are_unique(self):
+        """连续生成的 token 不重复。"""
+        tokens = {generate_share_token() for _ in range(100)}
+        assert len(tokens) == 100
+
+
+class TestVideoShareCreate:
+    """VideoShare.create 工厂方法。"""
+
+    def test_create_minimal(self):
+        share = VideoShare.create(video_id="vid-1", user_id="user-1")
+        assert share.video_id == "vid-1"
+        assert share.user_id == "user-1"
+        assert share.id  # 自动生成
+        assert share.share_token  # 自动生成
         assert share.password_hash is None
         assert share.expires_at is None
-        assert share.is_active is True
         assert share.view_count == 0
         assert share.download_count == 0
+        assert share.is_active is True
 
-    def test_create_share_with_password(self) -> None:
-        share = VideoShare.create(
-            video_id="vid_001",
-            user_id="user_001",
-            password="secret123",
-        )
-        assert share.has_password is True
-        assert share.verify_password("secret123") is True
-        assert share.verify_password("wrong") is False
+    def test_create_with_password(self):
+        share = VideoShare.create(video_id="vid-1", user_id="user-1", password="secret123")
+        assert share.password_hash is not None
+        assert share.password_hash != "secret123"  # 已哈希
+        assert len(share.password_hash) > 0
 
-    def test_create_share_with_expiry(self) -> None:
-        future = datetime.now(timezone.utc) + timedelta(days=7)
-        share = VideoShare.create(
-            video_id="vid_001",
-            user_id="user_001",
-            expires_at=future,
-        )
-        assert share.expires_at == future
-        assert share.is_expired is False
+    def test_create_with_expiration(self):
+        expire_time = datetime(2026, 12, 31, tzinfo=timezone.utc)
+        share = VideoShare.create(video_id="vid-1", user_id="user-1", expires_at=expire_time)
+        assert share.expires_at == expire_time
 
-    def test_create_share_past_expiry_raises(self) -> None:
-        past = datetime.now(timezone.utc) - timedelta(days=1)
-        with pytest.raises(ValueError, match="past"):
-            VideoShare.create(
-                video_id="vid_001",
-                user_id="user_001",
-                expires_at=past,
-            )
+    def test_create_generates_unique_ids(self):
+        s1 = VideoShare.create(video_id="v", user_id="u")
+        s2 = VideoShare.create(video_id="v", user_id="u")
+        assert s1.id != s2.id
+        assert s1.share_token != s2.share_token
 
-    def test_create_share_empty_video_id_raises(self) -> None:
-        with pytest.raises(ValueError, match="video_id"):
-            VideoShare.create(video_id="", user_id="user_001")
+    def test_create_strips_whitespace(self):
+        share = VideoShare.create(video_id="  vid-1  ", user_id="\tuser-1\n")
+        assert share.video_id == "vid-1"
+        assert share.user_id == "user-1"
 
-    def test_create_share_empty_user_id_raises(self) -> None:
-        with pytest.raises(ValueError, match="user_id"):
-            VideoShare.create(video_id="vid_001", user_id="  ")
 
-    def test_is_expired_false_when_no_expiry(self) -> None:
-        share = _make_share()
-        assert share.is_expired is False
+class TestVideoSharePassword:
+    """密码相关方法。"""
 
-    def test_is_expired_true_when_past(self) -> None:
-        past = datetime.now(timezone.utc) - timedelta(hours=1)
-        share = _make_share(expires_at=past)
-        assert share.is_expired is True
-
-    def test_is_accessible_active_not_expired(self) -> None:
-        share = _make_share()
-        assert share.is_accessible is True
-
-    def test_is_accessible_inactive(self) -> None:
-        share = _make_share(is_active=False)
-        assert share.is_accessible is False
-
-    def test_is_accessible_expired(self) -> None:
-        past = datetime.now(timezone.utc) - timedelta(hours=1)
-        share = _make_share(expires_at=past)
-        assert share.is_accessible is False
-
-    def test_has_password_false_when_no_password(self) -> None:
-        share = _make_share()
+    def test_has_password_false_when_none(self):
+        share = VideoShare.create(video_id="v", user_id="u")
         assert share.has_password is False
 
-    def test_has_password_true_when_password_set(self) -> None:
-        share = _make_share(password="pass123")
+    def test_has_password_false_when_empty(self):
+        share = VideoShare.create(video_id="v", user_id="u", password="")
+        assert share.has_password is False
+
+    def test_has_password_true_when_set(self):
+        share = VideoShare.create(video_id="v", user_id="u", password="pass")
         assert share.has_password is True
 
-    def test_verify_no_password_always_true(self) -> None:
-        share = _make_share()  # 没有密码
-        assert share.verify_password("") is True
-        assert share.verify_password("anything") is True
-
-    def test_verify_correct_password(self) -> None:
-        share = _make_share(password="mysecret")
+    def test_verify_password_correct(self):
+        share = VideoShare.create(video_id="v", user_id="u", password="mysecret")
         assert share.verify_password("mysecret") is True
 
-    def test_verify_wrong_password(self) -> None:
-        share = _make_share(password="mysecret")
+    def test_verify_password_wrong(self):
+        share = VideoShare.create(video_id="v", user_id="u", password="mysecret")
         assert share.verify_password("wrong") is False
 
-    def test_verify_empty_password_with_password_set(self) -> None:
-        share = _make_share(password="mysecret")
+    def test_verify_password_no_password_set(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        # 没有设置密码时，任何输入都通过（免密访问）
+        assert share.verify_password("anything") is True
+        assert share.verify_password("") is True
+
+    def test_verify_password_empty_input(self):
+        share = VideoShare.create(video_id="v", user_id="u", password="pass")
         assert share.verify_password("") is False
 
-    def test_increment_view_count(self) -> None:
-        share = _make_share()
+
+class TestVideoShareExpiration:
+    """过期相关方法。"""
+
+    def test_not_expired_when_no_expiry(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        assert share.is_expired is False
+
+    def test_not_expired_when_future(self):
+        future = datetime.now(timezone.utc) + timedelta(days=7)
+        share = VideoShare.create(video_id="v", user_id="u", expires_at=future)
+        assert share.is_expired is False
+
+    def test_expired_when_past(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        # 直接设置过去的过期时间（create 方法会阻止过期时间在过去）
+        share.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
+        assert share.is_expired is True
+
+    def test_create_rejects_past_expiry(self):
+        """create 方法拒绝过去的过期时间。"""
+        past = datetime.now(timezone.utc) - timedelta(days=1)
+        with pytest.raises(ValueError, match="expires_at cannot be in the past"):
+            VideoShare.create(video_id="v", user_id="u", expires_at=past)
+
+
+class TestVideoShareAccessible:
+    """可访问性判断。"""
+
+    def test_active_no_expiry_is_accessible(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        assert share.is_accessible is True
+
+    def test_inactive_not_accessible(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        share.revoke()
+        assert share.is_accessible is False
+
+    def test_expired_not_accessible(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        # 直接设置过去的过期时间
+        share.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
+        assert share.is_accessible is False
+
+
+class TestVideoShareCounts:
+    """计数相关方法。"""
+
+    def test_initial_view_count_zero(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        assert share.view_count == 0
+
+    def test_increment_view_count(self):
+        share = VideoShare.create(video_id="v", user_id="u")
         share.increment_view_count()
         assert share.view_count == 1
         share.increment_view_count()
-        assert share.view_count == 2
+        share.increment_view_count()
+        assert share.view_count == 3
 
-    def test_increment_download_count(self) -> None:
-        share = _make_share()
+    def test_initial_download_count_zero(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        assert share.download_count == 0
+
+    def test_increment_download_count(self):
+        share = VideoShare.create(video_id="v", user_id="u")
         share.increment_download_count()
-        assert share.download_count == 1
+        share.increment_download_count()
+        assert share.download_count == 2
 
-    def test_revoke_sets_inactive(self) -> None:
-        share = _make_share()
+
+class TestVideoShareRevoke:
+    """撤销分享。"""
+
+    def test_revoke_deactivates(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        assert share.is_active is True
         share.revoke()
         assert share.is_active is False
-        assert share.is_accessible is False
 
-
-class TestCreateShareUseCase:
-    def test_create_success(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        video_repo.get.return_value = _make_video()
-        share_repo.create.side_effect = lambda s: s
-
-        use_case = CreateShareUseCase(share_repo, video_repo)
-        cmd = CreateShareCommand(video_id="vid_001", user_id="user_001")
-        result = use_case.execute(cmd)
-
-        assert result.video_id == "vid_001"
-        assert result.user_id == "user_001"
-        share_repo.create.assert_called_once()
-
-    def test_create_with_password(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        video_repo.get.return_value = _make_video()
-        share_repo.create.side_effect = lambda s: s
-
-        use_case = CreateShareUseCase(share_repo, video_repo)
-        cmd = CreateShareCommand(
-            video_id="vid_001",
-            user_id="user_001",
-            password="secret",
-        )
-        result = use_case.execute(cmd)
-
-        assert result.has_password is True
-
-    def test_create_with_expiry(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        video_repo.get.return_value = _make_video()
-        share_repo.create.side_effect = lambda s: s
-
-        use_case = CreateShareUseCase(share_repo, video_repo)
-        future = datetime.now(timezone.utc) + timedelta(days=1)
-        cmd = CreateShareCommand(
-            video_id="vid_001",
-            user_id="user_001",
-            expires_at=future,
-        )
-        result = use_case.execute(cmd)
-
-        assert result.expires_at == future
-
-    def test_video_not_found_raises(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        video_repo.get.return_value = None
-
-        use_case = CreateShareUseCase(share_repo, video_repo)
-        cmd = CreateShareCommand(video_id="nonexistent", user_id="user_001")
-
-        with pytest.raises(VideoNotFoundError):
-            use_case.execute(cmd)
-
-    def test_wrong_user_cannot_share(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        video_repo.get.return_value = _make_video(user_id="other_user")
-
-        use_case = CreateShareUseCase(share_repo, video_repo)
-        cmd = CreateShareCommand(video_id="vid_001", user_id="user_001")
-
-        with pytest.raises(VideoNotFoundError):
-            use_case.execute(cmd)
-
-
-class TestGetShareByTokenUseCase:
-    def test_found_active_share(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = _make_share()
-
-        use_case = GetShareByTokenUseCase(repo)
-        result = use_case.execute("abc123xyz")
-        assert result.share_token == "abc123xyz"
-
-    def test_not_found_raises(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = None
-
-        use_case = GetShareByTokenUseCase(repo)
-        with pytest.raises(NotFoundError):
-            use_case.execute("nonexistent")
-
-    def test_inactive_share_raises_expired(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = _make_share(is_active=False)
-
-        use_case = GetShareByTokenUseCase(repo)
-        with pytest.raises(ShareExpiredError):
-            use_case.execute("token")
-
-    def test_expired_share_raises(self) -> None:
-        repo = MagicMock()
-        past = datetime.now(timezone.utc) - timedelta(days=1)
-        repo.get_by_token.return_value = _make_share(expires_at=past)
-
-        use_case = GetShareByTokenUseCase(repo)
-        with pytest.raises(ShareExpiredError):
-            use_case.execute("token")
-
-
-class TestAccessShareUseCase:
-    def test_access_no_password(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = _make_share()
-        video_repo.get.return_value = _make_video()
-        share_repo.increment_view.return_value = None
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-        result = use_case.execute("abc123xyz")
-
-        assert isinstance(result, ShareAccessResult)
-        assert result.video.id == "vid_001"
-        assert result.password_verified is True
-        assert result.share.view_count == 1  # 浏览量+1
-        share_repo.increment_view.assert_called_once()
-
-    def test_access_with_correct_password(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = _make_share(password="mypass")
-        video_repo.get.return_value = _make_video()
-        share_repo.increment_view.return_value = None
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-        result = use_case.execute("token", password="mypass")
-
-        assert result.password_verified is True
-
-    def test_access_password_required_but_not_provided(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = _make_share(password="secret")
-        video_repo.get.return_value = _make_video()
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-
-        with pytest.raises(PasswordRequiredError):
-            use_case.execute("token", password=None)
-
-    def test_access_wrong_password(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = _make_share(password="correct")
-        video_repo.get.return_value = _make_video()
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-
-        with pytest.raises(InvalidPasswordError):
-            use_case.execute("token", password="wrong")
-
-    def test_access_share_not_found(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = None
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-
-        with pytest.raises(NotFoundError):
-            use_case.execute("nonexistent")
-
-    def test_access_share_expired(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        past = datetime.now(timezone.utc) - timedelta(days=1)
-        share_repo.get_by_token.return_value = _make_share(expires_at=past)
-        video_repo.get.return_value = _make_video()
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-
-        with pytest.raises(ShareExpiredError):
-            use_case.execute("token")
-
-    def test_access_video_not_found(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = _make_share()
-        video_repo.get.return_value = None
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-
-        with pytest.raises(VideoNotFoundError):
-            use_case.execute("token")
-
-
-class TestListSharesByVideoUseCase:
-    def test_lists_shares(self) -> None:
-        repo = MagicMock()
-        expected = [_make_share(), _make_share(share_id="share_002", token="tok2")]
-        repo.list_by_video.return_value = expected
-
-        use_case = ListSharesByVideoUseCase(repo)
-        result = use_case.execute("vid_001", "user_001")
-
-        assert len(result) == 2
-        repo.list_by_video.assert_called_once_with("vid_001", "user_001")
-
-
-class TestListSharesByUserUseCase:
-    def test_lists_with_total(self) -> None:
-        repo = MagicMock()
-        items = [_make_share(), _make_share(share_id="s2", token="t2")]
-        repo.list_by_user.return_value = items
-        repo.count_by_user.return_value = 10
-
-        use_case = ListSharesByUserUseCase(repo)
-        result_items, total = use_case.execute("user_001", skip=0, limit=2)
-
-        assert len(result_items) == 2
-        assert total == 10
-        repo.list_by_user.assert_called_once_with("user_001", skip=0, limit=2)
-
-
-class TestUpdateShareUseCase:
-    def test_update_password(self) -> None:
-        repo = MagicMock()
-        share = _make_share()
-        repo.get_by_id.return_value = share
-        repo.update.side_effect = lambda s: s
-
-        use_case = UpdateShareUseCase(repo)
-        cmd = UpdateShareCommand(
-            share_id="share_001",
-            user_id="user_001",
-            password="newpass",
-        )
-        result = use_case.execute(cmd)
-
-        assert result.has_password is True
-        assert result.verify_password("newpass") is True
-        repo.update.assert_called_once()
-
-    def test_clear_password(self) -> None:
-        repo = MagicMock()
-        share = _make_share(password="oldpass")
-        repo.get_by_id.return_value = share
-        repo.update.side_effect = lambda s: s
-
-        use_case = UpdateShareUseCase(repo)
-        cmd = UpdateShareCommand(
-            share_id="share_001",
-            user_id="user_001",
-            password="",  # 空字符串=清除密码
-        )
-        result = use_case.execute(cmd)
-
-        assert result.has_password is False
-        assert result.password_hash is None
-
-    def test_password_none_does_not_change(self) -> None:
-        repo = MagicMock()
-        share = _make_share(password="existing")
-        repo.get_by_id.return_value = share
-        repo.update.side_effect = lambda s: s
-
-        use_case = UpdateShareUseCase(repo)
-        cmd = UpdateShareCommand(
-            share_id="share_001",
-            user_id="user_001",
-            password=None,  # None=不修改
-        )
-        result = use_case.execute(cmd)
-
-        assert result.verify_password("existing") is True
-
-    def test_update_expires_at(self) -> None:
-        repo = MagicMock()
-        share = _make_share()
-        repo.get_by_id.return_value = share
-        repo.update.side_effect = lambda s: s
-
-        use_case = UpdateShareUseCase(repo)
-        future = datetime.now(timezone.utc) + timedelta(days=3)
-        cmd = UpdateShareCommand(
-            share_id="share_001",
-            user_id="user_001",
-            expires_at=future,
-        )
-        result = use_case.execute(cmd)
-
-        assert result.expires_at == future
-
-    def test_not_found_raises(self) -> None:
-        repo = MagicMock()
-        repo.get_by_id.return_value = None
-
-        use_case = UpdateShareUseCase(repo)
-        cmd = UpdateShareCommand(share_id="no", user_id="u1", password="x")
-
-        with pytest.raises(NotFoundError):
-            use_case.execute(cmd)
-
-    def test_past_expiry_raises(self) -> None:
-        repo = MagicMock()
-        repo.get_by_id.return_value = _make_share()
-
-        use_case = UpdateShareUseCase(repo)
-        past = datetime.now(timezone.utc) - timedelta(days=1)
-        cmd = UpdateShareCommand(
-            share_id="share_001",
-            user_id="user_001",
-            expires_at=past,
-        )
-
-        with pytest.raises(ValueError, match="past"):
-            use_case.execute(cmd)
-
-
-class TestRevokeShareUseCase:
-    def test_revoke_success(self) -> None:
-        repo = MagicMock()
-        repo.get_by_id.return_value = _make_share()
-        repo.delete.return_value = True
-
-        use_case = RevokeShareUseCase(repo)
-        result = use_case.execute("share_001", "user_001")
-
-        assert result is True
-        repo.delete.assert_called_once_with("share_001", "user_001")
-
-    def test_revoke_not_found_raises(self) -> None:
-        repo = MagicMock()
-        repo.get_by_id.return_value = None
-
-        use_case = RevokeShareUseCase(repo)
-
-        with pytest.raises(NotFoundError):
-            use_case.execute("nonexistent", "user_001")
-
-
-class TestRecordShareDownloadUseCase:
-    def test_record_success(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = _make_share()
-        repo.increment_download.return_value = None
-
-        use_case = RecordShareDownloadUseCase(repo)
-        use_case.execute("token")
-
-        repo.increment_download.assert_called_once()
-
-    def test_record_with_password(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = _make_share(password="pass")
-        repo.increment_download.return_value = None
-
-        use_case = RecordShareDownloadUseCase(repo)
-        use_case.execute("token", password="pass")
-
-        repo.increment_download.assert_called_once()
-
-    def test_record_wrong_password_raises(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = _make_share(password="correct")
-
-        use_case = RecordShareDownloadUseCase(repo)
-
-        with pytest.raises(InvalidPasswordError):
-            use_case.execute("token", password="wrong")
-
-    def test_record_share_not_found(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = None
-
-        use_case = RecordShareDownloadUseCase(repo)
-
-        with pytest.raises(NotFoundError):
-            use_case.execute("nonexistent")
-
-    def test_record_expired_share(self) -> None:
-        repo = MagicMock()
-        past = datetime.now(timezone.utc) - timedelta(days=1)
-        repo.get_by_token.return_value = _make_share(expires_at=past)
-
-        use_case = RecordShareDownloadUseCase(repo)
-
-        with pytest.raises(ShareExpiredError):
-            use_case.execute("token")
+    def test_revoke_idempotent(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        share.revoke()
+        share.revoke()  # 再次调用不报错
+        assert share.is_active is False

tests/unit/test_voice_extractor_pure.py

@@ -0,0 +1,259 @@
+"""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