819 lines
48 KiB
Python
819 lines
48 KiB
Python
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
||
|
||
from app.schemas.common import NaiveDatetimeOptional
|
||
|
||
SHOT_REPLICATE_PROJECT_STATUS_DESCRIPTIONS: dict[str, str] = {
|
||
"pending": "已创建但未进入流程",
|
||
"waiting_user": "等待用户手动触发下一步",
|
||
"processing": "当前有步骤处理中",
|
||
"completed": "总任务完成",
|
||
"failed": "总任务失败",
|
||
"cancelled": "总任务取消",
|
||
}
|
||
|
||
SHOT_REPLICATE_STEP_STATUS_DESCRIPTIONS: dict[str, str] = {
|
||
"pending": "子任务待处理",
|
||
"waiting_user": "等待用户确认或触发",
|
||
"processing": "子任务处理中",
|
||
"completed": "子任务完成",
|
||
"failed": "子任务失败",
|
||
"cancelled": "子任务取消",
|
||
}
|
||
|
||
SHOT_REPLICATE_STEP_DESCRIPTIONS: list[dict[str, Any]] = [
|
||
{"step_index": 1, "step_code": "material_input", "name": "素材输入"},
|
||
{"step_index": 2, "step_code": "image_prompt_optimize", "name": "图片 AI 提词"},
|
||
{"step_index": 3, "step_code": "image_generate", "name": "图片生成"},
|
||
{"step_index": 4, "step_code": "video_prompt_optimize", "name": "视频 AI 提词 JSON schema"},
|
||
{"step_index": 5, "step_code": "video_generate", "name": "视频生成"},
|
||
]
|
||
|
||
SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION = "shot_replicate_step_io_v1"
|
||
|
||
SHOT_REPLICATE_STEP_IO_EXAMPLES: dict[str, dict[str, Any]] = {
|
||
"material_input": {
|
||
"input_json": {
|
||
"schema_version": SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION,
|
||
"step_code": "material_input",
|
||
"source": {"source_step_id": None, "parent_step_id": None},
|
||
"payload": {
|
||
"material_video_url": "https://example.com/source.mp4",
|
||
"material_image_url": "https://example.com/product.png",
|
||
"source_project_name": "参考素材项目名称",
|
||
"target_project_name": "新项目名称",
|
||
"core_content_point": "50字以内核心内容点",
|
||
},
|
||
"context": {},
|
||
},
|
||
"output_json": {
|
||
"schema_version": SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION,
|
||
"step_code": "material_input",
|
||
"status": "completed",
|
||
"payload": {},
|
||
"result": {"accepted": True, "message": "素材输入已提交", "next_step_code": "image_prompt_optimize"},
|
||
"usage": {},
|
||
"error": {},
|
||
},
|
||
},
|
||
"image_prompt_optimize": {
|
||
"input_json": {
|
||
"schema_version": SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION,
|
||
"step_code": "image_prompt_optimize",
|
||
"source": {"source_step_id": "第1步素材输入ID", "parent_step_id": "第1步素材输入ID"},
|
||
"payload": {"source_step_id": "第1步素材输入ID"},
|
||
"context": {},
|
||
},
|
||
"output_json": {
|
||
"schema_version": SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION,
|
||
"step_code": "image_prompt_optimize",
|
||
"status": "completed",
|
||
"payload": {
|
||
"optimized_prompt": "图片生成提示词",
|
||
"prompt": "兼容字段，同 optimized_prompt",
|
||
"original_prompt": "后端拼接的图片提词原始需求",
|
||
"references": [{"type": "video|image", "url": "...", "name": "..."}],
|
||
},
|
||
"result": {},
|
||
"usage": {
|
||
"input_tokens": 0,
|
||
"output_tokens": 0,
|
||
"total_tokens": 0,
|
||
"text_credits_cost": 0,
|
||
"credit_biz_key": "module_generation_step:{step_id}:attempt:1:text_prompt:charge",
|
||
},
|
||
"error": {},
|
||
},
|
||
},
|
||
"image_generate": {
|
||
"input_json": {
|
||
"schema_version": SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION,
|
||
"step_code": "image_generate",
|
||
"source": {"source_step_id": "第2步图片提词ID", "parent_step_id": "第2步图片提词ID"},
|
||
"payload": {
|
||
"engine_id": "图片引擎ID",
|
||
"params": {"image_size": "2K", "image_proportion": "1:1", "image_px": "2048x2048"},
|
||
"prompt": "图片生成提示词",
|
||
"media_references": [{"type": "image", "url": "新产品图片", "name": "新产品图片"}],
|
||
},
|
||
"context": {},
|
||
},
|
||
"output_json": {
|
||
"schema_version": SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION,
|
||
"step_code": "image_generate",
|
||
"status": "completed",
|
||
"payload": {},
|
||
"result": {"result_image_url": "/generate/images/xxx.png", "chat_task_id": "ChatGenerationTask ID"},
|
||
"usage": {},
|
||
"error": {},
|
||
},
|
||
},
|
||
"video_prompt_optimize": {
|
||
"input_json": {
|
||
"schema_version": SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION,
|
||
"step_code": "video_prompt_optimize",
|
||
"source": {"source_step_id": "第3步图片生成ID", "parent_step_id": "第3步图片生成ID"},
|
||
"payload": {
|
||
"source_step_id": "第3步图片生成ID",
|
||
"video_config": {"engine_id": "视频引擎ID", "duration": 8, "aspect_ratio": "9:16", "resolution": "1080p"},
|
||
"target_platform": "抖音",
|
||
},
|
||
"context": {},
|
||
},
|
||
"output_json": {
|
||
"schema_version": SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION,
|
||
"step_code": "video_prompt_optimize",
|
||
"status": "completed",
|
||
"payload": {
|
||
"prompt_schema": {"任务基础信息": {}, "最终提示词": {}},
|
||
"final_prompt": "展示用最终视频提示词",
|
||
"params_used_for_prompt": {"duration": 8, "aspect_ratio": "9:16", "resolution": "1080p"},
|
||
"target_platform": "抖音",
|
||
},
|
||
"result": {},
|
||
"usage": {
|
||
"input_tokens": 0,
|
||
"output_tokens": 0,
|
||
"total_tokens": 0,
|
||
"text_credits_cost": 0,
|
||
"credit_biz_key": "module_generation_step:{step_id}:attempt:1:text_prompt:charge",
|
||
},
|
||
"error": {},
|
||
},
|
||
},
|
||
"video_generate": {
|
||
"input_json": {
|
||
"schema_version": SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION,
|
||
"step_code": "video_generate",
|
||
"source": {"source_step_id": "第4步视频提词ID", "parent_step_id": "第4步视频提词ID"},
|
||
"payload": {
|
||
"engine_id": "视频引擎ID",
|
||
"params": {"duration": 8, "aspect_ratio": "9:16", "resolution": "1080p"},
|
||
"prompt_schema": {"任务基础信息": {}, "最终提示词": {}},
|
||
"final_prompt": "展示用最终提示词",
|
||
"media_references": [{"type": "image", "url": "第3步生成图片", "name": "新项目图片"}],
|
||
},
|
||
"context": {},
|
||
},
|
||
"output_json": {
|
||
"schema_version": SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION,
|
||
"step_code": "video_generate",
|
||
"status": "completed",
|
||
"payload": {},
|
||
"result": {
|
||
"result_video_url": "/generate/videos/xxx.mp4",
|
||
"result_video_cover_url": "/generate/covers/xxx.jpg",
|
||
"chat_task_id": "ChatGenerationTask ID",
|
||
},
|
||
"usage": {},
|
||
"error": {},
|
||
},
|
||
},
|
||
}
|
||
|
||
|
||
class ShotReplicateTaskCreate(BaseModel):
|
||
"""创建拆镜复刻总任务项目请求体。"""
|
||
|
||
model_config = ConfigDict(
|
||
json_schema_extra={
|
||
"example": {
|
||
"material_video_url": "https://example.com/source.mp4",
|
||
"material_image_url": "https://example.com/product.png",
|
||
"source_project_name": "参考素材项目名称",
|
||
"target_project_name": "新项目名称",
|
||
"core_content_point": "突出产品能帮助用户认识附近新朋友",
|
||
"idempotency_key": "frontend-submit-uuid-001",
|
||
}
|
||
}
|
||
)
|
||
|
||
material_video_url: str = Field(..., min_length=1, description="素材视频链接，参考素材，1份。由项目已有上传接口返回，本接口不负责上传，不做后端素材校验")
|
||
material_image_url: str = Field(..., min_length=1, description="素材图片链接，新产品图片，1份。由项目已有上传接口返回，本接口不负责上传，不做后端素材校验")
|
||
source_project_name: str = Field(..., min_length=1, max_length=20, description="视频素材内容项目名称")
|
||
target_project_name: str = Field(..., min_length=1, max_length=20, description="生成项目名称")
|
||
core_content_point: str = Field(..., min_length=1, max_length=50, description="生成的项目核心内容点，最多50字")
|
||
idempotency_key: str | None = Field(None, max_length=64, description="创建总任务幂等键。只用于 module_generation_projects，不用于 ChatGenerationTask")
|
||
|
||
@field_validator("material_video_url", "material_image_url", "source_project_name", "target_project_name", "core_content_point")
|
||
@classmethod
|
||
def _strip_required(cls, value: str) -> str:
|
||
value = str(value or "").strip()
|
||
if not value:
|
||
raise ValueError("字段不能为空")
|
||
return value
|
||
|
||
|
||
class ShotReplicateMaterialUpdateRequest(BaseModel):
|
||
"""修改拆镜复刻第1步素材输入请求体。
|
||
|
||
拆镜复刻的素材视频来自拆镜片段 segment_video_url，与片段强绑定，不允许修改。
|
||
本接口只允许修改新产品图片、参考素材项目名、生成项目名和核心内容点。
|
||
"""
|
||
|
||
model_config = ConfigDict(
|
||
extra="forbid",
|
||
json_schema_extra={
|
||
"example": {
|
||
"material_image_url": "https://example.com/new-product.png",
|
||
"source_project_name": "新的参考素材项目名称",
|
||
"target_project_name": "新的生成项目名称",
|
||
"core_content_point": "新的50字以内核心内容点",
|
||
}
|
||
},
|
||
)
|
||
|
||
material_image_url: str | None = Field(None, min_length=1, description="素材图片链接，未传则沿用旧值。用于新产品/目标素材图片，来自已有上传接口")
|
||
material_image_resource_id: str | None = Field(None, max_length=32, description="可选，拆镜模块上传接口返回的图片 UploadResource.id")
|
||
source_project_name: str | None = Field(None, min_length=1, max_length=20, description="视频素材内容项目名称，未传则沿用旧值")
|
||
target_project_name: str | None = Field(None, min_length=1, max_length=20, description="生成项目名称，未传则沿用旧值")
|
||
core_content_point: str | None = Field(None, min_length=1, max_length=50, description="生成项目核心内容点，最多50字，未传则沿用旧值")
|
||
|
||
@field_validator("material_image_url", "source_project_name", "target_project_name", "core_content_point", mode="before")
|
||
@classmethod
|
||
def _strip_optional(cls, value: str | None) -> str | None:
|
||
if value is None:
|
||
return None
|
||
value = str(value).strip()
|
||
if not value:
|
||
raise ValueError("字段不能为空字符串")
|
||
return value
|
||
|
||
@model_validator(mode="after")
|
||
def _require_at_least_one(self) -> "ShotReplicateMaterialUpdateRequest":
|
||
if not any(getattr(self, field) is not None for field in ("material_image_url", "source_project_name", "target_project_name", "core_content_point", "material_image_resource_id")):
|
||
raise ValueError("至少需要传入一个需要修改的字段")
|
||
return self
|
||
|
||
|
||
class ShotReplicateStepUpdate(BaseModel):
|
||
"""修改拆镜复刻子任务请求体。"""
|
||
|
||
material_video_url: str | None = Field(None, description="历史兼容字段：拆镜复刻项目素材视频与片段绑定，正式接口不允许修改")
|
||
material_image_url: str | None = Field(None, description="修改第1步素材图片链接")
|
||
source_project_name: str | None = Field(None, max_length=20, description="修改第1步视频素材内容项目名称")
|
||
target_project_name: str | None = Field(None, max_length=20, description="修改第1步生成项目名称")
|
||
core_content_point: str | None = Field(None, max_length=50, description="修改第1步生成项目核心内容点，最多50字")
|
||
prompt: str | None = Field(None, description="修改第2步图片提词或第4步视频最终提词")
|
||
prompt_schema: dict[str, Any] | None = Field(None, description="修改第4步视频提词 JSON schema。只对视频提词步骤有意义")
|
||
input_json: dict[str, Any] | None = Field(None, description="高级用法：合并修改当前步骤 input_json.payload")
|
||
output_json: dict[str, Any] | None = Field(None, description="高级用法：合并修改当前步骤 output_json.payload")
|
||
|
||
|
||
class ShotReplicateImagePromptUpdateRequest(BaseModel):
|
||
"""直接修改第2步图片 AI 优化提词请求体。
|
||
|
||
本接口不调用 AI、不扣积分；保存后会软删除第3、4、5步当前有效任务。
|
||
"""
|
||
|
||
model_config = ConfigDict(
|
||
extra="ignore",
|
||
json_schema_extra={"example": {"prompt": "用户手动修改后的图片生成提示词"}},
|
||
)
|
||
|
||
prompt: str = Field(..., min_length=1, description="用户手动修改后的图片生成提示词，不能为空")
|
||
|
||
@field_validator("prompt", mode="before")
|
||
@classmethod
|
||
def _strip_prompt(cls, value: str) -> str:
|
||
value = str(value or "").strip()
|
||
if not value:
|
||
raise ValueError("图片提示词不能为空")
|
||
return value
|
||
|
||
|
||
class ShotReplicateVideoPromptSchemaUpdateRequest(BaseModel):
|
||
"""修改第4步视频 AI 提词 JSON schema 请求体。
|
||
|
||
前端提交的 prompt_schema 只作为 patch：服务端会锁定视频时长、比例、清晰度、帧率、推荐分辨率、
|
||
动作/镜头/动态时间规划数组长度和时间段、输出规格限制、质量控制、合规控制、schema_version、schema_usage。
|
||
最终提示词允许修改，但保存前会清洗视频参数。
|
||
"""
|
||
|
||
model_config = ConfigDict(
|
||
extra="ignore",
|
||
json_schema_extra={
|
||
"example": {
|
||
"prompt_schema": {
|
||
"业务属性": {"产品名称": "脱单交友APP", "行动引导": "立即下载"},
|
||
"最终提示词": {"主提示词": "脱单交友APP推广短视频，突出认识附近新朋友和高效匹配"},
|
||
}
|
||
}
|
||
},
|
||
)
|
||
|
||
prompt_schema: dict[str, Any] = Field(..., description="前端修改后的视频提词 JSON schema。后端只按白名单回填允许修改字段")
|
||
|
||
@field_validator("prompt_schema")
|
||
@classmethod
|
||
def _validate_schema(cls, value: dict[str, Any]) -> dict[str, Any]:
|
||
if not isinstance(value, dict) or not value:
|
||
raise ValueError("prompt_schema 必须是非空 JSON 对象")
|
||
return value
|
||
|
||
|
||
class ShotReplicateGenerateImagePromptRequest(BaseModel):
|
||
"""手动生成第2步图片 AI 提词请求体。当前无需请求参数。"""
|
||
|
||
model_config = ConfigDict(extra="ignore")
|
||
|
||
|
||
class ShotReplicateGenerateImageRequest(BaseModel):
|
||
"""根据图片提词生成新项目图片请求体。
|
||
|
||
ChatGenerationTask.idempotency_key 由后端自动生成，接口不再接收前端幂等键。
|
||
"""
|
||
|
||
model_config = ConfigDict(
|
||
extra="ignore",
|
||
json_schema_extra={"example": {"engine_id": "image_engine_xxx", "image_size": "2K", "image_proportion": "1:1", "image_px": "2048x2048"}},
|
||
)
|
||
|
||
engine_id: str | None = Field(None, description="图片生成引擎ID。为空则使用当前启用且优先级最高的图片引擎")
|
||
image_size: str | None = Field(None, description="图片分辨率档位，例如 1K、2K。具体可选值来自图片引擎配置接口；为空使用引擎默认值")
|
||
image_proportion: str | None = Field(None, description="图片比例，例如 1:1、16:9、9:16。具体可选值来自图片引擎配置接口；为空使用默认值")
|
||
image_px: str | None = Field(None, description="图片像素尺寸，例如 1024x1024、2048x2048。具体可选值来自图片引擎配置接口；为空时按引擎支持尺寸自动匹配")
|
||
|
||
|
||
class ShotReplicateGenerateVideoPromptRequest(BaseModel):
|
||
"""手动生成第4步视频 AI 提词请求体。
|
||
|
||
视频时长、比例、分辨率集中在本步骤确定；第5步生成视频只选择视频引擎。
|
||
"""
|
||
|
||
model_config = ConfigDict(
|
||
extra="ignore",
|
||
json_schema_extra={"example": {"engine_id": "video_engine_xxx", "duration": 8, "aspect_ratio": "9:16", "resolution": "1080p", "target_platform": "抖音"}},
|
||
)
|
||
|
||
engine_id: str | None = Field(None, description="视频引擎ID。用于读取该引擎支持的视频时长、比例、分辨率配置；为空使用最高优先级启用引擎")
|
||
duration: int | None = Field(None, ge=1, description="希望用于视频提词规划的视频时长，单位秒。具体可选值来自视频引擎 supported_durations；为空时优先使用 SHOT_REPLICATE_DEFAULT_VIDEO_DURATION")
|
||
aspect_ratio: str | None = Field(None, description="希望用于视频提词规划的视频比例，例如 9:16、16:9、1:1。具体可选值来自视频引擎 supported_ratios；为空时优先使用 SHOT_REPLICATE_DEFAULT_VIDEO_RATIO")
|
||
resolution: str | None = Field(None, description="希望用于视频提词规划的视频分辨率，例如 480p、720p、1080p。具体可选值来自视频引擎 supported_resolutions；为空时优先使用 SHOT_REPLICATE_DEFAULT_VIDEO_RESOLUTION")
|
||
target_platform: str | None = Field(None, max_length=64, description="目标平台，例如抖音/快手/小红书。为空时使用 SHOT_REPLICATE_DEFAULT_TARGET_PLATFORM")
|
||
|
||
|
||
class ShotReplicateGenerateVideoRequest(BaseModel):
|
||
"""根据视频提词生成最终视频请求体。
|
||
|
||
只选择视频生成引擎。duration / aspect_ratio / resolution 从第4步视频提词优化结果读取。
|
||
ChatGenerationTask.original_prompt / optimized_prompt 都写入第4步生成的 prompt_schema JSON 字符串。
|
||
"""
|
||
|
||
model_config = ConfigDict(extra="ignore", json_schema_extra={"example": {"engine_id": "video_engine_xxx"}})
|
||
|
||
engine_id: str | None = Field(None, description="视频生成引擎ID。为空优先使用第4步视频提词时选择的 engine_id，再为空使用最高优先级启用视频引擎")
|
||
|
||
|
||
class ShotReplicateStepOut(BaseModel):
|
||
id: str = Field(..., description="子任务ID")
|
||
project_id: str = Field(..., description="总任务项目ID，即 module_generation_projects.id")
|
||
module: str = Field(..., description="模块标识，例如 shot_replicate")
|
||
step_index: int = Field(..., description="步骤序号：1素材输入、2图片提词、3图片生成、4视频提词、5视频生成")
|
||
step_code: str = Field(..., description="步骤编码：material_input/image_prompt_optimize/image_generate/video_prompt_optimize/video_generate")
|
||
status: str = Field(..., description="步骤状态：pending/waiting_user/processing/completed/failed/cancelled")
|
||
version: int = Field(..., description="步骤版本号。重新生成或修改上游步骤后 version+1")
|
||
is_current: bool = Field(..., description="是否当前有效步骤。旧步骤会软删除且 is_current=false")
|
||
parent_step_id: str | None = Field(None, description="上一个步骤ID")
|
||
source_step_id: str | None = Field(None, description="当前步骤基于哪个上游步骤生成")
|
||
chat_task_id: str | None = Field(None, description="关联的 ChatGenerationTask ID。第3步图片生成、第5步视频生成有值")
|
||
input: dict[str, Any] | None = Field(None, description=f"步骤输入 JSON，统一 schema_version={SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION}")
|
||
output: dict[str, Any] | None = Field(None, description=f"步骤输出 JSON，统一 schema_version={SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION}")
|
||
error_message: str | None = Field(None, description="步骤错误信息")
|
||
created_at: NaiveDatetimeOptional = Field(None, description="创建时间")
|
||
updated_at: NaiveDatetimeOptional = Field(None, description="更新时间")
|
||
completed_at: NaiveDatetimeOptional = Field(None, description="完成时间")
|
||
|
||
|
||
class ShotReplicateMaterialOut(BaseModel):
|
||
material_step_id: str | None = Field(None, description="第1步素材输入子任务ID")
|
||
material_video_url: str | None = Field(None, description="素材视频链接")
|
||
material_image_url: str | None = Field(None, description="素材图片链接")
|
||
source_project_name: str | None = Field(None, description="视频素材内容项目名称")
|
||
target_project_name: str | None = Field(None, description="生成项目名称")
|
||
core_content_point: str | None = Field(None, description="生成项目核心内容点")
|
||
|
||
|
||
class ShotReplicateImageGenerationOut(BaseModel):
|
||
prompt_step_id: str | None = Field(None, description="第2步图片 AI 提词子任务ID")
|
||
generate_step_id: str | None = Field(None, description="第3步图片生成子任务ID")
|
||
prompt: str | None = Field(None, description="图片优化提词")
|
||
engine_id: str | None = Field(None, description="图片生成引擎ID")
|
||
engine_name: str | None = Field(None, description="图片生成引擎名称")
|
||
params: dict[str, Any] | None = Field(None, description="图片生成参数")
|
||
chat_task_id: str | None = Field(None, description="图片生成 ChatGenerationTask ID")
|
||
status: str | None = Field(None, description="图片生成状态")
|
||
result_image_url: str | None = Field(None, description="新项目图片 URL")
|
||
error_message: str | None = Field(None, description="图片生成错误信息")
|
||
|
||
|
||
class ShotReplicateVideoGenerationOut(BaseModel):
|
||
prompt_step_id: str | None = Field(None, description="第4步视频 AI 提词子任务ID")
|
||
generate_step_id: str | None = Field(None, description="第5步视频生成子任务ID")
|
||
prompt_schema: dict[str, Any] | None = Field(None, description="视频提词 JSON schema。第5步 ChatGenerationTask 原始提词会使用该 JSON 字符串")
|
||
final_prompt: str | None = Field(None, description="视频最终提词，仅用于前端展示")
|
||
prompt_params: dict[str, Any] | None = Field(None, description="第4步生成视频提词时使用的视频配置，例如 duration、aspect_ratio、resolution")
|
||
schema_config_snapshot: dict[str, Any] | None = Field(None, description="第4步生成视频提词时备份的 VIDEO_SCHEMA 配置快照。修改时按快照判断可编辑字段")
|
||
schema_config_source: str | None = Field(None, description="VIDEO_SCHEMA 配置来源：database=后台配置，default=CLIENT_SCHEMA_V1 默认配置")
|
||
schema_config_version: str | None = Field(None, description="VIDEO_SCHEMA 配置版本")
|
||
schema_config_is_fallback: bool = Field(False, description="是否为接口运行时兼容历史数据临时补齐的默认 VIDEO_SCHEMA 配置。true 表示数据库原始第4步没有 schema_config_snapshot")
|
||
engine_id: str | None = Field(None, description="视频生成引擎ID")
|
||
engine_name: str | None = Field(None, description="视频生成引擎名称")
|
||
params: dict[str, Any] | None = Field(None, description="视频生成实际参数。第5步只传 engine_id，其它参数继承第4步")
|
||
chat_task_id: str | None = Field(None, description="视频生成 ChatGenerationTask ID")
|
||
status: str | None = Field(None, description="视频生成状态")
|
||
result_video_url: str | None = Field(None, description="最终视频 URL")
|
||
result_video_cover_url: str | None = Field(None, description="最终视频封面 URL")
|
||
error_message: str | None = Field(None, description="视频生成错误信息")
|
||
|
||
|
||
class ShotReplicateTaskDetailOut(BaseModel):
|
||
id: str = Field(..., description="总任务项目ID。这个ID就是前端项目ID")
|
||
project_id: str = Field(..., description="兼容前端命名，等同于 id")
|
||
user_id: str | None = Field(None, description="所属用户ID；管理员后台排查使用")
|
||
user_name: str | None = Field(None, description="所属用户名；管理员后台排查使用")
|
||
module: str = Field(..., description="模块标识，拆镜复刻固定为 shot_replicate")
|
||
title: str | None = Field(None, description="项目标题，默认取生成项目名称")
|
||
status: str = Field(..., description="总任务状态：pending/waiting_user/processing/completed/failed/cancelled")
|
||
current_step_code: str | None = Field(None, description="当前所处步骤编码")
|
||
final_image_url: str | None = Field(None, description="最终新项目图片 URL")
|
||
final_video_url: str | None = Field(None, description="最终视频 URL")
|
||
final_video_cover_url: str | None = Field(None, description="最终视频封面 URL")
|
||
error_message: str | None = Field(None, description="总任务错误信息")
|
||
material: ShotReplicateMaterialOut = Field(default_factory=ShotReplicateMaterialOut, description="素材和项目描述信息")
|
||
image_generation: ShotReplicateImageGenerationOut = Field(default_factory=ShotReplicateImageGenerationOut, description="图片提词、图片引擎参数和图片结果")
|
||
video_generation: ShotReplicateVideoGenerationOut = Field(default_factory=ShotReplicateVideoGenerationOut, description="视频提词、视频引擎参数和视频结果")
|
||
steps: list[ShotReplicateStepOut] = Field(default_factory=list, description="当前有效子任务列表")
|
||
created_at: NaiveDatetimeOptional = Field(None, description="创建时间")
|
||
updated_at: NaiveDatetimeOptional = Field(None, description="更新时间")
|
||
completed_at: NaiveDatetimeOptional = Field(None, description="完成时间")
|
||
|
||
|
||
class ShotReplicateTaskListItemOut(BaseModel):
|
||
id: str = Field(..., description="总任务项目ID。这个ID就是前端项目ID")
|
||
project_id: str = Field(..., description="兼容前端命名，等同于 id")
|
||
module: str = Field(..., description="模块标识")
|
||
title: str | None = Field(None, description="项目标题")
|
||
status: str = Field(..., description="总任务状态")
|
||
current_step_code: str | None = Field(None, description="当前步骤")
|
||
target_project_name: str | None = Field(None, description="生成项目名称，来源于第1步素材输入")
|
||
final_image_url: str | None = Field(None, description="最终图片 URL")
|
||
final_video_url: str | None = Field(None, description="最终视频 URL")
|
||
error_message: str | None = Field(None, description="错误信息")
|
||
created_at: NaiveDatetimeOptional = Field(None, description="创建时间")
|
||
updated_at: NaiveDatetimeOptional = Field(None, description="更新时间")
|
||
completed_at: NaiveDatetimeOptional = Field(None, description="完成时间")
|
||
|
||
|
||
class ShotReplicateTaskListOut(BaseModel):
|
||
total: int = Field(..., description="总数量")
|
||
items: list[ShotReplicateTaskListItemOut] = Field(default_factory=list, description="列表数据")
|
||
|
||
|
||
class ShotReplicateActionOut(BaseModel):
|
||
message: str = Field(..., description="操作结果提示")
|
||
project_id: str = Field(..., description="总任务项目ID")
|
||
step_id: str | None = Field(None, description="本次创建或修改的子任务ID")
|
||
next_step_id: str | None = Field(None, description="兼容字段：当前接口不自动生成下下个任务，一般为空")
|
||
detail: ShotReplicateTaskDetailOut | None = Field(None, description="操作后的总任务详情")
|
||
|
||
|
||
class ShotReplicateDeleteOut(BaseModel):
|
||
message: str = Field(..., description="删除结果提示")
|
||
project_id: str = Field(..., description="被软删除的内部拆镜复刻 ModuleGenerationProject ID")
|
||
deleted: bool = Field(..., description="是否已软删除")
|
||
released_size_bytes: int = Field(0, description="本次软删释放的用户容量占用字节数；UploadResource 真实文件在事务提交后清理")
|
||
upload_resource_released: int = Field(0, description="本次标记删除的上传资源数量")
|
||
pending_delete_resource_ids: list[str] = Field(default_factory=list, exclude=True, description="内部字段：commit 后待删除真实文件的 UploadResource ID")
|
||
|
||
|
||
class ShotSegmentDeleteOut(BaseModel):
|
||
message: str = Field(..., description="删除结果提示")
|
||
segment_id: str = Field(..., description="被软删除的拆镜片段ID")
|
||
task_set_id: str = Field(..., description="所属拆镜总任务集ID")
|
||
deleted: bool = Field(..., description="是否已软删除")
|
||
deleted_module_project_id: str | None = Field(None, description="联动软删除的拆镜复刻项目ID；没有关联项目时为空")
|
||
released_size_bytes: int = Field(0, description="本次释放的用户容量占用字节数；UploadResource 真实文件在事务提交后清理")
|
||
upload_resource_released: int = Field(0, description="本次标记删除的上传资源数量")
|
||
pending_delete_resource_ids: list[str] = Field(default_factory=list, exclude=True, description="内部字段：commit 后待删除真实文件的 UploadResource ID")
|
||
|
||
|
||
class ShotTaskSetDeleteOut(BaseModel):
|
||
message: str = Field(..., description="删除结果提示")
|
||
task_set_id: str = Field(..., description="被软删除的拆镜任务集ID")
|
||
deleted: bool = Field(..., description="是否已软删除")
|
||
deleted_segment_count: int = Field(0, description="本次软删除的片段数量")
|
||
deleted_module_project_count: int = Field(0, description="本次联动软删除的内部复刻项目数量")
|
||
released_size_bytes: int = Field(0, description="本次释放的用户容量占用字节数；UploadResource 真实文件在事务提交后清理")
|
||
upload_resource_released: int = Field(0, description="本次标记删除的上传资源数量")
|
||
pending_delete_resource_ids: list[str] = Field(default_factory=list, exclude=True, description="内部字段：commit 后待删除真实文件的 UploadResource ID")
|
||
|
||
|
||
|
||
# ========================
|
||
# 拆镜总任务集 / 片段 API Schema
|
||
# ========================
|
||
|
||
SHOT_TASK_SET_STATUS_DESCRIPTIONS: dict[str, str] = {
|
||
"pending_analysis": "已创建，等待原视频分析",
|
||
"analyzing": "原视频分析中",
|
||
"analysis_completed": "原视频分析完成",
|
||
"analysis_failed": "原视频分析失败",
|
||
"splitting": "拆镜处理中",
|
||
"split_completed": "拆镜全部完成",
|
||
"partial_failed": "部分片段失败",
|
||
"failed": "总任务失败",
|
||
"deleted": "已软删",
|
||
}
|
||
|
||
SHOT_ANALYSIS_STATUS_DESCRIPTIONS: dict[str, str] = {
|
||
"pending": "待分析",
|
||
"processing": "分析中",
|
||
"completed": "分析完成",
|
||
"failed": "分析失败",
|
||
}
|
||
|
||
SHOT_SPLIT_STATUS_DESCRIPTIONS: dict[str, str] = {
|
||
"none": "尚未拆镜",
|
||
"pending": "待拆镜",
|
||
"processing": "拆镜中",
|
||
"completed": "拆镜完成",
|
||
"failed": "拆镜失败",
|
||
"retry_waiting": "等待恢复重试",
|
||
}
|
||
|
||
SHOT_SEGMENT_SOURCE_MODE_DESCRIPTIONS: dict[str, str] = {
|
||
"ai_suggestion": "AI 建议拆镜",
|
||
"custom": "用户自定义拆镜",
|
||
}
|
||
|
||
SHOT_SEGMENT_ANALYSIS_STATUS_DESCRIPTIONS: dict[str, str] = {
|
||
"not_required": "不需要单独分析，通常用于 AI 建议拆镜",
|
||
"pending": "等待片段分析",
|
||
"processing": "片段分析中",
|
||
"completed": "片段分析完成",
|
||
"failed": "片段分析失败",
|
||
}
|
||
|
||
SHOT_SEGMENT_REPLICATE_STATUS_DESCRIPTIONS: dict[str, str] = {
|
||
"not_started": "未进入复刻流程",
|
||
"project_created": "已创建复刻项目",
|
||
"processing": "复刻流程处理中",
|
||
"completed": "复刻流程完成",
|
||
"failed": "复刻流程失败",
|
||
}
|
||
|
||
|
||
class ShotAISuggestionOut(BaseModel):
|
||
index: int = Field(..., description="AI 建议序号，从1开始")
|
||
start_second: float = Field(..., description="拆镜开始秒")
|
||
end_second: float = Field(..., description="拆镜结束秒")
|
||
duration_seconds: float = Field(..., description="片段时长")
|
||
time_node: str = Field(..., description="拆镜时间节点，例如 0-15秒")
|
||
content: str = Field(..., description="对应时间节点内的内容")
|
||
category: str = Field(..., description="片段分类")
|
||
audience: str = Field(..., description="片段受众人群")
|
||
|
||
|
||
class ShotTaskSetCreate(BaseModel):
|
||
model_config = ConfigDict(
|
||
json_schema_extra={
|
||
"example": {
|
||
"video_url": "/uploads/2026/06/11/demo.mp4",
|
||
"video_duration_seconds": 31.42,
|
||
"video_resource_id": "0019...",
|
||
"title": "游戏视频拆镜",
|
||
"idempotency_key": "frontend-shot-task-001",
|
||
}
|
||
}
|
||
)
|
||
|
||
video_url: str = Field(..., min_length=1, description="已有上传接口返回的视频链接。必须能反解到 storage/uploads 下文件")
|
||
video_duration_seconds: float = Field(..., gt=0, description="前端获取的视频时长秒数，允许浮点；后端会用 ffprobe 校验并以后端真实时长为准")
|
||
video_resource_id: str | None = Field(None, max_length=32, description="可选，拆镜模块上传接口返回的视频 UploadResource.id；创建任务集后用于绑定资源")
|
||
title: str | None = Field(None, max_length=160, description="拆镜总任务标题")
|
||
idempotency_key: str | None = Field(None, max_length=64, description="创建总任务幂等键")
|
||
|
||
@field_validator("video_url", "title", "idempotency_key", mode="before")
|
||
@classmethod
|
||
def _strip_optional_text(cls, value: str | None) -> str | None:
|
||
if value is None:
|
||
return None
|
||
value = str(value).strip()
|
||
if not value:
|
||
return None
|
||
return value
|
||
|
||
|
||
class ShotTaskSetListQuery(BaseModel):
|
||
status: str | None = Field(None, description="总任务状态筛选：pending_analysis/analyzing/analysis_completed/analysis_failed/splitting/split_completed/partial_failed/failed/deleted")
|
||
analysis_status: str | None = Field(None, description="分析状态筛选：pending/processing/completed/failed")
|
||
split_status: str | None = Field(None, description="拆镜状态筛选：none/pending/processing/completed/failed/retry_waiting")
|
||
keyword: str | None = Field(None, description="标题/内容关键词，模糊搜索")
|
||
user_id: str | None = Field(None, description="管理员专用：用户ID筛选")
|
||
user_name: str | None = Field(None, description="管理员专用：用户名模糊筛选")
|
||
created_start: NaiveDatetimeOptional = Field(None, description="管理员专用：创建时间开始")
|
||
created_end: NaiveDatetimeOptional = Field(None, description="管理员专用：创建时间结束")
|
||
page: int = Field(1, ge=1, description="页码")
|
||
page_size: int = Field(20, ge=1, le=100, description="每页数量")
|
||
|
||
|
||
class ShotTaskSetOut(BaseModel):
|
||
model_config = ConfigDict(from_attributes=True)
|
||
|
||
id: str = Field(..., description="拆镜总任务集ID，即 shot_replicate_task_sets.id")
|
||
user_id: str | None = Field(None, description="所属用户ID；管理员后台排查使用")
|
||
user_name: str | None = Field(None, description="所属用户名；管理员后台排查使用")
|
||
title: str | None = Field(None, description="拆镜总任务标题，可为空")
|
||
video_url: str = Field(..., description="原视频 URL，来自已有上传接口")
|
||
video_duration_seconds: float = Field(..., description="原视频时长，单位秒，允许浮点")
|
||
status: str = Field(..., description="总任务状态：pending_analysis=等待分析，analyzing=分析中，analysis_completed=分析完成，analysis_failed=分析失败，splitting=拆镜中，split_completed=拆镜完成，partial_failed=部分失败，failed=失败，deleted=已软删")
|
||
analysis_status: str = Field(..., description="原视频分析状态：pending=待分析，processing=分析中，completed=分析完成，failed=分析失败")
|
||
split_status: str = Field(..., description="拆镜状态：none=尚未拆镜，pending=待拆镜，processing=拆镜中，completed=拆镜完成，failed=拆镜失败，retry_waiting=等待恢复重试")
|
||
original_video_content: str | None = Field(None, description="AI 分析出的原视频整体内容描述")
|
||
original_video_category: str | None = Field(None, description="AI 分析出的原视频分类，例如游戏视频、产品广告、教程等")
|
||
original_video_audience: str | None = Field(None, description="AI 分析出的原视频受众人群")
|
||
segment_count: int = Field(0, description="当前有效拆镜片段总数")
|
||
completed_segment_count: int = Field(0, description="切割完成的片段数量")
|
||
failed_segment_count: int = Field(0, description="切割失败的片段数量")
|
||
analysis_error_message: str | None = Field(None, description="原视频 AI 分析失败原因")
|
||
split_error_message: str | None = Field(None, description="总任务级拆镜失败原因")
|
||
created_at: NaiveDatetimeOptional = Field(None, description="创建时间")
|
||
updated_at: NaiveDatetimeOptional = Field(None, description="更新时间")
|
||
|
||
|
||
class ShotTaskSetListOut(BaseModel):
|
||
total: int = Field(..., description="符合筛选条件的总任务集总数")
|
||
page: int = Field(..., description="当前页码")
|
||
page_size: int = Field(..., description="每页数量")
|
||
items: list[ShotTaskSetOut] = Field(default_factory=list, description="拆镜总任务集列表")
|
||
|
||
|
||
class ShotTaskSetDetailOut(ShotTaskSetOut):
|
||
ai_suggestions: list[ShotAISuggestionOut] = Field(default_factory=list, description="AI 建议拆镜时间段列表，split-by-ai 接口可按 index 选择")
|
||
analysis_result_json: dict[str, Any] | list[Any] | None = Field(None, description="原视频 AI 分析完整 JSON 结果，结构由模型响应决定")
|
||
|
||
|
||
class ShotSplitByAIRequest(BaseModel):
|
||
selected_indices: list[int] | None = Field(None, description="指定 AI 建议序号列表，序号来自 ai_suggestions[].index；不传则按全部 AI 建议拆镜")
|
||
replace_existing: bool = Field(False, description="是否软删旧 AI 建议片段后重新拆；false 时保留旧片段并新增未存在片段")
|
||
|
||
|
||
class ShotSplitCustomRequest(BaseModel):
|
||
start_second: float = Field(..., ge=0, description="自定义拆镜开始秒，允许浮点，必须大于等于0")
|
||
end_second: float = Field(..., gt=0, description="自定义拆镜结束秒，允许浮点，必须大于 start_second")
|
||
|
||
@model_validator(mode="after")
|
||
def _check_range(self) -> "ShotSplitCustomRequest":
|
||
if self.end_second <= self.start_second:
|
||
raise ValueError("end_second 必须大于 start_second")
|
||
return self
|
||
|
||
|
||
class ShotSegmentOut(BaseModel):
|
||
model_config = ConfigDict(from_attributes=True)
|
||
|
||
id: str = Field(..., description="拆镜片段ID，即 shot_replicate_segments.id")
|
||
task_set_id: str = Field(..., description="所属拆镜总任务集ID，即 shot_replicate_task_sets.id")
|
||
segment_index: int = Field(..., description="片段序号，从1开始")
|
||
segment_name: str | None = Field(None, description="片段名称，可为空")
|
||
source_mode: str = Field(..., description="片段来源：ai_suggestion=AI 建议拆镜，custom=用户自定义拆镜")
|
||
start_second: float = Field(..., description="片段开始秒")
|
||
end_second: float = Field(..., description="片段结束秒")
|
||
duration_seconds: float = Field(..., description="片段时长，单位秒")
|
||
time_node: str = Field(..., description="片段时间节点展示文案，例如 0-5秒")
|
||
split_status: str = Field(..., description="切割状态：none=尚未拆镜，pending=待拆镜，processing=拆镜中，completed=拆镜完成，failed=拆镜失败，retry_waiting=等待恢复重试")
|
||
analysis_status: str = Field(..., description="片段分析状态：not_required=无需单独分析，pending=等待分析，processing=分析中，completed=分析完成，failed=分析失败")
|
||
replicate_status: str = Field(..., description="片段复刻状态：not_started=未复刻，project_created=已创建复刻项目，processing=复刻处理中，completed=复刻完成，failed=复刻失败")
|
||
segment_video_url: str | None = Field(None, description="切割后的片段视频 URL。切割完成后有值，用作复刻项目锁定素材视频")
|
||
original_video_content: str | None = Field(None, description="原视频整体内容描述，来自总任务 AI 分析")
|
||
original_video_category: str | None = Field(None, description="原视频分类，来自总任务 AI 分析")
|
||
original_video_audience: str | None = Field(None, description="原视频受众，来自总任务 AI 分析")
|
||
segment_content: str | None = Field(None, description="当前片段内容描述")
|
||
segment_category: str | None = Field(None, description="当前片段分类")
|
||
segment_audience: str | None = Field(None, description="当前片段受众人群")
|
||
split_retry_count: int = Field(0, description="切割失败后的恢复重试次数")
|
||
split_last_error: str | None = Field(None, description="最近一次切割失败原因")
|
||
analysis_error_message: str | None = Field(None, description="片段分析失败原因")
|
||
module_project_id: str | None = Field(None, description="由该片段创建的拆镜复刻项目ID，即 module_generation_projects.id")
|
||
module_project_title: str | None = Field(None, description="关联拆镜复刻项目标题；后台片段列表展示使用")
|
||
module_project_status: str | None = Field(None, description="关联拆镜复刻项目状态；后台片段列表展示使用")
|
||
module_project_current_step_code: str | None = Field(None, description="关联拆镜复刻项目当前步骤；后台片段列表展示使用")
|
||
created_at: NaiveDatetimeOptional = Field(None, description="创建时间")
|
||
updated_at: NaiveDatetimeOptional = Field(None, description="更新时间")
|
||
|
||
|
||
class ShotSegmentDetailOut(ShotSegmentOut):
|
||
analysis_json: dict[str, Any] | list[Any] | None = Field(None, description="片段 AI 分析完整 JSON，custom 片段可能有值；AI 建议片段通常复用 ai_suggestion_json")
|
||
ai_suggestion_json: dict[str, Any] | list[Any] | None = Field(None, description="AI 建议拆镜原始 JSON，source_mode=ai_suggestion 时通常有值")
|
||
|
||
|
||
class ShotSegmentListOut(BaseModel):
|
||
total: int = Field(..., description="符合筛选条件的片段总数")
|
||
page: int = Field(..., description="当前页码")
|
||
page_size: int = Field(..., description="每页数量")
|
||
items: list[ShotSegmentOut] = Field(default_factory=list, description="拆镜片段列表")
|
||
|
||
|
||
|
||
|
||
class ShotReanalyzeRequest(BaseModel):
|
||
force: bool = Field(False, description="是否强制重跑；默认 false。当前仅允许失败/待处理数据重试，已完成数据不建议强制覆盖")
|
||
reason: str | None = Field(None, max_length=200, description="再次分析原因，会写入模块日志")
|
||
|
||
@field_validator("reason", mode="before")
|
||
@classmethod
|
||
def _strip_reason(cls, value: str | None) -> str | None:
|
||
if value is None:
|
||
return None
|
||
value = str(value).strip()
|
||
return value or None
|
||
|
||
|
||
class ShotReanalyzeOut(BaseModel):
|
||
message: str = Field(..., description="操作结果提示")
|
||
task_set_id: str | None = Field(None, description="拆镜总任务集ID")
|
||
segment_id: str | None = Field(None, description="拆镜片段ID")
|
||
analysis_status: str = Field(..., description="重置后的分析状态")
|
||
celery_task_name: str = Field(..., description="已投递或待投递的 Celery 任务名")
|
||
|
||
class ShotSplitRetryRequest(BaseModel):
|
||
force: bool = Field(False, description="是否强制重置；当前只用于兼容入参，已完成切片仍会拒绝重切，避免旧文件覆盖和资源释放复杂化")
|
||
reason: str | None = Field(None, max_length=200, description="切片失败重试原因，会写入模块日志")
|
||
|
||
@field_validator("reason", mode="before")
|
||
@classmethod
|
||
def _strip_reason(cls, value: str | None) -> str | None:
|
||
if value is None:
|
||
return None
|
||
value = str(value).strip()
|
||
return value or None
|
||
|
||
|
||
class ShotSegmentSplitRetryOut(BaseModel):
|
||
message: str = Field(..., description="操作结果提示")
|
||
task_set_id: str = Field(..., description="拆镜总任务集ID")
|
||
segment_id: str = Field(..., description="拆镜片段ID")
|
||
split_status: str = Field(..., description="重置后的切片状态")
|
||
celery_task_name: str = Field(..., description="已投递或待投递的 Celery 任务名")
|
||
|
||
|
||
class ShotSplitByAIOut(BaseModel):
|
||
task_set_id: str = Field(..., description="拆镜总任务集ID")
|
||
status: str = Field(..., description="总任务状态：pending_analysis/analyzing/analysis_completed/analysis_failed/splitting/split_completed/partial_failed/failed/deleted")
|
||
split_status: str = Field(..., description="拆镜状态：none/pending/processing/completed/failed/retry_waiting")
|
||
created_segment_count: int = Field(..., description="本次创建的拆镜片段数量")
|
||
segments: list[ShotSegmentOut] = Field(default_factory=list, description="本次创建或返回的拆镜片段列表")
|
||
|
||
|
||
class ShotSplitCustomOut(BaseModel):
|
||
task_set_id: str = Field(..., description="拆镜总任务集ID")
|
||
segment: ShotSegmentOut = Field(..., description="本次创建的自定义拆镜片段")
|
||
|
||
|
||
class ShotSegmentReplicationCreateRequest(BaseModel):
|
||
model_config = ConfigDict(
|
||
json_schema_extra={
|
||
"example": {
|
||
"target_project_name": "新产品推广视频",
|
||
"core_content_point": "突出产品附近交友和快速脱单",
|
||
"material_image_url": "/uploads/2026/06/11/product.png",
|
||
"material_image_resource_id": "0019...",
|
||
"idempotency_key": "optional-key",
|
||
}
|
||
}
|
||
)
|
||
|
||
target_project_name: str = Field(..., min_length=1, max_length=20, description="生成项目名称，最多20字；会写入 module_generation_projects.title 和第1步 material_input")
|
||
core_content_point: str = Field(..., min_length=1, max_length=50, description="生成项目核心内容点，最多50字；用于后续图片 AI 提词和视频 AI 提词")
|
||
material_image_url: str = Field(..., min_length=1, description="新产品/目标素材图片链接，来自已有上传接口；作为图片生成参考素材")
|
||
material_image_resource_id: str | None = Field(None, max_length=32, description="可选，拆镜模块上传接口返回的图片 UploadResource.id；创建复刻项目后用于绑定资源")
|
||
idempotency_key: str | None = Field(None, max_length=64, description="创建 ModuleGenerationProject 幂等键；为空时后端可按业务生成或不使用")
|
||
|
||
@field_validator("target_project_name", "core_content_point", "material_image_url", "material_image_resource_id", "idempotency_key", mode="before")
|
||
@classmethod
|
||
def _strip_text(cls, value: str | None) -> str | None:
|
||
if value is None:
|
||
return None
|
||
value = str(value).strip()
|
||
if not value:
|
||
return None
|
||
return value
|
||
|
||
|
||
class ShotReplicateSpecOut(BaseModel):
|
||
project_statuses: dict[str, str] = Field(default_factory=lambda: SHOT_REPLICATE_PROJECT_STATUS_DESCRIPTIONS, description="ModuleGenerationProject 总任务状态说明：pending/waiting_user/processing/completed/failed/cancelled")
|
||
step_statuses: dict[str, str] = Field(default_factory=lambda: SHOT_REPLICATE_STEP_STATUS_DESCRIPTIONS, description="ModuleGenerationStep 子任务状态说明：pending/waiting_user/processing/completed/failed/cancelled")
|
||
steps: list[dict[str, Any]] = Field(default_factory=lambda: SHOT_REPLICATE_STEP_DESCRIPTIONS, description="5个固定步骤说明：material_input/image_prompt_optimize/image_generate/video_prompt_optimize/video_generate")
|
||
step_io_schema_version: str = Field(default=SHOT_REPLICATE_STEP_IO_SCHEMA_VERSION, description="步骤 input_json/output_json 结构版本")
|
||
step_io_examples: dict[str, dict[str, Any]] = Field(default_factory=lambda: SHOT_REPLICATE_STEP_IO_EXAMPLES, description="每个步骤 input_json/output_json 示例")
|
||
task_set_statuses: dict[str, str] = Field(default_factory=lambda: SHOT_TASK_SET_STATUS_DESCRIPTIONS, description="拆镜总任务集状态说明")
|
||
analysis_statuses: dict[str, str] = Field(default_factory=lambda: SHOT_ANALYSIS_STATUS_DESCRIPTIONS, description="原视频/片段视频分析状态说明")
|
||
split_statuses: dict[str, str] = Field(default_factory=lambda: SHOT_SPLIT_STATUS_DESCRIPTIONS, description="ffmpeg 拆镜状态说明")
|
||
segment_source_modes: dict[str, str] = Field(default_factory=lambda: SHOT_SEGMENT_SOURCE_MODE_DESCRIPTIONS, description="拆镜片段来源说明")
|
||
segment_analysis_statuses: dict[str, str] = Field(default_factory=lambda: SHOT_SEGMENT_ANALYSIS_STATUS_DESCRIPTIONS, description="拆镜片段分析状态说明")
|
||
segment_replicate_statuses: dict[str, str] = Field(default_factory=lambda: SHOT_SEGMENT_REPLICATE_STATUS_DESCRIPTIONS, description="拆镜片段进入复刻流程后的状态说明")
|