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

@@ -82,6 +82,15 @@ 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)$",
+    ),
+    classification: Optional[str] = Query(
+        None,
+        description="按内容分类筛选：scenic=风景、product=产品、person=人物、animal=动物、food=美食、tech=科技、sport=运动、music=音乐、other=其他",
+    ),
     skip: int = Query(0, ge=0),
     limit: int = Query(100, ge=1, le=500),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
@@ -101,11 +110,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/classification 无法在 DB 层过滤）
+    needs_memory_filter = bool(keyword or gender or style or filter_tag_ids or smart_view or classification)
 
     def _apply_memory_filters(items):
-        """应用 keyword / gender / style / tag_ids 内存过滤。"""
+        """应用 keyword / gender / style / tag_ids / smart_view / classification 内存过滤。"""
         result = items
         if keyword:
             kw = keyword.lower()
@@ -114,9 +123,38 @@ def list_assets(
             result = [i for i in result if (i.metadata or {}).get("gender") == gender]
         if style:
             result = [i for i in result if (i.metadata or {}).get("style") == style]
+        if classification:
+            result = [i for i in result if (i.metadata or {}).get("classification") == classification]
         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:
+
+            def __meta(a):
+                return a.metadata or {}
+
+            def __use_count(a):
+                return int(__meta(a).get("generation_use_count") or 0)
+
+            def __review_status(a):
+                return __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/api/app/api/routes/titles.py

@@ -17,13 +17,18 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
 from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.title_library_repository import SQLAlchemyTitleLibraryRepository
-from packages.application.title_library.commands import CreateTitleLibraryCommand, UpdateTitleLibraryCommand
+from packages.application.title_library.commands import (
+    CreateTitleLibraryCommand,
+    PickTitleCommand,
+    UpdateTitleLibraryCommand,
+)
 from packages.application.title_library.use_cases import (
     CreateTitleLibraryUseCase,
     DeleteTitleLibraryUseCase,
     GetTitleLibraryUseCase,
     ListTitleLibraryUseCase,
     NotFoundError,
+    PickTitleUseCase,
     QuotaExceededError,
     UpdateTitleLibraryUseCase,
 )
@@ -70,6 +75,41 @@ def list_titles(
     )
 
 
+@router.post("/pick", response_model=TitleLibraryItemResponse)
+def pick_title(
+    category: Optional[str] = Query(None, description="按分类筛选，不填则从全部标题中选"),
+    exclude_ids: Optional[str] = Query(
+        None,
+        description="排除的标题ID（逗号分隔），用于批量生成时避免重复",
+    ),
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
+) -> TitleLibraryItemResponse:
+    """智能选择一个标题。
+
+    策略：优先使用次数少的，从最少的前5个中随机选一个，兼顾公平和多样性。
+    """
+    user_id = authenticated_user.user.id
+    exclude_list: list[str] = []
+    if exclude_ids:
+        exclude_list = [t.strip() for t in exclude_ids.split(",") if t.strip()]
+
+    use_case = PickTitleUseCase(title_repository)
+    item = use_case.execute(
+        PickTitleCommand(
+            user_id=user_id,
+            category=category,
+            exclude_ids=exclude_list,
+        )
+    )
+    if item is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="标题库为空，请先添加标题",
+        )
+    return _to_response(item)
+
+
 @router.get("/{title_id}", response_model=TitleLibraryItemResponse)
 def get_title(
     title_id: str,

apps/worker/worker_app/tasks/classification.py

@@ -68,6 +68,12 @@ def classify_asset(self, job_id: str) -> dict:
 
             # Update asset with classification status and result
             asset.classification_status = ClassificationStatus.COMPLETED
+            # 把分类结果写入 metadata，供列表筛选和智能视图使用
+            asset.metadata = {
+                **(asset.metadata or {}),
+                "classification": classification,
+                "classification_confidence": confidence,
+            }
             asset_repo.update(asset)
 
             session.commit()

apps/worker/worker_app/tasks/generation.py

@@ -1092,6 +1092,70 @@ 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)
+
+            # 5.2 更新素材使用次数 + 最近使用时间
+            try:
+                from worker_app.core.asset_usage import mark_asset_used_for_generation
+
+                _asset_session = SessionLocal()
+                try:
+                    from packages.adapters.sqlalchemy_impl.asset_repository import (
+                        SQLAlchemyAssetRepository,
+                    )
+                    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+                        SQLAlchemyGenerationTaskRepository,
+                    )
+
+                    _task_repo = SQLAlchemyGenerationTaskRepository(_asset_session)
+                    _asset_repo = SQLAlchemyAssetRepository(_asset_session)
+                    _gen_task = _task_repo.get(task_id)
+                    if _gen_task and _gen_task.asset_ids:
+                        for _aid in _gen_task.asset_ids:
+                            try:
+                                _asset = _asset_repo.get(_aid)
+                                if _asset:
+                                    mark_asset_used_for_generation(_asset)
+                                    _asset_repo.update(_asset)
+                            except Exception:
+                                logger.warning(
+                                    "[task_id=%s] 更新素材使用次数失败: asset_id=%s",
+                                    task_id,
+                                    _aid,
+                                    exc_info=True,
+                                )
+                finally:
+                    _asset_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,11 +1,14 @@
 """Title library application module."""
 
+from packages.application.title_library.commands import IncrementTitleUsageCommand, PickTitleCommand
 from packages.application.title_library.use_cases import (
     CreateTitleLibraryUseCase,
     DeleteTitleLibraryUseCase,
     GetTitleLibraryUseCase,
+    IncrementTitleUsageUseCase,
     ListTitleLibraryUseCase,
     NotFoundError,
+    PickTitleUseCase,
     QuotaExceededError,
     UpdateTitleLibraryUseCase,
 )
@@ -14,7 +17,11 @@ __all__ = [
     "CreateTitleLibraryUseCase",
     "DeleteTitleLibraryUseCase",
     "GetTitleLibraryUseCase",
+    "IncrementTitleUsageUseCase",
+    "IncrementTitleUsageCommand",
     "ListTitleLibraryUseCase",
+    "PickTitleUseCase",
+    "PickTitleCommand",
     "UpdateTitleLibraryUseCase",
     "QuotaExceededError",
     "NotFoundError",

packages/application/title_library/commands.py

@@ -28,3 +28,17 @@ 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
+
+
+@dataclass
+class PickTitleCommand:
+    user_id: str
+    category: Optional[str] = None
+    exclude_ids: List[str] = field(default_factory=list)

packages/application/title_library/use_cases.py

@@ -8,6 +8,8 @@ from typing import List, Optional
 from packages.adapters.sqlalchemy_impl.title_library_repository import SQLAlchemyTitleLibraryRepository
 from packages.application.title_library.commands import (
     CreateTitleLibraryCommand,
+    IncrementTitleUsageCommand,
+    PickTitleCommand,
     UpdateTitleLibraryCommand,
 )
 from packages.domain.quota import QuotaDimension, quota_checker
@@ -100,6 +102,68 @@ 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 PickTitleUseCase:
+    """智能选择一个标题。
+
+    策略：
+    1. 可选按 category 过滤
+    2. 排除指定的 title_ids（如本轮已用过的）
+    3. 按使用次数升序，取最少的前 5 个
+    4. 从中随机选一个，增加多样性
+    5. 无可用标题时返回 None
+    """
+
+    _CANDIDATE_POOL_SIZE = 5
+
+    def __init__(self, repository: SQLAlchemyTitleLibraryRepository) -> None:
+        self.repository = repository
+
+    def execute(self, command: PickTitleCommand) -> TitleLibraryItem | None:
+        import random
+
+        # 取该用户所有活跃标题（或指定分类）
+        all_titles = self.repository.list_by_user(
+            command.user_id,
+            category=command.category,
+            is_active=True,
+            skip=0,
+            limit=500,  # 取足够多的候选
+        )
+
+        if not all_titles:
+            return None
+
+        # 排除已使用/指定排除的
+        exclude_set = set(command.exclude_ids or [])
+        candidates = [t for t in all_titles if t.id not in exclude_set]
+        if not candidates:
+            # 排除后没了，就从全部里选
+            candidates = all_titles
+
+        # 按使用次数升序，取最少的前 N 个
+        candidates.sort(key=lambda t: t.usage_count)
+        pool = candidates[: self._CANDIDATE_POOL_SIZE]
+
+        # 随机选一个
+        return random.choice(pool)
+
+
 class QuotaExceededError(Exception):
     def __init__(self, dimension: str, limit: float, used: float) -> None:
         self.dimension = dimension