增加异步上传捕获异常

video-gen-api/app/api/v1/upload_material.py

@@ -55,144 +55,152 @@ async def async_batch_upload_material(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ) -> Any | dict:
-    if not req.tasks:
+    try:
+        if not req.tasks:
+            return {
+                "code": 0,
+                "message": "上传任务列表不能为空",
+                "task_ids": [],
+                "errors": [],
+            }
+
+        task_ids = []
+        errors = []
+        
+        source_model_map = {
+            "generation_records": "GenerationRecord",
+            "generated_resources": None,
+            "chat_generation_tasks": "ChatGenerationTask",
+        }
+
+        for task_index, task in enumerate(req.tasks, 1):
+            if not task.advertiser_ids:
+                errors.append({
+                    "task_index": task_index,
+                    "error": "广告主id数组不能为空",
+                })
+                continue
+            
+            if not task.resource_ids:
+                errors.append({
+                    "task_index": task_index,
+                    "error": "资源id数组不能为空",
+                })
+                continue
+            
+            if task.is_pre_test == "1" and not task.pre_test_template:
+                errors.append({
+                    "task_index": task_index,
+                    "error": "开启前测功能时，必须指定前测模板id",
+                })
+                continue
+            
+            if task.is_pre_test == "1":
+                template = await db.execute(
+                    select(PreTestTemplate).where(PreTestTemplate.id == task.pre_test_template).
+                    where(PreTestTemplate.deleted_at.is_(None)).
+                    where(PreTestTemplate.user_id == current_user.id)
+                )
+                template = template.scalar_one_or_none()
+                if not template:
+                    errors.append({
+                        "task_index": task_index,
+                        "error": "前测模板id不存在",
+                    })
+                    continue
+            
+            target_source_model = source_model_map.get(task.source_model)
+            
+            # 检查资源id是否存在，非资源id
+            if target_source_model:
+                query = (
+                    select(GeneratedResource.id)
+                    .where(GeneratedResource.source_model == target_source_model)
+                    .where(GeneratedResource.source_id.in_(task.resource_ids))
+                    .where(GeneratedResource.user_id == current_user.id)
+                    .where(GeneratedResource.deleted_at.is_(None))
+                )
+                result = await db.execute(query)
+                valid_resource_ids = [row[0] for row in result.all()]
+                
+                invalid_ids = set(task.resource_ids) - set(valid_resource_ids)
+                
+                if invalid_ids:
+                    invalid_ids_str = ", ".join(invalid_ids)
+                    errors.append({
+                        "task_index": task_index,
+                        "error": f"资源id [{invalid_ids_str}] 不可用或已删除",
+                    })
+                    continue
+                
+                resource_ids_to_upload = valid_resource_ids
+            else:
+                #用户提交的直接是资源id
+                query = (
+                    select(GeneratedResource.id)
+                    .where(GeneratedResource.id.in_(task.resource_ids))
+                    .where(GeneratedResource.user_id == current_user.id)
+                    .where(GeneratedResource.deleted_at.is_(None))
+                )
+                result = await db.execute(query)
+                valid_resource_ids = [row[0] for row in result.all()]
+                
+                invalid_ids = set(task.resource_ids) - set(valid_resource_ids)
+                
+                if invalid_ids:
+                    invalid_ids_str = ", ".join(invalid_ids)
+                    errors.append({
+                        "task_index": task_index,
+                        "error": f"资源id [{invalid_ids_str}] 不可用或已删除",
+                    })
+                    continue
+                
+                resource_ids_to_upload = valid_resource_ids
+            
+            for advertiser_id in task.advertiser_ids:
+                for resource_id in resource_ids_to_upload:
+                    other_info = {}
+                    if task.is_pre_test == "1":
+                        other_info["is_pre_test"] = task.is_pre_test
+                        other_info["pre_test_template"] = task.pre_test_template
+
+                    task_id = generate_id()
+                    upload_task = UploadTask(
+                        id=task_id,
+                        user_id=current_user.id,
+                        oauth_id=task.oauth_id,
+                        advertiser_id=advertiser_id,
+                        resource_id=resource_id,
+                        status=1,
+                        note=None,
+                        other_info=json.dumps(other_info) if other_info else None,
+                    )
+                    
+                    db.add(upload_task)
+                    await upload_queue.enqueue(task_id)
+                    task_ids.append(task_id)
+        
+        await db.commit()
+        
+        message = "上传任务已提交"
+        if errors:
+            message = f"部分任务提交成功，{len(errors)} 个任务失败"
+        
         return {
             "code": 0,
-            "message": "上传任务列表不能为空",
+            "message": message,
+            "task_ids": task_ids,
+            "errors": errors,
+        }
+
+    except Exception as e:
+        return {
+            "code": 1,
+            "message": str(e),
             "task_ids": [],
             "errors": [],
         }
 
-    task_ids = []
-    errors = []
-    
-    source_model_map = {
-        "generation_records": "GenerationRecord",
-        "generated_resources": None,
-        "chat_generation_tasks": "ChatGenerationTask",
-    }
-
-    for task_index, task in enumerate(req.tasks, 1):
-        if not task.advertiser_ids:
-            errors.append({
-                "task_index": task_index,
-                "error": "广告主id数组不能为空",
-            })
-            continue
-        
-        if not task.resource_ids:
-            errors.append({
-                "task_index": task_index,
-                "error": "资源id数组不能为空",
-            })
-            continue
-        
-        if task.is_pre_test == "1" and not task.pre_test_template:
-            errors.append({
-                "task_index": task_index,
-                "error": "开启前测功能时，必须指定前测模板id",
-            })
-            continue
-        
-        if task.is_pre_test == "1":
-            template = await db.execute(
-                select(PreTestTemplate).where(PreTestTemplate.id == task.pre_test_template).
-                where(PreTestTemplate.deleted_at.is_(None)).
-                where(PreTestTemplate.user_id == current_user.id)
-            )
-            template = template.scalar_one_or_none()
-            if not template:
-                errors.append({
-                    "task_index": task_index,
-                    "error": "前测模板id不存在",
-                })
-                continue
-        
-        target_source_model = source_model_map.get(task.source_model)
-        
-        # 检查资源id是否存在，非资源id
-        if target_source_model:
-            query = (
-                select(GeneratedResource.id)
-                .where(GeneratedResource.source_model == target_source_model)
-                .where(GeneratedResource.source_id.in_(task.resource_ids))
-                .where(GeneratedResource.user_id == current_user.id)
-                .where(GeneratedResource.deleted_at.is_(None))
-            )
-            result = await db.execute(query)
-            valid_resource_ids = [row[0] for row in result.all()]
-            
-            invalid_ids = set(task.resource_ids) - set(valid_resource_ids)
-            
-            if invalid_ids:
-                invalid_ids_str = ", ".join(invalid_ids)
-                errors.append({
-                    "task_index": task_index,
-                    "error": f"资源id [{invalid_ids_str}] 不可用或已删除",
-                })
-                continue
-            
-            resource_ids_to_upload = valid_resource_ids
-        else:
-            #用户提交的直接是资源id
-            query = (
-                select(GeneratedResource.id)
-                .where(GeneratedResource.id.in_(task.resource_ids))
-                .where(GeneratedResource.user_id == current_user.id)
-                .where(GeneratedResource.deleted_at.is_(None))
-            )
-            result = await db.execute(query)
-            valid_resource_ids = [row[0] for row in result.all()]
-            
-            invalid_ids = set(task.resource_ids) - set(valid_resource_ids)
-            
-            if invalid_ids:
-                invalid_ids_str = ", ".join(invalid_ids)
-                errors.append({
-                    "task_index": task_index,
-                    "error": f"资源id [{invalid_ids_str}] 不可用或已删除",
-                })
-                continue
-            
-            resource_ids_to_upload = valid_resource_ids
-        
-        for advertiser_id in task.advertiser_ids:
-            for resource_id in resource_ids_to_upload:
-                other_info = {}
-                if task.is_pre_test == "1":
-                    other_info["is_pre_test"] = task.is_pre_test
-                    other_info["pre_test_template"] = task.pre_test_template
-
-                task_id = generate_id()
-                upload_task = UploadTask(
-                    id=task_id,
-                    user_id=current_user.id,
-                    oauth_id=task.oauth_id,
-                    advertiser_id=advertiser_id,
-                    resource_id=resource_id,
-                    status=1,
-                    note=None,
-                    other_info=json.dumps(other_info) if other_info else None,
-                )
-                
-                db.add(upload_task)
-                await upload_queue.enqueue(task_id)
-                task_ids.append(task_id)
-    
-    await db.commit()
-    
-    message = "上传任务已提交"
-    if errors:
-        message = f"部分任务提交成功，{len(errors)} 个任务失败"
-    
-    return {
-        "code": 0,
-        "message": message,
-        "task_ids": task_ids,
-        "errors": errors,
-    }
-
-
 @router.post(
     "/batch-update-filename",
     summary="批量修改资源文件名",