124 lines
4.8 KiB
Python
124 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from enum import Enum
|
|
|
|
from app.enums.generation_task import ChatGenerationTaskStatus, GenerationMode, GenerationType
|
|
|
|
|
|
class GenerationHistorySourceEnum(str, Enum):
|
|
"""素材云历史接口支持的数据来源枚举。"""
|
|
|
|
CHAT_TASK = "chat_task"
|
|
GENERATION_RECORD = "generation_record"
|
|
HOT_OPENING_REPLICATE = "hot_opening_replicate"
|
|
SHOT_REPLICATE = "shot_replicate"
|
|
|
|
|
|
class GenerationHistoryResourceTypeEnum(str, Enum):
|
|
"""素材云历史接口支持的资源类型枚举。"""
|
|
|
|
IMAGE = GenerationType.IMAGE.value
|
|
VIDEO = GenerationType.VIDEO.value
|
|
|
|
|
|
GENERATION_HISTORY_SOURCE_LABELS: dict[GenerationHistorySourceEnum, str] = {
|
|
GenerationHistorySourceEnum.CHAT_TASK: "AI创作",
|
|
GenerationHistorySourceEnum.GENERATION_RECORD: "项目生成",
|
|
GenerationHistorySourceEnum.HOT_OPENING_REPLICATE: "爆款开头复刻",
|
|
GenerationHistorySourceEnum.SHOT_REPLICATE: "拆镜复刻",
|
|
}
|
|
"""素材云历史来源中文名称。"""
|
|
|
|
|
|
GENERATION_HISTORY_CHAT_TASK_SOURCES: tuple[GenerationHistorySourceEnum, ...] = (
|
|
GenerationHistorySourceEnum.CHAT_TASK,
|
|
GenerationHistorySourceEnum.HOT_OPENING_REPLICATE,
|
|
GenerationHistorySourceEnum.SHOT_REPLICATE,
|
|
)
|
|
"""来自 chat_generation_tasks 表的历史来源集合。"""
|
|
|
|
|
|
GENERATION_HISTORY_MODULE_SOURCES: tuple[GenerationHistorySourceEnum, ...] = (
|
|
GenerationHistorySourceEnum.HOT_OPENING_REPLICATE,
|
|
GenerationHistorySourceEnum.SHOT_REPLICATE,
|
|
)
|
|
"""需要回填 module_generation_projects/module_generation_steps 的模块来源集合。"""
|
|
|
|
|
|
GENERATION_HISTORY_SOURCE_TO_TASK_MODE: dict[GenerationHistorySourceEnum, GenerationMode] = {
|
|
GenerationHistorySourceEnum.CHAT_TASK: GenerationMode.CHATAPI_ASYNC,
|
|
GenerationHistorySourceEnum.HOT_OPENING_REPLICATE: GenerationMode.HOT_OPENING_REPLICATE,
|
|
GenerationHistorySourceEnum.SHOT_REPLICATE: GenerationMode.SHOT_REPLICATE,
|
|
}
|
|
"""history_source 到 ChatGenerationTask.generation_mode 的映射。"""
|
|
|
|
|
|
GENERATION_HISTORY_TASK_MODE_VALUE_TO_SOURCE: dict[str, GenerationHistorySourceEnum] = {
|
|
task_mode.value: history_source
|
|
for history_source, task_mode in GENERATION_HISTORY_SOURCE_TO_TASK_MODE.items()
|
|
}
|
|
"""ChatGenerationTask.generation_mode 字符串值到 history_source 的映射。"""
|
|
|
|
|
|
GENERATION_HISTORY_COMPLETED_STATUS = ChatGenerationTaskStatus.COMPLETED.value
|
|
"""素材云历史只展示生成成功的数据。"""
|
|
|
|
|
|
GENERATION_HISTORY_SOURCE_ALIASES: dict[str, GenerationHistorySourceEnum] = {
|
|
"": GenerationHistorySourceEnum.CHAT_TASK,
|
|
"chat": GenerationHistorySourceEnum.CHAT_TASK,
|
|
"chat_ai": GenerationHistorySourceEnum.CHAT_TASK,
|
|
"chat_task": GenerationHistorySourceEnum.CHAT_TASK,
|
|
"chat_generation_task": GenerationHistorySourceEnum.CHAT_TASK,
|
|
"chat_generation_tasks": GenerationHistorySourceEnum.CHAT_TASK,
|
|
"record": GenerationHistorySourceEnum.GENERATION_RECORD,
|
|
"records": GenerationHistorySourceEnum.GENERATION_RECORD,
|
|
"project": GenerationHistorySourceEnum.GENERATION_RECORD,
|
|
"generation_record": GenerationHistorySourceEnum.GENERATION_RECORD,
|
|
"generation_records": GenerationHistorySourceEnum.GENERATION_RECORD,
|
|
"hot_opening": GenerationHistorySourceEnum.HOT_OPENING_REPLICATE,
|
|
"hot_opening_replicate": GenerationHistorySourceEnum.HOT_OPENING_REPLICATE,
|
|
"shot": GenerationHistorySourceEnum.SHOT_REPLICATE,
|
|
"shot_replicate": GenerationHistorySourceEnum.SHOT_REPLICATE,
|
|
}
|
|
"""history_source 兼容别名映射。"""
|
|
|
|
|
|
def normalize_generation_history_source(value: str | None) -> GenerationHistorySourceEnum:
|
|
"""归一化素材云历史来源。"""
|
|
|
|
key = (value or "chat_task").lower().strip()
|
|
if key in GENERATION_HISTORY_SOURCE_ALIASES:
|
|
return GENERATION_HISTORY_SOURCE_ALIASES[key]
|
|
return GenerationHistorySourceEnum(key)
|
|
|
|
|
|
def get_generation_history_source_label(source: GenerationHistorySourceEnum | str | None) -> str | None:
|
|
"""获取素材云历史来源中文名称。"""
|
|
|
|
if source is None:
|
|
return None
|
|
source_enum = source if isinstance(source, GenerationHistorySourceEnum) else GenerationHistorySourceEnum(str(source))
|
|
return GENERATION_HISTORY_SOURCE_LABELS.get(source_enum)
|
|
|
|
|
|
def get_generation_history_task_mode(source: GenerationHistorySourceEnum) -> GenerationMode | None:
|
|
"""获取 history_source 对应的 ChatGenerationTask.generation_mode。"""
|
|
|
|
return GENERATION_HISTORY_SOURCE_TO_TASK_MODE.get(source)
|
|
|
|
|
|
def is_generation_history_chat_task_source(source: GenerationHistorySourceEnum) -> bool:
|
|
"""判断当前来源是否走 chat_generation_tasks 表。"""
|
|
|
|
return source in GENERATION_HISTORY_CHAT_TASK_SOURCES
|
|
|
|
|
|
def is_generation_history_module_source(source: GenerationHistorySourceEnum) -> bool:
|
|
"""判断当前来源是否需要回填模块项目信息。"""
|
|
|
|
return source in GENERATION_HISTORY_MODULE_SOURCES
|
|
|
|
|
|
MAX_BATCH_DELETE_COUNT = 30
|