alembic/versions/044_add_user_id_to_generated_videos.py

@@ -0,0 +1,34 @@
+"""add user_id to generated_videos
+
+Revision ID: 044_user_id_generated_videos
+Revises: 043_updated_at_generation_tasks
+Create Date: 2026-07-19 08:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "044_user_id_generated_videos"
+down_revision = "043_updated_at_generation_tasks"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "generated_videos",
+        sa.Column(
+            "user_id",
+            sa.String(36),
+            nullable=False,
+            server_default="",
+            index=True,
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("generated_videos", "user_id")

apps/api/app/api/routes/videos.py

@@ -57,7 +57,7 @@ def _to_video_response(item, storage: OSSStorageService | None = None) -> VideoI
 
 @router.get("/videos", response_model=ListVideosResponse)
 def list_videos(
-    project_id: str | None = Query(None, description="项目ID，不传则返回所有项目"),
+    project_id: str | None = Query(None, description="项目ID，可选过滤"),
     status: str | None = Query(None, description="按状态筛选"),
     review_status: str | None = Query(None, description="按复核状态筛选"),
     page: int = Query(1, ge=1, description="页码"),
@@ -66,9 +66,10 @@ def list_videos(
     storage: OSSStorageService = Depends(get_storage_service),
     current_user: AuthenticatedUser = Depends(get_current_user),
 ):
-    """成片列表，支持分页、按项目/状态/复核状态筛选。"""
+    """成片列表，默认返回当前用户的所有成片，支持按项目/状态/复核状态筛选。"""
     use_case = ListGeneratedVideosPaginatedUseCase(repo)
     items, total = use_case.execute(
+        user_id=current_user.user.id,
         project_id=project_id,
         status=status,
         review_status=review_status,

apps/worker/video_processing/dedup_helpers.py

@@ -18,6 +18,7 @@ def create_video_record_and_dedup(
     *,
     generation_task_id: str,
     project_id: str,
+    user_id: str = "",
     batch_id: str,
     file_url: str,
     file_size: int,
@@ -64,6 +65,7 @@ def create_video_record_and_dedup(
         generated_video = GeneratedVideo(
             id=video_id,
             project_id=project_id,
+            user_id=user_id,
             generation_task_id=generation_task_id,
             name=video_name,
             file_url=file_url,

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -151,6 +151,7 @@ def _finalize_render_success(
             create_video_record_and_dedup(
                 generation_task_id=generation_task_id,
                 project_id=project_id,
+                user_id=plan.created_by_user_id or "",
                 batch_id=batch_id,
                 file_url=output_url or "",
                 file_size=file_size,

apps/worker/worker_app/tasks/generation.py

@@ -1319,6 +1319,7 @@ def _upload_and_record(
     project_id: str,
     batch_id: str,
     editing_mode,
+    user_id: str = "",
 ) -> tuple[str, float, int, int]:
     """上传 OSS、创建视频记录并查重。
 
@@ -1366,6 +1367,7 @@ def _upload_and_record(
         video_count = create_video_record_and_dedup(
             generation_task_id=task_id,
             project_id=project_id,
+            user_id=user_id,
             batch_id=batch_id,
             file_url=file_url,
             file_size=file_size,
@@ -1511,6 +1513,7 @@ def generate_video(self, task_id: str) -> dict:
                 project_id=project_id,
                 batch_id=batch_id,
                 editing_mode=editing_mode,
+                user_id=user_id,
             )
 
             if gen_task:

docs/schema-metadata-snapshot.json

@@ -1232,6 +1232,14 @@
           "type": "VARCHAR(36)",
           "unique": false
         },
+        {
+          "index": true,
+          "name": "user_id",
+          "nullable": false,
+          "primary_key": false,
+          "type": "VARCHAR(36)",
+          "unique": false
+        },
         {
           "index": true,
           "name": "generation_task_id",
@@ -1405,6 +1413,13 @@
           ],
           "name": "ix_generated_videos_status",
           "unique": false
+        },
+        {
+          "columns": [
+            "user_id"
+          ],
+          "name": "ix_generated_videos_user_id",
+          "unique": false
         }
       ],
       "primary_key": [

packages/adapters/sqlalchemy_impl/generated_video_repository.py

@@ -14,6 +14,7 @@ class SQLAlchemyGeneratedVideoRepository:
         model = GeneratedVideoModel(
             id=video.id,
             project_id=video.project_id,
+            user_id=video.user_id,
             generation_task_id=video.generation_task_id,
             name=video.name,
             file_url=video.file_url,
@@ -43,6 +44,7 @@ class SQLAlchemyGeneratedVideoRepository:
         return GeneratedVideo(
             id=model.id,
             project_id=model.project_id,
+            user_id=getattr(model, "user_id", ""),
             generation_task_id=model.generation_task_id,
             name=model.name,
             file_url=model.file_url,
@@ -103,15 +105,18 @@ class SQLAlchemyGeneratedVideoRepository:
     def list_paginated(
         self,
         *,
+        user_id: str | None = None,
         project_id: str | None = None,
         status: str | None = None,
         review_status: str | None = None,
         page: int = 1,
         page_size: int = 20,
     ) -> tuple[list[GeneratedVideo], int]:
-        """分页查询成片列表，支持按项目、状态、复核状态筛选。"""
+        """分页查询成片列表，支持按用户、项目、状态、复核状态筛选。"""
         query = self.session.query(GeneratedVideoModel)
 
+        if user_id:
+            query = query.filter(GeneratedVideoModel.user_id == user_id)
         if project_id:
             query = query.filter(GeneratedVideoModel.project_id == project_id)
         if status:
@@ -183,6 +188,7 @@ class SQLAlchemyGeneratedVideoRepository:
         return GeneratedVideo(
             id=model.id,
             project_id=model.project_id,
+            user_id=getattr(model, "user_id", ""),
             generation_task_id=model.generation_task_id,
             name=model.name,
             file_url=model.file_url,

packages/adapters/sqlalchemy_impl/models.py

@@ -282,6 +282,7 @@ class GeneratedVideoModel(Base):
 
     id = Column(String(36), primary_key=True)
     project_id = Column(String(36), nullable=False, index=True)
+    user_id = Column(String(36), nullable=False, default="", index=True)
     generation_task_id = Column(String(36), nullable=False, index=True)
     name = Column(String(255), nullable=False)
     # file_url: 完整可访问的 URL，用于客户端直接访问视频

packages/application/generated_videos.py

@@ -21,6 +21,7 @@ class ListGeneratedVideosPaginatedUseCase:
     def execute(
         self,
         *,
+        user_id: str | None = None,
         project_id: str | None = None,
         status: str | None = None,
         review_status: str | None = None,
@@ -32,6 +33,7 @@ class ListGeneratedVideosPaginatedUseCase:
         if page_size < 1 or page_size > 100:
             page_size = 20
         return self.generated_video_repository.list_paginated(
+            user_id=user_id,
             project_id=project_id,
             status=status,
             review_status=review_status,

packages/domain/generated_video.py

@@ -18,6 +18,7 @@ class GeneratedVideo:
     width: int
     height: int
     fps: float
+    user_id: str = ""
     thumbnail_url: str | None = None
     status: str = "completed"
     review_status: str = "pending_review"
@@ -36,6 +37,7 @@ class GeneratedVideo:
         name: str,
         file_url: str,
         *,
+        user_id: str = "",
         file_size: int = 0,
         duration: float = 0.0,
         width: int = 0,
@@ -55,6 +57,7 @@ class GeneratedVideo:
         return cls(
             id=uuid4().hex,
             project_id=project_id.strip(),
+            user_id=user_id.strip(),
             generation_task_id=generation_task_id.strip(),
             name=name.strip(),
             file_url=file_url.strip(),

packages/ports/generated_video_repository.py

@@ -19,6 +19,7 @@ class GeneratedVideoRepository(Protocol):
     def list_paginated(
         self,
         *,
+        user_id: str | None = None,
         project_id: str | None = None,
         status: str | None = None,
         review_status: str | None = None,

tests/unit/test_generated_video_creation_logic.py

@@ -133,6 +133,7 @@ class TestVideoNameParameter:
         with pytest.raises(ValueError, match="name cannot be empty"):
             GeneratedVideo.create(
                 project_id="proj-1",
+                user_id="user-1",
                 generation_task_id="task-1",
                 name="",
                 file_url="https://example.com/test.mp4",
@@ -159,6 +160,134 @@ class TestVideoNameParameter:
         assert video_name_3 == "我的旅行vlog"
 
 
+class TestUserIdFilter:
+    """验证成片库按 user_id 过滤的核心逻辑。"""
+
+    def _setup_repo(self):
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+
+        from packages.adapters.sqlalchemy_impl.generated_video_repository import (
+            SQLAlchemyGeneratedVideoRepository,
+        )
+        from packages.adapters.sqlalchemy_impl.models import Base
+
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        session = sessionmaker(bind=engine)()
+        return SQLAlchemyGeneratedVideoRepository(session), session
+
+    def test_list_paginated_filters_by_user_id(self):
+        """list_paginated 传入 user_id 时只返回该用户的视频。"""
+        from packages.domain import GeneratedVideo
+
+        repo, session = self._setup_repo()
+        try:
+            # 用户A的2个视频
+            for i in range(2):
+                v = GeneratedVideo.create(
+                    project_id=f"proj-a-{i}",
+                    user_id="user-a",
+                    generation_task_id=f"task-a-{i}",
+                    name=f"video-a-{i}.mp4",
+                    file_url=f"https://oss.example.com/a-{i}.mp4",
+                )
+                repo.create(v)
+
+            # 用户B的3个视频
+            for i in range(3):
+                v = GeneratedVideo.create(
+                    project_id=f"proj-b-{i}",
+                    user_id="user-b",
+                    generation_task_id=f"task-b-{i}",
+                    name=f"video-b-{i}.mp4",
+                    file_url=f"https://oss.example.com/b-{i}.mp4",
+                )
+                repo.create(v)
+
+            # 查用户A → 2条
+            items, total = repo.list_paginated(user_id="user-a", page=1, page_size=10)
+            assert total == 2
+            assert len(items) == 2
+            assert all(it.user_id == "user-a" for it in items)
+
+            # 查用户B → 3条
+            items, total = repo.list_paginated(user_id="user-b", page=1, page_size=10)
+            assert total == 3
+            assert len(items) == 3
+            assert all(it.user_id == "user-b" for it in items)
+        finally:
+            session.close()
+
+    def test_list_paginated_user_id_plus_project_id(self):
+        """同时传 user_id 和 project_id 时两个条件同时过滤。"""
+        from packages.domain import GeneratedVideo
+
+        repo, session = self._setup_repo()
+        try:
+            # 用户A的proj-1视频
+            v1 = GeneratedVideo.create(
+                project_id="proj-1",
+                user_id="user-a",
+                generation_task_id="task-1",
+                name="v1.mp4",
+                file_url="https://oss.example.com/v1.mp4",
+            )
+            repo.create(v1)
+
+            # 用户B的proj-1视频（不同用户同项目）
+            v2 = GeneratedVideo.create(
+                project_id="proj-1",
+                user_id="user-b",
+                generation_task_id="task-2",
+                name="v2.mp4",
+                file_url="https://oss.example.com/v2.mp4",
+            )
+            repo.create(v2)
+
+            # 用户A + proj-1 → 只有1条
+            items, total = repo.list_paginated(user_id="user-a", project_id="proj-1", page=1, page_size=10)
+            assert total == 1
+            assert items[0].user_id == "user-a"
+            assert items[0].generation_task_id == "task-1"
+        finally:
+            session.close()
+
+    def test_generated_video_has_user_id_field(self):
+        """GeneratedVideo domain 对象有 user_id 字段。"""
+        from packages.domain import GeneratedVideo
+
+        v = GeneratedVideo.create(
+            project_id="proj-1",
+            user_id="user-123",
+            generation_task_id="task-1",
+            name="test.mp4",
+            file_url="https://example.com/test.mp4",
+        )
+        assert v.user_id == "user-123"
+
+    def test_created_video_persists_user_id(self):
+        """创建视频后 user_id 能正确持久化和读取。"""
+        from packages.domain import GeneratedVideo
+
+        repo, session = self._setup_repo()
+        try:
+            v = GeneratedVideo.create(
+                project_id="proj-1",
+                user_id="user-persist-test",
+                generation_task_id="task-persist",
+                name="persist.mp4",
+                file_url="https://example.com/persist.mp4",
+            )
+            repo.create(v)
+
+            fetched = repo.get(v.id)
+            assert fetched is not None
+            assert fetched.user_id == "user-persist-test"
+        finally:
+            session.close()
+
+
 class TestThumbnailInDedupHelpers:
     """验证 dedup_helpers 中缩略图相关逻辑（全 mock，不依赖 cv2）。"""