feat: 素材智能视图筛选 + 标题使用次数闭环

1. 素材智能视图筛选（smart_view 参数）
   - 新增 list_assets 接口 smart_view 查询参数
   - 支持 6 种视图：recommended / cautious / risky / unused / used / pending_review
   - 基于 quality_score + metadata.review_status + metadata.generation_use_count 筛选
   - 与现有内存过滤路径集成，无破坏性变更

2. 标题使用次数闭环
   - 标题库 repository 新增 increment_usage_count 方法
   - 新增 IncrementTitleUsageUseCase
   - 生成任务成功后自动遍历 title_ids，逐个递增 usage_count
   - 异常兜底：计数失败不影响主流程，仅打 warning 日志

apps/api/app/api/routes/assets.py

@@ -82,6 +82,11 @@ def list_assets(
     gender: Optional[str] = Query(None, description="按 metadata.gender 筛选"),
     style: Optional[str] = Query(None, description="按 metadata.style 筛选"),
     tag_ids: Optional[str] = Query(None, description="按标签 ID 筛选（逗号分隔，取交集）"),
+    smart_view: Optional[str] = Query(
+        None,
+        description="智能视图筛选：recommended=推荐(质量分≥80)、cautious=慎用(60-79)、risky=高风险(<60或已驳回)、unused=未使用、used=已使用、pending_review=待复核",
+        pattern="^(recommended|cautious|risky|unused|used|pending_review)$",
+    ),
     skip: int = Query(0, ge=0),
     limit: int = Query(100, ge=1, le=500),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
@@ -101,11 +106,11 @@ def list_assets(
         if not filter_tag_ids:
             filter_tag_ids = None
 
-    # 需要内存过滤的标志（keyword/gender/style/tag_ids 无法在 DB 层过滤）
-    needs_memory_filter = bool(keyword or gender or style or filter_tag_ids)
+    # 需要内存过滤的标志（keyword/gender/style/tag_ids/smart_view 无法在 DB 层过滤）
+    needs_memory_filter = bool(keyword or gender or style or filter_tag_ids or smart_view)
 
     def _apply_memory_filters(items):
-        """应用 keyword / gender / style / tag_ids 内存过滤。"""
+        """应用 keyword / gender / style / tag_ids / smart_view 内存过滤。"""
         result = items
         if keyword:
             kw = keyword.lower()
@@ -117,6 +122,26 @@ def list_assets(
         if filter_tag_ids:
             tag_set = set(filter_tag_ids)
             result = [i for i in result if tag_set.issubset(set(getattr(i, "tag_ids", [])))]
+        if smart_view:
+            meta = lambda a: a.metadata or {}
+            use_count = lambda a: int(meta(a).get("generation_use_count") or 0)
+            review_status = lambda a: meta(a).get("review_status", "")
+            if smart_view == "recommended":
+                result = [i for i in result if i.quality_score is not None and i.quality_score >= 80]
+            elif smart_view == "cautious":
+                result = [i for i in result if i.quality_score is not None and 60 <= i.quality_score < 80]
+            elif smart_view == "risky":
+                result = [
+                    i for i in result
+                    if (i.quality_score is not None and i.quality_score < 60)
+                    or review_status(i) == "rejected"
+                ]
+            elif smart_view == "unused":
+                result = [i for i in result if use_count(i) == 0]
+            elif smart_view == "used":
+                result = [i for i in result if use_count(i) > 0]
+            elif smart_view == "pending_review":
+                result = [i for i in result if review_status(i) == "pending_review"]
         return result
 
     # ── 优化路径：无内存过滤时，使用 DB 级分页 ──

apps/worker/worker_app/tasks/generation.py

@@ -1092,6 +1092,35 @@ def generate_video(self, task_id: str) -> dict:
             # ── 5. 标记完成 ──────────────────────────────────────────────────
             _update_task_status(task_id, "mark_completed", result_count=video_count)
 
+            # 5.1 更新标题使用次数
+            try:
+                _title_session = SessionLocal()
+                try:
+                    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+                        SQLAlchemyGenerationTaskRepository,
+                    )
+                    from packages.adapters.sqlalchemy_impl.title_library_repository import (
+                        SQLAlchemyTitleLibraryRepository,
+                    )
+
+                    _task_repo = SQLAlchemyGenerationTaskRepository(_title_session)
+                    _gen_task = _task_repo.get(task_id)
+                    if _gen_task and _gen_task.title_ids and _gen_task.created_by_user_id:
+                        _title_repo = SQLAlchemyTitleLibraryRepository(_title_session)
+                        for _tid in _gen_task.title_ids:
+                            try:
+                                _title_repo.increment_usage_count(
+                                    _tid, _gen_task.created_by_user_id
+                                )
+                            except Exception:
+                                logger.warning(
+                                    "[task_id=%s] 更新标题使用次数失败: title_id=%s",
+                                    task_id, _tid, exc_info=True,
+                                )
+                finally:
+                    _title_session.close()
+            except Exception:
+                logger.warning("[task_id=%s] 更新标题使用次数异常（不影响主流程）", task_id, exc_info=True)
             if gen_task:
                 gen_task.append_log(
                     "任务完成",

packages/adapters/sqlalchemy_impl/title_library_repository.py

@@ -103,6 +103,25 @@ class SQLAlchemyTitleLibraryRepository:
         self.session.commit()
         return True
 
+    def increment_usage_count(self, title_id: str, user_id: str, increment: int = 1) -> bool:
+        """递增标题使用次数。返回是否成功。"""
+        from sqlalchemy import func
+
+        model = (
+            self.session.query(TitleLibraryModel)
+            .filter(
+                TitleLibraryModel.id == title_id,
+                TitleLibraryModel.user_id == user_id,
+            )
+            .first()
+        )
+        if model is None:
+            return False
+        model.usage_count = (model.usage_count or 0) + increment
+        model.updated_at = func.now()
+        self.session.commit()
+        return True
+
     def count_by_user(self, user_id: str, is_active: bool = True) -> int:
         return (
             self.session.query(TitleLibraryModel)

packages/application/title_library/__init__.py

@@ -1,9 +1,11 @@
 """Title library application module."""
 
+from packages.application.title_library.commands import IncrementTitleUsageCommand
 from packages.application.title_library.use_cases import (
     CreateTitleLibraryUseCase,
     DeleteTitleLibraryUseCase,
     GetTitleLibraryUseCase,
+    IncrementTitleUsageUseCase,
     ListTitleLibraryUseCase,
     NotFoundError,
     QuotaExceededError,
@@ -14,6 +16,8 @@ __all__ = [
     "CreateTitleLibraryUseCase",
     "DeleteTitleLibraryUseCase",
     "GetTitleLibraryUseCase",
+    "IncrementTitleUsageUseCase",
+    "IncrementTitleUsageCommand",
     "ListTitleLibraryUseCase",
     "UpdateTitleLibraryUseCase",
     "QuotaExceededError",

packages/application/title_library/commands.py

@@ -28,3 +28,10 @@ class UpdateTitleLibraryCommand:
     tags: Optional[List[str]] = None
     is_active: Optional[bool] = None
     metadata_: Optional[dict] = None
+
+
+@dataclass
+class IncrementTitleUsageCommand:
+    title_id: str
+    user_id: str
+    increment: int = 1

packages/application/title_library/use_cases.py

@@ -8,6 +8,7 @@ from typing import List, Optional
 from packages.adapters.sqlalchemy_impl.title_library_repository import SQLAlchemyTitleLibraryRepository
 from packages.application.title_library.commands import (
     CreateTitleLibraryCommand,
+    IncrementTitleUsageCommand,
     UpdateTitleLibraryCommand,
 )
 from packages.domain.quota import QuotaDimension, quota_checker
@@ -100,6 +101,22 @@ class DeleteTitleLibraryUseCase:
         return self.repository.delete(title_id, user_id)
 
 
+class IncrementTitleUsageUseCase:
+    """递增标题使用次数。用于生成视频成功后，更新标题的使用统计。"""
+
+    def __init__(self, repository: SQLAlchemyTitleLibraryRepository) -> None:
+        self.repository = repository
+
+    def execute(self, command: IncrementTitleUsageCommand) -> bool:
+        if command.increment <= 0:
+            return False
+        return self.repository.increment_usage_count(
+            command.title_id,
+            command.user_id,
+            increment=command.increment,
+        )
+
+
 class QuotaExceededError(Exception):
     def __init__(self, dimension: str, limit: float, used: float) -> None:
         self.dimension = dimension