xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 16 Pull Requests 11 Actions 34.2k Packages Projects Releases 7 Wiki Activity

feat(phase7): align asset flow and scaffold generation pipeline

Browse Source
This commit is contained in:
Xiaoxia AI
2026-06-18 19:59:37 +08:00
parent 9c3fb6d4ff
commit 758c2c7e13
37 changed files with 1202 additions and 543 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/app/api/router.py
+12
View File
@@ -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",
apps/api/app/api/routes/asset_libraries.py
+14 -19
View File
@@ -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)
apps/api/app/api/routes/assets.py
+38 -25
View File
@@ -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)
apps/api/app/api/routes/generated_videos.py
+61
View File
@@ -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,
)
apps/api/app/api/routes/generation_tasks.py
+95
View File
@@ -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
]
)
apps/api/app/dependencies.py
+10
View File
@@ -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)
apps/api/app/schemas/asset.py
+22
View File
@@ -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):
apps/api/app/schemas/asset_library.py
+3 -1
View File
@@ -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):
apps/api/app/schemas/generated_video.py
+20
View File
@@ -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]
apps/api/app/schemas/generation_task.py
+23
View File
@@ -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
apps/worker/tasks.py
+40 -146
View File
@@ -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()
docs/PHASE7-PROGRESS.md
+153
View File
@@ -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 验证阶段。
packages/adapters/in_memory/asset_library_repository.py
+16 -46
View File
@@ -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)
packages/adapters/in_memory/asset_repository.py
+13 -51
View File
@@ -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
])
packages/adapters/postgres/__init__.py
+6 -4
View File
@@ -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",
packages/adapters/postgres/asset_repository.py
+44 -56
View File
@@ -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,
)
packages/adapters/sqlalchemy_impl/__init__.py
+4
View File
@@ -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",
packages/adapters/sqlalchemy_impl/asset_library_repository.py
+6
View File
@@ -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()
packages/adapters/sqlalchemy_impl/asset_repository.py
+38 -10
View File
@@ -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,
)
packages/adapters/sqlalchemy_impl/generated_video_repository.py
+59
View File
@@ -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]
packages/adapters/sqlalchemy_impl/generation_task_repository.py
+68
View File
@@ -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
packages/adapters/sqlalchemy_impl/models.py
+38
View File
@@ -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"
packages/application/__init__.py
+8
View File
@@ -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",
packages/application/assets.py
+23 -1
View File
@@ -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)
packages/application/generated_videos.py
+32
View File
@@ -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())
packages/application/generation_tasks.py
+40
View File
@@ -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)
packages/domain/__init__.py
+19 -1
View File
@@ -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",
packages/domain/asset.py
+10 -98
View File
@@ -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",
]
packages/domain/asset_library.py
+12 -48
View File
@@ -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",
]
packages/domain/entities.py
+57
View File
@@ -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)
packages/domain/generated_video.py
+64
View File
@@ -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,
)
packages/domain/generation_task.py
+59
View File
@@ -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(),
)
packages/ports/asset_library_repository.py
+8 -19
View File
@@ -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
packages/ports/asset_repository.py
+8 -18
View File
@@ -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
packages/ports/generated_video_repository.py
+19
View File
@@ -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]:
...
packages/ports/generation_task_repository.py
+19
View File
@@ -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:
...
tests/integration/test_generation_pipeline.py
+41
View File
@@ -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