素材云多模块API追加

2026-07-01 10:28:30 +08:00
parent 838215736f
commit 87c0739863
6 changed files with 561 additions and 35 deletions
@@ -7,6 +7,7 @@ from app.enums.user import *
 from app.enums.credit_record import *
 from app.enums.token_usage import *
 from app.enums.generation_task import *
+from app.enums.generation_history import *
 from app.enums.generation_status import *
 from app.enums.sms import *
 from app.enums.notification import *
@@ -0,0 +1,120 @@
+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