Upload 7 files

annotation.py

@@ -0,0 +1,255 @@
+import re
+import os
+import json
+from huggingface_hub import HfApi, hf_hub_download
+from collections import defaultdict
+import config
+
+HF_TOKEN = os.getenv("HF_TOKEN")
+REPO_ID = config.SAVE_REPO_ID
+
+api = HfApi()
+
+
+def get_user_annotation_filename(username: str) -> str:
+    """生成用户标注文件名"""
+    safe_username = re.sub(r'[\\/*?:"<>|]', "_", username)
+    return f"annotation_results_{safe_username}.json"
+
+
+def get_aggregated_filename() -> str:
+    """聚合结果文件名"""
+    return "aggregated_annotations.json"
+
+
+def save_annotations(username_state, annotation_results_state, tasks):
+    """保存个人标注结果并更新聚合结果"""
+    try:
+        individual_result = save_individual_annotations(username_state, annotation_results_state, tasks)
+
+        aggregated_result = update_aggregated_annotations(tasks)
+
+        return f"{individual_result}\n{aggregated_result}"
+
+    except Exception as e:
+        return f"❌ 保存失败: {str(e)}"
+
+
+def save_individual_annotations(username_state, annotation_results_state, tasks):
+    """保存个人标注结果"""
+    save_data = {
+        "total_tasks": len(tasks),
+        "completed_tasks": len(annotation_results_state),
+        "username": username_state,
+        "annotations": []
+    }
+
+    for task_id, choice in annotation_results_state.items():
+        task = tasks[task_id]
+        save_data["annotations"].append({
+            "task_id": task_id,
+            "text": task["text"],
+            "instruction": task["instruction"],
+            "comparison": f"{task['audioA_source']} vs {task['audioB_source']}",
+            "audioA_source": task["audioA_source"],
+            "audioB_source": task["audioB_source"],
+            "original_index": task["original_index"],
+            "choice": choice,
+            "username": username_state
+        })
+
+    save_str = json.dumps(save_data, ensure_ascii=False, indent=2)
+    filename = get_user_annotation_filename(username_state)
+
+    api.upload_file(
+        path_or_fileobj=save_str.encode("utf-8"),
+        path_in_repo=filename,
+        repo_id=REPO_ID,
+        repo_type="dataset",
+        token=HF_TOKEN
+    )
+
+    return f"✅ 个人标注已保存: {filename} ({len(annotation_results_state)}/{len(tasks)})"
+
+
+def update_aggregated_annotations(tasks):
+    """更新聚合标注结果"""
+    try:
+        aggregated_data = load_aggregated_annotations()
+
+        all_annotations = collect_all_annotations()
+
+        aggregated_data = build_aggregated_results(all_annotations, tasks)
+
+        save_str = json.dumps(aggregated_data, ensure_ascii=False, indent=2)
+        filename = get_aggregated_filename()
+
+        api.upload_file(
+            path_or_fileobj=save_str.encode("utf-8"),
+            path_in_repo=filename,
+            repo_id=REPO_ID,
+            repo_type="dataset",
+            token=HF_TOKEN
+        )
+
+        return f"✅ 聚合结果已更新: {filename}"
+
+    except Exception as e:
+        return f"❌ 聚合结果更新失败: {str(e)}"
+
+
+def collect_all_annotations():
+    """收集所有用户的标注结果"""
+    try:
+        files_info = api.list_repo_files(repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN)
+
+        all_annotations = []
+
+        for filename in files_info:
+            if filename.startswith("annotation_results_") and filename.endswith(".json"):
+                try:
+                    # 下载并加载用户标注
+                    local_path = hf_hub_download(
+                        repo_id=REPO_ID,
+                        filename=filename,
+                        repo_type="dataset",
+                        token=HF_TOKEN,
+                        force_download=True
+                    )
+
+                    with open(local_path, "r", encoding="utf-8") as f:
+                        user_data = json.load(f)
+                        all_annotations.extend(user_data.get("annotations", []))
+
+                except Exception as e:
+                    print(f"加载文件 {filename} 失败: {e}")
+                    continue
+
+        return all_annotations
+
+    except Exception as e:
+        print(f"收集标注失败: {e}")
+        return []
+
+
+def build_aggregated_results(all_annotations, tasks):
+    """构建聚合结果"""
+    groups = defaultdict(lambda: {
+        "text": "",
+        "instruction": "",
+        "comparisons": defaultdict(lambda: {"win": 0, "tie": 0, "lose": 0, "annotators": []})
+    })
+
+    for ann in all_annotations:
+        original_index = ann.get("original_index")
+        comparison = ann.get("comparison")
+        choice = ann.get("choice")
+        username = ann.get("username")
+        text = ann.get("text", "")
+        instruction = ann.get("instruction", "")
+
+        if original_index is not None and comparison and choice:
+            key = original_index
+
+            groups[key]["text"] = text
+            groups[key]["instruction"] = instruction
+
+            if choice in ["win", "tie", "lose"]:
+                groups[key]["comparisons"][comparison][choice] += 1
+                if username not in groups[key]["comparisons"][comparison]["annotators"]:
+                    groups[key]["comparisons"][comparison]["annotators"].append(username)
+
+    aggregated_results = []
+    for original_index, group_data in groups.items():
+        result_item = {
+            "original_index": original_index,
+            "text": group_data["text"],
+            "instruction": group_data["instruction"],
+            "comparisons": {}
+        }
+
+        for comparison, votes in group_data["comparisons"].items():
+            result_item["comparisons"][comparison] = {
+                "votes（win tie lose）": [votes["win"], votes["tie"], votes["lose"]],
+                "total_annotators": len(votes["annotators"]),
+                "annotators": votes["annotators"]
+            }
+
+        aggregated_results.append(result_item)
+
+    aggregated_results.sort(key=lambda x: x["original_index"])
+
+    return {
+        "total_groups": len(aggregated_results),
+        "total_annotations": len(all_annotations),
+        "results": aggregated_results
+    }
+
+
+def load_aggregated_annotations():
+    """加载现有的聚合结果"""
+    try:
+        filename = get_aggregated_filename()
+        local_path = hf_hub_download(
+            repo_id=REPO_ID,
+            filename=filename,
+            repo_type="dataset",
+            token=HF_TOKEN,
+            force_download=True
+        )
+
+        with open(local_path, "r", encoding="utf-8") as f:
+            return json.load(f)
+
+    except Exception:
+        return {"total_groups": 0, "total_annotations": 0, "results": []}
+
+
+def load_annotations(username):
+    """从 Hugging Face Hub 加载用户特定的标注结果"""
+    try:
+        filename = get_user_annotation_filename(username)
+        local_path = hf_hub_download(
+            repo_id=REPO_ID,
+            filename=filename,
+            repo_type="dataset",
+            token=HF_TOKEN,
+            force_download=True
+        )
+        with open(local_path, "r", encoding="utf-8") as f:
+            save_data = json.load(f)
+        annotation_results = {ann["task_id"]: ann["choice"] for ann in save_data.get("annotations", [])}
+        return annotation_results
+    except Exception:
+        return {}
+
+
+def get_aggregated_stats():
+    """获取聚合统计信息"""
+    try:
+        aggregated_data = load_aggregated_annotations()
+
+        stats = {
+            "total_groups": aggregated_data.get("total_groups", 0),
+            "total_annotations": aggregated_data.get("total_annotations", 0),
+            "comparison_summary": {}
+        }
+
+        for result in aggregated_data.get("results", []):
+            for comparison, data in result.get("comparisons", {}).items():
+                if comparison not in stats["comparison_summary"]:
+                    stats["comparison_summary"][comparison] = {
+                        "total_votes": 0,
+                        "win": 0, "tie": 0, "lose": 0
+                    }
+
+                votes = data.get("votes", [0, 0, 0])
+                stats["comparison_summary"][comparison]["win"] += votes[0]
+                stats["comparison_summary"][comparison]["tie"] += votes[1]
+                stats["comparison_summary"][comparison]["lose"] += votes[2]
+                stats["comparison_summary"][comparison]["total_votes"] += sum(votes)
+
+        return stats
+
+    except Exception as e:
+        return {"error": str(e)}

app.py

@@ -0,0 +1,198 @@
+import gradio as gr
+import config
+from ui_components import create_ui
+from annotation import load_annotations
+from task_manager import get_current_task, get_current_task_with_annotations, get_total_tasks
+
+css = """
+.center { text-align: center; }
+.audio-container { margin: 10px; padding: 15px; }
+
+/* 胜负选择样式 - 绿色 */
+.selected {
+    border: 3px solid #4CAF50 !important;
+    background-color: #e8f5e9 !important;
+}
+
+/* 平局选择样式 - 橙色 */
+.tie-selected {
+    border: 3px solid #FF9800 !important;
+    background-color: #fff3e0 !important;
+}
+
+.user-task-info {
+    font-size: 16px;
+    color: #333;
+    padding: 10px;
+    background-color: #f0f0f0;
+    border-radius: 5px;
+    display: flex;
+    justify-content: space-between;
+    align-items: center;
+}
+
+.progress-bar {
+    background-color: #e0e0e0;
+    border-radius: 10px;
+    height: 8px;
+    width: 200px;
+    margin: 0 10px;
+    position: relative;
+}
+
+.progress-fill {
+    background-color: #4CAF50;
+    height: 100%;
+    border-radius: 10px;
+    transition: width 0.3s ease;
+}
+"""
+
+
+def calculate_completion_stats(annotation_results):
+    """计算任务完成统计信息"""
+    if not annotation_results:
+        return 0, 0, 0.0
+
+    total_tasks = get_total_tasks()
+    completed_tasks = len(annotation_results)
+    completion_rate = (completed_tasks / total_tasks * 100) if total_tasks > 0 else 0.0
+
+    return completed_tasks, total_tasks, completion_rate
+
+
+def get_initial_task_position(annotation_results):
+    """根据用户的标注历史确定初始任务位置"""
+    if not annotation_results:
+        return 0
+
+    max_annotated = max(annotation_results.keys()) if annotation_results else -1
+    next_task = max_annotated + 1
+
+    total_tasks = get_total_tasks()
+    return min(next_task, total_tasks - 1)
+
+
+def create_task_info_html(username, annotation_results, current_task_num):
+    """创建包含用户信息和完成度的HTML"""
+    completed_tasks, total_tasks, completion_rate = calculate_completion_stats(annotation_results)
+
+    # 创建进度条HTML
+    progress_bar_html = f"""
+    <div class="progress-bar">
+        <div class="progress-fill" style="width: {completion_rate}%;"></div>
+    </div>
+    """
+
+    task_info_html = f"""
+    <div class="user-task-info">
+        <span>👤 当前用户: {username}</span>
+        <div style="display: flex; align-items: center;">
+            <span>完成度: {completed_tasks}/{total_tasks} ({completion_rate:.1f}%)</span>
+            {progress_bar_html}
+        </div>
+        <span><strong>当前任务: {current_task_num}</strong></span>
+    </div>
+    """
+
+    return task_info_html
+
+
+def set_user_info(request: gr.Request):
+    """设置用户信息到 State 并加载用户特定的标注"""
+    username = request.username if hasattr(request, 'username') else "unknown"
+    annotation_results = load_annotations(username)
+    print(f"加载用户 {username} 的标注结果：{annotation_results}")
+
+    user_current_task = get_initial_task_position(annotation_results)
+
+    inst, text, audioA_update, audioB_update, prev_disabled, next_disabled, task_num = get_current_task_with_annotations(
+        annotation_results, user_current_task)
+
+    # 创建包含完成度信息的HTML
+    task_info_html = create_task_info_html(username, annotation_results, task_num)
+
+    return (
+        username,
+        annotation_results,
+        user_current_task,
+        inst,
+        text,
+        audioA_update,
+        audioB_update,
+        gr.update(interactive=not prev_disabled),
+        gr.update(interactive=not next_disabled),
+        gr.update(value=task_info_html)
+    )
+
+
+def update_task_info_after_action(username, annotation_results, current_task_num):
+    """在用户操作后更新任务信息显示"""
+    return create_task_info_html(username, annotation_results, current_task_num)
+
+
+if __name__ == "__main__":
+    print("启动应用...")
+    with gr.Blocks(css=css) as demo:
+        username = gr.State(value="unknown")
+        annotation_results = gr.State(value={})
+        user_current_task = gr.State(value=0)
+
+        ui_components = create_ui(get_current_task(), username, annotation_results)
+
+        demo.load(
+            set_user_info,
+            inputs=None,
+            outputs=[
+                username,
+                annotation_results,
+                user_current_task,
+                ui_components["instruction"],
+                ui_components["text_box"],
+                ui_components["audioA"],
+                ui_components["audioB"],
+                ui_components["btn_prev"],
+                ui_components["btn_next"],
+                ui_components["task_number"]
+            ]
+        )
+
+        # 修改按钮点击事件，添加任务信息更新
+        ui_components["btn_win"].click(
+            ui_components["select_result"],
+            inputs=[gr.State("win"), ui_components["audioA"], ui_components["audioB"], annotation_results, username,
+                    user_current_task],
+            outputs=[ui_components["audioA"], ui_components["audioB"], annotation_results, ui_components["task_number"]]
+        )
+
+        ui_components["btn_tie"].click(
+            ui_components["select_result"],
+            inputs=[gr.State("tie"), ui_components["audioA"], ui_components["audioB"], annotation_results, username,
+                    user_current_task],
+            outputs=[ui_components["audioA"], ui_components["audioB"], annotation_results, ui_components["task_number"]]
+        )
+
+        ui_components["btn_lose"].click(
+            ui_components["select_result"],
+            inputs=[gr.State("lose"), ui_components["audioA"], ui_components["audioB"], annotation_results, username,
+                    user_current_task],
+            outputs=[ui_components["audioA"], ui_components["audioB"], annotation_results, ui_components["task_number"]]
+        )
+
+        ui_components["btn_prev"].click(
+            ui_components["change_task"],
+            inputs=[gr.State("prev"), annotation_results, username, user_current_task],
+            outputs=[ui_components["instruction"], ui_components["text_box"], ui_components["audioA"],
+                     ui_components["audioB"], ui_components["btn_prev"], ui_components["btn_next"],
+                     ui_components["task_number"], annotation_results, user_current_task]
+        )
+
+        ui_components["btn_next"].click(
+            ui_components["change_task"],
+            inputs=[gr.State("next"), annotation_results, username, user_current_task],
+            outputs=[ui_components["instruction"], ui_components["text_box"], ui_components["audioA"],
+                     ui_components["audioB"], ui_components["btn_prev"], ui_components["btn_next"],
+                     ui_components["task_number"], annotation_results, user_current_task]
+        )
+
+    demo.launch(auth=config.ANNOTATOR)

config.py

@@ -0,0 +1,12 @@
+# config.py
+
+# Hugging Face 数据集配置
+PROCESS_REPO_ID = "sunnyzjx/Test_dataset"
+SAVE_REPO_ID = "sunnyzjx/annotation_results"
+
+# 数据集字段名配置
+AUDIO_FIELDS = ['model1', 'model2']
+FIELD_TEXT = "text"
+FIELD_INSTRUCTION = "instruction"
+
+ANNOTATOR = [('zjx', '123'), ('wy', '123')]

data_processing.py

@@ -0,0 +1,176 @@
+import numpy as np
+from datasets import load_dataset
+import os
+import config
+from itertools import combinations
+import random
+
+os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "true"
+
+dataset = load_dataset(config.PROCESS_REPO_ID, split="train")
+
+
+def process_audio(audio_obj):
+    """处理音频对象，返回音频数据和采样率"""
+    try:
+        if hasattr(audio_obj, 'get_all_samples'):
+            samples = audio_obj.get_all_samples()
+            audio_data = samples.data
+            if not isinstance(audio_data, np.ndarray):
+                audio_data = np.array(audio_data, dtype=np.float32)
+            sample_rate = samples.sample_rate
+            if not isinstance(sample_rate, int):
+                sample_rate = int(sample_rate)
+            if len(audio_data.shape) > 1:
+                audio_data = audio_data.mean(axis=0)
+            return audio_data, sample_rate
+        else:
+            print("音频对象缺少 get_all_samples 方法")
+            return None, None
+    except Exception as e:
+        print(f"处理音频失败: {e}")
+        return None, None
+
+
+def generate_random_pairs(audio_fields, include_reverse=True, shuffle_order=True):
+    """
+    生成随机的音频对比较对
+
+    Args:
+        audio_fields: 音频字段列表
+        include_reverse: 是否包含反向比较（A vs B 和 B vs A）
+        shuffle_order: 是否随机打乱比较对的顺序
+
+    Returns:
+        比较对的列表
+    """
+    basic_combinations = list(combinations(audio_fields, 2))
+
+    if include_reverse:
+        pairs = []
+        for combo in basic_combinations:
+            if random.choice([True, False]):
+                pairs.append((combo[1], combo[0]))
+            else:
+                pairs.append(combo)
+    else:
+        pairs = basic_combinations
+
+    if shuffle_order:
+        random.shuffle(pairs)
+
+    return pairs
+
+
+def generate_all_permutations(audio_fields, shuffle_order=True):
+    """
+    生成所有可能的有序对（包括正向和反向）
+
+    Args:
+        audio_fields: 音频字段列表
+        shuffle_order: 是否随机打乱顺序
+
+    Returns:
+        所有有序对的列表
+    """
+    pairs = []
+    for i, field_a in enumerate(audio_fields):
+        for j, field_b in enumerate(audio_fields):
+            if i != j:  # 不与自己比较
+                pairs.append((field_a, field_b))
+
+    if shuffle_order:
+        random.shuffle(pairs)
+
+    return pairs
+
+
+def load_tasks(comparison_mode="random_reverse", seed=None):
+    """
+    使用config配置的音频字段进行两两比较
+
+    Args:
+        comparison_mode: 比较模式
+            - "fixed": 固定顺序的组合（原始模式）
+            - "random_reverse": 随机决定是否反转每个组合的顺序
+            - "all_permutations": 生成所有可能的有序对
+        seed: 随机种子，仅在需要复现结果时使用
+    """
+    if seed is not None:
+        random.seed(seed)
+        print(f"使用随机种子: {seed}")
+    else:
+        print("使用真随机模式")
+
+    print("处理数据集...")
+
+    audio_fields = config.AUDIO_FIELDS
+    text_field = config.FIELD_TEXT
+    instruction_field = config.FIELD_INSTRUCTION
+
+    print(f"使用音频字段: {audio_fields}")
+    print(f"文本字段: {text_field}")
+    print(f"指令字段: {instruction_field}")
+    print(f"比较模式: {comparison_mode}")
+
+    tasks = []
+
+    for i, row in enumerate(dataset):
+        processed_audios = {}
+        for field in audio_fields:
+            if field not in row or row[field] is None:
+                print(f"任务 {i} 缺少音频字段: {field}")
+                continue
+
+            audio_data, audio_rate = process_audio(row[field])
+            if (audio_data is not None and audio_rate is not None and
+                    isinstance(audio_data, np.ndarray) and isinstance(audio_rate, int)):
+                processed_audios[field] = (audio_data, audio_rate)
+            else:
+                print(f"任务 {i} 的音频字段 {field} 处理失败")
+
+        if len(processed_audios) < 2:
+            print(f"跳过任务 {i}：有效音频数量不足")
+            continue
+
+        text = row.get(text_field, '')
+        instruction = row.get(instruction_field, '请比较这两个音频的质量')
+
+        available_fields = list(processed_audios.keys())
+
+        if comparison_mode == "fixed":
+            pairs = list(combinations(available_fields, 2))
+        elif comparison_mode == "random_reverse":
+            pairs = generate_random_pairs(available_fields, include_reverse=True, shuffle_order=True)
+        elif comparison_mode == "all_permutations":
+            pairs = generate_all_permutations(available_fields, shuffle_order=True)
+        else:
+            raise ValueError(f"未知的比较模式: {comparison_mode}")
+
+        for field_a, field_b in pairs:
+            tasks.append({
+                "instruction": instruction,
+                "text": text,
+                "audioA": processed_audios[field_a],
+                "audioB": processed_audios[field_b],
+                "audioA_source": field_a,
+                "audioB_source": field_b,
+                "comparison": f"{field_a} vs {field_b}",
+                "original_index": i
+            })
+
+    print(f"成功生成 {len(tasks)} 个比较任务")
+    if len(tasks) == 0:
+        print("没有可用任务！")
+        exit()
+
+    comparison_counts = {}
+    for task in tasks:
+        comp = task["comparison"]
+        comparison_counts[comp] = comparison_counts.get(comp, 0) + 1
+
+    print("比较任务统计：")
+    for comp, count in sorted(comparison_counts.items()):
+        print(f"  {comp}: {count} 个任务")
+
+    return tasks

requirements.txt

@@ -0,0 +1,4 @@
+gradio==4.44.0
+numpy
+datasets
+huggingface_hub

task_manager.py

@@ -0,0 +1,161 @@
+import gradio as gr
+from data_processing import load_tasks
+from annotation import save_annotations
+
+tasks = load_tasks(comparison_mode="random_reverse", seed=42)
+
+
+def get_current_task_with_annotations(annotation_results, user_current_task=0):
+    """获取当前任务信息，应用已有标注的样式（用于初始加载）"""
+    task = tasks[user_current_task]
+    current_choice = annotation_results.get(user_current_task) if annotation_results else None
+
+    # 基础音频数据
+    audioA_data = (task["audioA"][1], task["audioA"][0])  # (rate, data)
+    audioB_data = (task["audioB"][1], task["audioB"][0])  # (rate, data)
+
+    # 根据选择结果应用样式
+    if current_choice == "win":
+        # A胜过B - A高亮绿色，B显示败北
+        audioA_styled = gr.update(value=audioA_data, elem_classes="selected")
+        audioB_styled = gr.update(value=audioB_data, elem_classes="")
+    elif current_choice == "lose":
+        # A输给B - B高亮绿色，A显示败北
+        audioA_styled = gr.update(value=audioA_data, elem_classes="")
+        audioB_styled = gr.update(value=audioB_data, elem_classes="selected")
+    elif current_choice == "tie":
+        # 平局 - 两个都用特殊样式
+        audioA_styled = gr.update(value=audioA_data, elem_classes="tie-selected")
+        audioB_styled = gr.update(value=audioB_data, elem_classes="tie-selected")
+    else:
+        # 未选择
+        audioA_styled = gr.update(value=audioA_data, elem_classes="")
+        audioB_styled = gr.update(value=audioB_data, elem_classes="")
+
+    return (
+        task["instruction"],
+        task["text"],
+        audioA_styled,
+        audioB_styled,
+        user_current_task == 0,
+        user_current_task == len(tasks) - 1,
+        user_current_task + 1
+    )
+
+
+def get_current_task(user_current_task=0, annotation_results=None, styled=False):
+    """获取当前任务信息，可选择是否应用样式"""
+    task = tasks[user_current_task]
+
+    if styled and annotation_results is not None:
+        current_choice = annotation_results.get(user_current_task)
+        audioA_data = (task["audioA"][1], task["audioA"][0])  # (rate, data)
+        audioB_data = (task["audioB"][1], task["audioB"][0])  # (rate, data)
+
+        if current_choice == "win":
+            audioA_styled = gr.update(value=audioA_data, elem_classes="selected")
+            audioB_styled = gr.update(value=audioB_data, elem_classes="")
+        elif current_choice == "lose":
+            audioA_styled = gr.update(value=audioA_data, elem_classes="")
+            audioB_styled = gr.update(value=audioB_data, elem_classes="selected")
+        elif current_choice == "tie":
+            audioA_styled = gr.update(value=audioA_data, elem_classes="tie-selected")
+            audioB_styled = gr.update(value=audioB_data, elem_classes="tie-selected")
+        else:
+            audioA_styled = gr.update(value=audioA_data, elem_classes="")
+            audioB_styled = gr.update(value=audioB_data, elem_classes="")
+
+        return (
+            task["instruction"],
+            task["text"],
+            audioA_styled,
+            audioB_styled,
+            user_current_task == 0,
+            user_current_task == len(tasks) - 1,
+            user_current_task + 1
+        )
+    else:
+        return (
+            task["instruction"],
+            task["text"],
+            task["audioA"][0],
+            task["audioA"][1],
+            task["audioB"][0],
+            task["audioB"][1],
+            user_current_task == 0,
+            user_current_task == len(tasks) - 1,
+            user_current_task + 1
+        )
+
+
+def apply_selection_style(audioA, audioB, choice):
+    """根据选择结果应用样式"""
+    if choice == "win":
+        # A胜过B
+        return (
+            gr.update(value=audioA, elem_classes="selected"),
+            gr.update(value=audioB, elem_classes="")
+        )
+    elif choice == "lose":
+        # A输给B
+        return (
+            gr.update(value=audioA, elem_classes=""),
+            gr.update(value=audioB, elem_classes="selected")
+        )
+    elif choice == "tie":
+        # 平局
+        return (
+            gr.update(value=audioA, elem_classes="tie-selected"),
+            gr.update(value=audioB, elem_classes="tie-selected")
+        )
+    else:
+        # 清除选择
+        return (
+            gr.update(value=audioA, elem_classes=""),
+            gr.update(value=audioB, elem_classes="")
+        )
+
+
+def select_result(choice, audioA, audioB, annotation_results, username, user_current_task):
+    """记录选择结果并更新UI高亮，自动保存标注结果"""
+    annotation_results[user_current_task] = choice
+
+    # 自动保存标注结果
+    save_result = save_annotations(username, annotation_results, tasks)
+    print(f"自动保存结果: {save_result}")
+
+    audioA_update, audioB_update = apply_selection_style(audioA, audioB, choice)
+    return audioA_update, audioB_update, annotation_results
+
+
+def change_task(direction, annotation_results, username, user_current_task):
+    """切换任务"""
+    new_user_current_task = user_current_task
+
+    if direction == "prev" and user_current_task > 0:
+        new_user_current_task = user_current_task - 1
+    elif direction == "next" and user_current_task < len(tasks) - 1:
+        new_user_current_task = user_current_task + 1
+
+    inst, text, audioA_update, audioB_update, prev_disabled, next_disabled, task_num = get_current_task(
+        new_user_current_task, annotation_results, styled=True)
+
+    total_tasks = get_total_tasks()
+
+    combined_task_info = f'<div class="user-task-info"><span>👤 当前用户: {username}</span><span><strong>任务编号: {task_num} / {total_tasks}</strong></span></div>'
+
+    return (
+        inst, text,
+        audioA_update,
+        audioB_update,
+        gr.update(interactive=not prev_disabled),
+        gr.update(interactive=not next_disabled),
+        gr.update(value=combined_task_info),
+        annotation_results,
+        new_user_current_task
+    )
+
+
+def get_total_tasks():
+    """返回总任务数"""
+    return len(tasks)

ui_components.py

@@ -0,0 +1,119 @@
+import gradio as gr
+from task_manager import get_current_task, select_result, change_task, tasks, get_total_tasks
+from annotation import save_annotations
+
+
+def calculate_completion_stats(annotation_results):
+    """计算任务完成统计信息"""
+    if not annotation_results:
+        return 0, 0, 0.0
+
+    total_tasks = get_total_tasks()
+    completed_tasks = len(annotation_results)
+    completion_rate = (completed_tasks / total_tasks * 100) if total_tasks > 0 else 0.0
+
+    return completed_tasks, total_tasks, completion_rate
+
+
+def create_task_info_html(username, annotation_results, current_task_num):
+    """创建包含用户信息和完成度的HTML"""
+    completed_tasks, total_tasks, completion_rate = calculate_completion_stats(annotation_results)
+
+    # 使用更简洁的HTML结构
+    task_info_html = f"""
+    <div style="font-size: 16px; color: #333; padding: 10px; background-color: #f0f0f0; border-radius: 5px; display: flex; justify-content: space-between; align-items: center;">
+        <span>👤 用户: {username}</span>
+        <span>完成度: {completed_tasks}/{total_tasks} ({completion_rate:.1f}%)</span>
+        <span><strong>当前任务: {current_task_num}</strong></span>
+    </div>
+    """
+
+    return task_info_html
+
+
+def create_ui(init_task, username, annotation_results):
+    """创建 Gradio 界面组件"""
+    init_inst, init_text, init_audioA_data, init_audioA_rate, init_audioB_data, init_audioB_rate, init_prev_dis, init_next_dis, init_task_num = init_task
+
+    gr.Markdown('<div class="center"><h2>🎵 音频对比标注平台</h2></div>')
+    user_display = gr.Markdown()
+
+    instruction = gr.Textbox(label="🎯 指令", value=init_inst, interactive=False)
+    text_box = gr.Textbox(label="📋 转录文本", value=init_text, interactive=False)
+
+    # 初始化时创建完成度显示 - 使用HTML组件而不是Markdown
+    initial_task_info = create_task_info_html("unknown", {}, init_task_num)
+    task_number = gr.HTML(value=initial_task_info, elem_classes="center")
+
+    with gr.Row():
+        with gr.Column(elem_classes="audio-container"):
+            audioA = gr.Audio(
+                label="🔊 音频 A",
+                value=(init_audioA_rate, init_audioA_data),
+                interactive=False,
+                type="numpy"
+            )
+        with gr.Column(elem_classes="audio-container"):
+            audioB = gr.Audio(
+                label="🔊 音频 B",
+                value=(init_audioB_rate, init_audioB_data),
+                interactive=False,
+                type="numpy"
+            )
+
+    with gr.Row():
+        btn_win = gr.Button("🏆 Win", variant="primary")
+        btn_tie = gr.Button("🤝 Tie", variant="secondary")
+        btn_lose = gr.Button("❌ Lose", variant="stop")
+
+    with gr.Row():
+        btn_prev = gr.Button("⬅️ 上一题", interactive=not init_prev_dis)
+        btn_next = gr.Button("➡️ 下一题", interactive=not init_next_dis)
+
+    def wrapped_select_result(choice, audioA, audioB, annotation_results, username, user_current_task):
+        """包装 select_result 函数以适配新的参数签名"""
+        result_audioA, result_audioB, updated_annotation_results = select_result(choice, audioA, audioB,
+                                                                                 annotation_results, username,
+                                                                                 user_current_task)
+
+        # 更新完成度显示 - 保持当前任务编号不变，因为还在同一个任务上
+        current_task_display = user_current_task + 1  # 假设任务编号从1开始显示
+        updated_task_info = create_task_info_html(username, updated_annotation_results, current_task_display)
+
+        return result_audioA, result_audioB, updated_annotation_results, gr.update(value=updated_task_info)
+
+    def wrapped_change_task(direction, annotation_results, username, user_current_task):
+        """包装 change_task 函数以适配新的参数签名"""
+        result = change_task(direction, annotation_results, username, user_current_task)
+
+        if len(result) >= 9:  # 确保返回值包含所有必需的字段
+            inst, text, audioA, audioB, btn_prev, btn_next, original_task_display, updated_annotation_results, updated_current_task = result[
+                                                                                                                                      :9]
+
+            # 使用更新后的当前任务索引来计算显示编号
+            current_task_display = updated_current_task + 1  # 转换为1-based显示
+
+            # 创建新的完成度显示
+            updated_task_info = create_task_info_html(username, updated_annotation_results, current_task_display)
+
+            return inst, text, audioA, audioB, btn_prev, btn_next, gr.update(
+                value=updated_task_info), updated_annotation_results, updated_current_task
+        else:
+            return result
+
+    return {
+        "user_display": user_display,
+        "instruction": instruction,
+        "text_box": text_box,
+        "task_number": task_number,
+        "audioA": audioA,
+        "audioB": audioB,
+        "btn_win": btn_win,
+        "btn_tie": btn_tie,
+        "btn_lose": btn_lose,
+        "btn_prev": btn_prev,
+        "btn_next": btn_next,
+        "select_result": wrapped_select_result,
+        "change_task": wrapped_change_task,
+        "save_annotations": lambda u, a: save_annotations(u, a, tasks)
+    }