feat: 并行视频生成 + 更新数据集路径

- 更新数据集路径: video-model-evaluator-cuti/video-evaluations
- 添加异步并行视频生成，使用 asyncio.gather
- 限制 Pollo API 并发数为 5（使用 Semaphore）
- 性能提升：4个模型从串行20+分钟降至并行5-6分钟
- 提交阶段快速并行，轮询阶段受信号量限制

app.py

@@ -35,9 +35,10 @@ logger = logging.getLogger(__name__)
 
 # 配置常量
 MAX_DAILY_CALLS = 4  # 每个用户每天最多调用次数
-DATASET_REPO_ID = "learnmlf/video-evaluations"  # Private Dataset 名称
+DATASET_REPO_ID = "video-model-evaluator-cuti/video-evaluations"  # Private Dataset 名称
 HF_TOKEN = os.getenv("HF_TOKEN", "")  # 从 Space Settings 获取
 API_KEY = os.getenv("API_KEY", "")  # 从 Space Settings 获取
+MAX_POLLO_CONCURRENCY = 5  # Pollo API 最大并发数
 
 # 支持的模型列表
 MODELS_TO_CALL = [
@@ -375,9 +376,145 @@ def check_user_access(request: gr.Request) -> Tuple[str, bool]:
     return username, True
 
 
+async def _generate_single_video_async(
+    model_name: str,
+    prompt: str,
+    image_url: Optional[str],
+    semaphore: asyncio.Semaphore
+) -> Tuple[str, Dict[str, Any], Optional[str], str]:
+    """
+    异步生成单个模型的视频（使用信号量限制并发）
+    
+    Args:
+        model_name: 模型名称
+        prompt: 提示词
+        image_url: 图片URL（可选）
+        semaphore: asyncio信号量，用于限制并发数
+        
+    Returns:
+        (model_name, model_result, video_url, status_message) 元组
+    """
+    display_name = MODEL_DISPLAY_NAMES.get(model_name, model_name)
+    
+    try:
+        logger.info(f"开始生成视频: {display_name} ({model_name}), 提示词: {prompt[:50]}...")
+        
+        # 获取对应模型的服务实例
+        service = get_pollo_service(model_name)
+        
+        # 根据是否有图片选择模式
+        mode = "i2v" if image_url else "t2v"
+        
+        # 提交任务（快速，不需要限制并发）
+        loop = asyncio.get_event_loop()
+        result = await loop.run_in_executor(
+            None,
+            lambda: service.generate_video(
+                prompt=prompt,
+                mode=mode,
+                input_image_path=image_url if image_url else None,
+                video_length=5,
+                width=1280,
+                height=720
+            )
+        )
+        
+        task_id = result.get('pollo_task_id')
+        if not task_id:
+            raise Exception("未获取到任务ID")
+        
+        logger.info(f"{display_name}: 任务已提交，task_id={task_id}")
+        
+        # 使用信号量限制轮询并发数
+        async with semaphore:
+            logger.info(f"{display_name}: 开始轮询（当前并发槽位已占用）")
+            
+            # 轮询任务结果
+            max_polls = 60
+            poll_interval = 10
+            
+            for i in range(max_polls):
+                # 在线程池中执行同步的轮询操作
+                poll_result = await loop.run_in_executor(
+                    None,
+                    service.poll_task_result,
+                    task_id
+                )
+                
+                if poll_result['status'] == 'completed':
+                    pollo_video_url = poll_result.get('video_url')
+                    if pollo_video_url:
+                        # 下载视频并上传到S3（在线程池中执行）
+                        logger.info(f"{display_name}: 下载视频并上传到S3: {pollo_video_url}")
+                        s3_video_url = await loop.run_in_executor(
+                            None,
+                            s3_utils.download_and_upload_video,
+                            pollo_video_url
+                        )
+                        
+                        if s3_video_url:
+                            model_result = {
+                                'status': 'success',
+                                'task_id': task_id,
+                                'video_url': s3_video_url,
+                                'pollo_video_url': pollo_video_url
+                            }
+                            status_message = f"✅ {display_name}: 生成成功并已保存到S3"
+                            logger.info(f"{display_name}: 完成，释放并发槽位")
+                            return model_name, model_result, s3_video_url, status_message
+                        else:
+                            # 如果S3上传失败，使用原始URL
+                            logger.warning(f"{display_name}: S3上传失败，使用原始URL: {pollo_video_url}")
+                            model_result = {
+                                'status': 'success',
+                                'task_id': task_id,
+                                'video_url': pollo_video_url,
+                                'warning': 'S3上传失败，使用临时URL'
+                            }
+                            status_message = f"✅ {display_name}: 生成成功（S3上传失败）"
+                            logger.info(f"{display_name}: 完成，释放并发槽位")
+                            return model_name, model_result, pollo_video_url, status_message
+                    break
+                    
+                elif poll_result['status'] == 'failed':
+                    error_msg = poll_result.get('error_message', '未知错误')
+                    model_result = {
+                        'status': 'failed',
+                        'error': error_msg
+                    }
+                    status_message = f"❌ {display_name}: {error_msg}"
+                    logger.info(f"{display_name}: 失败，释放并发槽位")
+                    return model_name, model_result, None, status_message
+                    
+                else:
+                    # 处理中，继续等待
+                    if i == max_polls - 1:
+                        model_result = {
+                            'status': 'timeout',
+                            'error': '任务超时'
+                        }
+                        status_message = f"⏱️ {display_name}: 任务超时"
+                        logger.info(f"{display_name}: 超时，释放并发槽位")
+                        return model_name, model_result, None, status_message
+                    else:
+                        await asyncio.sleep(poll_interval)
+        
+        # 如果没有返回结果，说明出现异常
+        raise Exception("轮询未返回有效结果")
+        
+    except Exception as e:
+        logger.error(f"生成视频失败 ({display_name}): {e}")
+        model_result = {
+            'status': 'error',
+            'error': str(e)
+        }
+        status_message = f"❌ {display_name}: {str(e)}"
+        return model_name, model_result, None, status_message
+
+
 def generate_videos(prompt: str, input_image: Optional[str], request: gr.Request) -> Tuple[str, Dict[str, Any], Dict[str, str]]:
     """
-    生成视频（调用多个模型）
+    生成视频（并行调用多个模型，限制Pollo API并发数为5）
     
     Args:
         prompt: 提示词
@@ -412,115 +549,54 @@ def generate_videos(prompt: str, input_image: Optional[str], request: gr.Request
             logger.warning(f"更新用户调用次数失败: {e}")
     
     try:
-        model_results = {}
-        video_urls = {}
-        
-        # 使用配置的模型列表
-        models = MODELS_TO_CALL
-        
-        status_messages = []
-        
         # 处理图片上传（如果提供）
-        image_path = None
         image_url = None
         if input_image:
-            # Gradio 返回的是临时文件路径
-            image_path = input_image
-            
             # 上传图片到S3，获取公网URL（Pollo API需要URL）
             logger.info("上传图片到S3...")
-            image_url = s3_utils.upload_image_from_path(image_path)
+            image_url = s3_utils.upload_image_from_path(input_image)
             
             if not image_url:
                 return "❌ 图片上传到S3失败，请检查S3配置", {}, {}
             
             logger.info(f"图片已上传到S3: {image_url}")
         
-        for model_name in models:
-            try:
-                display_name = MODEL_DISPLAY_NAMES.get(model_name, model_name)
-                logger.info(f"开始生成视频: {display_name} ({model_name}), 提示词: {prompt[:50]}...")
-                
-                # 获取对应模型的服务实例
-                service = get_pollo_service(model_name)
-                
-                # 根据是否有图片选择模式
-                mode = "i2v" if image_url else "t2v"
-                
-                # 使用S3 URL而不是本地路径
-                result = service.generate_video(
-                    prompt=prompt,
-                    mode=mode,  # 根据是否有图片自动选择 i2v 或 t2v
-                    input_image_path=image_url if image_url else None,  # 使用S3 URL
-                    video_length=5,
-                    width=1280,
-                    height=720
-                )
-                
-                task_id = result.get('pollo_task_id')
-                if task_id:
-                    # 轮询任务结果
-                    max_polls = 60
-                    poll_interval = 10
-                    
-                    for i in range(max_polls):
-                        poll_result = service.poll_task_result(task_id)
-                        
-                        if poll_result['status'] == 'completed':
-                            pollo_video_url = poll_result.get('video_url')
-                            if pollo_video_url:
-                                # 下载视频并上传到S3（Pollo的视频只保存一段时间）
-                                logger.info(f"下载视频并上传到S3: {pollo_video_url}")
-                                s3_video_url = s3_utils.download_and_upload_video(pollo_video_url)
-                                
-                                if s3_video_url:
-                                    video_urls[model_name] = s3_video_url
-                                    model_results[model_name] = {
-                                        'status': 'success',
-                                        'task_id': task_id,
-                                        'video_url': s3_video_url,
-                                        'pollo_video_url': pollo_video_url  # 保留原始URL
-                                    }
-                                    status_messages.append(f"✅ {display_name}: 生成成功并已保存到S3")
-                                else:
-                                    # 如果S3上传失败，使用原始URL
-                                    logger.warning(f"S3上传失败，使用原始URL: {pollo_video_url}")
-                                    video_urls[model_name] = pollo_video_url
-                                    model_results[model_name] = {
-                                        'status': 'success',
-                                        'task_id': task_id,
-                                        'video_url': pollo_video_url,
-                                        'warning': 'S3上传失败，使用临时URL'
-                                    }
-                                    status_messages.append(f"✅ {display_name}: 生成成功（S3上传失败）")
-                                break
-                        elif poll_result['status'] == 'failed':
-                            error_msg = poll_result.get('error_message', '未知错误')
-                            model_results[model_name] = {
-                                'status': 'failed',
-                                'error': error_msg
-                            }
-                            status_messages.append(f"❌ {display_name}: {error_msg}")
-                            break
-                        else:
-                            # 处理中，继续等待
-                            if i == max_polls - 1:
-                                model_results[model_name] = {
-                                    'status': 'timeout',
-                                    'error': '任务超时'
-                                }
-                                status_messages.append(f"⏱️ {display_name}: 任务超时")
-                            else:
-                                time.sleep(poll_interval)
-                
-            except Exception as e:
-                display_name = MODEL_DISPLAY_NAMES.get(model_name, model_name)
-                logger.error(f"生成视频失败 ({display_name}): {e}")
-                model_results[model_name] = {
-                    'status': 'error',
-                    'error': str(e)
-                }
-                status_messages.append(f"❌ {display_name}: {str(e)}")
+        # 使用配置的模型列表
+        models = MODELS_TO_CALL
+        
+        # 创建信号量限制并发数
+        semaphore = asyncio.Semaphore(MAX_POLLO_CONCURRENCY)
+        
+        # 创建异步任务列表
+        async def run_parallel_generation():
+            tasks = [
+                _generate_single_video_async(model_name, prompt, image_url, semaphore)
+                for model_name in models
+            ]
+            # 并行执行所有任务
+            return await asyncio.gather(*tasks, return_exceptions=True)
+        
+        # 运行异步任务
+        logger.info(f"开始并行生成视频，最大并发数: {MAX_POLLO_CONCURRENCY}")
+        results = asyncio.run(run_parallel_generation())
+        
+        # 整理结果
+        model_results = {}
+        video_urls = {}
+        status_messages = []
+        
+        for result in results:
+            if isinstance(result, Exception):
+                # 捕获异常
+                logger.error(f"任务执行异常: {result}")
+                status_messages.append(f"❌ 任务异常: {str(result)}")
+            else:
+                # 正常结果
+                model_name, model_result, video_url, status_message = result
+                model_results[model_name] = model_result
+                if video_url:
+                    video_urls[model_name] = video_url
+                status_messages.append(status_message)
         
         status_message = "\n".join(status_messages) if status_messages else "生成完成"