feat: 优化视频解码任务处理逻辑和数据库结构

- 在requirements.txt中添加socksio依赖
- 修改DecodeParam和DecodeRecord模型，分离video_id和video_url字段
- 调整任务调度器执行间隔和实例数量限制
- 重构decode_video接口支持批量任务创建
- 改进decode_task_status_handler和script_task_status_handler的任务处理逻辑
- 增强脚本任务处理的数据解析容错能力

decode_task/decodeTask.py

@@ -78,7 +78,6 @@ def invoke_decode_workflow(video_list: List[DecodeParam]):
         output_data = save_decode_result(results, timestamp, total, success_count, fail_count)
         # print(f"   保存结果到文件... [{success_count}成功/{fail_count}失败/{total}总计]")
         if output_data:
-            print(f"✅ 结果已实时保存到: {output_path}")
             return output_data
         else:
             print(f"❌ 保存结果失败，但将继续处理")
@@ -94,7 +93,7 @@ def get_decode_result_by_id(task_id:str):
     return tasks['decode_result'], tasks['task_status']
 
 
-def  decode_task_status_handler(task_id:str):
+def  decode_task_status_handler():
     # 从数据库中获取任务，每次获取一个
     sql = "SELECT * FROM decode_record WHERE task_status = 1 "
     """json"""
@@ -116,7 +115,7 @@ def  decode_task_status_handler(task_id:str):
         
         # 获取任务结果
         try:
-            decode_result = invoke_decode_workflow(task_params)
+            decode_result = invoke_decode_workflow(task_params['video_list'])
 
             if decode_result:
                 # 更新任务状态为2，任务完成
@@ -125,12 +124,12 @@ def  decode_task_status_handler(task_id:str):
                 logger.info(f"task_id = {task_id} , decode_result = {decode_result}")
             else:
                 if current_timestamp - task_create_timestamp > 1000 * 60 * 60:
-                    sql = "UPDATE decode_record SET task_status = 3, decode_result = %s WHERE task_id = %s"
+                    sql = "UPDATE decode_record SET task_status = 5, decode_result = %s WHERE task_id = %s"
                     mysql.execute(sql, (json.dumps(decode_result, ensure_ascii=False), task_id))
                     logger.info(f"task_id = {task_id} ,任务状态异常")
         except Exception as e:
             logger.error(f"task_id = {task_id} , error = {e}")
-            sql = "UPDATE decode_record SET task_status = 3, decode_result = '任务异常' WHERE task_id = %s"
+            sql = "UPDATE decode_record SET task_status = 5, decode_result = '任务异常' WHERE task_id = %s"
             mysql.execute(sql,  (task_id))
             logger.info(f"task_id = {task_id} ,任务异常")
             continue

decode_task/scriptTask.py

@@ -21,7 +21,7 @@ def invoke_script_workflow(decode_result: Dict[str, Any], script_result: Dict[st
     raw_results: List[Dict[str, Any]] = decode_result.get("results", []) or []
 
     # 读取已有的脚本理解输出，支持增量追加
-    output_data = script_result
+    output_data = script_result if isinstance(script_result, dict) else None
     if not output_data:
         output_data = {
             "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
@@ -80,10 +80,8 @@ def invoke_script_workflow(decode_result: Dict[str, Any], script_result: Dict[st
         output_data["results"].append(record)
         output_data["total"] = output_data.get("total", 0) + 1
 
-        # 处理完一条就保存一次，避免长任务中途失败导致全部丢失
-
-
-        return json.dump(output_data, ensure_ascii=False, indent=2)
+    # 返回序列化后的 JSON 字符串（供数据库存储）
+    return json.dumps(output_data, ensure_ascii=False)
 
     
 
@@ -94,10 +92,10 @@ def get_script_result_by_id(task_id:str):
     if not tasks:
         logger.info(f"task_id = {task_id} , 任务不存在")
         return None
-    return tasks['script_result'], tasks['decode_result'], tasks['task_status']
+    return tasks['decode_result'], tasks['script_result'], tasks['task_status']
 
 
-def  script_task_status_handler(task_id:str):
+def  script_task_status_handler():
     # 从数据库中获取任务，每次获取一个
     sql = "SELECT * FROM decode_record WHERE task_status = 2 "
     """json"""
@@ -105,16 +103,39 @@ def  script_task_status_handler(task_id:str):
     # tasks = mysql.fetchall(sql,())
 
     if not tasks:
-        logger.info("任务列表为空")
+        logger.info("script任务列表为空")
         return
     for task in tasks:
         task_id = task['task_id']
-        decode_result = json.loads(task['decode_result'])
-        script_result = json.loads(task['script_result'])
+        # 解析 decode_result（可能为 None 或已是对象）
+        decode_result_raw = task.get('decode_result')
+        if isinstance(decode_result_raw, (dict, list)):
+            decode_result = decode_result_raw
+        elif decode_result_raw:
+            try:
+                decode_result = json.loads(decode_result_raw)
+            except Exception as e:
+                logger.error(f"task_id = {task_id} , decode_result 解析失败: {e}")
+                decode_result = {"results": []}
+        else:
+            decode_result = {"results": []}
+
+        # 解析 script_result（可能为 None 或已是对象）
+        script_result_raw = task.get('script_result')
+        if isinstance(script_result_raw, (dict, list)):
+            script_result = script_result_raw
+        elif script_result_raw:
+            try:
+                script_result = json.loads(script_result_raw)
+            except Exception as e:
+                logger.error(f"task_id = {task_id} , script_result 解析失败: {e}")
+                script_result = None
+        else:
+            script_result = None
 
         logger.info(f"task_id = {task_id} , decode_result = {decode_result}")
         #如何任务超过30分钟，则认为任务超时，更新任务状态为3
-        task_create_timestamp = task['script_timestamp']
+        task_create_timestamp = task['create_timestamp']
         current_timestamp = int(time.time() * 1000)
 
 
@@ -147,4 +168,3 @@ def  script_task_status_handler(task_id:str):
             mysql.execute(sql,  (task_id))
             logger.info(f"task_id = {task_id} ,任务异常")
             continue
-

main.py

@@ -47,20 +47,30 @@ def startup_event():
 
 @app.post("/decodeVideo/create")
 def decode_video(param:DecodeListParam):
-    task_id = str(uuid.uuid4())
-    DecodeRecord(
-        task_id=task_id,
-        task_params=json.dumps(param.model_dump()),
-        decode_result=None,
-        script_result=None,
-        task_status = 1
-    ).save()
+
+    video_list = param.video_list
+    data_list = []
+    for video in video_list:
+        video_id = video.channel_content_id
+        video_url = video.video
+        task_id = str(uuid.uuid4())
+        DecodeRecord(
+            task_id=task_id,
+            video_id=video_id,
+            video_url=video_url,
+            task_params=json.dumps(param.model_dump()),
+            decode_result=None,
+            script_result=None,
+            task_status = 1
+        ).save()
+        data_list.append({
+            "task_id": task_id,
+            "video_id": video_id,
+        })
     return {
         "code": 0,
         "message": "success",
-        "data": {
-            "task_id": task_id
-        }
+        "data": data_list
     }
 
 
@@ -104,4 +114,3 @@ def get_script_result(param: TaskStatusParam):
         }
 
 
-

models/decode_record.py

@@ -10,7 +10,8 @@ class DecodeRecord(BaseModel):
     table_name:       Annotated[Optional[str], Field(description='表名', exclude=True)] = 'decode_record'
 
     task_id:          Annotated[str, Field(description='任务ID')]
-    model:            Annotated[str, Field(description='模型名称')]
+    video_id:            Annotated[str, Field(description='视频ID')]
+    video_url:            Annotated[str, Field(description='视频地址')]
     task_params:      Annotated[Optional[str], Field(description='任务参数', default=None)]
     task_status:      Annotated[Optional[int], Field(description='任务状态', default=1)]  # 1: 进行中, 2: 已完成, 3: 失败
     decode_result:      Annotated[Optional[str], Field(description='解构结果', default=None)]

requirements.txt

@@ -22,4 +22,5 @@ PyMySQL>=1.1.0
 DBUtils>=3.0.3
 dashscope>=1.14.0
 oss2>=2.17.0
-python-dotenv
+python-dotenv
+socksio>=1.0.0

task_schedule.py

@@ -15,18 +15,18 @@ class TaskScheduler:
 
         self.scheduler.add_job(
             func=decode_task_status_handler,
-            trigger=IntervalTrigger(seconds=60),
+            trigger=IntervalTrigger(seconds=30),
             id='decode_task_status_handler',
-            name='任务状态刷新',
-            max_instances=5  # 防止重复执行
+            name='decode_task任务刷新',
+            max_instances=1  # 防止重复执行
         )
 
         self.scheduler.add_job(
             func=script_task_status_handler,
-            trigger=IntervalTrigger(seconds=60),
+            trigger=IntervalTrigger(seconds=30),
             id='script_task_status_handler',
-            name='script_task_status_handler',
-            max_instances=5  # 防止重复执行
+            name='script_task任务刷新',
+            max_instances=1  # 防止重复执行
         )
 
     def start(self):

utils/params.py

@@ -8,7 +8,7 @@ class TaskStatusParam(BaseModel):
 
 class DecodeParam(BaseModel):
     channel_content_id: str
-    video_url: str
+    video: str
     title: str
 
 class DecodeListParam(BaseModel):