app.py

import gradio as gr
import os
import json
from datetime import datetime
from huggingface_hub import hf_hub_download
import random
import csv

# 全局变量存储数据集
VIDEO_DATA = None
REPO_ID = "WenjiaWang/videoforuser"

# 从Hugging Face dataset加载视频
def load_videos_from_huggingface():
    global VIDEO_DATA
    
    try:
        print("正在加载数据集: WenjiaWang/videoforuser...")
        
        # 定义视频文件列表（手动指定，避免列出所有文件）
        # 根据你的数据集结构，这里列出所有场景和方法
        scenes_methods = {
            "0O_YyxcC-kA_scene-35_2": ["gen3c", "svc", "trajattn", "stage12_new_epoch-9"],
            "3IC8kkbZbsk_scene-61_2": ["gen3c", "svc", "trajattn", "stage12_new_epoch-9"],
            "3JZFI0yKr6Y_scene-940_2": ["gen3c", "svc", "trajattn", "stage12_new_epoch-9"],
            "9Ixk_PSj__Q_scene-92_2": ["gen3c", "svc", "trajattn", "stage12_new_epoch-9"],
            "CEPKsNuSDiQ_scene-31_1": ["gen3c", "svc", "trajattn", "stage12_new_epoch-9"],
            "CEPKsNuSDiQ_scene-50_1": ["gen3c", "svc", "trajattn", "stage12_new_epoch-9"],
            "F4bBJG9UKL0_scene-35_2": ["gen3c", "svc", "trajattn", "stage12_new_epoch-9"],
            "I3EG5SDZn1s_scene-101_2": ["gen3c", "svc", "trajattn", "stage12_new_epoch-9"],
            "RTaUgpxLBag_scene-284_1": ["gen3c", "svc", "trajattn", "stage12_new_epoch-9"],
            "c4OnR033eMk_scene-158_3": ["gen3c", "svc", "trajattn", "stage12_new_epoch-9"],
        }
        
        # 组织视频数据：按场景分组
        VIDEO_DATA = {}
        
        for scene_name, methods in scenes_methods.items():
            VIDEO_DATA[scene_name] = {}
            
            for method_name in methods:
                video_filename = f"{method_name}.mp4"
                video_path = f"{scene_name}/{video_filename}"
                
                try:
                    # 下载视频文件到本地缓存
                    print(f"下载视频: {video_path}")
                    local_path = hf_hub_download(
                        repo_id=REPO_ID,
                        filename=video_path,
                        repo_type="dataset"
                    )
                    
                    # 存储视频信息
                    VIDEO_DATA[scene_name][method_name] = {
                        'path': video_path,
                        'local_path': local_path
                    }
                    print(f"✓ 已下载: {video_path} -> {local_path}")
                    
                except Exception as e:
                    print(f"✗ 下载失败: {video_path} - {e}")
                    continue
        
        print(f"组织完成，共 {len(VIDEO_DATA)} 个场景")
        return True
        
    except Exception as e:
        print(f"加载数据集失败: {e}")
        import traceback
        traceback.print_exc()
        return False

# 获取所有场景列表
def get_question_folders():
    if VIDEO_DATA is None:
        success = load_videos_from_huggingface()
        if not success:
            return []
    
    return sorted(list(VIDEO_DATA.keys()))

# 获取某个场景的所有视频
def get_videos_for_question(scene_name):
    if VIDEO_DATA is None or scene_name not in VIDEO_DATA:
        return {}, {}
    
    scene_videos = VIDEO_DATA[scene_name]
    
    # 创建方法名到真实名称的映射
    method_names = list(scene_videos.keys())
    
    # 随机打乱顺序以匿名化
    shuffled_methods = method_names.copy()
    random.shuffle(shuffled_methods)
    
    videos = {}
    method_mapping = {}
    
    for i, method_name in enumerate(shuffled_methods):
        display_name = f"Method {chr(65+i)}"  # Method A, B, C, D
        
        # 获取视频数据
        video_info = scene_videos[method_name]
        
        # 使用本地路径
        videos[display_name] = video_info['local_path']
        
        method_mapping[display_name] = method_name
    
    return videos, method_mapping

# 保存评分数据
def save_ratings(scene_name, ratings_data, method_mapping):
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    
    # 将显示名称映射到真实方法名
    mapped_ratings = {}
    for display_name, ratings in ratings_data.items():
        real_method = method_mapping.get(display_name, display_name)
        mapped_ratings[real_method] = ratings
    
    # 读取现有数据
    all_data = []
    if os.path.exists("ratings_data.json"):
        try:
            with open("ratings_data.json", "r", encoding="utf-8") as f:
                all_data = json.load(f)
        except:
            all_data = []
    
    # 添加新数据
    entry = {
        "timestamp": timestamp,
        "scene": scene_name,
        "ratings": mapped_ratings
    }
    all_data.append(entry)
    
    # 保存数据
    with open("ratings_data.json", "w", encoding="utf-8") as f:
        json.dump(all_data, f, ensure_ascii=False, indent=2)
    
    return f"✓ 评分已保存 / Ratings saved"

# 创建Gradio界面
def create_video_survey_app():
    # 预加载数据集
    print("初始化应用...")
    load_videos_from_huggingface()
    question_folders = get_question_folders()
    
    if not question_folders:
        print("错误: 没有找到任何场景数据")
        return None
    
    print(f"找到 {len(question_folders)} 个场景")
    
    with gr.Blocks(title="视频生成质量用户研究", theme=gr.themes.Soft()) as demo:
        # 问卷内容区域
        with gr.Column(visible=True) as survey_area:
            gr.Markdown("# 🎬 视频生成质量用户研究 / Video Generation Quality User Study")
            gr.Markdown("""
            ### 说明 / Instructions:
            - 请观看每个视频并进行评分 / Please watch each video and rate them
            - **💡 提示：可以点击视频右下角全屏按钮观看清晰细节** / Tip: Click fullscreen button to view details
            - 评分标准 / Rating criteria:
              - **动态生成质量** / Dynamic Generation Quality: 视频中物体运动的流畅性和真实性
              - **静态一致性** / Static Consistency: 视频中静态物体的稳定性和一致性
              - **整体质量** / Overall Quality: 视频的整体观感
            - 评分范围：1-5分（5分最好）/ Rating scale: 1-5 (5 = Best)
            - 完成所有评分后，点击底部的"提交所有评分"按钮 / After rating all videos, click "Submit All Ratings" at the bottom
            """)
            
            # 存储所有评分组件
            all_scene_components = []
            
            # 为每个场景创建界面
            for scene_idx, scene_name in enumerate(question_folders):
                videos, method_mapping = get_videos_for_question(scene_name)
                
                with gr.Group():
                    gr.Markdown(f"## 场景 {scene_idx + 1} / Scene {scene_idx + 1}: `{scene_name}`")
                
                # 4个视频横向排列
                with gr.Row(equal_height=True, variant="compact"):
                    scene_ratings = []
                    for i in range(4):
                        method_name = f"Method {chr(65+i)}"
                        
                        with gr.Column(scale=1, min_width=0):
                            gr.Markdown(f"### 🎥 {method_name}")
                            
                            # 视频 - 使用 autoplay=False 确保一致性
                            if method_name in videos:
                                video = gr.Video(
                                    value=videos[method_name], 
                                    label="", 
                                    height=400,
                                    autoplay=False,
                                    show_label=False
                                )
                            else:
                                video = gr.Video(
                                    value=None, 
                                    label="", 
                                    height=400, 
                                    visible=False,
                                    show_label=False
                                )
                            
                            # 三个评分滑块（垂直排列）
                            dynamic = gr.Slider(
                                minimum=1, maximum=5, step=1, value=3,
                                label="动态质量 / Dynamic"
                            )
                            static = gr.Slider(
                                minimum=1, maximum=5, step=1, value=3,
                                label="静态一致性 / Static"
                            )
                            overall = gr.Slider(
                                minimum=1, maximum=5, step=1, value=3,
                                label="整体质量 / Overall"
                            )
                            
                            scene_ratings.append({
                                "method": method_name,
                                "dynamic": dynamic,
                                "static": static,
                                "overall": overall
                            })
                
                    all_scene_components.append({
                        "scene_name": scene_name,
                        "method_mapping": method_mapping,
                        "ratings": scene_ratings
                    })
            
            # 提交按钮
            gr.Markdown("---")
            submit_btn = gr.Button("📤 提交所有评分 / Submit All Ratings", variant="primary", size="lg")
        
        # 感谢页面（初始隐藏）
        with gr.Column(visible=False) as thank_you_page:
            gr.Markdown("""
            # ✅ 提交成功！/ Submission Successful!
            
            ## 🎉 感谢您的参与！/ Thank you for participating!
            
            您的评分已成功保存。您的反馈对我们非常重要！
            
            Your ratings have been saved successfully. Your feedback is very important to us!
            
            ---
            
            您现在可以关闭此页面。
            
            You may now close this page.
            """)
        
        status_output = gr.Markdown("")
        
        # 提交函数
        def submit_all_ratings(*all_rating_values):
            try:
                # 解析所有评分
                value_idx = 0
                for scene_data in all_scene_components:
                    scene_name = scene_data["scene_name"]
                    method_mapping = scene_data["method_mapping"]
                    
                    ratings = {}
                    for rating_comp in scene_data["ratings"]:
                        method_name = rating_comp["method"]
                        ratings[method_name] = {
                            "dynamic_quality": all_rating_values[value_idx],
                            "static_consistency": all_rating_values[value_idx + 1],
                            "overall_quality": all_rating_values[value_idx + 2]
                        }
                        value_idx += 3
                    
                    # 保存评分
                    save_ratings(scene_name, ratings, method_mapping)
                
                # 隐藏问卷，显示感谢页面
                return gr.update(visible=False), gr.update(visible=True), ""
            except Exception as e:
                return gr.update(visible=True), gr.update(visible=False), f"❌ **保存失败** / Save failed: {str(e)}"
        
        # 收集所有评分输入
        all_rating_inputs = []
        for scene_data in all_scene_components:
            for rating_comp in scene_data["ratings"]:
                all_rating_inputs.extend([
                    rating_comp["dynamic"],
                    rating_comp["static"],
                    rating_comp["overall"]
                ])
        
        # 绑定提交按钮
        submit_btn.click(
            submit_all_ratings,
            inputs=all_rating_inputs,
            outputs=[survey_area, thank_you_page, status_output]
        )
    
    return demo

if __name__ == "__main__":
    app = create_video_survey_app()
    if app:
        # 添加密码保护的管理员结果查看和分析页面
        with gr.Blocks() as admin_app:
            gr.Markdown("# 🔐 管理员登录")
            gr.Markdown("请输入管理员密码查看结果")
            
            # 管理员密码（从环境变量读取，保护隐私）
            # 在 Hugging Face Space 设置中添加 Secret: ADMIN_PASSWORD
            ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin123")  # 默认密码仅用于本地测试
            
            with gr.Column(visible=True) as login_box:
                password_input = gr.Textbox(
                    label="密码",
                    type="password",
                    placeholder="请输入管理员密码"
                )
                login_btn = gr.Button("🔓 登录", variant="primary")
                login_status = gr.Markdown("")
            
            with gr.Column(visible=False) as admin_content:
                gr.Markdown("# 📊 管理员 - 查看结果")
                gr.Markdown("查看所有用户提交的评分数据，并下载原始 JSON 文件")
                
                def load_results():
                    if os.path.exists("ratings_data.json"):
                        with open("ratings_data.json", "r", encoding="utf-8") as f:
                            data = json.load(f)
                        
                        # 格式化显示
                        formatted_text = f"📊 总共收集了 {len(data)} 条评分记录\n\n"
                        formatted_text += "=" * 80 + "\n\n"
                        
                        for i, entry in enumerate(data, 1):
                            formatted_text += f"【记录 {i}】\n"
                            formatted_text += f"时间: {entry['timestamp']}\n"
                            formatted_text += f"场景: {entry['scene']}\n"
                            formatted_text += f"评分:\n"
                            
                            for method, ratings in entry['ratings'].items():
                                formatted_text += f"  • {method}:\n"
                                formatted_text += f"    - 动态质量: {ratings['dynamic_quality']}\n"
                                formatted_text += f"    - 静态一致性: {ratings['static_consistency']}\n"
                                formatted_text += f"    - 整体质量: {ratings['overall_quality']}\n"
                            
                            formatted_text += "\n" + "-" * 80 + "\n\n"
                        
                        return formatted_text, "ratings_data.json"
                    else:
                        return "📭 暂无数据\n\n还没有用户提交评分。", None
                
                def get_statistics():
                    if not os.path.exists("ratings_data.json"):
                        return "📭 暂无数据", None, None
                    
                    with open("ratings_data.json", "r", encoding="utf-8") as f:
                        data = json.load(f)
                    
                    if not data:
                        return "📭 暂无数据", None, None
                    
                    # 统计每个方法的平均分
                    method_stats = {}
                    scene_stats = {}
                    
                    for entry in data:
                        scene = entry['scene']
                        if scene not in scene_stats:
                            scene_stats[scene] = {}
                        
                        for method, ratings in entry['ratings'].items():
                            if method not in method_stats:
                                method_stats[method] = {
                                    'dynamic': [],
                                    'static': [],
                                    'overall': []
                                }
                            
                            method_stats[method]['dynamic'].append(ratings['dynamic_quality'])
                            method_stats[method]['static'].append(ratings['static_consistency'])
                            method_stats[method]['overall'].append(ratings['overall_quality'])
                            
                            # 按场景统计
                            if method not in scene_stats[scene]:
                                scene_stats[scene][method] = {
                                    'dynamic': [],
                                    'static': [],
                                    'overall': []
                                }
                            scene_stats[scene][method]['dynamic'].append(ratings['dynamic_quality'])
                            scene_stats[scene][method]['static'].append(ratings['static_consistency'])
                            scene_stats[scene][method]['overall'].append(ratings['overall_quality'])
                    
                    # 计算平均值和提交人数
                    num_submissions = len(data)
                    stats_text = f"📈 统计分析\n\n"
                    stats_text += f"👥 **提交人数**: {num_submissions}\n\n"
                    stats_text += "=" * 80 + "\n\n"
                    stats_text += "【总体统计 - 按方法】\n\n"
                    
                    for method, scores in sorted(method_stats.items()):
                        avg_dynamic = sum(scores['dynamic']) / len(scores['dynamic'])
                        avg_static = sum(scores['static']) / len(scores['static'])
                        avg_overall = sum(scores['overall']) / len(scores['overall'])
                        avg_total = (avg_dynamic + avg_static + avg_overall) / 3
                        
                        stats_text += f"🎬 {method}\n"
                        stats_text += f"  样本数: {len(scores['dynamic'])} (来自 {num_submissions} 人)\n"
                        stats_text += f"  动态质量: {avg_dynamic:.2f}\n"
                        stats_text += f"  静态一致性: {avg_static:.2f}\n"
                        stats_text += f"  整体质量: {avg_overall:.2f}\n"
                        stats_text += f"  综合平均: {avg_total:.2f}\n"
                        stats_text += "\n"
                    
                    # 生成CSV文件 - 总体统计
                    csv_summary_path = "ratings_summary.csv"
                    with open(csv_summary_path, 'w', newline='', encoding='utf-8-sig') as f:
                        writer = csv.writer(f)
                        writer.writerow(['方法', '样本数', '动态质量均分', '静态一致性均分', '整体质量均分', '综合均分'])
                        
                        for method, scores in sorted(method_stats.items()):
                            avg_dynamic = sum(scores['dynamic']) / len(scores['dynamic'])
                            avg_static = sum(scores['static']) / len(scores['static'])
                            avg_overall = sum(scores['overall']) / len(scores['overall'])
                            avg_total = (avg_dynamic + avg_static + avg_overall) / 3
                            
                            writer.writerow([
                                method,
                                len(scores['dynamic']),
                                f"{avg_dynamic:.2f}",
                                f"{avg_static:.2f}",
                                f"{avg_overall:.2f}",
                                f"{avg_total:.2f}"
                            ])
                    
                    # 生成CSV文件 - 详细数据
                    csv_detail_path = "ratings_detail.csv"
                    with open(csv_detail_path, 'w', newline='', encoding='utf-8-sig') as f:
                        writer = csv.writer(f)
                        writer.writerow(['时间戳', '场景', '方法', '动态质量', '静态一致性', '整体质量'])
                        
                        for entry in data:
                            timestamp = entry['timestamp']
                            scene = entry['scene']
                            for method, ratings in entry['ratings'].items():
                                writer.writerow([
                                    timestamp,
                                    scene,
                                    method,
                                    ratings['dynamic_quality'],
                                    ratings['static_consistency'],
                                    ratings['overall_quality']
                                ])
                    
                    return stats_text, csv_summary_path, csv_detail_path
                
                with gr.Tabs():
                    with gr.Tab("📊 统计分析"):
                        with gr.Row():
                            stats_refresh_btn = gr.Button("🔄 刷新统计", variant="primary")
                            clear_data_btn = gr.Button("🗑️ 清空所有数据", variant="stop")
                        
                        stats_display = gr.Textbox(
                            label="统计结果",
                            lines=25,
                            max_lines=50,
                            interactive=False
                        )
                        
                        with gr.Row():
                            csv_summary_file = gr.File(label="📥 下载汇总统计 CSV", interactive=False)
                            csv_detail_file = gr.File(label="📥 下载详细数据 CSV", interactive=False)
                            json_file = gr.File(label="📥 下载原始 JSON", interactive=False)
                        
                        gr.Markdown("""
                        **说明**：
                        - **汇总统计 CSV**：每个方法的平均分和样本数
                        - **详细数据 CSV**：所有原始评分记录，可用于进一步分析
                        - **原始 JSON**：完整的原始数据
                        - **清空数据**：删除所有评分记录（谨慎操作！）
                        """)
                        
                        # 清空数据函数
                        def clear_all_data():
                            try:
                                if os.path.exists("ratings_data.json"):
                                    os.remove("ratings_data.json")
                                return "✅ 所有数据已清空", None, None, None
                            except Exception as e:
                                return f"❌ 清空失败: {str(e)}", None, None, None
                        
                        # 获取统计和文件
                        def get_stats_and_files():
                            stats, csv_summary, csv_detail = get_statistics()
                            json_path = "ratings_data.json" if os.path.exists("ratings_data.json") else None
                            return stats, csv_summary, csv_detail, json_path
                        
                        stats_refresh_btn.click(
                            get_stats_and_files,
                            outputs=[stats_display, csv_summary_file, csv_detail_file, json_file]
                        )
                        
                        clear_data_btn.click(
                            clear_all_data,
                            outputs=[stats_display, csv_summary_file, csv_detail_file, json_file]
                        )
            
            # 登录验证
            def check_password(password):
                if password == ADMIN_PASSWORD:
                    stats, csv_summary, csv_detail = get_statistics()
                    json_path = "ratings_data.json" if os.path.exists("ratings_data.json") else None
                    return (
                        gr.Column(visible=False),  # 隐藏登录框
                        gr.Column(visible=True),   # 显示管理内容
                        "✅ 登录成功！",
                        stats,
                        csv_summary,
                        csv_detail,
                        json_path
                    )
                else:
                    return (
                        gr.Column(visible=True),   # 保持登录框显示
                        gr.Column(visible=False),  # 隐藏管理内容
                        "❌ 密码错误，请重试",
                        "", None, None, None
                    )
            
            login_btn.click(
                check_password,
                inputs=[password_input],
                outputs=[login_box, admin_content, login_status, stats_display, csv_summary_file, csv_detail_file, json_file]
            )
        
        # 使用 TabbedInterface 组合两个界面
        combined_app = gr.TabbedInterface(
            [app, admin_app],
            ["📝 用户问卷 / Survey", "🔐 管理员 / Admin"]
        )
        
        combined_app.launch(
            server_name="0.0.0.0", 
            server_port=7860, 
            share=False,
            allowed_paths=[os.path.expanduser("~/.cache/huggingface")]
        )
    else:
        print("应用初始化失败 / App initialization failed")