tabs/data_viewer_side_by_side_tab.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Data-Viewer Side-by-Side tab for ICBCBench.
"""

import gradio as gr
import random
import re

from tabs.shared_data import get_entries_for_task, get_index


def make_user_task_markdown(item_id, prompt):
    return f"""### User Task 🎯

**Task ID:** {item_id}

**Description:** {prompt}"""


def make_article_markdown(article: str) -> str:
    if article and isinstance(article, str):
        processed_article = re.sub(r'\n{2,}', '\n\n', article)
        table_pattern = r'(\|[^\n]*\n(?:[\|\s\-:]+\n)?(?:\|[^\n]*\n)*)'
        tables = []

        def replace_table(match):
            tables.append(match.group(1))
            return f'__TABLE_PLACEHOLDER_{len(tables)-1}__'

        processed_article = re.sub(table_pattern, replace_table, processed_article)
        processed_article = re.sub(r'(?<!\n)\*\s*\*\*([^*]+?)\*\*:', r'\n\n* **\1**:', processed_article)
        processed_article = re.sub(r'\*\s*\*\*([^*]+?)\*\*:\s*([^*]*?)\s*\*\s*\*\*', r'* **\1**: \2\n  * **', processed_article)
        processed_article = re.sub(r'(?<!\n)\[\d+[^\]]*\]\*\s*\*\*', r'\n\n* **', processed_article)
        lines = processed_article.split('\n')
        result_lines = []
        for i, line in enumerate(lines):
            result_lines.append(line)
            if (i < len(lines) - 1 and
                line.strip() and
                lines[i + 1].strip() and
                not line.strip().startswith('*') and
                not lines[i + 1].strip().startswith('*') and
                not line.strip().startswith('#')):
                if i + 1 < len(lines) and lines[i + 1].strip():
                    result_lines.append('')
        processed_article = '\n'.join(result_lines)
        for i, table in enumerate(tables):
            processed_article = processed_article.replace(f'__TABLE_PLACEHOLDER_{i}__', table)
    else:
        processed_article = article if article is not None else ""
    return f"""### Generated Article 📖

{processed_article}"""


def make_scores_html(entry: dict) -> str:
    """Build score cards for ICBCBench side-by-side viewer."""
    track = entry.get("track", "subjective")

    overall = entry.get("overall_score")
    objective = entry.get("objective_score")
    subjective = entry.get("subjective_score")
    expert = entry.get("expert_score")
    citation = entry.get("citation_score")
    source = entry.get("source_quality_score")
    confidence = entry.get("confidence")
    correct = entry.get("correct")

    comp = entry.get("comprehensiveness_score")
    insight = entry.get("insight_score")
    inst = entry.get("instruction_following_score")
    read = entry.get("readability_score")

    def fmt(val):
        if val is None:
            return "N/A"
        try:
            return f"{float(val):.2f}"
        except (TypeError, ValueError):
            return str(val)

    if track == "objective":
        scores_data = [
            ("Overall<br>Score", fmt(overall)),
            ("Objective<br>Score", fmt(objective)),
            ("Confidence", fmt(confidence)),
            ("Correct", "Yes" if correct is True else ("No" if correct is False else "N/A")),
        ]
    else:
        scores_data = [
            ("Overall<br>Score", fmt(overall)),
            ("Subjective<br>Score", fmt(subjective)),
            ("Expert<br>Score", fmt(expert)),
            ("Citation", fmt(citation)),
            ("Source<br>Quality", fmt(source)),
        ]
        if subjective is None and any(v is not None for v in [comp, insight, inst, read]):
            scores_data = [
                ("Overall<br>Score", fmt(overall)),
                ("Comprehen-<br>siveness", fmt(comp)),
                ("Insight<br>Score", fmt(insight)),
                ("Instruction<br>Following", fmt(inst)),
                ("Readability<br>Score", fmt(read)),
            ]

    html_items_str = ""
    for title, score in scores_data:
        html_items_str += f"""
        <div style="text-align: center; padding: 10px 3px; flex-grow: 1; flex-basis: 19%; min-width: 0;">
            <h4 style="margin: 0 0 5px 0; font-size: 1em; color: #4a4a4a; font-weight: 600; line-height: 1.2;">{title}</h4>
            <p style="margin: 0; font-size: 1.1em; font-weight: bold; color: #333;">{score}</p>
        </div>
        """
    return f"""
<div style="background:#fff; border:1px solid #e0e0e0; border-radius:8px; padding: 15px 10px; margin:18px 0; box-shadow:0 2px 4px rgba(0,0,0,.06);">
    <div style="display: flex; justify-content: space-around; align-items: stretch;">
        {html_items_str}
    </div>
</div>"""


# ---------- 生成 Tab ----------
def create_data_viewer_side_by_side_tab():
    with gr.Tab("⚔️Side-by-Side Viewer"):
        gr.HTML(
            """<style>
            .card{background:#fff;border:1px solid #e0e0e0;border-radius:8px;padding:22px 24px;margin:18px 0;box-shadow:0 2px 4px rgba(0,0,0,.06);}
            .scrollable-sm{max-height:180px;overflow-y:auto;}
            .scrollable-lg{max-height:550px;overflow-y:auto;}
            .card p{color:#424242 !important;line-height:1.75;margin:0 0 14px 0;text-align:justify;}
            .card ul,.card ol{margin:12px 0 12px 24px;color:#424242 !important;}
            .card li{margin:4px 0;color:#424242 !important;}
            .card blockquote{border-left:4px solid #3498db;margin:18px 0;padding:14px 18px;background:#f8f9fa;font-style:italic;color:#555 !important;}
            .card pre{background:#f8f8f8;color:#333 !important;padding:18px;border-radius:6px;overflow-x:auto;border:1px solid #e0e0e0;}
            .card strong,.card b{font-weight:700 !important;}
            .card::-webkit-scrollbar{width:10px}
            .card::-webkit-scrollbar-track{background:#f5f5f5;border-radius:5px}
            .card::-webkit-scrollbar-thumb{background:#c0c0c0;border-radius:5px}
            .card::-webkit-scrollbar-thumb:hover{background:#a0a0a0}
            </style>"""
        )

        with gr.Row():
            task_dd = gr.Dropdown(label="Select Task", choices=[], interactive=True)

        user_task_display_md = gr.Markdown(value="Loading data…", elem_classes=["card", "scrollable-sm"])

        with gr.Row():
            with gr.Column(scale=1):
                model_a_dd = gr.Dropdown(label="Select Model A", choices=[], interactive=True)
                article_a_md = gr.Markdown(elem_classes=["card", "scrollable-lg"])
                scores_a_html = gr.HTML()
            with gr.Column(scale=1):
                model_b_dd = gr.Dropdown(label="Select Model B", choices=[], interactive=True)
                article_b_md = gr.Markdown(elem_classes=["card", "scrollable-lg"])
                scores_b_html = gr.HTML()

        def fetch_side_by_side_data(selected_task_display, model_a_name, model_b_name):
            empty_article = make_article_markdown("")
            empty_scores = make_scores_html({})

            if not selected_task_display:
                no_task_msg = "请选择一个任务。"
                return make_user_task_markdown("--", no_task_msg), \
                       empty_article, empty_scores, empty_article, empty_scores

            item_id_str = selected_task_display.split(".", 1)[0].strip()
            index = get_index()
            task_prompt = next(
                (task.get("prompt", "") for task in index.get("tasks", []) if str(task.get("id")) == item_id_str),
                "任务描述未找到。"
            )
            user_task_md_content = make_user_task_markdown(item_id_str, task_prompt)

            outputs_a = [make_article_markdown("模型A未选择或数据未找到"), empty_scores]
            outputs_b = [make_article_markdown("模型B未选择或数据未找到"), empty_scores]
            selected_models = {name for name in (model_a_name, model_b_name) if name}
            entries = get_entries_for_task(item_id_str, selected_models)

            if model_a_name:
                entry_a = entries.get(model_a_name)
                if entry_a:
                    outputs_a[0] = make_article_markdown(entry_a.get("article", ""))
                    outputs_a[1] = make_scores_html(entry_a)

            if model_b_name:
                entry_b = entries.get(model_b_name)
                if entry_b:
                    outputs_b[0] = make_article_markdown(entry_b.get("article", ""))
                    outputs_b[1] = make_scores_html(entry_b)

            return user_task_md_content, outputs_a[0], outputs_a[1], outputs_b[0], outputs_b[1]

        def _build_task_choices(tasks):
            return [
                f"{task['id']}. {task.get('prompt', '')[:60] + ('…' if len(task.get('prompt', '')) > 60 else '')}"
                for task in tasks
            ]

        def on_load():
            index = get_index()
            empty_article = make_article_markdown("")
            empty_scores = make_scores_html({})
            all_models = index.get("models", [])
            tasks = index.get("tasks", [])
            if not all_models or not tasks:
                return (gr.update(choices=[], value=None),
                        gr.update(choices=[], value=None), gr.update(choices=[], value=None),
                        make_user_task_markdown("--", "No data"),
                        empty_article, empty_scores, empty_article, empty_scores)

            task_choices = _build_task_choices(tasks)
            init_task = random.choice(task_choices) if task_choices else None
            init_a = random.choice(all_models) if all_models else None
            init_b = random.choice([m for m in all_models if m != init_a]) if len(all_models) > 1 else init_a

            data = (
                make_user_task_markdown("--", "请选择任务"),
                empty_article, empty_scores, empty_article, empty_scores
            )

            return (gr.update(choices=task_choices, value=init_task),
                    gr.update(choices=all_models, value=init_a),
                    gr.update(choices=all_models, value=init_b),
                    *data)

        task_dd.change(fetch_side_by_side_data, inputs=[task_dd, model_a_dd, model_b_dd],
                       outputs=[user_task_display_md, article_a_md, scores_a_html, article_b_md, scores_b_html])
        model_a_dd.change(fetch_side_by_side_data, inputs=[task_dd, model_a_dd, model_b_dd],
                          outputs=[user_task_display_md, article_a_md, scores_a_html, article_b_md, scores_b_html])
        model_b_dd.change(fetch_side_by_side_data, inputs=[task_dd, model_a_dd, model_b_dd],
                          outputs=[user_task_display_md, article_a_md, scores_a_html, article_b_md, scores_b_html])

        return on_load, [task_dd, model_a_dd, model_b_dd, user_task_display_md, article_a_md, scores_a_html, article_b_md, scores_b_html]