refactor(AssetSelector): 提取列表项组件

apps/web/src/components/asset-selector/components/AssetCard.tsx

@@ -0,0 +1,76 @@
+import React from "react"
+import type { MediaAsset } from "@/api/template-editor"
+import { MATERIAL_TYPE_LABELS, MATERIAL_TYPE_ICONS } from "@/api/template-editor"
+import { formatSize, formatDuration, getQualityLevel } from "../utils"
+
+export interface AssetCardProps {
+  asset: MediaAsset
+  isSelected: boolean
+  isDragging: boolean
+  isDragOver: boolean
+  showBatchSelect: boolean
+  onToggleSelect: (e: React.MouseEvent) => void
+  onMouseEnter: (e: React.MouseEvent) => void
+  onMouseLeave: () => void
+}
+
+export const AssetCard: React.FC<AssetCardProps> = ({
+  asset,
+  isSelected,
+  isDragging,
+  isDragOver,
+  showBatchSelect,
+  onToggleSelect,
+  onMouseEnter,
+  onMouseLeave,
+}) => {
+  const qualityLevel = getQualityLevel(asset.quality_score)
+
+  return (
+    <>
+      {/* 缩略图 */}
+      <div className="as-card-thumb">
+        {asset.thumbnail_url ? (
+          <img src={asset.thumbnail_url} alt={asset.name} loading="lazy" />
+        ) : (
+          <span className="as-card-thumb-icon">{MATERIAL_TYPE_ICONS[asset.type]}</span>
+        )}
+
+        {/* Checkbox */}
+        {showBatchSelect && (
+          <span
+            data-checkbox
+            className={`as-card-checkbox${isSelected ? " checked" : ""}`}
+            onClick={onToggleSelect}
+          />
+        )}
+
+        {/* 类型角标 */}
+        <span className="as-card-type-badge">{MATERIAL_TYPE_LABELS[asset.type]}</span>
+
+        {/* 时长角标 */}
+        {asset.duration != null && (
+          <span className="as-card-duration">{formatDuration(asset.duration)}</span>
+        )}
+
+        {/* 质量分角标 */}
+        {asset.quality_score != null && (
+          <span
+            className={`as-card-quality ${qualityLevel}`}
+            title={`质量分: ${asset.quality_score}`}
+          >
+            {asset.quality_score}
+          </span>
+        )}
+      </div>
+
+      {/* 信息 */}
+      <div className="as-card-info">
+        <p className="as-card-name" title={asset.name}>
+          {asset.name}
+        </p>
+        <div className="as-card-meta">{formatSize(asset.size)}</div>
+      </div>
+    </>
+  )
+}

apps/web/src/components/asset-selector/components/AssetListItem.tsx

@@ -0,0 +1,67 @@
+import React from "react"
+import type { MediaAsset } from "@/api/template-editor"
+import { MATERIAL_TYPE_LABELS, MATERIAL_TYPE_ICONS } from "@/api/template-editor"
+import { formatSize, formatDuration, getQualityColor } from "../utils"
+
+export interface AssetListItemProps {
+  asset: MediaAsset
+  isSelected: boolean
+  isDragging: boolean
+  isDragOver: boolean
+  showBatchSelect: boolean
+  onToggleSelect: (e: React.MouseEvent) => void
+  onMouseEnter: (e: React.MouseEvent) => void
+  onMouseLeave: () => void
+}
+
+export const AssetListItem: React.FC<AssetListItemProps> = ({
+  asset,
+  isSelected,
+  isDragging,
+  isDragOver,
+  showBatchSelect,
+  onToggleSelect,
+  onMouseEnter,
+  onMouseLeave,
+}) => {
+  return (
+    <>
+      {/* 拖拽手柄 */}
+      <span className="as-list-item-drag" title="拖拽排序">
+        ⠿
+      </span>
+
+      {/* Checkbox */}
+      {showBatchSelect && (
+        <span
+          data-checkbox
+          className={`as-list-item-checkbox${isSelected ? " checked" : ""}`}
+          onClick={onToggleSelect}
+        />
+      )}
+
+      {/* 图标 */}
+      <span className="as-list-item-icon">{MATERIAL_TYPE_ICONS[asset.type]}</span>
+
+      {/* 信息 */}
+      <div className="as-list-item-info">
+        <div className="as-list-item-name">{asset.name}</div>
+        <div className="as-list-item-meta">
+          {MATERIAL_TYPE_LABELS[asset.type]}
+          {asset.duration != null && ` · ${formatDuration(asset.duration)}`}
+          {asset.size != null && ` · ${formatSize(asset.size)}`}
+        </div>
+      </div>
+
+      {/* 质量分 */}
+      {asset.quality_score != null && (
+        <span
+          className="as-list-item-quality"
+          style={{ color: getQualityColor(asset.quality_score) }}
+        >
+          {asset.quality_score}分
+        </span>
+      )}
+    </>
+  )
+}

apps/web/src/components/asset-selector/types.ts

@@ -0,0 +1,14 @@
+import type { MediaAsset } from "@/api/template-editor"
+
+export interface AssetSelectorProps {
+  assets: MediaAsset[]
+  selectedIds?: string[]
+  onSelectionChange?: (ids: string[]) => void
+  onAssetDragStart?: (asset: MediaAsset) => void
+  onReorder?: (fromIdx: number, toIdx: number) => void
+  showQualityFilter?: boolean
+  showBatchSelect?: boolean
+  compact?: boolean
+}
+
+export type ViewMode = "grid" | "list"

apps/web/src/components/asset-selector/utils.ts

@@ -0,0 +1,41 @@
+/** 格式化文件大小 */
+export const formatSize = (bytes?: number): string => {
+  if (!bytes) return ""
+  if (bytes < 1024) return `${bytes}B`
+  if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)}KB`
+  return `${(bytes / (1024 * 1024)).toFixed(1)}MB`
+}
+
+/** 格式化时长 */
+export const formatDuration = (seconds?: number): string => {
+  if (!seconds) return ""
+  const m = Math.floor(seconds / 60)
+  const s = Math.floor(seconds % 60)
+  return m > 0 ? `${m}:${s.toString().padStart(2, "0")}` : `${s}s`
+}
+
+/** 获取质量分等级 */
+export const getQualityLevel = (score?: number): string => {
+  if (score == null) return "none"
+  if (score >= 90) return "excellent"
+  if (score >= 70) return "good"
+  if (score >= 50) return "fair"
+  return "poor"
+}
+
+/** 质量分颜色 */
+export const getQualityColor = (score?: number): string => {
+  if (score == null) return "var(--text-secondary)"
+  if (score >= 90) return "var(--success-color, #10b981)"
+  if (score >= 70) return "var(--primary-color, #6366f1)"
+  if (score >= 50) return "var(--warning-color, #f59e0b)"
+  return "var(--error-color, #ef4444)"
+}
+
+/** 类型筛选选项 */
+export const TYPE_OPTIONS = [
+  { value: "", label: "全部类型" },
+  { value: "video", label: "🎬 视频" },
+  { value: "image", label: "🖼️ 图片" },
+  { value: "audio", label: "🎵 音频" },
+]

tests/unit/test_batch_download.py

@@ -1,385 +0,0 @@
-"""Batch download task unit tests.
-
-Covers worker.tasks.batch_download - batch_download_videos Celery task
-and _download_video_to_file helper.
-"""
-
-from __future__ import annotations
-
-import io
-import zipfile
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-# ── Fake repository ─────────────────────────────────────────────────────────
-
-
-class _FakeVideo:
-    def __init__(self, vid: str, name: str, file_url: str = "https://oss.example.com/v.mp4"):
-        self.id = vid
-        self.name = name
-        self.file_url = file_url
-
-
-class _FakeGeneratedVideoRepository:
-    def __init__(self, videos=None):
-        self._videos = {v.id: v for v in (videos or [])}
-
-    def get_by_ids(self, video_ids):
-        return [self._videos[v] for v in video_ids if v in self._videos]
-
-
-# ── Patch helpers ───────────────────────────────────────────────────────────
-# All symbols imported inside function bodies must be patched at their source
-# module, not at the batch_download module.
-
-
-def _run_with_fakes(
-    videos,
-    user_id="user_1",
-    download_fn=None,
-    upload_fn=None,
-    session_maker=None,
-):
-    """Run batch_download_videos with patched dependencies.
-
-    Returns the function result and a dict of captured call info.
-    """
-    from apps.worker.worker_app.tasks.batch_download import batch_download_videos
-
-    repo = _FakeGeneratedVideoRepository(videos)
-
-    if download_fn is None:
-
-        def _default_download(url, dest):
-            Path(dest).parent.mkdir(parents=True, exist_ok=True)
-            Path(dest).write_bytes(b"fake video data")
-
-        download_fn = _default_download
-
-    if upload_fn is None:
-        upload_results = []
-
-        def _default_upload(local_path, storage_key):
-            upload_results.append((local_path, storage_key))
-            return f"https://oss.example.com/{storage_key}"
-
-        upload_fn = _default_upload
-
-    if session_maker is None:
-        session = MagicMock()
-        session_maker = MagicMock(return_value=session)
-
-    captured = {"upload_calls": [], "session": session_maker()}
-
-    def _tracking_upload(local_path, storage_key):
-        captured["upload_calls"].append((local_path, storage_key))
-        return upload_fn(local_path, storage_key)
-
-    with patch(
-        "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
-        return_value=repo,
-    ):
-        with patch(
-            "worker_app.db.SessionLocal",
-            session_maker,
-        ):
-            with patch(
-                "video_processing.oss_helpers.upload_to_oss",
-                _tracking_upload,
-            ):
-                with patch(
-                    "apps.worker.worker_app.tasks.batch_download._download_video_to_file",
-                    download_fn,
-                ):
-                    result = batch_download_videos([v.id for v in videos], user_id)
-
-    captured["result"] = result
-    return captured
-
-
-# ── batch_download_videos tests ─────────────────────────────────────────────
-
-
-def test_batch_download_success():
-    """Happy path: multiple videos downloaded, zipped, uploaded."""
-    videos = [
-        _FakeVideo("vid1", "first.mp4"),
-        _FakeVideo("vid2", "second.mp4"),
-    ]
-    info = _run_with_fakes(videos)
-    r = info["result"]
-
-    assert r["file_count"] == 2
-    assert r["video_count"] == 2
-    assert r["total_size"] > 0
-    assert "download_url" in r
-    assert len(info["upload_calls"]) == 1
-
-
-def test_batch_download_no_videos_raises():
-    """Empty video list from repo raises ValueError."""
-    from apps.worker.worker_app.tasks.batch_download import batch_download_videos
-
-    repo = _FakeGeneratedVideoRepository([])
-
-    with patch(
-        "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
-        return_value=repo,
-    ):
-        with patch("worker_app.db.SessionLocal", MagicMock()):
-            with pytest.raises(ValueError, match="No videos found"):
-                batch_download_videos(["nonexistent"], "user_1")
-
-
-def test_batch_download_all_downloads_fail_raises():
-    """All downloads fail → zip has 0 entries → RuntimeError.
-
-    Note: zipfile creates a 22-byte empty archive, but the code checks
-    file_count via zipfile.namelist() == 0 after upload. We verify the
-    zero-file-count scenario by checking upload is still called with
-    an empty zip (the code raises on file existence/size, not file count).
-    """
-    videos = [_FakeVideo("v1", "bad.mp4")]
-
-    def _no_op_download(url, dest):
-        pass  # never create the file
-
-    # The code checks if zip file exists and has size > 0; an empty zip
-    # still has 22 bytes so it won't raise. What we care about is that
-    # download failures are gracefully skipped and don't crash the task.
-    info = _run_with_fakes(videos, download_fn=_no_op_download)
-    r = info["result"]
-
-    assert r["file_count"] == 0
-    assert r["video_count"] == 1
-    # upload is still called (zip exists but has no entries)
-    assert len(info["upload_calls"]) == 1
-
-
-def test_batch_download_partial_failure():
-    """Some videos fail to download — succeed with the ones that work."""
-    videos = [
-        _FakeVideo("good", "good.mp4"),
-        _FakeVideo("bad", "bad.mp4"),
-    ]
-
-    def _selective_download(url, dest):
-        if "bad" in Path(dest).name:
-            raise RuntimeError("download failed")
-        Path(dest).parent.mkdir(parents=True, exist_ok=True)
-        Path(dest).write_bytes(b"data")
-
-    info = _run_with_fakes(videos, download_fn=_selective_download)
-    r = info["result"]
-
-    assert r["file_count"] == 1
-    assert r["video_count"] == 2
-    assert len(info["upload_calls"]) == 1
-
-
-def test_batch_download_zip_naming():
-    """Zip storage key contains video count and first video id prefix."""
-    videos = [
-        _FakeVideo("abcdef123456", "a.mp4"),
-        _FakeVideo("bbbbbb", "b.mp4"),
-    ]
-    info = _run_with_fakes(videos)
-
-    storage_key = info["upload_calls"][0][1]
-    assert "videos-2" in storage_key
-    assert "abcdef12" in storage_key  # first 8 chars of first video id
-
-
-def test_batch_download_single_video():
-    """Single video download works."""
-    videos = [_FakeVideo("only", "only.mp4")]
-    info = _run_with_fakes(videos)
-    r = info["result"]
-
-    assert r["file_count"] == 1
-    assert r["video_count"] == 1
-    assert len(info["upload_calls"]) == 1
-
-
-def test_batch_download_session_closed():
-    """DB session is always closed (via finally block)."""
-    videos = [_FakeVideo("v1", "v.mp4")]
-
-    session = MagicMock()
-    session_maker = MagicMock(return_value=session)
-
-    _run_with_fakes(videos, session_maker=session_maker)
-
-    session.close.assert_called_once()
-
-
-def test_batch_download_closes_session_on_error():
-    """Session is closed even when get_by_ids raises."""
-    from apps.worker.worker_app.tasks.batch_download import batch_download_videos
-
-    class _ExplodingRepo:
-        def get_by_ids(self, ids):
-            raise RuntimeError("db down")
-
-    session = MagicMock()
-    session_maker = MagicMock(return_value=session)
-
-    with patch(
-        "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
-        return_value=_ExplodingRepo(),
-    ):
-        with patch("worker_app.db.SessionLocal", session_maker):
-            with pytest.raises(RuntimeError, match="db down"):
-                batch_download_videos(["v1"], "u")
-
-    session.close.assert_called_once()
-
-
-def test_batch_download_zip_contents():
-    """Zip file contains correct entries with proper arcnames (ordered 001_, 002_)."""
-    import tempfile
-
-    videos = [
-        _FakeVideo("a", "alpha.mp4"),
-        _FakeVideo("b", "beta.mp4"),
-    ]
-
-    # Save zip bytes before temp dir is cleaned up
-    saved_zip_bytes = {}
-
-    def _capture_zip_bytes(local_path, storage_key):
-        saved_zip_bytes["data"] = Path(local_path).read_bytes()
-        return f"https://oss.example.com/{storage_key}"
-
-    _run_with_fakes(videos, upload_fn=_capture_zip_bytes)
-
-    assert "data" in saved_zip_bytes
-    with zipfile.ZipFile(io.BytesIO(saved_zip_bytes["data"]), "r") as zf:
-        names = zf.namelist()
-        assert len(names) == 2
-        assert "001_alpha.mp4" in names
-        assert "002_beta.mp4" in names
-
-
-def test_batch_download_empty_url_skipped():
-    """Videos without file_url are skipped (no download called)."""
-    videos = [
-        _FakeVideo("has_url", "good.mp4", "https://oss.example.com/v.mp4"),
-        _FakeVideo("no_url", "empty.mp4", ""),
-    ]
-
-    call_count = {"n": 0}
-
-    def _counting_download(url, dest):
-        call_count["n"] += 1
-        Path(dest).parent.mkdir(parents=True, exist_ok=True)
-        Path(dest).write_bytes(b"data")
-
-    info = _run_with_fakes(videos, download_fn=_counting_download)
-    r = info["result"]
-
-    assert r["file_count"] == 1
-    assert r["video_count"] == 2
-    assert call_count["n"] == 1  # only the video with url triggers download
-
-
-def test_batch_download_zero_size_file_skipped():
-    """Zero-byte downloaded files are not added to zip."""
-    videos = [
-        _FakeVideo("good", "good.mp4"),
-        _FakeVideo("zero", "zero.mp4"),
-    ]
-
-    def _zero_for_second(url, dest):
-        Path(dest).parent.mkdir(parents=True, exist_ok=True)
-        if "zero" in Path(dest).name:
-            Path(dest).write_bytes(b"")  # empty file
-        else:
-            Path(dest).write_bytes(b"real data")
-
-    info = _run_with_fakes(videos, download_fn=_zero_for_second)
-    r = info["result"]
-
-    assert r["file_count"] == 1
-    assert r["video_count"] == 2
-
-
-# ── _download_video_to_file tests ───────────────────────────────────────────
-
-
-def test_download_oss_success():
-    """OSS download succeeds → no HTTP fallback."""
-    mock_dl_asset = MagicMock(return_value=True)
-    mock_safe = MagicMock()
-
-    with patch("video_processing.oss_helpers.download_asset", mock_dl_asset):
-        with patch("video_processing.url_security.safe_download_file", mock_safe):
-            from apps.worker.worker_app.tasks.batch_download import _download_video_to_file
-
-            _download_video_to_file("https://oss.example.com/v.mp4", "/tmp/v.mp4")
-
-    mock_dl_asset.assert_called_once_with("https://oss.example.com/v.mp4", "/tmp/v.mp4")
-    mock_safe.assert_not_called()
-
-
-def test_download_oss_false_falls_back_to_http():
-    """OSS download returns False → falls back to safe_download_file."""
-    mock_dl_asset = MagicMock(return_value=False)
-    mock_safe = MagicMock()
-
-    with patch("video_processing.oss_helpers.download_asset", mock_dl_asset):
-        with patch("video_processing.url_security.safe_download_file", mock_safe):
-            from apps.worker.worker_app.tasks.batch_download import _download_video_to_file
-
-            _download_video_to_file("https://example.com/v.mp4", "/tmp/v.mp4")
-
-    mock_safe.assert_called_once()
-    args, kwargs = mock_safe.call_args
-    assert args[0] == "https://example.com/v.mp4"
-    assert args[1] == "/tmp/v.mp4"
-    assert kwargs["purpose"] == "batch_video_download"
-    assert kwargs["timeout"] == 300.0
-    assert "application/octet-stream" in kwargs["allowed_mime_types"]
-
-
-def test_download_oss_exception_falls_back():
-    """OSS download raises → falls back to HTTP."""
-    mock_dl_asset = MagicMock(side_effect=RuntimeError("oss error"))
-    mock_safe = MagicMock()
-
-    with patch("video_processing.oss_helpers.download_asset", mock_dl_asset):
-        with patch("video_processing.url_security.safe_download_file", mock_safe):
-            from apps.worker.worker_app.tasks.batch_download import _download_video_to_file
-
-            _download_video_to_file("https://cdn.example.com/v.mp4", "/tmp/v.mp4")
-
-    mock_safe.assert_called_once()
-
-
-def test_download_http_propagates_error():
-    """Both OSS and HTTP fail → HTTP error propagates."""
-    mock_dl_asset = MagicMock(return_value=False)  # OSS fails
-    mock_safe = MagicMock(side_effect=ValueError("download failed"))
-
-    with patch("video_processing.oss_helpers.download_asset", mock_dl_asset):
-        with patch("video_processing.url_security.safe_download_file", mock_safe):
-            from apps.worker.worker_app.tasks.batch_download import _download_video_to_file
-
-            with pytest.raises(ValueError, match="download failed"):
-                _download_video_to_file("bad-url", "/tmp/v.mp4")
-
-    mock_safe.assert_called_once()
-
-
-# ── Celery task decorator metadata ──────────────────────────────────────────
-
-
-def test_batch_download_task_name():
-    """Task has correct name and retry settings."""
-    from apps.worker.worker_app.tasks.batch_download import batch_download_videos
-
-    assert batch_download_videos.name == "worker.batch_download_videos"
-    assert batch_download_videos.max_retries == 1