test(unit): wave88 - 清理未使用导入和变量，修复code quality

packages/domain/video_concat.py

@@ -0,0 +1,175 @@
+"""视频拼接配置 — 纯数据结构与解析逻辑.
+
+从 concat_engine.py 抽离的纯逻辑，负责：
+- ConcatSegment: 单个拼接片段数据结构
+- ConcatConfig: 拼接配置数据结构
+- 从 dict 安全解析配置
+- 有效片段计算
+
+所有函数均为纯函数，不依赖 FFmpeg、数据库或外部 IO。
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+# ── 拼接片段 ────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class ConcatSegment:
+    """单个拼接片段."""
+
+    video_path: str  # 视频文件路径
+    start_time: float = 0.0  # 开始时间（秒）
+    duration: float = 0.0  # 持续时长（秒），0 表示取到末尾
+    has_audio: bool = True  # 是否包含音频
+
+    @classmethod
+    def from_dict(cls, seg: dict[str, Any]) -> "ConcatSegment":
+        """从字典创建拼接片段，带安全类型转换.
+
+        Args:
+            seg: 片段配置字典
+
+        Returns:
+            ConcatSegment 实例（字段解析失败时使用安全默认值）
+        """
+        try:
+            start_time = max(0.0, float(seg.get("start_time", 0.0)))
+        except (TypeError, ValueError):
+            start_time = 0.0
+
+        try:
+            duration = max(0.0, float(seg.get("duration", 0.0)))
+        except (TypeError, ValueError):
+            duration = 0.0
+
+        return cls(
+            video_path=str(seg.get("video_path", "")),
+            start_time=start_time,
+            duration=duration,
+            has_audio=bool(seg.get("has_audio", True)),
+        )
+
+    @property
+    def is_valid(self) -> bool:
+        """是否为有效片段（有视频路径）."""
+        return bool(self.video_path)
+
+
+# ── 拼接配置 ────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class ConcatConfig:
+    """视频拼接配置."""
+
+    segments: list[ConcatSegment] = field(default_factory=list)
+    output_width: int = 0  # 输出宽度（0=自动取第一段）
+    output_height: int = 0  # 输出高度（0=自动取第一段）
+    output_fps: float = 0.0  # 输出帧率（0=自动取第一段）
+    force_reencode: bool = False  # 强制重新编码
+    transition: str = "none"  # 转场效果（none/crossfade）
+    transition_duration: float = 0.3  # 转场时长（秒）
+
+    @classmethod
+    def from_config_dict(cls, config: dict[str, Any] | None) -> "ConcatConfig":
+        """从配置字典创建 ConcatConfig.
+
+        容错策略：字段解析失败时使用默认值，无效片段自动跳过。
+
+        Args:
+            config: 配置字典（可为 None）
+
+        Returns:
+            ConcatConfig 实例
+        """
+        if not config or not isinstance(config, dict):
+            return cls()
+
+        segments_raw = config.get("segments", [])
+        segments: list[ConcatSegment] = []
+
+        if isinstance(segments_raw, list):
+            for s in segments_raw:
+                if isinstance(s, dict) and s.get("video_path"):
+                    try:
+                        seg = ConcatSegment.from_dict(s)
+                        if seg.is_valid:
+                            segments.append(seg)
+                    except Exception:
+                        logger.warning("[concat] skip invalid segment: %s", s)
+                        continue
+
+        try:
+            output_width = max(0, int(config.get("output_width", 0)))
+        except (TypeError, ValueError):
+            output_width = 0
+
+        try:
+            output_height = max(0, int(config.get("output_height", 0)))
+        except (TypeError, ValueError):
+            output_height = 0
+
+        try:
+            output_fps = max(0.0, float(config.get("output_fps", 0.0)))
+        except (TypeError, ValueError):
+            output_fps = 0.0
+
+        try:
+            transition_duration = max(0.1, float(config.get("transition_duration", 0.3)))
+        except (TypeError, ValueError):
+            transition_duration = 0.3
+
+        return cls(
+            segments=segments,
+            output_width=output_width,
+            output_height=output_height,
+            output_fps=output_fps,
+            force_reencode=bool(config.get("force_reencode", False)),
+            transition=str(config.get("transition", "none")),
+            transition_duration=transition_duration,
+        )
+
+    @property
+    def valid_segments(self) -> list[ConcatSegment]:
+        """有效片段列表（有视频路径的）."""
+        return [s for s in self.segments if s.is_valid]
+
+    @property
+    def has_effect(self) -> bool:
+        """是否有足够的有效片段需要拼接（>=2个）."""
+        return len(self.valid_segments) >= 2
+
+    @property
+    def total_segments(self) -> int:
+        """有效片段数量."""
+        return len(self.valid_segments)
+
+    @property
+    def total_duration(self) -> float:
+        """估算总时长（各片段 duration 之和，duration=0 的不计）。
+
+        注意：这是粗略估算，实际时长需要 probe 后才能确定。
+        """
+        total = 0.0
+        for seg in self.valid_segments:
+            if seg.duration > 0:
+                total += seg.duration
+        return total
+
+    @property
+    def has_transition(self) -> bool:
+        """是否启用了转场效果."""
+        return self.transition != "none" and self.transition_duration > 0
+
+    @property
+    def auto_output_size(self) -> bool:
+        """是否自动判断输出分辨率（宽或高为 0）."""
+        return self.output_width == 0 or self.output_height == 0

tests/unit/test_asset_usage_deep.py

@@ -8,8 +8,6 @@ from __future__ import annotations
 
 from datetime import datetime, timezone
 
-import pytest
-
 from worker_app.core.asset_usage import mark_asset_used_for_generation
 
 from packages.domain import Asset, AssetStatus

tests/unit/test_title_usage_pure.py

@@ -6,9 +6,7 @@
 from __future__ import annotations
 
 from datetime import datetime, timezone
-from unittest.mock import MagicMock, patch
-
-import pytest
+from unittest.mock import MagicMock
 
 from packages.adapters.sqlalchemy_impl.models import TitleLibraryModel
 

tests/unit/test_tts_streaming_service.py

@@ -9,8 +9,7 @@ mock 掉 WebSocket 和 CosyVoiceService，验证核心逻辑：
 
 from __future__ import annotations
 
-import asyncio
-from unittest.mock import AsyncMock, MagicMock, call, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -362,7 +361,6 @@ class TestStreamLongText:
             await streaming_service._stream_long_text(mock_ws, {"text": text, "voice_id": "v1"})
 
         calls = mock_ws.send_json.call_args_list
-        last_msg = calls[-1][0][0]
         # 应该有错误
         error_msgs = [c for c in calls if c[0][0].get("type") == "error"]
         assert len(error_msgs) >= 1

tests/unit/test_video_concat.py

@@ -0,0 +1,340 @@
+"""video_concat 模块单元测试."""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.domain.video_concat import ConcatConfig, ConcatSegment
+
+
+# ── ConcatSegment ──────────────────────────────────────────────────────────
+
+
+class TestConcatSegment:
+    def test_default_values(self):
+        seg = ConcatSegment(video_path="/tmp/test.mp4")
+        assert seg.video_path == "/tmp/test.mp4"
+        assert seg.start_time == 0.0
+        assert seg.duration == 0.0
+        assert seg.has_audio is True
+
+    def test_full_values(self):
+        seg = ConcatSegment(
+            video_path="/tmp/test.mp4",
+            start_time=5.0,
+            duration=10.0,
+            has_audio=False,
+        )
+        assert seg.start_time == 5.0
+        assert seg.duration == 10.0
+        assert seg.has_audio is False
+
+    def test_is_valid_with_path(self):
+        seg = ConcatSegment(video_path="/tmp/test.mp4")
+        assert seg.is_valid is True
+
+    def test_is_valid_empty_path(self):
+        seg = ConcatSegment(video_path="")
+        assert seg.is_valid is False
+
+    def test_from_dict_basic(self):
+        seg = ConcatSegment.from_dict({"video_path": "/tmp/test.mp4"})
+        assert seg.video_path == "/tmp/test.mp4"
+        assert seg.start_time == 0.0
+        assert seg.duration == 0.0
+        assert seg.has_audio is True
+
+    def test_from_dict_with_all_fields(self):
+        seg = ConcatSegment.from_dict({
+            "video_path": "/tmp/test.mp4",
+            "start_time": 3.5,
+            "duration": 7.2,
+            "has_audio": False,
+        })
+        assert seg.video_path == "/tmp/test.mp4"
+        assert seg.start_time == 3.5
+        assert seg.duration == 7.2
+        assert seg.has_audio is False
+
+    def test_from_dict_negative_start_time_clamped(self):
+        seg = ConcatSegment.from_dict({"video_path": "x.mp4", "start_time": -5.0})
+        assert seg.start_time == 0.0
+
+    def test_from_dict_negative_duration_clamped(self):
+        seg = ConcatSegment.from_dict({"video_path": "x.mp4", "duration": -3.0})
+        assert seg.duration == 0.0
+
+    def test_from_dict_invalid_start_time_falls_back(self):
+        seg = ConcatSegment.from_dict({"video_path": "x.mp4", "start_time": "abc"})
+        assert seg.start_time == 0.0
+
+    def test_from_dict_invalid_duration_falls_back(self):
+        seg = ConcatSegment.from_dict({"video_path": "x.mp4", "duration": "abc"})
+        assert seg.duration == 0.0
+
+    def test_from_dict_empty_dict(self):
+        seg = ConcatSegment.from_dict({})
+        assert seg.video_path == ""
+        assert seg.start_time == 0.0
+        assert seg.duration == 0.0
+        assert seg.has_audio is True
+
+    def test_from_dict_path_converted_to_string(self):
+        seg = ConcatSegment.from_dict({"video_path": 123})
+        assert seg.video_path == "123"
+
+    def test_from_dict_has_audio_falsy_values(self):
+        seg = ConcatSegment.from_dict({"video_path": "x.mp4", "has_audio": False})
+        assert seg.has_audio is False
+        seg2 = ConcatSegment.from_dict({"video_path": "x.mp4", "has_audio": 0})
+        assert seg2.has_audio is False
+        seg3 = ConcatSegment.from_dict({"video_path": "x.mp4", "has_audio": ""})
+        assert seg3.has_audio is False
+
+
+# ── ConcatConfig ────────────────────────────────────────────────────────────
+
+
+class TestConcatConfigDefaults:
+    def test_default_values(self):
+        cfg = ConcatConfig()
+        assert cfg.segments == []
+        assert cfg.output_width == 0
+        assert cfg.output_height == 0
+        assert cfg.output_fps == 0.0
+        assert cfg.force_reencode is False
+        assert cfg.transition == "none"
+        assert cfg.transition_duration == 0.3
+
+    def test_has_effect_no_segments(self):
+        cfg = ConcatConfig()
+        assert cfg.has_effect is False
+
+    def test_total_segments_no_segments(self):
+        cfg = ConcatConfig()
+        assert cfg.total_segments == 0
+
+    def test_valid_segments_empty(self):
+        cfg = ConcatConfig()
+        assert cfg.valid_segments == []
+
+    def test_total_duration_empty(self):
+        cfg = ConcatConfig()
+        assert cfg.total_duration == 0.0
+
+    def test_has_transition_default(self):
+        cfg = ConcatConfig()
+        assert cfg.has_transition is False
+
+    def test_auto_output_size_default(self):
+        cfg = ConcatConfig()
+        assert cfg.auto_output_size is True
+
+
+class TestConcatConfigWithSegments:
+    def test_single_segment_no_effect(self):
+        cfg = ConcatConfig(segments=[ConcatSegment(video_path="a.mp4")])
+        assert cfg.has_effect is False
+        assert cfg.total_segments == 1
+
+    def test_two_segments_has_effect(self):
+        cfg = ConcatConfig(segments=[
+            ConcatSegment(video_path="a.mp4"),
+            ConcatSegment(video_path="b.mp4"),
+        ])
+        assert cfg.has_effect is True
+        assert cfg.total_segments == 2
+
+    def test_invalid_segments_filtered(self):
+        cfg = ConcatConfig(segments=[
+            ConcatSegment(video_path="a.mp4"),
+            ConcatSegment(video_path=""),
+            ConcatSegment(video_path="b.mp4"),
+            ConcatSegment(video_path=""),
+        ])
+        assert cfg.total_segments == 2
+        assert len(cfg.valid_segments) == 2
+        assert cfg.valid_segments[0].video_path == "a.mp4"
+        assert cfg.valid_segments[1].video_path == "b.mp4"
+
+    def test_total_duration_with_durations(self):
+        cfg = ConcatConfig(segments=[
+            ConcatSegment(video_path="a.mp4", duration=5.0),
+            ConcatSegment(video_path="b.mp4", duration=3.0),
+            ConcatSegment(video_path="c.mp4", duration=0),  # 不计
+        ])
+        assert cfg.total_duration == pytest.approx(8.0)
+
+    def test_has_transition_enabled(self):
+        cfg = ConcatConfig(transition="crossfade", transition_duration=0.5)
+        assert cfg.has_transition is True
+
+    def test_has_transition_none_transition(self):
+        cfg = ConcatConfig(transition="none", transition_duration=0.5)
+        assert cfg.has_transition is False
+
+    def test_has_transition_zero_duration(self):
+        cfg = ConcatConfig(transition="crossfade", transition_duration=0.0)
+        assert cfg.has_transition is False
+
+    def test_auto_output_size_with_width_only(self):
+        cfg = ConcatConfig(output_width=1920)
+        assert cfg.auto_output_size is True
+
+    def test_auto_output_size_with_height_only(self):
+        cfg = ConcatConfig(output_height=1080)
+        assert cfg.auto_output_size is True
+
+    def test_auto_output_size_false_when_both_set(self):
+        cfg = ConcatConfig(output_width=1920, output_height=1080)
+        assert cfg.auto_output_size is False
+
+
+class TestConcatConfigFromDict:
+    def test_none_config(self):
+        cfg = ConcatConfig.from_config_dict(None)
+        assert cfg.segments == []
+        assert cfg.output_width == 0
+
+    def test_empty_dict(self):
+        cfg = ConcatConfig.from_config_dict({})
+        assert cfg.segments == []
+        assert cfg.output_width == 0
+
+    def test_not_dict_returns_default(self):
+        cfg = ConcatConfig.from_config_dict("not a dict")
+        assert cfg.segments == []
+
+    def test_with_segments(self):
+        cfg = ConcatConfig.from_config_dict({
+            "segments": [
+                {"video_path": "a.mp4", "duration": 5.0},
+                {"video_path": "b.mp4", "duration": 3.0},
+            ]
+        })
+        assert cfg.total_segments == 2
+        assert cfg.segments[0].duration == 5.0
+        assert cfg.segments[1].duration == 3.0
+
+    def test_segments_not_list_ignored(self):
+        cfg = ConcatConfig.from_config_dict({"segments": "not a list"})
+        assert cfg.segments == []
+
+    def test_invalid_segment_skipped(self):
+        cfg = ConcatConfig.from_config_dict({
+            "segments": [
+                {"video_path": "a.mp4"},
+                "not a dict",
+                {"video_path": ""},  # 空路径
+                {"video_path": "b.mp4"},
+            ]
+        })
+        assert cfg.total_segments == 2
+
+    def test_output_dimensions(self):
+        cfg = ConcatConfig.from_config_dict({
+            "output_width": 1920,
+            "output_height": 1080,
+            "output_fps": 30.0,
+        })
+        assert cfg.output_width == 1920
+        assert cfg.output_height == 1080
+        assert cfg.output_fps == pytest.approx(30.0)
+
+    def test_negative_dimensions_clamped(self):
+        cfg = ConcatConfig.from_config_dict({
+            "output_width": -100,
+            "output_height": -50,
+            "output_fps": -10.0,
+        })
+        assert cfg.output_width == 0
+        assert cfg.output_height == 0
+        assert cfg.output_fps == 0.0
+
+    def test_invalid_dimensions_fall_back(self):
+        cfg = ConcatConfig.from_config_dict({
+            "output_width": "abc",
+            "output_height": "xyz",
+            "output_fps": "not_a_number",
+        })
+        assert cfg.output_width == 0
+        assert cfg.output_height == 0
+        assert cfg.output_fps == 0.0
+
+    def test_force_reencode(self):
+        cfg = ConcatConfig.from_config_dict({"force_reencode": True})
+        assert cfg.force_reencode is True
+
+    def test_transition_settings(self):
+        cfg = ConcatConfig.from_config_dict({
+            "transition": "crossfade",
+            "transition_duration": 0.8,
+        })
+        assert cfg.transition == "crossfade"
+        assert cfg.transition_duration == pytest.approx(0.8)
+
+    def test_transition_duration_clamped_minimum(self):
+        cfg = ConcatConfig.from_config_dict({"transition_duration": 0.0})
+        assert cfg.transition_duration == pytest.approx(0.1)
+
+    def test_transition_duration_invalid_falls_back(self):
+        cfg = ConcatConfig.from_config_dict({"transition_duration": "bad"})
+        assert cfg.transition_duration == pytest.approx(0.3)
+
+    def test_transition_converted_to_string(self):
+        cfg = ConcatConfig.from_config_dict({"transition": 123})
+        assert cfg.transition == "123"
+
+    def test_full_config(self):
+        cfg = ConcatConfig.from_config_dict({
+            "segments": [
+                {"video_path": "a.mp4", "duration": 3.0},
+                {"video_path": "b.mp4", "duration": 4.0},
+                {"video_path": "c.mp4", "duration": 5.0},
+            ],
+            "output_width": 1280,
+            "output_height": 720,
+            "output_fps": 25.0,
+            "force_reencode": True,
+            "transition": "crossfade",
+            "transition_duration": 0.4,
+        })
+        assert cfg.total_segments == 3
+        assert cfg.total_duration == pytest.approx(12.0)
+        assert cfg.output_width == 1280
+        assert cfg.output_height == 720
+        assert cfg.output_fps == 25.0
+        assert cfg.force_reencode is True
+        assert cfg.has_transition is True
+        assert cfg.auto_output_size is False
+        assert cfg.has_effect is True
+
+
+class TestConcatConfigEdgeCases:
+    def test_segment_with_start_time_only(self):
+        cfg = ConcatConfig(segments=[
+            ConcatSegment(video_path="a.mp4", start_time=2.0),
+        ])
+        assert cfg.total_segments == 1
+        assert cfg.segments[0].start_time == 2.0
+
+    def test_mixed_valid_invalid_long_list(self):
+        segments = []
+        for i in range(10):
+            if i % 2 == 0:
+                segments.append(ConcatSegment(video_path=f"v{i}.mp4"))
+            else:
+                segments.append(ConcatSegment(video_path=""))
+        cfg = ConcatConfig(segments=segments)
+        assert cfg.total_segments == 5
+        assert cfg.valid_segments[0].video_path == "v0.mp4"
+        assert cfg.valid_segments[-1].video_path == "v8.mp4"
+
+    def test_total_duration_mixed_zero_and_nonzero(self):
+        cfg = ConcatConfig(segments=[
+            ConcatSegment(video_path="a.mp4", duration=0),
+            ConcatSegment(video_path="b.mp4", duration=5.0),
+            ConcatSegment(video_path="c.mp4", duration=0),
+            ConcatSegment(video_path="d.mp4", duration=3.0),
+        ])
+        assert cfg.total_duration == pytest.approx(8.0)