Update app.py

app.py

@@ -1,7 +1,293 @@
+"""
+VideoEval Movie-Level 问卷应用（Hugging Face Spaces）
+仅保留 Movie-Level 评测，并支持方法级别统计输出。
+"""
+
+import json
+import os
+import threading
+from collections import defaultdict
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Tuple
+
 import gradio as gr
 
-def greet(name):
-    return "Hello " + name + "!!"
+# 路径配置（按用户要求）
+ROOT_DIR = Path(os.environ.get("VIDEOEVAL_ROOT", "MemDirector"))
+INPUT_DIR = ROOT_DIR / "user_study_input"
+OUTPUT_DIR = ROOT_DIR / "user_study_results"
+STORY_DIR = INPUT_DIR / "clip_movie_story"
+VIDEO_DIR = INPUT_DIR / "video"
+
+Path(OUTPUT_DIR).mkdir(parents=True, exist_ok=True)
+
+# Movie-Level 指标定义
+MOVIE_CRITERIA: List[Tuple[str, str, str]] = [
+    ("SF", "Script Faithfulness (剧本忠实度)", "生成的视觉内容与原始剧本描述的吻合程度。"),
+    ("NC", "Narrative Coherence (叙事连贯性)", "镜头间情节发展的逻辑性，确保故事表达清晰、不破碎。"),
+    ("VQ", "Visual Quality (视觉质量)", "画面的清晰度、噪点控制、光影效果等基础图像质量。"),
+    ("CC", "Character Consistency (角色一致性)", "同一角色在不同镜头、不同角度下的外貌、服装及特征的稳定性。"),
+    ("PLC", "Physical Law Compliance (物理规律符合度)", "运动、重力、碰撞等是否符合现实物理逻辑，是否存在严重 AI 幻觉。"),
+    ("V_AQ", "Voice/Audio Quality (语音/音频质量)", "配音、背景音乐和音效的清晰度、自然度及技术品质。"),
+    ("CT", "Cinematic Techniques (电影技巧)", "镜头运动、景深控制及构图的专业性。"),
+    ("AVR", "Audio-Visual Richness (视听丰富度)", "画面细节精细度以及音频层次（音效、氛围音）的丰富程度。"),
+    ("NP", "Narrative Pacing (叙事节奏)", "镜头剪辑长短切换是否契合故事情节张力需求。"),
+    ("VAC", "Video-Audio Coordination (视听协调性)", "画面动作与音效、音乐卡点的同步率。"),
+    ("CD", "Compelling Degree (引人入胜程度)", "吸引注意力并引发情感共鸣或沉浸感的能力。"),
+    ("OQ", "Overall Quality (整体质量)", "对生成视频作为“电影作品”的综合观感评分。"),
+]
+
+BASE_METRIC_KEYS = [k for k, _, _ in MOVIE_CRITERIA]
+SAVE_LOCK = threading.Lock()
+
+
+def _safe_read_text(path: Path) -> str:
+    if not path.exists():
+        return ""
+    return path.read_text(encoding="utf-8-sig").strip()
+
+
+def load_dataset_index() -> List[Dict[str, Any]]:
+    """扫描输入目录，构建可评测样本列表。"""
+    stories = {p.stem: _safe_read_text(p) for p in sorted(STORY_DIR.glob("*.txt"))}
+    samples: List[Dict[str, Any]] = []
+
+    if not VIDEO_DIR.exists():
+        return samples
+
+    for method_dir in sorted([d for d in VIDEO_DIR.iterdir() if d.is_dir()]):
+        method = method_dir.name
+        for story_dir in sorted([d for d in method_dir.iterdir() if d.is_dir()]):
+            story_name = story_dir.name
+            # 一个 story 文件夹里可能多个 mp4，逐个作为样本
+            for video_path in sorted(story_dir.glob("*.mp4")):
+                sample_id = f"{method}__{story_name}__{video_path.stem}"
+                samples.append(
+                    {
+                        "sample_id": sample_id,
+                        "method": method,
+                        "story_name": story_name,
+                        "video_name": video_path.name,
+                        "video_path": str(video_path.resolve()),
+                        "story_text": stories.get(story_name, ""),
+                    }
+                )
+    return samples
+
+
+def compute_derived(scores: Dict[str, float]) -> Dict[str, float]:
+    """计算 CL / CRh / AVG。"""
+    cl = (
+        (scores["SF"] + scores["NC"] + scores["VQ"] + scores["CC"] + scores["PLC"]) / 5.0
+        + 0.5 * ((scores["CT"] + scores["AVR"]) / 2.0)
+    )
+    crh = (
+        (scores["V_AQ"] + scores["NP"] + scores["VAC"] + scores["CD"] + scores["OQ"]) / 5.0
+        + 0.5 * ((scores["CT"] + scores["AVR"]) / 2.0)
+    )
+    avg = sum(scores[k] for k in BASE_METRIC_KEYS) / len(BASE_METRIC_KEYS)
+    return {"CL": cl, "CRh": crh, "AVG": avg}
+
+
+def save_single_result(sample: Dict[str, Any], evaluator_id: str, scores: Dict[str, int], reasons: Dict[str, str], summary: str) -> Path:
+    """保存单个问卷结果。"""
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    result_dir = OUTPUT_DIR / "raw_results" / sample["method"] / sample["story_name"]
+    result_dir.mkdir(parents=True, exist_ok=True)
+    out_path = result_dir / f"{sample['video_name'].replace('.mp4', '')}_{evaluator_id}_{ts}.json"
+
+    score_float = {k: float(v) for k, v in scores.items()}
+    derived = compute_derived(score_float)
+
+    payload = {
+        "timestamp": datetime.now().isoformat(),
+        "evaluator_id": evaluator_id,
+        "sample": sample,
+        "scores": scores,
+        "reasons": reasons,
+        "summary": summary,
+        "derived": derived,
+    }
+    with open(out_path, "w", encoding="utf-8") as f:
+        json.dump(payload, f, ensure_ascii=False, indent=2)
+    return out_path
+
+
+def recompute_method_aggregates() -> Path:
+    """
+    统计每个方法各维度均分，并输出 method_aggregates.json。
+    同时给出 CL/CRh/AVG 的方法均值。
+    """
+    raw_root = OUTPUT_DIR / "raw_results"
+    method_scores: Dict[str, Dict[str, List[float]]] = defaultdict(lambda: defaultdict(list))
+    method_count: Dict[str, int] = defaultdict(int)
+
+    if raw_root.exists():
+        for fp in raw_root.rglob("*.json"):
+            with open(fp, "r", encoding="utf-8-sig") as f:
+                data = json.load(f)
+            method = data.get("sample", {}).get("method", "UNKNOWN")
+            scores = data.get("scores", {})
+            if not all(k in scores for k in BASE_METRIC_KEYS):
+                continue
+            method_count[method] += 1
+            for k in BASE_METRIC_KEYS:
+                method_scores[method][k].append(float(scores[k]))
+
+            # 衍生指标也参与方法均值统计
+            derived = compute_derived({k: float(scores[k]) for k in BASE_METRIC_KEYS})
+            for d_key, d_val in derived.items():
+                method_scores[method][d_key].append(float(d_val))
+
+    agg = {
+        "updated_at": datetime.now().isoformat(),
+        "metric_keys": BASE_METRIC_KEYS,
+        "derived_keys": ["CL", "CRh", "AVG"],
+        "methods": {},
+    }
+    for method in sorted(method_scores.keys()):
+        metric_avg = {}
+        for key, vals in method_scores[method].items():
+            metric_avg[key] = round(sum(vals) / len(vals), 4) if vals else None
+        agg["methods"][method] = {
+            "num_submissions": method_count[method],
+            "avg_scores": metric_avg,
+        }
+
+    out_path = OUTPUT_DIR / "method_aggregates.json"
+    with open(out_path, "w", encoding="utf-8") as f:
+        json.dump(agg, f, ensure_ascii=False, indent=2)
+    return out_path
+
+
+def build_sample_brief(sample: Dict[str, Any], index: int, total: int) -> str:
+    story = sample.get("story_text") or "(未找到对应 story 文本，请检查 clip_movie_story 下是否有同名 txt)"
+    return (
+        f"### 当前样本 {index + 1}/{total}\n"
+        f"- **Method**: `{sample['method']}`\n"
+        f"- **Story**: `{sample['story_name']}`\n"
+        f"- **Video**: `{sample['video_name']}`\n\n"
+        f"### Story Description\n{story}"
+    )
+
+
+def create_app():
+    samples = load_dataset_index()
+    sample_map = {s["sample_id"]: s for s in samples}
+
+    with gr.Blocks(title="VideoEval Movie-Level Evaluation") as app:
+        gr.Markdown("# VideoEval - Movie-Level Evaluation")
+        gr.Markdown(
+            f"- 输入目录: `{INPUT_DIR}`  \n"
+            f"- 输出目录: `{OUTPUT_DIR}`  \n"
+            "- 指标: SF/NC/VQ/CC/PLC/V_AQ/CT/AVR/NP/VAC/CD/OQ + CL/CRh/AVG"
+        )
+
+        current_idx = gr.State(0)
+        evaluator_state = gr.State("anonymous")
+
+        with gr.Row():
+            evaluator_input = gr.Textbox(label="Evaluator ID", value="anonymous")
+            sample_dropdown = gr.Dropdown(
+                label="选择评测样本",
+                choices=[s["sample_id"] for s in samples],
+                value=samples[0]["sample_id"] if samples else None,
+                interactive=True,
+            )
+
+        sample_info = gr.Markdown("无可用样本" if not samples else build_sample_brief(samples[0], 0, len(samples)))
+        movie_video = gr.Video(label="Movie Video", value=samples[0]["video_path"] if samples else None, height=420)
+
+        gr.Markdown("## Movie-Level 评分（1-5）")
+        score_widgets: Dict[str, gr.Radio] = {}
+        reason_widgets: Dict[str, gr.Textbox] = {}
+        for key, name, desc in MOVIE_CRITERIA:
+            with gr.Group():
+                gr.Markdown(f"**{key} - {name}**\n\n{desc}")
+                score_widgets[key] = gr.Radio(choices=[1, 2, 3, 4, 5], label=f"{key} Score")
+                reason_widgets[key] = gr.Textbox(label=f"{key} Reason", lines=2, placeholder="可选：补充评分理由")
+
+        final_summary = gr.Textbox(label="Final Summary", lines=4, placeholder="可选：整体评价总结")
+
+        with gr.Row():
+            prev_btn = gr.Button("← Previous")
+            next_btn = gr.Button("Next →")
+            submit_btn = gr.Button("提交当前评分并统计", variant="primary")
+        status = gr.Markdown("")
+
+        def _sync_sample_from_dropdown(sample_id: str) -> Tuple[str, str, int]:
+            if not sample_id or sample_id not in sample_map:
+                return None, "未找到样本", 0
+            idx = next(i for i, s in enumerate(samples) if s["sample_id"] == sample_id)
+            sample = samples[idx]
+            return sample["video_path"], build_sample_brief(sample, idx, len(samples)), idx
+
+        def _go_prev(idx: int) -> Tuple[str, str, str, int]:
+            if not samples:
+                return None, "无可用样本", None, 0
+            idx = max(0, idx - 1)
+            sample = samples[idx]
+            return sample["video_path"], build_sample_brief(sample, idx, len(samples)), sample["sample_id"], idx
+
+        def _go_next(idx: int) -> Tuple[str, str, str, int]:
+            if not samples:
+                return None, "无可用样本", None, 0
+            idx = min(len(samples) - 1, idx + 1)
+            sample = samples[idx]
+            return sample["video_path"], build_sample_brief(sample, idx, len(samples)), sample["sample_id"], idx
+
+        def _submit(evaluator_id: str, sample_id: str, summary: str, *score_reason_vals):
+            if not samples:
+                return "❌ 没有可提交样本。"
+            if not sample_id or sample_id not in sample_map:
+                return "❌ 请先选择样本。"
+            sample = sample_map[sample_id]
+            evaluator_id = (evaluator_id or "anonymous").strip() or "anonymous"
+
+            scores: Dict[str, int] = {}
+            reasons: Dict[str, str] = {}
+            for i, key in enumerate(BASE_METRIC_KEYS):
+                score = score_reason_vals[i * 2]
+                reason = score_reason_vals[i * 2 + 1]
+                if score is None:
+                    return f"❌ 请为 `{key}` 打分。"
+                scores[key] = int(score)
+                reasons[key] = (reason or "").strip()
+
+            with SAVE_LOCK:
+                single_path = save_single_result(sample, evaluator_id, scores, reasons, summary or "")
+                agg_path = recompute_method_aggregates()
+
+            return f"✅ 已保存: `{single_path}`\n\n✅ 已更新方法统计: `{agg_path}`"
+
+        sample_dropdown.change(
+            _sync_sample_from_dropdown,
+            inputs=[sample_dropdown],
+            outputs=[movie_video, sample_info, current_idx],
+        )
+        prev_btn.click(_go_prev, inputs=[current_idx], outputs=[movie_video, sample_info, sample_dropdown, current_idx])
+        next_btn.click(_go_next, inputs=[current_idx], outputs=[movie_video, sample_info, sample_dropdown, current_idx])
+
+        submit_inputs = [evaluator_input, sample_dropdown, final_summary]
+        for key in BASE_METRIC_KEYS:
+            submit_inputs.append(score_widgets[key])
+            submit_inputs.append(reason_widgets[key])
+        submit_btn.click(_submit, inputs=submit_inputs, outputs=[status])
+
+        app.load(lambda x: x, inputs=[evaluator_input], outputs=[evaluator_state])
+
+    return app
+
+
+demo = create_app()
 
-demo = gr.Interface(fn=greet, inputs="text", outputs="text")
-demo.launch()
+if __name__ == "__main__":
+    allowed_paths = [str(INPUT_DIR.resolve())] if INPUT_DIR.exists() else None
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=False,
+        show_error=True,
+        allowed_paths=allowed_paths,
+    )