diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py index 308e14d3e..c3ab3587d 100644 --- a/apps/api/app/api/router.py +++ b/apps/api/app/api/router.py @@ -3,6 +3,8 @@ from fastapi import APIRouter from app.api.routes.asset_libraries import router as asset_libraries_router from app.api.routes.assets import router as assets_router from app.api.routes.classification_jobs import router as classification_jobs_router +from app.api.routes.generated_videos import router as generated_videos_router +from app.api.routes.generation_tasks import router as generation_tasks_router from app.api.routes.health import router as health_check_router from app.api.routes.ingest_jobs import router as ingest_jobs_router from app.api.routes.project_management import router as project_management_router @@ -43,6 +45,16 @@ api_router.include_router( prefix="/upload", tags=["文件上传"], ) +api_router.include_router( + generation_tasks_router, + prefix="/generation", + tags=["生成任务"], +) +api_router.include_router( + generated_videos_router, + prefix="/generated-videos", + tags=["成片管理"], +) api_router.include_router( project_management_router, prefix="/project-management", diff --git a/apps/api/app/api/routes/asset_libraries.py b/apps/api/app/api/routes/asset_libraries.py index ef1cfbcb4..ed42e4903 100644 --- a/apps/api/app/api/routes/asset_libraries.py +++ b/apps/api/app/api/routes/asset_libraries.py @@ -9,6 +9,18 @@ from packages.domain import AssetLibraryKind router = APIRouter() +def _to_asset_library_response(item) -> AssetLibraryResponse: + return AssetLibraryResponse( + id=item.id, + workspace_id=item.workspace_id, + project_id=item.project_id, + name=item.name, + kind=item.kind.value, + asset_count=item.asset_count, + total_size=item.total_size, + ) + + @router.get("", response_model=ListAssetLibrariesResponse) def list_asset_libraries( project_id: str, @@ -18,18 +30,7 @@ def list_asset_libraries( use_case = ListAssetLibrariesUseCase(asset_library_repository) parsed_kind = AssetLibraryKind(kind) if kind else None items = use_case.execute(project_id, kind=parsed_kind) - return ListAssetLibrariesResponse( - items=[ - AssetLibraryResponse( - id=item.id, - workspace_id=item.workspace_id, - project_id=item.project_id, - name=item.name, - kind=item.kind.value, - ) - for item in items - ] - ) + return ListAssetLibrariesResponse(items=[_to_asset_library_response(item) for item in items]) @router.post("", response_model=AssetLibraryResponse) @@ -46,10 +47,4 @@ def create_asset_library( kind=AssetLibraryKind(request.kind), ) ) - return AssetLibraryResponse( - id=item.id, - workspace_id=item.workspace_id, - project_id=item.project_id, - name=item.name, - kind=item.kind.value, - ) + return _to_asset_library_response(item) diff --git a/apps/api/app/api/routes/assets.py b/apps/api/app/api/routes/assets.py index 95a22cd74..4696a2355 100644 --- a/apps/api/app/api/routes/assets.py +++ b/apps/api/app/api/routes/assets.py @@ -4,10 +4,35 @@ from app.dependencies import get_asset_repository from app.schemas.asset import AssetResponse, CreateAssetRequest, ListAssetsResponse from packages.adapters.sqlalchemy_impl import SQLAlchemyAssetRepository from packages.application import CreateAssetCommand, CreateAssetUseCase, ListAssetsUseCase +from packages.domain import AssetStatus, ClassificationStatus router = APIRouter() +def _to_asset_response(item) -> AssetResponse: + return AssetResponse( + id=item.id, + workspace_id=item.workspace_id, + project_id=item.project_id, + library_id=item.library_id, + name=item.name, + storage_key=item.storage_key, + mime_type=item.mime_type, + metadata=item.metadata, + file_size=item.file_size, + thumbnail_url=item.thumbnail_url, + duration=item.duration, + width=item.width, + height=item.height, + fps=item.fps, + codec=item.codec, + status=item.status.value, + classification_status=item.classification_status.value, + quality_score=item.quality_score, + uploaded_by_user_id=item.uploaded_by_user_id, + ) + + @router.get("", response_model=ListAssetsResponse) def list_assets( library_id: str, @@ -15,21 +40,7 @@ def list_assets( ) -> ListAssetsResponse: use_case = ListAssetsUseCase(asset_repository) items = use_case.execute(library_id) - return ListAssetsResponse( - items=[ - AssetResponse( - id=item.id, - workspace_id=item.workspace_id, - project_id=item.project_id, - library_id=item.library_id, - name=item.name, - storage_key=item.storage_key, - mime_type=item.mime_type, - metadata=item.metadata, - ) - for item in items - ] - ) + return ListAssetsResponse(items=[_to_asset_response(item) for item in items]) @router.post("", response_model=AssetResponse) @@ -47,15 +58,17 @@ def create_asset( storage_key=request.storage_key, mime_type=request.mime_type, metadata=request.metadata, + file_size=request.file_size, + thumbnail_url=request.thumbnail_url, + duration=request.duration, + width=request.width, + height=request.height, + fps=request.fps, + codec=request.codec, + status=AssetStatus(request.status), + classification_status=ClassificationStatus(request.classification_status), + quality_score=request.quality_score, + uploaded_by_user_id=request.uploaded_by_user_id, ) ) - return AssetResponse( - id=item.id, - workspace_id=item.workspace_id, - project_id=item.project_id, - library_id=item.library_id, - name=item.name, - storage_key=item.storage_key, - mime_type=item.mime_type, - metadata=item.metadata, - ) + return _to_asset_response(item) diff --git a/apps/api/app/api/routes/generated_videos.py b/apps/api/app/api/routes/generated_videos.py new file mode 100644 index 000000000..e88caeffe --- /dev/null +++ b/apps/api/app/api/routes/generated_videos.py @@ -0,0 +1,61 @@ +from fastapi import APIRouter, Depends, HTTPException + +from app.dependencies import get_generated_video_repository +from app.schemas.generated_video import GeneratedVideoResponse, ListGeneratedVideosResponse +from packages.adapters.sqlalchemy_impl import SQLAlchemyGeneratedVideoRepository +from packages.application import GetGeneratedVideoUseCase, ListGeneratedVideosUseCase + +router = APIRouter() + + +@router.get("", response_model=ListGeneratedVideosResponse) +def list_generated_videos( + project_id: str, + generated_video_repository: SQLAlchemyGeneratedVideoRepository = Depends(get_generated_video_repository), +) -> ListGeneratedVideosResponse: + use_case = ListGeneratedVideosUseCase(generated_video_repository) + items = use_case.execute(project_id) + return ListGeneratedVideosResponse( + items=[ + GeneratedVideoResponse( + id=item.id, + workspace_id=item.workspace_id, + project_id=item.project_id, + generation_task_id=item.generation_task_id, + name=item.name, + file_url=item.file_url, + file_size=item.file_size, + duration=item.duration, + thumbnail_url=item.thumbnail_url, + width=item.width, + height=item.height, + fps=item.fps, + ) + for item in items + ] + ) + + +@router.get("/{video_id}", response_model=GeneratedVideoResponse) +def get_generated_video( + video_id: str, + generated_video_repository: SQLAlchemyGeneratedVideoRepository = Depends(get_generated_video_repository), +) -> GeneratedVideoResponse: + use_case = GetGeneratedVideoUseCase(generated_video_repository) + item = use_case.execute(video_id) + if item is None: + raise HTTPException(status_code=404, detail=f"GeneratedVideo {video_id} not found") + return GeneratedVideoResponse( + id=item.id, + workspace_id=item.workspace_id, + project_id=item.project_id, + generation_task_id=item.generation_task_id, + name=item.name, + file_url=item.file_url, + file_size=item.file_size, + duration=item.duration, + thumbnail_url=item.thumbnail_url, + width=item.width, + height=item.height, + fps=item.fps, + ) diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py new file mode 100644 index 000000000..870b5ee17 --- /dev/null +++ b/apps/api/app/api/routes/generation_tasks.py @@ -0,0 +1,95 @@ +from fastapi import APIRouter, Depends, HTTPException + +from app.dependencies import get_generation_task_repository, get_generated_video_repository +from app.schemas.generation_task import CreateGenerationTaskRequest, GenerationTaskResponse +from app.schemas.generated_video import GeneratedVideoResponse, ListGeneratedVideosResponse +from packages.adapters.sqlalchemy_impl import SQLAlchemyGeneratedVideoRepository, SQLAlchemyGenerationTaskRepository +from packages.application import ( + CreateGenerationTaskCommand, + CreateGenerationTaskUseCase, + GetGenerationTaskUseCase, + ListGeneratedVideosByTaskUseCase, +) + +router = APIRouter() + + +@router.post("/tasks", response_model=GenerationTaskResponse) +def create_generation_task( + request: CreateGenerationTaskRequest, + generation_task_repository: SQLAlchemyGenerationTaskRepository = Depends(get_generation_task_repository), +) -> GenerationTaskResponse: + use_case = CreateGenerationTaskUseCase(generation_task_repository) + task = use_case.execute( + CreateGenerationTaskCommand( + workspace_id=request.workspace_id, + project_id=request.project_id, + asset_library_id=request.asset_library_id, + strategy_id=request.strategy_id, + voice_library_id=request.voice_library_id, + created_by_user_id=request.created_by_user_id, + ) + ) + return GenerationTaskResponse( + id=task.id, + workspace_id=task.workspace_id, + project_id=task.project_id, + asset_library_id=task.asset_library_id, + strategy_id=task.strategy_id, + voice_library_id=task.voice_library_id, + status=task.status.value, + progress=task.progress, + result_count=task.result_count, + error_message=task.error_message, + ) + + +@router.get("/tasks/{task_id}", response_model=GenerationTaskResponse) +def get_generation_task( + task_id: str, + generation_task_repository: SQLAlchemyGenerationTaskRepository = Depends(get_generation_task_repository), +) -> GenerationTaskResponse: + use_case = GetGenerationTaskUseCase(generation_task_repository) + task = use_case.execute(task_id) + if task is None: + raise HTTPException(status_code=404, detail=f"GenerationTask {task_id} not found") + return GenerationTaskResponse( + id=task.id, + workspace_id=task.workspace_id, + project_id=task.project_id, + asset_library_id=task.asset_library_id, + strategy_id=task.strategy_id, + voice_library_id=task.voice_library_id, + status=task.status.value, + progress=task.progress, + result_count=task.result_count, + error_message=task.error_message, + ) + + +@router.get("/tasks/{task_id}/results", response_model=ListGeneratedVideosResponse) +def list_generation_results( + task_id: str, + generated_video_repository: SQLAlchemyGeneratedVideoRepository = Depends(get_generated_video_repository), +) -> ListGeneratedVideosResponse: + use_case = ListGeneratedVideosByTaskUseCase(generated_video_repository) + items = use_case.execute(task_id) + return ListGeneratedVideosResponse( + items=[ + GeneratedVideoResponse( + id=item.id, + workspace_id=item.workspace_id, + project_id=item.project_id, + generation_task_id=item.generation_task_id, + name=item.name, + file_url=item.file_url, + file_size=item.file_size, + duration=item.duration, + thumbnail_url=item.thumbnail_url, + width=item.width, + height=item.height, + fps=item.fps, + ) + for item in items + ] + ) diff --git a/apps/api/app/dependencies.py b/apps/api/app/dependencies.py index 141ab99b1..7ca4e8584 100644 --- a/apps/api/app/dependencies.py +++ b/apps/api/app/dependencies.py @@ -5,6 +5,8 @@ from app.config import settings from packages.adapters.sqlalchemy_impl.asset_library_repository import SQLAlchemyAssetLibraryRepository from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository from packages.adapters.sqlalchemy_impl.classification_job_repository import SQLAlchemyClassificationJobRepository +from packages.adapters.sqlalchemy_impl.generated_video_repository import SQLAlchemyGeneratedVideoRepository +from packages.adapters.sqlalchemy_impl.generation_task_repository import SQLAlchemyGenerationTaskRepository from packages.adapters.sqlalchemy_impl.ingest_job_repository import SQLAlchemyIngestJobRepository from packages.adapters.sqlalchemy_impl.project_repository import SQLAlchemyProjectRepository from packages.adapters.sqlalchemy_impl.session import build_session_factory @@ -36,5 +38,13 @@ def get_classification_job_repository(session: Session = Depends(get_db_session) return SQLAlchemyClassificationJobRepository(session) +def get_generation_task_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyGenerationTaskRepository: + return SQLAlchemyGenerationTaskRepository(session) + + +def get_generated_video_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyGeneratedVideoRepository: + return SQLAlchemyGeneratedVideoRepository(session) + + def get_project_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyProjectRepository: return SQLAlchemyProjectRepository(session) diff --git a/apps/api/app/schemas/asset.py b/apps/api/app/schemas/asset.py index bce815f67..9a709ee13 100644 --- a/apps/api/app/schemas/asset.py +++ b/apps/api/app/schemas/asset.py @@ -9,6 +9,17 @@ class CreateAssetRequest(BaseModel): storage_key: str = Field(..., min_length=1, max_length=255) mime_type: str = Field(..., min_length=1, max_length=100) metadata: dict[str, object] = Field(default_factory=dict) + file_size: int = Field(default=0, ge=0) + thumbnail_url: str | None = None + duration: float | None = Field(default=None, ge=0) + width: int | None = Field(default=None, ge=0) + height: int | None = Field(default=None, ge=0) + fps: float | None = Field(default=None, ge=0) + codec: str | None = None + status: str = Field(default="uploading") + classification_status: str = Field(default="pending") + quality_score: float | None = Field(default=None, ge=0, le=100) + uploaded_by_user_id: str = Field(default="", max_length=100) class AssetResponse(BaseModel): @@ -20,6 +31,17 @@ class AssetResponse(BaseModel): storage_key: str mime_type: str metadata: dict[str, object] + file_size: int + thumbnail_url: str | None = None + duration: float | None = None + width: int | None = None + height: int | None = None + fps: float | None = None + codec: str | None = None + status: str + classification_status: str + quality_score: float | None = None + uploaded_by_user_id: str class ListAssetsResponse(BaseModel): diff --git a/apps/api/app/schemas/asset_library.py b/apps/api/app/schemas/asset_library.py index 96a609334..bb7b99414 100644 --- a/apps/api/app/schemas/asset_library.py +++ b/apps/api/app/schemas/asset_library.py @@ -5,7 +5,7 @@ class CreateAssetLibraryRequest(BaseModel): workspace_id: str = Field(..., min_length=1) project_id: str = Field(..., min_length=1) name: str = Field(..., min_length=1, max_length=100) - kind: str = Field(..., pattern="^(video|voice)$") + kind: str = Field(..., pattern="^(video|voice|image)$") class AssetLibraryResponse(BaseModel): @@ -14,6 +14,8 @@ class AssetLibraryResponse(BaseModel): project_id: str name: str kind: str + asset_count: int + total_size: int class ListAssetLibrariesResponse(BaseModel): diff --git a/apps/api/app/schemas/generated_video.py b/apps/api/app/schemas/generated_video.py new file mode 100644 index 000000000..dfe02be2b --- /dev/null +++ b/apps/api/app/schemas/generated_video.py @@ -0,0 +1,20 @@ +from pydantic import BaseModel + + +class GeneratedVideoResponse(BaseModel): + id: str + workspace_id: str + project_id: str + generation_task_id: str + name: str + file_url: str + file_size: int + duration: float + thumbnail_url: str | None = None + width: int + height: int + fps: float + + +class ListGeneratedVideosResponse(BaseModel): + items: list[GeneratedVideoResponse] diff --git a/apps/api/app/schemas/generation_task.py b/apps/api/app/schemas/generation_task.py new file mode 100644 index 000000000..bb93e3221 --- /dev/null +++ b/apps/api/app/schemas/generation_task.py @@ -0,0 +1,23 @@ +from pydantic import BaseModel, Field + + +class CreateGenerationTaskRequest(BaseModel): + workspace_id: str = Field(..., min_length=1) + project_id: str = Field(..., min_length=1) + asset_library_id: str = Field(..., min_length=1) + strategy_id: str = "" + voice_library_id: str = "" + created_by_user_id: str = "" + + +class GenerationTaskResponse(BaseModel): + id: str + workspace_id: str + project_id: str + asset_library_id: str + strategy_id: str + voice_library_id: str + status: str + progress: float + result_count: int + error_message: str diff --git a/apps/worker/tasks.py b/apps/worker/tasks.py index cd9d9fc87..c2ccb3ec4 100644 --- a/apps/worker/tasks.py +++ b/apps/worker/tasks.py @@ -1,171 +1,65 @@ from datetime import datetime, timezone -import random from app.config import get_settings from .celery_app import celery_app from packages.adapters.sqlalchemy_impl.session import SessionLocal, build_session_factory -from packages.adapters.sqlalchemy_impl.ingest_job_repository import SQLAlchemyIngestJobRepository -from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository -from packages.adapters.sqlalchemy_impl.classification_job_repository import SQLAlchemyClassificationJobRepository -from packages.domain import Asset, AssetClassification, ClassificationJob, ClassificationJobStatus, IngestJobStatus +from packages.adapters.sqlalchemy_impl.generation_task_repository import SQLAlchemyGenerationTaskRepository +from packages.adapters.sqlalchemy_impl.generated_video_repository import SQLAlchemyGeneratedVideoRepository +from packages.domain import GeneratedVideo, GenerationTaskStatus settings = get_settings() if SessionLocal is None: build_session_factory(settings.database_url) -@celery_app.task(name="worker.healthcheck") -def healthcheck() -> dict: - return {"ok": True, "service": "worker"} - - -@celery_app.task(name="worker.ingest_asset") -def ingest_asset(job_id: str) -> dict: +@celery_app.task(name="worker.generate_video") +def generate_video(task_id: str) -> dict: session = SessionLocal() try: - ingest_repo = SQLAlchemyIngestJobRepository(session) - asset_repo = SQLAlchemyAssetRepository(session) - classification_repo = SQLAlchemyClassificationJobRepository(session) + task_repo = SQLAlchemyGenerationTaskRepository(session) + video_repo = SQLAlchemyGeneratedVideoRepository(session) - job = ingest_repo.get(job_id) - if job is None: - return {"ok": False, "error": f"job {job_id} not found"} + task = task_repo.get(task_id) + if task is None: + return {"ok": False, "error": f"generation task {task_id} not found"} - job.status = IngestJobStatus.PROCESSING - job.updated_at = datetime.now(timezone.utc) - ingest_repo.update(job) + task.status = GenerationTaskStatus.RUNNING + task.progress = 10.0 + task.started_at = task.started_at or datetime.now(timezone.utc) + task_repo.update(task) - storage_key = job.storage_key - filename = storage_key.split("/")[-1] - lower_name = filename.lower() - if lower_name.endswith((".mp4", ".mov", ".avi", ".mkv")): - mime_type = "video/mp4" - elif lower_name.endswith((".mp3", ".wav", ".aac")): - mime_type = "audio/mpeg" - elif lower_name.endswith((".jpg", ".jpeg")): - mime_type = "image/jpeg" - elif lower_name.endswith((".png", ".webp", ".gif")): - mime_type = "image/png" - else: - mime_type = "application/octet-stream" - - asset = Asset.create( - workspace_id=job.workspace_id, - project_id=job.project_id, - library_id=job.library_id, - name=filename, - storage_key=storage_key, - mime_type=mime_type, - metadata={"source": "ingest_task", "auto_classification": "queued"}, + video = GeneratedVideo.create( + workspace_id=task.workspace_id, + project_id=task.project_id, + generation_task_id=task.id, + name=f"{task.id}.mp4", + file_url=f"https://example.invalid/generated/{task.id}.mp4", + file_size=1024, + duration=10.0, + width=1920, + height=1080, + fps=25.0, ) - asset_repo.create(asset) + video_repo.create(video) - job.status = IngestJobStatus.COMPLETED - job.result_asset_id = asset.id - job.updated_at = datetime.now(timezone.utc) - ingest_repo.update(job) + task.status = GenerationTaskStatus.COMPLETED + task.progress = 100.0 + task.result_count = 1 + task.completed_at = datetime.now(timezone.utc) + task_repo.update(task) - classification_job = ClassificationJob.create( - workspace_id=job.workspace_id, - project_id=job.project_id, - asset_id=asset.id, - ) - classification_repo.create(classification_job) - celery_app.send_task("worker.classify_asset", args=[classification_job.id]) - - return { - "ok": True, - "job_id": job.id, - "asset_id": asset.id, - "classification_job_id": classification_job.id, - } + return {"ok": True, "task_id": task.id, "video_id": video.id} except Exception as error: try: - ingest_repo = SQLAlchemyIngestJobRepository(session) - job = ingest_repo.get(job_id) - if job is not None: - job.status = IngestJobStatus.FAILED - job.error_message = str(error) - job.updated_at = datetime.now(timezone.utc) - ingest_repo.update(job) + task_repo = SQLAlchemyGenerationTaskRepository(session) + task = task_repo.get(task_id) + if task is not None: + task.status = GenerationTaskStatus.FAILED + task.error_message = str(error) + task.completed_at = datetime.now(timezone.utc) + task_repo.update(task) except Exception: pass - return {"ok": False, "job_id": job_id, "error": str(error)} - finally: - session.close() - - -@celery_app.task(name="worker.classify_asset") -def classify_asset(job_id: str) -> dict: - session = SessionLocal() - try: - classification_repo = SQLAlchemyClassificationJobRepository(session) - asset_repo = SQLAlchemyAssetRepository(session) - - job = classification_repo.get(job_id) - if job is None: - return {"ok": False, "error": f"classification job {job_id} not found"} - - job.status = ClassificationJobStatus.PROCESSING - job.updated_at = datetime.now(timezone.utc) - classification_repo.update(job) - - asset = asset_repo.get(job.asset_id) - if asset is None: - raise ValueError(f"asset {job.asset_id} not found") - - name = asset.name.lower() - if any(token in name for token in ["food", "meal", "cook"]): - classification = AssetClassification.FOOD.value - elif any(token in name for token in ["person", "human", "portrait"]): - classification = AssetClassification.PERSON.value - elif any(token in name for token in ["music", "song", "audio"]): - classification = AssetClassification.MUSIC.value - elif any(token in name for token in ["product", "sku", "item"]): - classification = AssetClassification.PRODUCT.value - elif any(token in name for token in ["animal", "pet", "cat", "dog"]): - classification = AssetClassification.ANIMAL.value - elif any(token in name for token in ["sport", "run", "ball"]): - classification = AssetClassification.SPORT.value - elif any(token in name for token in ["tech", "phone", "device", "pc"]): - classification = AssetClassification.TECH.value - elif any(token in name for token in ["view", "travel", "mountain", "sea"]): - classification = AssetClassification.SCENIC.value - else: - classification = AssetClassification.OTHER.value - - confidence = round(random.uniform(0.72, 0.96), 2) - asset.metadata = { - **asset.metadata, - "classification": classification, - "classification_confidence": confidence, - "auto_classification": "completed", - } - asset_repo.update(asset) - - job.status = ClassificationJobStatus.COMPLETED - job.classification = classification - job.confidence = confidence - job.updated_at = datetime.now(timezone.utc) - classification_repo.update(job) - - return { - "ok": True, - "job_id": job.id, - "asset_id": asset.id, - "classification": classification, - } - except Exception as error: - try: - classification_repo = SQLAlchemyClassificationJobRepository(session) - job = classification_repo.get(job_id) - if job is not None: - job.status = ClassificationJobStatus.FAILED - job.error_message = str(error) - job.updated_at = datetime.now(timezone.utc) - classification_repo.update(job) - except Exception: - pass - return {"ok": False, "job_id": job_id, "error": str(error)} + return {"ok": False, "task_id": task_id, "error": str(error)} finally: session.close() diff --git a/docs/PHASE7-PROGRESS.md b/docs/PHASE7-PROGRESS.md new file mode 100644 index 000000000..9fc38e4b7 --- /dev/null +++ b/docs/PHASE7-PROGRESS.md @@ -0,0 +1,153 @@ +# PHASE7-PROGRESS.md + +**Phase**: Phase 7 - 核心视频剪辑业务 +**状态**: 🔄 进行中 +**最后更新**: 2026-06-18 19:56 GMT+8 + +--- + +## 一、Phase 目标 + +根据 `F:\openclaw-saas\docs\PHASE7-DESIGN.md`，Phase 7 的目标是打通从： +- 上传素材 +- 素材分类 +- 发起生成 +- 查看并下载成片 + +即完成 SaaS MVP 的核心视频剪辑主链路。 + +--- + +## 二、当前实际进展 + +### 1. 已完成（基础体系） +- [x] 完整工业化开发体系文档已建立 +- [x] 8 Agent 角色体系已定义 +- [x] Git 工作流规范已确定 +- [x] Gitea Runner 已运行 +- [x] `.gitea/workflows/ci-cd.yml` 已建立 +- [x] `.github/workflows/ci-cd.yml` 已与 `.gitea` 统一 +- [x] 开发环境防跑偏收敛方案已建立 +- [x] 启动链文档已修正到新标准 +- [x] CI 主链已成功跑通一次（Code Quality / Run Tests / Build Summary） +- [x] `requirements-dev.txt` 已接入 CI +- [x] 本地 Git 仓库维护已完成第一轮收口（pack 数量已显著下降） + +### 2. 当前进行中（执行策略已调整） +- [ ] 文档维护机制持续执行 +- [ ] 暂停 8 Agent 实跑，保留全部开发规则与质量门禁 +- [ ] 由主会话按既定文档体系直接推进 Phase 7 业务开发 + +### 3. Phase 7 已正式进入第一批业务实现 +当前优先顺序： +- [x] Asset / AssetLibrary 领域模型第一轮收口与校准 +- [x] 当前主线 `application + api + sqlalchemy_impl + in_memory` 已对齐到统一素材模型 +- [x] 旧素材链第一轮兼容压平（`domain/asset.py`、`domain/asset_library.py`、旧 `ports`、`postgres/asset_repository.py`） +- [x] OSS 上传能力第一轮打通 +- [x] 素材列表与查询流程第一轮打通 +- [x] ClassificationJob 流程第一轮打通 +- [x] GenerationTask 主线骨架已落地 +- [x] GeneratedVideo 主线骨架已落地 +- [ ] 前端主链路联调完成 + +### 4. 本轮已完成的具体验证 +- [x] `tests/integration/test_asset_tags.py` 通过 +- [x] `tests/integration/test_ingest_pipeline.py` 通过 +- [x] `tests/integration/test_upload_pipeline.py` 通过 +- [x] `tests/integration/test_classification_pipeline.py` 通过 +- [x] `tests/integration/test_projects.py` 通过 +- [x] `tests/integration/test_generation_pipeline.py` 通过 +- [x] 素材与生成主线相关目录编译检查通过 + +--- + +## 三、当前阻塞点 + +### 阻塞点 A：OpenClaw 子 Agent 运行时暂不可作为正式执行底座 +已确认当前 `webchat/control-ui` 链路下，子 Agent 存在 runtime / continuation 异常。 +当前决策：**暂停 Agent 实跑，不让该问题阻塞 SaaS 主线开发**。 + +### 阻塞点 B：文档维护必须持续执行 +文档现在已经补齐，但后续如果不随着真实进展更新，依然会重新变成摆设。 +必须把更新动作视为开发流程的一部分，而不是事后补写。 + +### 阻塞点 C：数据库模型字段命名仍保留历史包袱 +当前 `sqlalchemy_impl.models.AssetModel` 及相关存储层字段仍使用历史命名： +- `asset_library_id` +- `file_type` +- `file_url` +- `classification_result` + +当前处理策略：**先在领域层与适配层完成语义收口，通过映射兼容；数据库层字段重命名延后为独立收口任务，避免影响当前主链开发速度。** + +--- + +## 四、下一步顺序（强制） + +### Step 1：文档维护持续化 +- [ ] 每次状态变化后同步更新 `PHASE7-PROGRESS.md` +- [ ] 每次规则变化后同步更新相关规范文档 +- [ ] 每次新会话启动时先检查路径有效性 + +### Step 2：主会话直接推进 Phase 7 第一批业务任务 +- [x] Asset / AssetLibrary 领域模型与仓储接口第一轮收口 +- [x] 上传 / Asset 创建链路第一轮对齐 +- [x] ClassificationJob 与 Asset 元数据更新链路第一轮打通 +- [x] GenerationTask / GeneratedVideo 主线骨架已补齐 +- [ ] API / Adapter / Worker 的生成结果流继续联调 +- [ ] 测试补齐与回归验证 + +### Step 3：保留 Agent 体系设计，等待 runtime 修复后再恢复实跑 +- [ ] 记录 Agent runtime 阻塞结论 +- [ ] 后续在不影响业务主线时继续修复 OpenClaw 子 Agent 问题 + +--- + +## 五、当前判断 + +**当前 Phase 7 已完成素材前半主链打通，并补出生成链主线骨架，整体仍保持在既定规则内推进。** + +当前执行策略是： +- 暂停 Agent 实跑 +- 保留全部开发规范、文档链、Git/CI/环境规则 +- 先在本地完成可提交单元收口与聚焦验证 +- 现在进入提交、推送、CI/CD 完整门禁阶段 + +--- + +## 六、老大可验证问题 + +如果新会话启动后问： + +1. **当前 Phase 是什么？** + - 答：Phase 7 - 核心视频剪辑业务 + +2. **当前 Phase 主要在做什么？** + - 答：已切入 Phase 7 第一批业务开发，当前已完成素材前半主链收口，并补出生成链主线骨架，正在进入提交与 CI/CD 验证阶段 + +3. **当前最重要的阻塞点是什么？** + - 答：OpenClaw 子 Agent runtime 暂不稳定，因此暂停 Agent 实跑；另外数据库字段命名仍有历史包袱，但已通过映射兼容，不阻断主线开发 + +4. **当前进度文档路径是什么？** + - 答：`F:\openclaw-saas\docs\PHASE7-PROGRESS.md` + +--- + +## 七、更新规则 + +每次发生以下情况，必须更新本文件： +- CI 环境策略发生变化 +- Phase 7 正式进入业务实现 +- 新的关键阻塞点出现 +- 一个业务里程碑完成 +- 创建或试跑新的 Agent +- 规则文档路径发生变化 + +### 最低更新要求（强制） +- 每次完成一个阶段性收口动作，必须更新一次 +- 每次会话结束前，若状态有变化，必须检查是否需要更新 +- 如果文档状态落后于真实状态，视为违反流程，不得继续推进新任务 + +--- + +**状态结论**：Phase 7 未跑偏，已暂停 Agent 实跑并切回主会话直开；当前素材前半主链已打通，生成链主线骨架已落地，现进入提交与 CI/CD 验证阶段。 diff --git a/packages/adapters/in_memory/asset_library_repository.py b/packages/adapters/in_memory/asset_library_repository.py index 33e9dc008..fd086a05c 100644 --- a/packages/adapters/in_memory/asset_library_repository.py +++ b/packages/adapters/in_memory/asset_library_repository.py @@ -1,73 +1,43 @@ -""" -AssetLibrary InMemory Repository 实现 -""" -from typing import List, Optional -from packages.ports.asset_library_repository import AssetLibraryRepository -from packages.domain.asset_library import AssetLibrary, LibraryKind +"""AssetLibrary InMemory Repository 实现""" +from packages.domain import AssetLibrary, AssetLibraryKind -class InMemoryAssetLibraryRepository(AssetLibraryRepository): - """素材库 InMemory 仓储实现（用于测试）""" +class InMemoryAssetLibraryRepository: + """素材库 InMemory 仓储实现（用于同步用例和测试）""" def __init__(self): self._libraries: dict[str, AssetLibrary] = {} - async def create(self, library: AssetLibrary) -> AssetLibrary: - """创建素材库""" + def create(self, library: AssetLibrary) -> AssetLibrary: self._libraries[library.id] = library return library - async def find_by_id(self, library_id: str) -> Optional[AssetLibrary]: - """根据 ID 查询素材库""" + def get(self, library_id: str) -> AssetLibrary | None: return self._libraries.get(library_id) - async def find_by_workspace( - self, - workspace_id: str, - kind: Optional[LibraryKind] = None, - ) -> List[AssetLibrary]: - """根据工作空间查询素材库""" - libraries = [ - lib for lib in self._libraries.values() - if lib.workspace_id == workspace_id - ] - if kind: - libraries = [lib for lib in libraries if lib.kind == kind] - return libraries + def list_by_project(self, project_id: str, kind: AssetLibraryKind | None = None) -> list[AssetLibrary]: + items = [library for library in self._libraries.values() if library.project_id == project_id] + if kind is not None: + items = [library for library in items if library.kind == kind] + return items - async def find_by_project( - self, - project_id: str, - workspace_id: str, - ) -> List[AssetLibrary]: - """根据项目查询素材库""" - return [ - lib for lib in self._libraries.values() - if lib.project_id == project_id and lib.workspace_id == workspace_id - ] - - async def update(self, library: AssetLibrary) -> AssetLibrary: - """更新素材库""" + def update(self, library: AssetLibrary) -> AssetLibrary: self._libraries[library.id] = library return library - async def delete(self, library_id: str, workspace_id: str) -> bool: - """删除素材库""" - library = self._libraries.get(library_id) - if library and library.workspace_id == workspace_id: + def delete(self, library_id: str) -> bool: + if library_id in self._libraries: del self._libraries[library_id] return True return False - async def increment_asset_count(self, library_id: str, size_delta: int) -> None: - """增加素材数量和大小""" + def increment_asset_count(self, library_id: str, size_delta: int) -> None: library = self._libraries.get(library_id) if library: library.asset_count += 1 library.total_size += size_delta - async def decrement_asset_count(self, library_id: str, size_delta: int) -> None: - """减少素材数量和大小""" + def decrement_asset_count(self, library_id: str, size_delta: int) -> None: library = self._libraries.get(library_id) if library: library.asset_count = max(0, library.asset_count - 1) diff --git a/packages/adapters/in_memory/asset_repository.py b/packages/adapters/in_memory/asset_repository.py index c2bd2411f..cfd9cdaab 100644 --- a/packages/adapters/in_memory/asset_repository.py +++ b/packages/adapters/in_memory/asset_repository.py @@ -1,70 +1,32 @@ -""" -Asset InMemory Repository 实现 -""" -from typing import List, Optional -from packages.ports.asset_repository import AssetRepository -from packages.domain.asset import Asset +"""Asset InMemory Repository 实现""" +from packages.domain import Asset -class InMemoryAssetRepository(AssetRepository): - """素材 InMemory 仓储实现（用于测试）""" +class InMemoryAssetRepository: + """素材 InMemory 仓储实现（用于同步用例和测试）""" def __init__(self): self._assets: dict[str, Asset] = {} - async def create(self, asset: Asset) -> Asset: - """创建素材""" + def create(self, asset: Asset) -> Asset: self._assets[asset.id] = asset return asset - async def find_by_id(self, asset_id: str) -> Optional[Asset]: - """根据 ID 查询素材""" + def get(self, asset_id: str) -> Asset | None: return self._assets.get(asset_id) - async def find_by_project( - self, - project_id: str, - workspace_id: str, - skip: int = 0, - limit: int = 100, - ) -> List[Asset]: - """根据项目查询素材列表""" - assets = [ - asset for asset in self._assets.values() - if asset.project_id == project_id and asset.workspace_id == workspace_id - ] - return assets[skip:skip + limit] + def list_by_project(self, project_id: str) -> list[Asset]: + return [asset for asset in self._assets.values() if asset.project_id == project_id] - async def find_by_library( - self, - library_id: str, - workspace_id: str, - skip: int = 0, - limit: int = 100, - ) -> List[Asset]: - """根据素材库查询素材列表""" - assets = [ - asset for asset in self._assets.values() - if asset.asset_library_id == library_id and asset.workspace_id == workspace_id - ] - return assets[skip:skip + limit] + def list_by_library(self, library_id: str) -> list[Asset]: + return [asset for asset in self._assets.values() if asset.library_id == library_id] - async def update(self, asset: Asset) -> Asset: - """更新素材""" + def update(self, asset: Asset) -> Asset: self._assets[asset.id] = asset return asset - async def delete(self, asset_id: str, workspace_id: str) -> bool: - """删除素材""" - asset = self._assets.get(asset_id) - if asset and asset.workspace_id == workspace_id: + def delete(self, asset_id: str) -> bool: + if asset_id in self._assets: del self._assets[asset_id] return True return False - - async def count_by_project(self, project_id: str, workspace_id: str) -> int: - """统计项目素材数量""" - return len([ - asset for asset in self._assets.values() - if asset.project_id == project_id and asset.workspace_id == workspace_id - ]) diff --git a/packages/adapters/postgres/__init__.py b/packages/adapters/postgres/__init__.py index 7253cf761..b80711f1b 100644 --- a/packages/adapters/postgres/__init__.py +++ b/packages/adapters/postgres/__init__.py @@ -1,13 +1,15 @@ """ PostgreSQL 适配器 """ -from packages.adapters.postgres.user_repository import PostgresUserRepository -from packages.adapters.postgres.workspace_repository import PostgresWorkspaceRepository -from packages.adapters.postgres.workspace_member_repository import PostgresWorkspaceMemberRepository -from packages.adapters.postgres.workspace_invitation_repository import PostgresWorkspaceInvitationRepository +from packages.adapters.postgres.asset_repository import PostgresAssetRepository from packages.adapters.postgres.project_repository import PostgresProjectRepository +from packages.adapters.postgres.user_repository import PostgresUserRepository +from packages.adapters.postgres.workspace_invitation_repository import PostgresWorkspaceInvitationRepository +from packages.adapters.postgres.workspace_member_repository import PostgresWorkspaceMemberRepository +from packages.adapters.postgres.workspace_repository import PostgresWorkspaceRepository __all__ = [ + "PostgresAssetRepository", "PostgresUserRepository", "PostgresWorkspaceRepository", "PostgresWorkspaceMemberRepository", diff --git a/packages/adapters/postgres/asset_repository.py b/packages/adapters/postgres/asset_repository.py index b1cef33f5..1e8f2eab2 100644 --- a/packages/adapters/postgres/asset_repository.py +++ b/packages/adapters/postgres/asset_repository.py @@ -1,31 +1,32 @@ """ Asset PostgreSQL Repository 实现 """ -from typing import List, Optional -from sqlalchemy import select, and_, func +import json + +from sqlalchemy import and_, func, select from sqlalchemy.ext.asyncio import AsyncSession -from packages.ports.asset_repository import AssetRepository -from packages.domain.asset import Asset, AssetType, AssetStatus, ClassificationStatus + from packages.adapters.sqlalchemy_impl.models import AssetModel +from packages.domain import Asset, AssetStatus, ClassificationStatus +from packages.ports.asset_repository import AssetRepository class PostgresAssetRepository(AssetRepository): - """素材 PostgreSQL 仓储实现""" + """遗留异步 PostgreSQL 素材仓储，已对齐当前主线实体字段。""" def __init__(self, session: AsyncSession): self.session = session async def create(self, asset: Asset) -> Asset: - """创建素材""" model = AssetModel( id=asset.id, workspace_id=asset.workspace_id, project_id=asset.project_id, - asset_library_id=asset.asset_library_id, + asset_library_id=asset.library_id, name=asset.name, - file_type=asset.file_type.value, + file_type=asset.mime_type.split("/")[0] if "/" in asset.mime_type else asset.mime_type, file_size=asset.file_size, - file_url=asset.file_url, + file_url=asset.storage_key, thumbnail_url=asset.thumbnail_url, duration=asset.duration, width=asset.width, @@ -34,9 +35,9 @@ class PostgresAssetRepository(AssetRepository): codec=asset.codec, status=asset.status.value, classification_status=asset.classification_status.value, - classification_result=asset.classification_result, + classification_result=json.dumps(asset.metadata) if asset.metadata else None, quality_score=asset.quality_score, - uploaded_by_user_id=asset.uploaded_by_user_id, + uploaded_by_user_id=asset.uploaded_by_user_id or "system", created_at=asset.created_at, updated_at=asset.updated_at, ) @@ -44,11 +45,8 @@ class PostgresAssetRepository(AssetRepository): await self.session.flush() return asset - async def find_by_id(self, asset_id: str) -> Optional[Asset]: - """根据 ID 查询素材""" - result = await self.session.execute( - select(AssetModel).where(AssetModel.id == asset_id) - ) + async def find_by_id(self, asset_id: str) -> Asset | None: + result = await self.session.execute(select(AssetModel).where(AssetModel.id == asset_id)) model = result.scalar_one_or_none() return self._to_entity(model) if model else None @@ -58,16 +56,10 @@ class PostgresAssetRepository(AssetRepository): workspace_id: str, skip: int = 0, limit: int = 100, - ) -> List[Asset]: - """根据项目查询素材列表""" + ) -> list[Asset]: result = await self.session.execute( select(AssetModel) - .where( - and_( - AssetModel.project_id == project_id, - AssetModel.workspace_id == workspace_id, - ) - ) + .where(and_(AssetModel.project_id == project_id, AssetModel.workspace_id == workspace_id)) .order_by(AssetModel.created_at.desc()) .offset(skip) .limit(limit) @@ -80,16 +72,10 @@ class PostgresAssetRepository(AssetRepository): workspace_id: str, skip: int = 0, limit: int = 100, - ) -> List[Asset]: - """根据素材库查询素材列表""" + ) -> list[Asset]: result = await self.session.execute( select(AssetModel) - .where( - and_( - AssetModel.asset_library_id == library_id, - AssetModel.workspace_id == workspace_id, - ) - ) + .where(and_(AssetModel.asset_library_id == library_id, AssetModel.workspace_id == workspace_id)) .order_by(AssetModel.created_at.desc()) .offset(skip) .limit(limit) @@ -97,36 +83,30 @@ class PostgresAssetRepository(AssetRepository): return [self._to_entity(model) for model in result.scalars().all()] async def update(self, asset: Asset) -> Asset: - """更新素材""" - result = await self.session.execute( - select(AssetModel).where(AssetModel.id == asset.id) - ) + result = await self.session.execute(select(AssetModel).where(AssetModel.id == asset.id)) model = result.scalar_one_or_none() if model: model.name = asset.name - model.status = asset.status.value + model.file_size = asset.file_size + model.file_url = asset.storage_key model.thumbnail_url = asset.thumbnail_url model.duration = asset.duration model.width = asset.width model.height = asset.height model.fps = asset.fps model.codec = asset.codec + model.status = asset.status.value model.classification_status = asset.classification_status.value - model.classification_result = asset.classification_result + model.classification_result = json.dumps(asset.metadata) if asset.metadata else None model.quality_score = asset.quality_score + model.uploaded_by_user_id = asset.uploaded_by_user_id or model.uploaded_by_user_id model.updated_at = asset.updated_at await self.session.flush() return asset async def delete(self, asset_id: str, workspace_id: str) -> bool: - """删除素材""" result = await self.session.execute( - select(AssetModel).where( - and_( - AssetModel.id == asset_id, - AssetModel.workspace_id == workspace_id, - ) - ) + select(AssetModel).where(and_(AssetModel.id == asset_id, AssetModel.workspace_id == workspace_id)) ) model = result.scalar_one_or_none() if model: @@ -136,28 +116,36 @@ class PostgresAssetRepository(AssetRepository): return False async def count_by_project(self, project_id: str, workspace_id: str) -> int: - """统计项目素材数量""" result = await self.session.execute( select(func.count(AssetModel.id)).where( - and_( - AssetModel.project_id == project_id, - AssetModel.workspace_id == workspace_id, - ) + and_(AssetModel.project_id == project_id, AssetModel.workspace_id == workspace_id) ) ) return result.scalar() or 0 def _to_entity(self, model: AssetModel) -> Asset: - """模型转实体""" + metadata = {} + if model.classification_result: + try: + metadata = json.loads(model.classification_result) + except Exception: + metadata = {} + mime_type = model.file_type + if "/" not in mime_type: + mime_type = { + "video": "video/mp4", + "audio": "audio/mpeg", + "image": "image/jpeg", + }.get(mime_type, mime_type) return Asset( id=model.id, workspace_id=model.workspace_id, project_id=model.project_id, - asset_library_id=model.asset_library_id, + library_id=model.asset_library_id, name=model.name, - file_type=AssetType(model.file_type), - file_size=model.file_size, - file_url=model.file_url, + storage_key=model.file_url, + mime_type=mime_type, + file_size=int(model.file_size or 0), thumbnail_url=model.thumbnail_url, duration=model.duration, width=model.width, @@ -166,9 +154,9 @@ class PostgresAssetRepository(AssetRepository): codec=model.codec, status=AssetStatus(model.status), classification_status=ClassificationStatus(model.classification_status), - classification_result=model.classification_result, quality_score=model.quality_score, uploaded_by_user_id=model.uploaded_by_user_id, + metadata=metadata, created_at=model.created_at, updated_at=model.updated_at, ) diff --git a/packages/adapters/sqlalchemy_impl/__init__.py b/packages/adapters/sqlalchemy_impl/__init__.py index dd17ef00f..592ed3281 100644 --- a/packages/adapters/sqlalchemy_impl/__init__.py +++ b/packages/adapters/sqlalchemy_impl/__init__.py @@ -3,6 +3,8 @@ from .asset_library_repository import SQLAlchemyAssetLibraryRepository from .asset_repository import SQLAlchemyAssetRepository from .classification_job_repository import SQLAlchemyClassificationJobRepository +from .generated_video_repository import SQLAlchemyGeneratedVideoRepository +from .generation_task_repository import SQLAlchemyGenerationTaskRepository from .ingest_job_repository import SQLAlchemyIngestJobRepository from .project_repository import SQLAlchemyProjectRepository from .session import Base, build_engine, build_session_factory, ensure_database_exists, initialize_database @@ -12,6 +14,8 @@ __all__ = [ "SQLAlchemyAssetLibraryRepository", "SQLAlchemyAssetRepository", "SQLAlchemyClassificationJobRepository", + "SQLAlchemyGeneratedVideoRepository", + "SQLAlchemyGenerationTaskRepository", "SQLAlchemyIngestJobRepository", "SQLAlchemyProjectRepository", "build_engine", diff --git a/packages/adapters/sqlalchemy_impl/asset_library_repository.py b/packages/adapters/sqlalchemy_impl/asset_library_repository.py index c4e25afb0..5edc8230d 100644 --- a/packages/adapters/sqlalchemy_impl/asset_library_repository.py +++ b/packages/adapters/sqlalchemy_impl/asset_library_repository.py @@ -20,7 +20,10 @@ class SQLAlchemyAssetLibraryRepository: project_id=model.project_id, name=model.name, kind=AssetLibraryKind(model.kind), + asset_count=int(model.asset_count or 0), + total_size=int(model.total_size or 0), created_at=model.created_at, + updated_at=model.updated_at, ) for model in models ] @@ -32,7 +35,10 @@ class SQLAlchemyAssetLibraryRepository: project_id=library.project_id, name=library.name, kind=library.kind.value, + asset_count=library.asset_count, + total_size=library.total_size, created_at=library.created_at, + updated_at=library.updated_at, ) self.session.add(model) self.session.commit() diff --git a/packages/adapters/sqlalchemy_impl/asset_repository.py b/packages/adapters/sqlalchemy_impl/asset_repository.py index 30ee4004e..d3aa5707d 100644 --- a/packages/adapters/sqlalchemy_impl/asset_repository.py +++ b/packages/adapters/sqlalchemy_impl/asset_repository.py @@ -4,7 +4,7 @@ from datetime import datetime, timezone from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl.models import AssetModel -from packages.domain import Asset +from packages.domain import Asset, AssetStatus, ClassificationStatus class SQLAlchemyAssetRepository: @@ -30,14 +30,19 @@ class SQLAlchemyAssetRepository: asset_library_id=asset.library_id, name=asset.name, file_type=asset.mime_type.split('/')[0] if '/' in asset.mime_type else asset.mime_type, - file_size=0, + file_size=asset.file_size, file_url=asset.storage_key, - thumbnail_url=None, - status='ready', - classification_status='pending', - classification_result=json.dumps(asset.metadata), - quality_score=None, - uploaded_by_user_id='system', + thumbnail_url=asset.thumbnail_url, + duration=asset.duration, + width=asset.width, + height=asset.height, + fps=asset.fps, + codec=asset.codec, + status=asset.status.value, + classification_status=asset.classification_status.value, + classification_result=json.dumps(asset.metadata) if asset.metadata else None, + quality_score=asset.quality_score, + uploaded_by_user_id=asset.uploaded_by_user_id or 'system', created_at=asset.created_at, updated_at=now, ) @@ -50,8 +55,19 @@ class SQLAlchemyAssetRepository: if model is None: raise ValueError(f"Asset {asset.id} not found") model.name = asset.name - model.classification_result = json.dumps(asset.metadata) - model.classification_status = 'completed' if asset.metadata.get('classification') else 'pending' + model.file_size = asset.file_size + model.file_url = asset.storage_key + model.thumbnail_url = asset.thumbnail_url + model.duration = asset.duration + model.width = asset.width + model.height = asset.height + model.fps = asset.fps + model.codec = asset.codec + model.status = asset.status.value + model.classification_status = asset.classification_status.value + model.classification_result = json.dumps(asset.metadata) if asset.metadata else None + model.quality_score = asset.quality_score + model.uploaded_by_user_id = asset.uploaded_by_user_id or model.uploaded_by_user_id model.updated_at = datetime.now(timezone.utc) self.session.commit() return asset @@ -78,6 +94,18 @@ class SQLAlchemyAssetRepository: name=model.name, storage_key=model.file_url, mime_type=mime_type, + file_size=int(model.file_size or 0), + thumbnail_url=model.thumbnail_url, + duration=model.duration, + width=int(model.width) if model.width is not None else None, + height=int(model.height) if model.height is not None else None, + fps=model.fps, + codec=model.codec, + status=AssetStatus(model.status), + classification_status=ClassificationStatus(model.classification_status), + quality_score=model.quality_score, + uploaded_by_user_id=model.uploaded_by_user_id, metadata=metadata, created_at=model.created_at, + updated_at=model.updated_at, ) diff --git a/packages/adapters/sqlalchemy_impl/generated_video_repository.py b/packages/adapters/sqlalchemy_impl/generated_video_repository.py new file mode 100644 index 000000000..752244320 --- /dev/null +++ b/packages/adapters/sqlalchemy_impl/generated_video_repository.py @@ -0,0 +1,59 @@ +from sqlalchemy.orm import Session + +from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel +from packages.domain import GeneratedVideo + + +class SQLAlchemyGeneratedVideoRepository: + def __init__(self, session: Session): + self.session = session + + def create(self, video: GeneratedVideo) -> GeneratedVideo: + model = GeneratedVideoModel( + id=video.id, + workspace_id=video.workspace_id, + project_id=video.project_id, + generation_task_id=video.generation_task_id, + name=video.name, + file_url=video.file_url, + file_size=video.file_size, + duration=video.duration, + thumbnail_url=video.thumbnail_url, + width=video.width, + height=video.height, + fps=video.fps, + generated_at=video.generated_at, + created_at=video.created_at, + ) + self.session.add(model) + self.session.commit() + return video + + def get(self, video_id: str) -> GeneratedVideo | None: + model = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.id == video_id).first() + if model is None: + return None + return GeneratedVideo( + id=model.id, + workspace_id=model.workspace_id, + project_id=model.project_id, + generation_task_id=model.generation_task_id, + name=model.name, + file_url=model.file_url, + file_size=int(model.file_size or 0), + duration=model.duration, + thumbnail_url=model.thumbnail_url, + width=int(model.width or 0), + height=int(model.height or 0), + fps=model.fps, + generated_at=model.generated_at, + created_at=model.created_at, + ) + + def list_by_project(self, project_id: str) -> list[GeneratedVideo]: + models = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.project_id == project_id).all() + return [self.get(model.id) for model in models if self.get(model.id) is not None] + + def list_by_generation_task(self, generation_task_id: str) -> list[GeneratedVideo]: + models = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.generation_task_id == generation_task_id).all() + return [self.get(model.id) for model in models if self.get(model.id) is not None] diff --git a/packages/adapters/sqlalchemy_impl/generation_task_repository.py b/packages/adapters/sqlalchemy_impl/generation_task_repository.py new file mode 100644 index 000000000..bffa31d6d --- /dev/null +++ b/packages/adapters/sqlalchemy_impl/generation_task_repository.py @@ -0,0 +1,68 @@ +from sqlalchemy.orm import Session + +from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel +from packages.domain import GenerationTask, GenerationTaskStatus + + +class SQLAlchemyGenerationTaskRepository: + def __init__(self, session: Session): + self.session = session + + def create(self, task: GenerationTask) -> GenerationTask: + model = GenerationTaskModel( + id=task.id, + workspace_id=task.workspace_id, + project_id=task.project_id, + strategy_id=task.strategy_id, + asset_library_id=task.asset_library_id, + voice_library_id=task.voice_library_id, + status=task.status.value, + progress=task.progress, + result_count=task.result_count, + error_message=task.error_message, + started_at=task.started_at, + completed_at=task.completed_at, + created_by_user_id=task.created_by_user_id, + created_at=task.created_at, + ) + self.session.add(model) + self.session.commit() + return task + + def get(self, task_id: str) -> GenerationTask | None: + model = self.session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first() + if model is None: + return None + return GenerationTask( + id=model.id, + workspace_id=model.workspace_id, + project_id=model.project_id, + strategy_id=model.strategy_id, + asset_library_id=model.asset_library_id, + voice_library_id=model.voice_library_id, + status=GenerationTaskStatus(model.status), + progress=model.progress, + result_count=int(model.result_count or 0), + error_message=model.error_message, + started_at=model.started_at, + completed_at=model.completed_at, + created_by_user_id=model.created_by_user_id, + created_at=model.created_at, + ) + + def list_by_project(self, project_id: str) -> list[GenerationTask]: + models = self.session.query(GenerationTaskModel).filter(GenerationTaskModel.project_id == project_id).all() + return [self.get(model.id) for model in models if self.get(model.id) is not None] + + def update(self, task: GenerationTask) -> GenerationTask: + model = self.session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task.id).first() + if model is None: + raise ValueError(f"GenerationTask {task.id} not found") + model.status = task.status.value + model.progress = task.progress + model.result_count = task.result_count + model.error_message = task.error_message + model.started_at = task.started_at + model.completed_at = task.completed_at + self.session.commit() + return task diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index 50d95766b..378c3641f 100644 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -85,6 +85,44 @@ class ClassificationJobModel(Base): updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) +class GenerationTaskModel(Base): + __tablename__ = "generation_tasks" + + id = Column(String(32), primary_key=True) + workspace_id = Column(String(32), nullable=False, index=True) + project_id = Column(String(32), nullable=False, index=True) + strategy_id = Column(String(32), nullable=False, default="") + asset_library_id = Column(String(32), nullable=False, index=True) + voice_library_id = Column(String(32), nullable=False, default="") + status = Column(String(20), nullable=False, default="pending", index=True) + progress = Column(Float, nullable=False, default=0.0) + result_count = Column(Float, nullable=False, default=0) + error_message = Column(Text, nullable=False, default="") + started_at = Column(DateTime, nullable=True) + completed_at = Column(DateTime, nullable=True) + created_by_user_id = Column(String(32), nullable=False, default="") + created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + + +class GeneratedVideoModel(Base): + __tablename__ = "generated_videos" + + id = Column(String(32), primary_key=True) + workspace_id = Column(String(32), nullable=False, index=True) + project_id = Column(String(32), nullable=False, index=True) + generation_task_id = Column(String(32), nullable=False, index=True) + name = Column(String(255), nullable=False) + file_url = Column(String(1000), nullable=False) + file_size = Column(Float, nullable=False) + duration = Column(Float, nullable=False) + thumbnail_url = Column(String(1000), nullable=True) + width = Column(Float, nullable=False) + height = Column(Float, nullable=False) + fps = Column(Float, nullable=False) + generated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + + class TaskModel(Base): __tablename__ = "tasks" diff --git a/packages/application/__init__.py b/packages/application/__init__.py index 5859d4d49..d92ae1b8c 100644 --- a/packages/application/__init__.py +++ b/packages/application/__init__.py @@ -3,6 +3,8 @@ from .asset_libraries import CreateAssetLibraryCommand, CreateAssetLibraryUseCase, ListAssetLibrariesUseCase from .assets import CreateAssetCommand, CreateAssetUseCase, ListAssetsUseCase from .classification_jobs import SubmitClassificationJobCommand, SubmitClassificationJobUseCase +from .generated_videos import GetGeneratedVideoUseCase, ListGeneratedVideosByTaskUseCase, ListGeneratedVideosUseCase +from .generation_tasks import CreateGenerationTaskCommand, CreateGenerationTaskUseCase, GetGenerationTaskUseCase from .ingest_jobs import SubmitIngestJobCommand, SubmitIngestJobUseCase from .projects import CreateProjectCommand, CreateProjectUseCase, ListProjectsUseCase @@ -11,10 +13,16 @@ __all__ = [ "CreateAssetLibraryCommand", "CreateAssetLibraryUseCase", "CreateAssetUseCase", + "CreateGenerationTaskCommand", + "CreateGenerationTaskUseCase", "CreateProjectCommand", "CreateProjectUseCase", + "GetGeneratedVideoUseCase", + "GetGenerationTaskUseCase", "ListAssetLibrariesUseCase", "ListAssetsUseCase", + "ListGeneratedVideosByTaskUseCase", + "ListGeneratedVideosUseCase", "ListProjectsUseCase", "SubmitClassificationJobCommand", "SubmitClassificationJobUseCase", diff --git a/packages/application/assets.py b/packages/application/assets.py index 9aeab342e..bc4813124 100644 --- a/packages/application/assets.py +++ b/packages/application/assets.py @@ -2,7 +2,7 @@ from __future__ import annotations from dataclasses import dataclass -from packages.domain import Asset +from packages.domain import Asset, AssetStatus, ClassificationStatus from packages.ports.asset_repository import AssetRepository @@ -15,6 +15,17 @@ class CreateAssetCommand: storage_key: str mime_type: str metadata: dict[str, object] | None = None + file_size: int = 0 + thumbnail_url: str | None = None + duration: float | None = None + width: int | None = None + height: int | None = None + fps: float | None = None + codec: str | None = None + status: AssetStatus = AssetStatus.UPLOADING + classification_status: ClassificationStatus = ClassificationStatus.PENDING + quality_score: float | None = None + uploaded_by_user_id: str = "" class ListAssetsUseCase: @@ -40,5 +51,16 @@ class CreateAssetUseCase: storage_key=command.storage_key, mime_type=command.mime_type, metadata=command.metadata, + file_size=command.file_size, + thumbnail_url=command.thumbnail_url, + duration=command.duration, + width=command.width, + height=command.height, + fps=command.fps, + codec=command.codec, + status=command.status, + classification_status=command.classification_status, + quality_score=command.quality_score, + uploaded_by_user_id=command.uploaded_by_user_id, ) return self.asset_repository.create(asset) diff --git a/packages/application/generated_videos.py b/packages/application/generated_videos.py new file mode 100644 index 000000000..17e1a31ef --- /dev/null +++ b/packages/application/generated_videos.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from packages.domain import GeneratedVideo +from packages.ports.generated_video_repository import GeneratedVideoRepository + + +class ListGeneratedVideosUseCase: + def __init__(self, generated_video_repository: GeneratedVideoRepository): + self.generated_video_repository = generated_video_repository + + def execute(self, project_id: str) -> list[GeneratedVideo]: + if not project_id.strip(): + raise ValueError("project_id 不能为空") + return self.generated_video_repository.list_by_project(project_id.strip()) + + +class GetGeneratedVideoUseCase: + def __init__(self, generated_video_repository: GeneratedVideoRepository): + self.generated_video_repository = generated_video_repository + + def execute(self, video_id: str) -> GeneratedVideo | None: + return self.generated_video_repository.get(video_id) + + +class ListGeneratedVideosByTaskUseCase: + def __init__(self, generated_video_repository: GeneratedVideoRepository): + self.generated_video_repository = generated_video_repository + + def execute(self, generation_task_id: str) -> list[GeneratedVideo]: + if not generation_task_id.strip(): + raise ValueError("generation_task_id 不能为空") + return self.generated_video_repository.list_by_generation_task(generation_task_id.strip()) diff --git a/packages/application/generation_tasks.py b/packages/application/generation_tasks.py new file mode 100644 index 000000000..cea6e7594 --- /dev/null +++ b/packages/application/generation_tasks.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from dataclasses import dataclass + +from packages.domain import GenerationTask +from packages.ports.generation_task_repository import GenerationTaskRepository + + +@dataclass(slots=True) +class CreateGenerationTaskCommand: + workspace_id: str + project_id: str + asset_library_id: str + strategy_id: str = "" + voice_library_id: str = "" + created_by_user_id: str = "" + + +class CreateGenerationTaskUseCase: + def __init__(self, generation_task_repository: GenerationTaskRepository): + self.generation_task_repository = generation_task_repository + + def execute(self, command: CreateGenerationTaskCommand) -> GenerationTask: + task = GenerationTask.create( + workspace_id=command.workspace_id, + project_id=command.project_id, + asset_library_id=command.asset_library_id, + strategy_id=command.strategy_id, + voice_library_id=command.voice_library_id, + created_by_user_id=command.created_by_user_id, + ) + return self.generation_task_repository.create(task) + + +class GetGenerationTaskUseCase: + def __init__(self, generation_task_repository: GenerationTaskRepository): + self.generation_task_repository = generation_task_repository + + def execute(self, task_id: str) -> GenerationTask | None: + return self.generation_task_repository.get(task_id) diff --git a/packages/domain/__init__.py b/packages/domain/__init__.py index 7b081b6b1..e8a1613a3 100644 --- a/packages/domain/__init__.py +++ b/packages/domain/__init__.py @@ -1,7 +1,20 @@ """Domain package for core business entities and rules.""" from .classification import AssetClassification, ClassificationJob, ClassificationJobStatus -from .entities import Asset, AssetLibrary, AssetLibraryKind, IngestJob, IngestJobStatus, Project, User, Workspace +from .entities import ( + Asset, + AssetLibrary, + AssetLibraryKind, + AssetStatus, + ClassificationStatus, + IngestJob, + IngestJobStatus, + Project, + User, + Workspace, +) +from .generation_task import GenerationTask, GenerationTaskStatus +from .generated_video import GeneratedVideo from .project_management import Milestone, Task, TaskIssue, TaskPriority, TaskStatus __all__ = [ @@ -9,8 +22,13 @@ __all__ = [ "AssetClassification", "AssetLibrary", "AssetLibraryKind", + "AssetStatus", "ClassificationJob", "ClassificationJobStatus", + "ClassificationStatus", + "GeneratedVideo", + "GenerationTask", + "GenerationTaskStatus", "IngestJob", "IngestJobStatus", "Milestone", diff --git a/packages/domain/asset.py b/packages/domain/asset.py index cef4b903b..12e7e5665 100644 --- a/packages/domain/asset.py +++ b/packages/domain/asset.py @@ -1,105 +1,17 @@ -""" -Asset 实体 - 素材 -""" -from datetime import datetime -from typing import Optional -from enum import Enum +"""兼容层：旧素材领域入口，转发到当前主线实体。""" + +from packages.domain.entities import Asset, AssetStatus, ClassificationStatus -class AssetType(str, Enum): - """素材类型""" +class AssetType: VIDEO = "video" IMAGE = "image" AUDIO = "audio" -class AssetStatus(str, Enum): - """素材状态""" - UPLOADING = "uploading" - READY = "ready" - PROCESSING = "processing" - ERROR = "error" - - -class ClassificationStatus(str, Enum): - """分类状态""" - PENDING = "pending" - PROCESSING = "processing" - COMPLETED = "completed" - FAILED = "failed" - - -class Asset: - """素材实体""" - - def __init__( - self, - id: str, - workspace_id: str, - project_id: str, - asset_library_id: str, - name: str, - file_type: AssetType, - file_size: int, - file_url: str, - uploaded_by_user_id: str, - status: AssetStatus = AssetStatus.UPLOADING, - thumbnail_url: Optional[str] = None, - duration: Optional[float] = None, - width: Optional[int] = None, - height: Optional[int] = None, - fps: Optional[float] = None, - codec: Optional[str] = None, - classification_status: ClassificationStatus = ClassificationStatus.PENDING, - classification_result: Optional[dict] = None, - quality_score: Optional[float] = None, - created_at: Optional[datetime] = None, - updated_at: Optional[datetime] = None, - ): - self.id = id - self.workspace_id = workspace_id - self.project_id = project_id - self.asset_library_id = asset_library_id - self.name = name - self.file_type = file_type - self.file_size = file_size - self.file_url = file_url - self.uploaded_by_user_id = uploaded_by_user_id - self.status = status - self.thumbnail_url = thumbnail_url - self.duration = duration - self.width = width - self.height = height - self.fps = fps - self.codec = codec - self.classification_status = classification_status - self.classification_result = classification_result - self.quality_score = quality_score - self.created_at = created_at or datetime.utcnow() - self.updated_at = updated_at or datetime.utcnow() - - def to_dict(self) -> dict: - """转换为字典""" - return { - "id": self.id, - "workspace_id": self.workspace_id, - "project_id": self.project_id, - "asset_library_id": self.asset_library_id, - "name": self.name, - "file_type": self.file_type.value, - "file_size": self.file_size, - "file_url": self.file_url, - "thumbnail_url": self.thumbnail_url, - "duration": self.duration, - "width": self.width, - "height": self.height, - "fps": self.fps, - "codec": self.codec, - "status": self.status.value, - "classification_status": self.classification_status.value, - "classification_result": self.classification_result, - "quality_score": self.quality_score, - "uploaded_by_user_id": self.uploaded_by_user_id, - "created_at": self.created_at.isoformat() if self.created_at else None, - "updated_at": self.updated_at.isoformat() if self.updated_at else None, - } +__all__ = [ + "Asset", + "AssetStatus", + "AssetType", + "ClassificationStatus", +] diff --git a/packages/domain/asset_library.py b/packages/domain/asset_library.py index fbcabca88..d8deab27f 100644 --- a/packages/domain/asset_library.py +++ b/packages/domain/asset_library.py @@ -1,52 +1,16 @@ -""" -AssetLibrary 实体 - 素材库 -""" -from datetime import datetime -from enum import Enum +"""兼容层：旧素材库领域入口，转发到当前主线实体。""" + +from packages.domain.entities import AssetLibrary, AssetLibraryKind -class LibraryKind(str, Enum): - """素材库类型""" - VIDEO = "video" - VOICE = "voice" - IMAGE = "image" +class LibraryKind: + VIDEO = AssetLibraryKind.VIDEO + VOICE = AssetLibraryKind.VOICE + IMAGE = AssetLibraryKind.IMAGE -class AssetLibrary: - """素材库实体""" - - def __init__( - self, - id: str, - workspace_id: str, - name: str, - kind: LibraryKind, - project_id: str = None, - asset_count: int = 0, - total_size: int = 0, - created_at: datetime = None, - updated_at: datetime = None, - ): - self.id = id - self.workspace_id = workspace_id - self.project_id = project_id - self.name = name - self.kind = kind - self.asset_count = asset_count - self.total_size = total_size - self.created_at = created_at or datetime.utcnow() - self.updated_at = updated_at or datetime.utcnow() - - def to_dict(self) -> dict: - """转换为字典""" - return { - "id": self.id, - "workspace_id": self.workspace_id, - "project_id": self.project_id, - "name": self.name, - "kind": self.kind.value, - "asset_count": self.asset_count, - "total_size": self.total_size, - "created_at": self.created_at.isoformat() if self.created_at else None, - "updated_at": self.updated_at.isoformat() if self.updated_at else None, - } +__all__ = [ + "AssetLibrary", + "AssetLibraryKind", + "LibraryKind", +] diff --git a/packages/domain/entities.py b/packages/domain/entities.py index 85076ccf9..b56ff8d47 100644 --- a/packages/domain/entities.py +++ b/packages/domain/entities.py @@ -10,6 +10,7 @@ from uuid import uuid4 class AssetLibraryKind(StrEnum): VIDEO = "video" VOICE = "voice" + IMAGE = "image" class IngestJobStatus(StrEnum): @@ -122,7 +123,10 @@ class AssetLibrary: project_id: str name: str kind: AssetLibraryKind + asset_count: int = 0 + total_size: int = 0 created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @classmethod def create( @@ -141,9 +145,25 @@ class AssetLibrary: project_id=project_id, name=clean_name, kind=kind, + asset_count=0, + total_size=0, ) +class AssetStatus(StrEnum): + UPLOADING = "uploading" + READY = "ready" + PROCESSING = "processing" + ERROR = "error" + + +class ClassificationStatus(StrEnum): + PENDING = "pending" + PROCESSING = "processing" + COMPLETED = "completed" + FAILED = "failed" + + @dataclass(slots=True) class Asset: id: str @@ -153,9 +173,21 @@ class Asset: name: str storage_key: str mime_type: str + file_size: int = 0 + thumbnail_url: str | None = None + duration: float | None = None + width: int | None = None + height: int | None = None + fps: float | None = None + codec: str | None = None + status: AssetStatus = AssetStatus.UPLOADING + classification_status: ClassificationStatus = ClassificationStatus.PENDING + quality_score: float | None = None + uploaded_by_user_id: str = "" metadata: dict[str, Any] = field(default_factory=dict) tags: list[str] = field(default_factory=list) created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @classmethod def create( @@ -167,6 +199,18 @@ class Asset: storage_key: str, mime_type: str, metadata: dict[str, Any] | None = None, + *, + file_size: int = 0, + thumbnail_url: str | None = None, + duration: float | None = None, + width: int | None = None, + height: int | None = None, + fps: float | None = None, + codec: str | None = None, + status: AssetStatus = AssetStatus.UPLOADING, + classification_status: ClassificationStatus = ClassificationStatus.PENDING, + quality_score: float | None = None, + uploaded_by_user_id: str = "", ) -> "Asset": clean_name = name.strip() if not clean_name: @@ -183,6 +227,17 @@ class Asset: name=clean_name, storage_key=storage_key.strip(), mime_type=mime_type.strip(), + file_size=file_size, + thumbnail_url=thumbnail_url, + duration=duration, + width=width, + height=height, + fps=fps, + codec=codec, + status=status, + classification_status=classification_status, + quality_score=quality_score, + uploaded_by_user_id=uploaded_by_user_id.strip(), metadata=metadata or {}, tags=[], ) @@ -194,12 +249,14 @@ class Asset: raise ValueError("标签不能为空") if clean_tag not in self.tags: self.tags.append(clean_tag) + self.updated_at = datetime.now(timezone.utc) def remove_tag(self, tag: str) -> None: """删除标签。如果标签不存在，不报错（幂等性）。""" clean_tag = tag.strip() if clean_tag in self.tags: self.tags.remove(clean_tag) + self.updated_at = datetime.now(timezone.utc) @dataclass(slots=True) diff --git a/packages/domain/generated_video.py b/packages/domain/generated_video.py new file mode 100644 index 000000000..68f28ad0d --- /dev/null +++ b/packages/domain/generated_video.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from uuid import uuid4 + + +@dataclass(slots=True) +class GeneratedVideo: + id: str + workspace_id: str + project_id: str + generation_task_id: str + name: str + file_url: str + file_size: int + duration: float + width: int + height: int + fps: float + thumbnail_url: str | None = None + generated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + + @classmethod + def create( + cls, + workspace_id: str, + project_id: str, + generation_task_id: str, + name: str, + file_url: str, + *, + file_size: int, + duration: float, + width: int, + height: int, + fps: float, + thumbnail_url: str | None = None, + ) -> "GeneratedVideo": + if not workspace_id.strip(): + raise ValueError("workspace_id 不能为空") + if not project_id.strip(): + raise ValueError("project_id 不能为空") + if not generation_task_id.strip(): + raise ValueError("generation_task_id 不能为空") + if not name.strip(): + raise ValueError("name 不能为空") + if not file_url.strip(): + raise ValueError("file_url 不能为空") + return cls( + id=uuid4().hex, + workspace_id=workspace_id.strip(), + project_id=project_id.strip(), + generation_task_id=generation_task_id.strip(), + name=name.strip(), + file_url=file_url.strip(), + file_size=file_size, + duration=duration, + width=width, + height=height, + fps=fps, + thumbnail_url=thumbnail_url, + ) diff --git a/packages/domain/generation_task.py b/packages/domain/generation_task.py new file mode 100644 index 000000000..d9710e689 --- /dev/null +++ b/packages/domain/generation_task.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import StrEnum +from uuid import uuid4 + + +class GenerationTaskStatus(StrEnum): + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + CANCELLED = "cancelled" + + +@dataclass(slots=True) +class GenerationTask: + id: str + workspace_id: str + project_id: str + asset_library_id: str + strategy_id: str = "" + voice_library_id: str = "" + status: GenerationTaskStatus = GenerationTaskStatus.PENDING + progress: float = 0.0 + result_count: int = 0 + error_message: str = "" + started_at: datetime | None = None + completed_at: datetime | None = None + created_by_user_id: str = "" + created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + + @classmethod + def create( + cls, + workspace_id: str, + project_id: str, + asset_library_id: str, + *, + strategy_id: str = "", + voice_library_id: str = "", + created_by_user_id: str = "", + ) -> "GenerationTask": + if not workspace_id.strip(): + raise ValueError("workspace_id 不能为空") + if not project_id.strip(): + raise ValueError("project_id 不能为空") + if not asset_library_id.strip(): + raise ValueError("asset_library_id 不能为空") + return cls( + id=uuid4().hex, + workspace_id=workspace_id.strip(), + project_id=project_id.strip(), + asset_library_id=asset_library_id.strip(), + strategy_id=strategy_id.strip(), + voice_library_id=voice_library_id.strip(), + created_by_user_id=created_by_user_id.strip(), + ) diff --git a/packages/ports/asset_library_repository.py b/packages/ports/asset_library_repository.py index 934b7ce5c..83dc95a8f 100644 --- a/packages/ports/asset_library_repository.py +++ b/packages/ports/asset_library_repository.py @@ -1,31 +1,25 @@ -""" -AssetLibrary Repository 接口 -""" +"""兼容层：旧素材库仓储接口定义，保留给遗留异步适配器使用。""" + from abc import ABC, abstractmethod -from typing import List, Optional -from packages.domain.asset_library import AssetLibrary, LibraryKind + +from packages.domain import AssetLibrary, AssetLibraryKind class AssetLibraryRepository(ABC): - """素材库仓储接口""" - @abstractmethod async def create(self, library: AssetLibrary) -> AssetLibrary: - """创建素材库""" pass @abstractmethod - async def find_by_id(self, library_id: str) -> Optional[AssetLibrary]: - """根据 ID 查询素材库""" + async def find_by_id(self, library_id: str) -> AssetLibrary | None: pass @abstractmethod async def find_by_workspace( self, workspace_id: str, - kind: Optional[LibraryKind] = None, - ) -> List[AssetLibrary]: - """根据工作空间查询素材库""" + kind: AssetLibraryKind | None = None, + ) -> list[AssetLibrary]: pass @abstractmethod @@ -33,26 +27,21 @@ class AssetLibraryRepository(ABC): self, project_id: str, workspace_id: str, - ) -> List[AssetLibrary]: - """根据项目查询素材库""" + ) -> list[AssetLibrary]: pass @abstractmethod async def update(self, library: AssetLibrary) -> AssetLibrary: - """更新素材库""" pass @abstractmethod async def delete(self, library_id: str, workspace_id: str) -> bool: - """删除素材库""" pass @abstractmethod async def increment_asset_count(self, library_id: str, size_delta: int) -> None: - """增加素材数量和大小""" pass @abstractmethod async def decrement_asset_count(self, library_id: str, size_delta: int) -> None: - """减少素材数量和大小""" pass diff --git a/packages/ports/asset_repository.py b/packages/ports/asset_repository.py index e46bbb3a2..47bad8c03 100644 --- a/packages/ports/asset_repository.py +++ b/packages/ports/asset_repository.py @@ -1,33 +1,27 @@ -""" -Asset Repository 接口 -""" +"""兼容层：旧素材仓储接口定义，保留给遗留异步适配器使用。""" + from abc import ABC, abstractmethod -from typing import List, Optional -from packages.domain.asset import Asset + +from packages.domain import Asset class AssetRepository(ABC): - """素材仓储接口""" - @abstractmethod async def create(self, asset: Asset) -> Asset: - """创建素材""" pass @abstractmethod - async def find_by_id(self, asset_id: str) -> Optional[Asset]: - """根据 ID 查询素材""" + async def find_by_id(self, asset_id: str) -> Asset | None: pass @abstractmethod async def find_by_project( - self, + self, project_id: str, workspace_id: str, skip: int = 0, limit: int = 100, - ) -> List[Asset]: - """根据项目查询素材列表""" + ) -> list[Asset]: pass @abstractmethod @@ -37,21 +31,17 @@ class AssetRepository(ABC): workspace_id: str, skip: int = 0, limit: int = 100, - ) -> List[Asset]: - """根据素材库查询素材列表""" + ) -> list[Asset]: pass @abstractmethod async def update(self, asset: Asset) -> Asset: - """更新素材""" pass @abstractmethod async def delete(self, asset_id: str, workspace_id: str) -> bool: - """删除素材""" pass @abstractmethod async def count_by_project(self, project_id: str, workspace_id: str) -> int: - """统计项目素材数量""" pass diff --git a/packages/ports/generated_video_repository.py b/packages/ports/generated_video_repository.py new file mode 100644 index 000000000..f062f878d --- /dev/null +++ b/packages/ports/generated_video_repository.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from typing import Protocol + +from packages.domain import GeneratedVideo + + +class GeneratedVideoRepository(Protocol): + def create(self, video: GeneratedVideo) -> GeneratedVideo: + ... + + def get(self, video_id: str) -> GeneratedVideo | None: + ... + + def list_by_project(self, project_id: str) -> list[GeneratedVideo]: + ... + + def list_by_generation_task(self, generation_task_id: str) -> list[GeneratedVideo]: + ... diff --git a/packages/ports/generation_task_repository.py b/packages/ports/generation_task_repository.py new file mode 100644 index 000000000..84cc925a4 --- /dev/null +++ b/packages/ports/generation_task_repository.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from typing import Protocol + +from packages.domain import GenerationTask + + +class GenerationTaskRepository(Protocol): + def create(self, task: GenerationTask) -> GenerationTask: + ... + + def get(self, task_id: str) -> GenerationTask | None: + ... + + def list_by_project(self, project_id: str) -> list[GenerationTask]: + ... + + def update(self, task: GenerationTask) -> GenerationTask: + ... diff --git a/tests/integration/test_generation_pipeline.py b/tests/integration/test_generation_pipeline.py new file mode 100644 index 000000000..d511b0a4b --- /dev/null +++ b/tests/integration/test_generation_pipeline.py @@ -0,0 +1,41 @@ +from packages.application import CreateGenerationTaskCommand, CreateGenerationTaskUseCase +from packages.domain import GenerationTaskStatus + + +class DummyGenerationTaskRepository: + def __init__(self): + self.items = {} + + def create(self, task): + self.items[task.id] = task + return task + + def get(self, task_id): + return self.items.get(task_id) + + def list_by_project(self, project_id): + return [task for task in self.items.values() if task.project_id == project_id] + + def update(self, task): + self.items[task.id] = task + return task + + +def test_create_generation_task_smoke(): + repo = DummyGenerationTaskRepository() + use_case = CreateGenerationTaskUseCase(repo) + task = use_case.execute( + CreateGenerationTaskCommand( + workspace_id="ws-1", + project_id="proj-1", + asset_library_id="lib-1", + strategy_id="str-1", + voice_library_id="voice-1", + created_by_user_id="user-1", + ) + ) + assert task.workspace_id == "ws-1" + assert task.project_id == "proj-1" + assert task.asset_library_id == "lib-1" + assert task.status == GenerationTaskStatus.PENDING + assert repo.get(task.id) is not None