models/decode_task_result.py

@@ -38,8 +38,13 @@ class WorkflowDecodeTaskResult(BaseModel):
         return [url.strip() for url in self.images.split(',') if url.strip()]
 
     @staticmethod
-    def create_result(task_id: str, content: 'ContentParam') -> 'WorkflowDecodeTaskResult':
-        """创建并初始化结果记录"""
+    def create_result(task_id: str, content: 'ContentParam', table_name: Optional[str] = None) -> 'WorkflowDecodeTaskResult':
+        """创建并初始化结果记录
+
+        :param task_id: 任务ID
+        :param content: 内容参数
+        :param table_name: 可选，自定义结果表名；为空时使用默认的 workflow_decode_task_result
+        """
         # 处理 images 列表，使用逗号分隔的字符串格式存储
         # 数据库字段建议使用 TEXT 类型，可存储大量 URL
         # 如果有 video_url，也将其加入到 images 字段中
@@ -63,5 +68,8 @@ class WorkflowDecodeTaskResult(BaseModel):
             result_payload=None,
             result_size=0
         )
+        # 如果指定了表名，则覆盖默认表名
+        if table_name:
+            result.table_name = table_name
         result.save()
         return result

tasks/decode.py

@@ -54,7 +54,7 @@ def _create_workflow_task(scene: SceneEnum, content_type: ContentTypeEnum) -> Op
 
 
 def _initialize_task_result(task_id: str, content) -> Optional[WorkflowDecodeTaskResult]:
-    """初始化任务结果"""
+    """初始化选题解构任务结果"""
     try:
         result = WorkflowDecodeTaskResult.create_result(
             task_id=task_id,
@@ -67,6 +67,21 @@ def _initialize_task_result(task_id: str, content) -> Optional[WorkflowDecodeTas
         return None
 
 
+def _initialize_script_task_result(task_id: str, content) -> Optional[WorkflowDecodeTaskResult]:
+    """初始化创作解构任务结果（写入 workflow_script_task_result 表）"""
+    try:
+        result = WorkflowDecodeTaskResult.create_result(
+            task_id=task_id,
+            content=content,
+            table_name="workflow_script_task_result",
+        )
+        logger.info(f"初始化创作解构任务结果成功，task_id: {task_id}")
+        return result
+    except Exception as e:
+        logger.error(f"初始化创作解构任务结果失败，task_id: {task_id}, error: {str(e)}")
+        return None
+
+
 def _check_quota(scene: SceneEnum, capability: CapabilityEnum = CapabilityEnum.DECODE, content_type: ContentTypeEnum = ContentTypeEnum.TEXT) -> bool:
     """检查配额是否充足（并发安全版本）"""
     try:
@@ -196,10 +211,11 @@ def decode_topic(param: DecodeContentParam) -> Dict[str, Any]:
 def begin_decode_task(param: DecodeContentParam) -> Dict[str, Any]:
     """根据场景分发任务"""
     if param.scene == SceneEnum.TOPIC:
+        # 选题解构
         return decode_topic(param)
     elif param.scene == SceneEnum.CREATION:
-        # TODO: 实现创作场景
-        return _build_error_response(ERROR_CODE_FAILED, "创作场景暂未实现")
+        # 创作解构：流程与选题相同，只是结果表不同
+        return decode_creation(param)
     elif param.scene == SceneEnum.PRODUCTION:
         # TODO: 实现制作场景
         return _build_error_response(ERROR_CODE_FAILED, "制作场景暂未实现")
@@ -207,6 +223,38 @@ def begin_decode_task(param: DecodeContentParam) -> Dict[str, Any]:
         return _build_error_response(ERROR_CODE_FAILED, f"未知场景: {param.scene}")
 
 
+def decode_creation(param: DecodeContentParam) -> Dict[str, Any]:
+    """创作解构方法"""
+    try:
+        # 步骤1: 创建工作流task任务（scene 为 CREATION）
+        task = _create_workflow_task(param.scene, param.content_type)
+        if not task or not task.task_id:
+            return _build_error_response(
+                ERROR_CODE_TASK_CREATE_FAILED,
+                "创建创作解构任务失败"
+            )
+
+        # 步骤2: 初始化创作解构任务结果到 workflow_script_task_result
+        result = _initialize_script_task_result(task.task_id, param.content)
+        if not result or not result.task_id:
+            return _build_error_response(
+                ERROR_CODE_FAILED,
+                "初始化创作解构任务结果失败"
+            )
+
+        # 步骤3:（可选）触发创作解构工作流，目前按选题逻辑仅返回 task_id
+
+        # 所有步骤成功
+        return _build_success_response(task.task_id)
+
+    except Exception as e:
+        logger.error(f"创作解构失败: {str(e)}")
+        return _build_error_response(
+            ERROR_CODE_TASK_CREATE_FAILED,
+            f"创作解构任务创建失败: {str(e)}"
+        )
+
+
 def _trigger_topic_decode_workflow(task_id: str) -> Dict[str, Any]:
     """发起解构任务（调用上游工作流服务）"""
     try:

tasks/detail.py

@@ -50,6 +50,33 @@ def _parse_web_url(web_url: Optional[str]) -> Any:
         return web_url
 
 
+def _build_script_point_url(result: Any) -> Dict[str, str]:
+    """
+    从创作解构的 result 中构建链接字段：
+    1. 优先使用 result['script_elements_table'] 作为链接
+    2. 如果没有该字段，则任意取一个字段的值作为链接
+    3. 统一返回结构为 { "pointUrl": "<链接或空字符串>" }
+    """
+    point_url = ""
+
+    # 优先从 script_elements_table 字段取值
+    if isinstance(result, dict):
+        value = result.get("script_elements_table")
+        if isinstance(value, str) and value:
+            point_url = value
+        else:
+            # 没有 script_elements_table 时，从任意字段中取一个非空字符串值
+            for v in result.values():
+                if isinstance(v, str) and v:
+                    point_url = v
+                    break
+    elif isinstance(result, str):
+        # 如果整体就是一个字符串，直接作为链接使用
+        point_url = result
+
+    return {"pointUrl": point_url}
+
+
 def _fetch_decode_result(task_id: str) -> Optional[Dict[str, Any]]:
     """获取选题解构任务结果"""
     sql = "SELECT result_payload, error_message, web_url FROM workflow_decode_task_result WHERE task_id = %s"
@@ -73,10 +100,15 @@ def _fetch_script_decode_result(task_id: str) -> Optional[Dict[str, Any]]:
     if not result_record:
         return None
 
+    parsed_result = _parse_result_payload(result_record.get("result_payload"))
+    url_result = _parse_result_payload(result_record.get("web_url"))
+    # 创作解构的链接从 result 中提取，返回 { "pointUrl": "<链接>" } 结构
+    url = _build_script_point_url(url_result)
+
     return {
-        "result": _parse_result_payload(result_record.get("result_payload")),
+        "result": parsed_result,
         "error_message": result_record.get("error_message"),
-        "url": _parse_web_url(result_record.get("web_url")),
+        "url": url,
     }