145 lines
5.6 KiB
Python
145 lines
5.6 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_MODES: dict[GenerationHistorySourceEnum, tuple[GenerationMode, ...]] = {
|
|
GenerationHistorySourceEnum.CHAT_TASK: (
|
|
GenerationMode.CHATAPI_ASYNC,
|
|
GenerationMode.CHATAPI_CHILD,
|
|
),
|
|
GenerationHistorySourceEnum.HOT_OPENING_REPLICATE: (GenerationMode.HOT_OPENING_REPLICATE,),
|
|
GenerationHistorySourceEnum.SHOT_REPLICATE: (GenerationMode.SHOT_REPLICATE,),
|
|
}
|
|
"""history_source 到 ChatGenerationTask.generation_mode 集合的映射。"""
|
|
|
|
|
|
GENERATION_HISTORY_SOURCE_TO_TASK_MODE: dict[GenerationHistorySourceEnum, GenerationMode] = {
|
|
history_source: task_modes[0]
|
|
for history_source, task_modes in GENERATION_HISTORY_SOURCE_TO_TASK_MODES.items()
|
|
}
|
|
"""兼容旧调用的单一模式映射；新查询应使用 GENERATION_HISTORY_SOURCE_TO_TASK_MODES。"""
|
|
|
|
|
|
GENERATION_HISTORY_TASK_MODE_VALUE_TO_SOURCE: dict[str, GenerationHistorySourceEnum] = {
|
|
task_mode.value: history_source
|
|
for history_source, task_modes in GENERATION_HISTORY_SOURCE_TO_TASK_MODES.items()
|
|
for task_mode in task_modes
|
|
}
|
|
"""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 对应的第一个任务模式。"""
|
|
|
|
return GENERATION_HISTORY_SOURCE_TO_TASK_MODE.get(source)
|
|
|
|
|
|
def get_generation_history_task_modes(source: GenerationHistorySourceEnum) -> tuple[GenerationMode, ...]:
|
|
"""获取 history_source 对应的全部 ChatGenerationTask.generation_mode。"""
|
|
|
|
return GENERATION_HISTORY_SOURCE_TO_TASK_MODES.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
|
|
|
|
|
|
HISTORY_DAY_PAGE_SIZE_MAX = 10
|
|
HISTORY_GROUP_ITEM_LIMIT = 10
|