lite_app.py

import json
import shutil
import tempfile
import traceback
import uuid
import zipfile
from pathlib import Path


APP_ROOT = Path(__file__).resolve().parent
RUNTIME_ARCHIVE = APP_ROOT / "lite_runtime.zip"


def _extract_runtime():
    with zipfile.ZipFile(RUNTIME_ARCHIVE) as archive:
        for member in archive.infolist():
            target = (APP_ROOT / member.filename).resolve()
            if APP_ROOT not in target.parents and target != APP_ROOT:
                raise RuntimeError("The evaluator runtime archive contains an unsafe path.")
        archive.extractall(APP_ROOT)


_extract_runtime()

import gradio as gr

from chinatravel.ui import content
from chinatravel.ui.archive import SubmissionArchiveError, extract_submission_archive
from chinatravel.ui.leaderboard import build_leaderboards
from eval_exp import evaluate


EVALUATION_SPLITS = ("easy", "human")
LEADERBOARD_SPLITS = ("easy", "human", "human1000")
LANGUAGE_SPLITS = {"zh": EVALUATION_SPLITS, "en": EVALUATION_SPLITS}
DEFAULT_SPLIT = "easy"
OVERVIEW_URL = (
    "https://huggingface.co/spaces/LAMDA-NeSy/ChinaTravel/resolve/"
    "main/images/overview.png"
)


class Arguments:
    def __init__(self, splits, result_dir, lang):
        self.splits = splits
        self.result_dir = result_dir
        self.lang = lang
        self.oracle_translation = True


def _upload_path(zip_file):
    if isinstance(zip_file, dict):
        return zip_file.get("path") or zip_file.get("name")
    return getattr(zip_file, "name", zip_file)


def splits_for_language(lang):
    return LANGUAGE_SPLITS.get(lang, ())


def update_evaluation_splits(language, current_split):
    choices = list(splits_for_language(language))
    value = current_split if current_split in choices else choices[0]
    return gr.update(choices=choices, value=value)


def handle_submission(zip_file, dataset_choice, language_choice):
    empty_progress = content.render_progress()
    empty_metrics = content.render_metrics()

    if zip_file is None:
        yield (
            content.render_status(
                "error",
                "Archive required",
                "Upload a ZIP archive containing prediction JSON files.",
            ),
            empty_progress,
            empty_metrics,
            None,
        )
        return
    if dataset_choice not in EVALUATION_SPLITS:
        yield (
            content.render_status(
                "error",
                "Unsupported split",
                "Browser evaluation is available for easy and human.",
            ),
            empty_progress,
            empty_metrics,
            None,
        )
        return
    if language_choice not in LANGUAGE_SPLITS:
        yield (
            content.render_status(
                "error",
                "Invalid language",
                "Select either Chinese or English evaluation.",
            ),
            empty_progress,
            empty_metrics,
            None,
        )
        return

    job_dir = Path(tempfile.mkdtemp(prefix="chinatravel-lite-"))
    submission_dir = job_dir / "submission"

    try:
        summary = extract_submission_archive(_upload_path(zip_file), submission_dir)
        yield (
            content.render_status(
                "running",
                "Evaluation started",
                f"Validated {summary.json_files} prediction files.",
            ),
            empty_progress,
            empty_metrics,
            None,
        )

        args = Arguments(dataset_choice, str(submission_dir), language_choice)
        stage_values = {"schema": 0, "commonsense": 0, "logic": 0}
        result = {}

        for update in evaluate(args, result):
            stage = update.get("stage", "")
            if stage in stage_values:
                stage_values[stage] = max(
                    0,
                    min(100, round(update.get("progress", 0))),
                )
                labels = {
                    "schema": "Checking output schema",
                    "commonsense": "Checking environment grounding",
                    "logic": "Checking hard constraints",
                }
                yield (
                    content.render_status(
                        "running",
                        labels[stage],
                        f"Split: {dataset_choice} · Language: {language_choice}",
                    ),
                    content.render_progress(
                        stage_values["schema"],
                        stage_values["commonsense"],
                        stage_values["logic"],
                    ),
                    empty_metrics,
                    None,
                )
            elif stage == "final":
                result.update(update.get("result", {}))

        output_path = job_dir / (
            f"evaluation-{dataset_choice}-{language_choice}-"
            f"{uuid.uuid4().hex[:8]}.json"
        )
        output_path.write_text(
            json.dumps(result, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )

        matched = len(result.get("matched_uid", []))
        unmatched = len(result.get("unmatched_uid", []))
        yield (
            content.render_status(
                "complete",
                "Evaluation complete",
                f"Matched {matched} predictions; {unmatched} expected UIDs were missing.",
            ),
            content.render_progress(100, 100, 100),
            content.render_metrics(result),
            str(output_path),
        )
    except SubmissionArchiveError as exc:
        yield (
            content.render_status("error", "Archive rejected", str(exc)),
            empty_progress,
            empty_metrics,
            None,
        )
    except Exception as exc:
        traceback.print_exc()
        yield (
            content.render_status("error", "Evaluation failed", str(exc)),
            empty_progress,
            empty_metrics,
            None,
        )
    finally:
        shutil.rmtree(submission_dir, ignore_errors=True)


leaderboards = build_leaderboards(LEADERBOARD_SPLITS)

with gr.Blocks(
    title="ChinaTravel Benchmark",
    theme=gr.themes.Base(
        primary_hue=gr.themes.colors.red,
        secondary_hue=gr.themes.colors.emerald,
        neutral_hue=gr.themes.colors.gray,
    ),
    css=content.CSS,
    fill_width=True,
) as demo:
    gr.HTML(content.HEADER_HTML)

    with gr.Tabs(elem_id="primary-tabs"):
        with gr.Tab("Leaderboard"):
            gr.HTML(content.LEADERBOARD_HEADER_HTML)
            with gr.Tabs(elem_id="track-tabs"):
                with gr.Tab("Standard"), gr.Tabs(elem_classes=["ct-split-tabs"]):
                    for split in LEADERBOARD_SPLITS:
                        with gr.Tab(split):
                            gr.HTML(leaderboards[split]["standard"])
                with gr.Tab("Oracle-assisted"), gr.Tabs(
                    elem_classes=["ct-split-tabs"]
                ):
                    for split in LEADERBOARD_SPLITS:
                        with gr.Tab(split):
                            gr.HTML(leaderboards[split]["oracle"])
            gr.HTML(content.LEADERBOARD_NOTE_HTML)

        with gr.Tab("Evaluate"):
            gr.HTML(content.EVALUATOR_HEADER_HTML)
            with gr.Row(elem_classes=["ct-workbench"]):
                with gr.Column(scale=4, min_width=280):
                    dataset_choice = gr.Dropdown(
                        list(EVALUATION_SPLITS),
                        label="Evaluation split",
                        value=DEFAULT_SPLIT,
                        interactive=True,
                    )
                    language_choice = gr.Radio(
                        choices=[("Chinese", "zh"), ("English", "en")],
                        value="zh",
                        label="Evaluation language",
                        interactive=True,
                    )
                    language_choice.change(
                        update_evaluation_splits,
                        inputs=[language_choice, dataset_choice],
                        outputs=dataset_choice,
                    )
                    zip_input = gr.File(
                        label="Prediction archive",
                        file_types=[".zip"],
                        type="filepath",
                    )
                    submit_btn = gr.Button(
                        "Run evaluation",
                        variant="primary",
                        elem_classes=["ct-submit-button"],
                    )
                    gr.HTML(content.EVALUATOR_RESOURCES_HTML)

                with gr.Column(scale=7, min_width=340):
                    status = gr.HTML(content.render_status())
                    progress = gr.HTML(content.render_progress())
                    metrics = gr.HTML(content.render_metrics())
                    result_file = gr.File(
                        label="Evaluation report",
                        interactive=False,
                    )

            submit_btn.click(
                handle_submission,
                inputs=[zip_input, dataset_choice, language_choice],
                outputs=[status, progress, metrics, result_file],
            )

        with gr.Tab("Benchmark"):
            gr.HTML(content.ABOUT_HTML)
            gr.Image(
                value=OVERVIEW_URL,
                show_label=False,
                interactive=False,
                elem_classes=["overview-image"],
            )

    gr.HTML(content.FOOTER_HTML)


demo.launch(show_error=False)