""" Hugging Face Space: Video Model Evaluator 视频生成模型评估系统 - 支持 Prompt、模型、视频的评估和评分 """ import os import gradio as gr import pandas as pd from datetime import datetime, timedelta from typing import Optional, Dict, Any, List, Tuple import json import logging from huggingface_hub import HfApi import tempfile import time import asyncio import nest_asyncio import warnings # 抑制 asyncio 事件循环清理时的警告(这是 Gradio 6.0 的已知问题,不影响功能) warnings.filterwarnings('ignore', category=RuntimeWarning, module='asyncio') # 应用 nest_asyncio 以解决事件循环嵌套问题 nest_asyncio.apply() # 导入视频生成服务 from pollo_service_single import PolloAIService, get_pollo_service # 导入S3工具 from s3_utils import S3Utils # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 配置常量 MAX_DAILY_CALLS = 4 # 每个用户每天最多调用次数 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 = [ "sora/sora-2-pro", # Sora 2 pro (修正路径) "bytedance/seedance-pro", # Seedance Pro "google/veo3", # Veo 3.1 "kling-ai/kling-v2-6", # Kling 2.6 ] # 模型显示名称映射 MODEL_DISPLAY_NAMES = { "sora/sora-2-pro": "Sora 2 pro", "bytedance/seedance-pro": "Seedance Pro", "google/veo3": "Veo 3.1", "kling-ai/kling-v2-6": "Kling 2.6", } # 初始化 HF API hf_api = HfApi(token=HF_TOKEN) if HF_TOKEN else None # 初始化视频生成服务 video_service = None if API_KEY: os.environ['POLLO_API_KEY'] = API_KEY video_service = get_pollo_service("bytedance/seedance-pro") else: logger.warning("API_KEY 未设置,视频生成功能将不可用") # 初始化 S3 工具 s3_utils = S3Utils() class DatasetManager: """管理 Private Dataset 的读写操作""" def __init__(self, repo_id: str, hf_token: str): self.repo_id = repo_id self.hf_token = hf_token self.api = HfApi(token=hf_token) if hf_token else None def get_user_calls_today(self, username: str) -> int: """ 获取用户今天的调用次数(从调用计数文件读取) Args: username: Hugging Face 用户名 Returns: 今天的调用次数 """ if not self.api or not self.repo_id: logger.warning("Dataset API 未配置,无法检查调用次数") return 0 try: today = datetime.now().strftime("%Y-%m-%d") count_file = f"call_counts/{today}/{username}.json" try: # 下载计数文件 local_path = self.api.hf_hub_download( repo_id=self.repo_id, filename=count_file, repo_type="dataset", token=self.hf_token ) # 读取计数 with open(local_path, 'r', encoding='utf-8') as f: data = json.load(f) return data.get('count', 0) except Exception: # 文件不存在,返回0 return 0 except Exception as e: logger.error(f"获取用户调用次数失败: {e}") return 0 def increment_user_calls(self, username: str) -> bool: """ 增加用户今天的调用次数(点击生成视频时调用) Args: username: Hugging Face 用户名 Returns: 是否成功 """ if not self.api or not self.repo_id: logger.warning("Dataset API 未配置,无法更新调用次数") return False try: today = datetime.now().strftime("%Y-%m-%d") count_file = f"call_counts/{today}/{username}.json" # 获取当前计数 current_count = self.get_user_calls_today(username) new_count = current_count + 1 # 创建计数数据 count_data = { 'user': username, 'date': today, 'count': new_count, 'last_updated': datetime.now().isoformat() } # 保存到临时文件 import tempfile with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as f: json.dump(count_data, f, ensure_ascii=False, indent=2) temp_path = f.name try: # 上传到 Dataset self.api.upload_file( path_or_fileobj=temp_path, path_in_repo=count_file, repo_id=self.repo_id, repo_type="dataset", token=self.hf_token ) logger.info(f"用户 {username} 调用次数已更新: {new_count}") return True finally: # 清理临时文件 os.unlink(temp_path) except Exception as e: logger.error(f"更新用户调用次数失败: {e}") return False def get_user_history(self, username: str, limit: int = 10) -> List[Dict]: """ 获取用户的历史评分记录 Args: username: Hugging Face 用户名 limit: 返回记录数量限制 Returns: 历史记录列表 """ if not self.api or not self.repo_id: logger.warning("Dataset API 未配置,无法获取历史记录") return [] try: # 列出评分文件 files = self.api.list_repo_files( repo_id=self.repo_id, repo_type="dataset", token=self.hf_token ) # 筛选该用户的评分文件(按用户分组:evaluations/{username}/时间戳.jsonl) user_files = [f for f in files if f.startswith(f"evaluations/{username}/") and f.endswith('.jsonl')] # 按文件名倒序排序(文件名就是时间戳,自然排序) user_files.sort(reverse=True) history = [] for file_path in user_files[:limit]: # 只读需要的数量 try: # 下载文件 local_path = self.api.hf_hub_download( repo_id=self.repo_id, filename=file_path, repo_type="dataset", token=self.hf_token ) # 读取数据 with open(local_path, 'r', encoding='utf-8') as f: data = json.load(f) # 只返回该用户的记录 if data.get('user') == username: history.append({ 'timestamp': data.get('timestamp'), 'prompt': data.get('prompt', ''), 'scores': data.get('scores', {}), 'video_urls': data.get('video_urls', {}) }) if len(history) >= limit: break except Exception as e: logger.warning(f"读取历史记录失败 {file_path}: {e}") continue return history except Exception as e: logger.error(f"获取用户历史记录失败: {e}") return [] def save_evaluation( self, username: str, prompt: str, model_results: Dict[str, Any], scores: Dict[str, float], video_urls: Dict[str, str] ) -> bool: """ 保存评分数据到 Private Dataset Args: username: 用户名 prompt: 提示词 model_results: 模型结果字典 scores: 评分字典 {model_name: score} video_urls: 视频URL字典 {model_name: url} Returns: 是否保存成功 """ if not self.api or not self.repo_id: logger.warning("Dataset API 未配置,无法保存数据") return False try: # 准备数据 data = { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "user": username, "prompt": prompt, "scores": json.dumps(scores, ensure_ascii=False), "video_urls": json.dumps(video_urls, ensure_ascii=False), "model_results": json.dumps(model_results, ensure_ascii=False, default=str) } # 创建临时文件 with tempfile.NamedTemporaryFile(mode='w', suffix='.jsonl', delete=False, encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False) temp_file = f.name try: # 按用户分组存储(方便查询单个用户的历史记录) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"evaluations/{username}/{timestamp}.jsonl" # 上传到 Dataset self.api.upload_file( path_or_fileobj=temp_file, path_in_repo=filename, repo_id=self.repo_id, repo_type="dataset", token=self.hf_token ) logger.info(f"成功保存评分数据: {filename}") return True finally: # 清理临时文件 if os.path.exists(temp_file): os.remove(temp_file) except Exception as e: logger.error(f"保存评分数据失败: {e}") return False # 初始化 Dataset Manager dataset_manager = DatasetManager(DATASET_REPO_ID, HF_TOKEN) if DATASET_REPO_ID and HF_TOKEN else None def check_user_access(request: gr.Request) -> Tuple[str, bool]: """ 检查用户访问权限 Args: request: Gradio Request 对象 Returns: (username, has_access) 元组 """ if not request: return "", False # 获取登录用户名 # 在 Hugging Face Space 中,OAuth 登录后用户信息存储在 Starlette Request 的 session 中 username = None # 从 Starlette Request 的 session 中获取 OAuth 用户信息 if hasattr(request, 'request') and hasattr(request.request, 'session'): session = request.request.session # OAuth 用户信息在 session['oauth_info'] 中 if 'oauth_info' in session: oauth_info = session.get('oauth_info', {}) # 用户信息嵌套在 userinfo 字典中 userinfo = oauth_info.get('userinfo', {}) # 尝试多个可能的用户名字段 username = ( userinfo.get('preferred_username') or userinfo.get('name') or userinfo.get('sub') ) if username: logger.info(f"从 session['oauth_info']['userinfo'] 获取用户名: {username}") # 或者在 session['user'] 中 elif 'user' in session: user_info = session.get('user', {}) logger.info(f"user 内容: {user_info}") username = user_info.get('preferred_username') or user_info.get('name') or user_info.get('username') if username: logger.info(f"从 session['user'] 获取用户名: {username}") # 如果 session 中没有用户信息,打印调试信息 if not username: logger.warning("无法获取用户名,请确保:") logger.warning("1. 已在 Space Settings 中启用 OAuth (hf_oauth: true)") logger.warning("2. 用户已通过 'Login with Hugging Face' 按钮登录") if not username: return "", False # 检查今天的调用次数 if dataset_manager: calls_today = dataset_manager.get_user_calls_today(username) if calls_today >= MAX_DAILY_CALLS: return username, False 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: 提示词 input_image: 输入图片路径(可选) request: Gradio Request 对象 Returns: (status_message, model_results, video_urls) 元组 """ # 检查用户权限 username, has_access = check_user_access(request) if not has_access: if not username: return "❌ 请先登录 Hugging Face 账户", {}, {} else: calls_today = dataset_manager.get_user_calls_today(username) if dataset_manager else 0 return f"❌ 您今天的调用次数已用完({calls_today}/{MAX_DAILY_CALLS}),请明天再试", {}, {} if not video_service: return "❌ 视频生成服务未配置,请联系管理员", {}, {} if not prompt or not prompt.strip(): return "❌ 请输入提示词", {}, {} # 增加用户调用计数(在生成视频前就计数,避免失败后不计数的问题) if dataset_manager and username: try: dataset_manager.increment_user_calls(username) logger.info(f"用户 {username} 调用次数+1") except Exception as e: logger.warning(f"更新用户调用次数失败: {e}") try: # 处理图片上传(如果提供) image_url = None if input_image: # 上传图片到S3,获取公网URL(Pollo API需要URL) logger.info("上传图片到S3...") image_url = s3_utils.upload_image_from_path(input_image) if not image_url: return "❌ 图片上传到S3失败,请检查S3配置", {}, {} logger.info(f"图片已上传到S3: {image_url}") # 使用配置的模型列表 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 "生成完成" return status_message, model_results, video_urls except Exception as e: logger.error(f"生成视频异常: {e}") return f"❌ 生成视频失败: {str(e)}", {}, {} def submit_evaluation( prompt: str, ranks: Dict[str, int], model_results: Dict[str, Any], video_urls: Dict[str, str], request: gr.Request ) -> str: """ 提交排名 Args: prompt: 提示词 ranks: 排名字典 {model_name: rank} model_results: 模型结果 video_urls: 视频URL request: Gradio Request 对象 Returns: 状态消息 """ username, has_access = check_user_access(request) if not username: return "❌ 请先登录 Hugging Face 账户" if not ranks: return "❌ 请至少为一个模型选择排名" # 验证排名:检查是否有重复的排名 rank_values = [r for r in ranks.values() if r is not None] if len(rank_values) != len(set(rank_values)): return "❌ 排名不能重复!每个模型的排名必须不同" # 验证排名:如果有成功生成的视频,必须对所有成功的模型排名 successful_models = [m for m in video_urls.keys()] if len(rank_values) != len(successful_models): return f"❌ 请为所有 {len(successful_models)} 个成功生成的模型选择排名" try: # 保存到 Dataset if dataset_manager: success = dataset_manager.save_evaluation( username=username, prompt=prompt, model_results=model_results, scores=ranks, # 这里用ranks替代scores,但key名保持兼容 video_urls=video_urls ) if success: return f"✅ 排名已保存!感谢 {username} 的反馈" else: return "❌ 保存排名失败,请重试" else: return "❌ Dataset 未配置,无法保存排名" except Exception as e: logger.error(f"提交排名失败: {e}") return f"❌ 提交排名失败: {str(e)}" def get_user_info(request: gr.Request = None) -> str: """获取用户信息""" # 如果没有 request,返回提示 if not request: return "⏳ 正在检查登录状态..." username, has_access = check_user_access(request) if not username: return "❌ 请先登录 Hugging Face 账户\n\n提示:点击右上角的 'Login with Hugging Face' 按钮登录" calls_today = dataset_manager.get_user_calls_today(username) if dataset_manager else 0 remaining = max(0, MAX_DAILY_CALLS - calls_today) status_icon = "✅" if has_access else "❌" return f"{status_icon} 用户: {username}\n📊 今日已用: {calls_today}/{MAX_DAILY_CALLS}\n✨ 剩余次数: {remaining}" def get_history_html(request: gr.Request = None) -> str: """获取用户历史记录的HTML""" if not request: return "
⏳ 正在加载...
" username, _ = check_user_access(request) if not username: return "❌ 请先登录 Hugging Face 账户
" if not dataset_manager: return "❌ Dataset 未配置
" try: history = dataset_manager.get_user_history(username, limit=10) if not history: return f"📭 暂无历史记录(用户: {username})
" html = f"共 {len(history)} 条记录
" html += "提示词: {prompt[:100]}{'...' if len(prompt) > 100 else ''}
排名:
生成的视频:
" for model, url in video_urls.items(): display_name = MODEL_DISPLAY_NAMES.get(model, model) html += f'' html += "❌ 获取历史记录失败: {str(e)}
" # 创建 Gradio 界面 with gr.Blocks(title="Video Model Evaluator") as demo: gr.Markdown(""" # 🎬 视频生成模型评估系统 欢迎使用视频生成模型评估系统!请先登录您的 Hugging Face 账户。 **功能说明:** - 每个用户每天最多可调用 4 次 - 支持多个模型同时生成视频(Sora 2 pro, Seedance Pro, Veo 3.1, Kling 2.6) - 可以对生成结果进行排名比较 - 排名数据将保存到 Private Dataset **评估流程:** 1. 上传输入图片(可选) 2. 输入提示词(Prompt) 3. 系统调用多个模型生成视频 4. 对每个模型的生成结果进行排名(1=最好,4=最差) 5. 提交排名,数据自动保存 """) # 添加 HF 登录按钮(Gradio 6.0 OAuth 支持) # 只在 Space 环境中显示(本地测试时会失败) if os.getenv("SPACE_ID"): # SPACE_ID 只在 HF Space 中存在 login_btn = gr.LoginButton() # 用户信息区域(独立一行,更显眼) with gr.Row(): user_info = gr.Textbox( label="👤 用户状态(点击生成视频时统计次数,提交评分不额外计数)", interactive=False, value="⏳ 等待检查登录状态...", scale=9, lines=2 ) refresh_user_btn = gr.Button("🔄 刷新", size="sm", scale=1) with gr.Tabs(): with gr.Tab("🎬 生成视频"): with gr.Row(): with gr.Column(scale=1): prompt_input = gr.Textbox( label="提示词 (Prompt)", placeholder="输入视频生成的提示词...", lines=3 ) input_image = gr.Image( label="输入图片 (可选,用于图生视频 i2v)", type="filepath", sources=["upload"] ) generate_btn = gr.Button("🚀 生成视频", variant="primary", size="lg") status_output = gr.Textbox( label="生成状态", interactive=False, lines=5 ) with gr.Column(scale=1): # 视频展示区域 gr.Markdown("### 🎬 生成的视频") # 动态创建视频和排名组件 video_components = {} rank_components = {} for model_name in MODELS_TO_CALL: display_name = MODEL_DISPLAY_NAMES.get(model_name, model_name) video_components[model_name] = gr.HTML( label=display_name, visible=False ) rank_components[model_name] = gr.Radio( label=f"{display_name} 排名", choices=["1 (最好)", "2", "3", "4 (最差)"], value=None, visible=False ) # 排序区域 gr.Markdown("### 🏆 排名 (选择每个视频的排名)") gr.Markdown("💡 提示:给每个模型选择排名(1=最好,4=最差),不能有重复排名") submit_btn = gr.Button("💾 提交排名", variant="secondary", visible=False) submit_status = gr.Textbox( label="提交状态", interactive=False ) # 历史记录Tab with gr.Tab("📜 历史记录"): gr.Markdown(""" ### 📜 我的生成历史 这里显示您最近的视频生成记录(最多10条) """) history_refresh_btn = gr.Button("🔄 刷新历史记录", variant="primary") history_output = gr.HTML(label="历史记录", value="点击刷新按钮加载历史记录
") # 存储中间数据 model_results_state = gr.State({}) video_urls_state = gr.State({}) prompt_state = gr.State("") # 事件处理 def on_generate(prompt, image, request: gr.Request): # 更新用户信息 user_info_text = get_user_info(request) # 生成视频 status, results, urls = generate_videos(prompt, image, request) # 构建输出列表 outputs = [user_info_text, status] has_results = len(urls) > 0 # 为每个模型添加视频和评分组件更新 for model_name in MODELS_TO_CALL: video_value = urls.get(model_name, None) # 返回视频组件更新(使用HTML video标签避免卡顿) # 重要:每次都要明确设置value,确保覆盖之前的内容 if video_value: # 使用HTML video标签,更可控且不会卡顿 display_name = MODEL_DISPLAY_NAMES.get(model_name, model_name) video_html = f""" """ outputs.append(gr.update(value=video_html, visible=True)) # 该模型成功,显示排名选择器 outputs.append(gr.update(visible=True)) else: # 该模型失败或未生成,清空并隐藏 display_name = MODEL_DISPLAY_NAMES.get(model_name, model_name) failed_html = f"""视频生成失败或未完成