app.py

import logging
import os

os.makedirs("tmp", exist_ok=True)
os.environ["TMP_DIR"] = "tmp"
import subprocess
import shutil
import glob
import gradio as gr
import numpy as np
from src.radial.radial import create_plot
from apscheduler.schedulers.background import BackgroundScheduler
from gradio_leaderboard import Leaderboard, SelectColumns
from gradio_space_ci import enable_space_ci
import json
from io import BytesIO


def handle_file_upload(file):
    file_path = file.name.split("/")[-1] if "/" in file.name else file.name
    logging.info("File uploaded: %s", file_path)
    with open(file.name, "r") as f:
        v = json.load(f)
    return v, file_path


def submit_file(v, file_path, mn, profile: gr.OAuthProfile | None):
    """
    Обрабатывает загрузку файлов с результатами пользовательских моделей

    Args:
        v: Загруженные данные результатов из JSON
        file_path: Путь к загруженному файлу
        mn: Имя модели, указанное пользователем
        profile: Профиль пользователя HuggingFace

    Returns:
        str: Сообщение об успехе или ошибке
    """
    if profile is None:
        return "Hub Login Required: Войдите в HuggingFace, чтобы загрузить результаты"

    try:
        # Проверяем наличие обязательных полей
        required_fields = ["score", "math_score", "physics_score"]
        for field in required_fields:
            if field not in v:
                return f"Error: Отсутствует обязательное поле '{field}' в JSON файле"

        # Создаем новый объект для сохранения в базе данных
        new_file = {
            "model_name": profile.username + "/" + mn,
            "score": float(v["score"]),
            "math_score": float(v["math_score"]),
            "physics_score": float(v["physics_score"]),
            "total_tokens": int(v.get("total_tokens", 0)),
            "evaluation_time": float(v.get("evaluation_time", 0.0)),
            "system_prompt": v.get(
                "system_prompt", "Вы - полезный помощник по математике и физике. Ответьте на русском языке."
            ),
        }

        # Проверка значений на корректность
        if not (
            0 <= new_file["score"] <= 1 and 0 <= new_file["math_score"] <= 1 and 0 <= new_file["physics_score"] <= 1
        ):
            return "Error: Все значения оценок должны быть в диапазоне от 0 до 1"

        # Создаем уникальное имя файла на основе username и имени модели
        safe_filename = profile.username + "_" + mn.replace("/", "_").replace(" ", "_") + ".json"

        # Создаем JSON в памяти и загружаем его в репозиторий
        buf = BytesIO()
        buf.write(json.dumps(new_file, ensure_ascii=False).encode("utf-8"))

        # Загружаем файл в репозиторий
        API.upload_file(
            path_or_fileobj=buf,
            path_in_repo="model_data/external/" + safe_filename,
            repo_id="Vikhrmodels/DeathMath-leaderboard-data",
            repo_type="dataset",
        )

        # Устанавливаем флаг для обновления лидерборда
        os.environ[RESET_JUDGEMENT_ENV] = "1"

        # Логируем успешную загрузку
        logging.info(f"Successfully uploaded model results: {new_file['model_name']}")

        return f"Success! Результаты модели '{mn}' успешно отправлены в лидерборд DeathMath."

    except Exception as e:
        logging.error(f"Error submitting file: {e}")
        return f"Error: Произошла ошибка при отправке файла: {str(e)}"


from src.display.about import INTRODUCTION_TEXT, TITLE, LLM_BENCHMARKS_TEXT
from src.display.css_html_js import custom_css
from src.display.utils import (
    AutoEvalColumn,
    fields,
)
from src.envs import API, H4_TOKEN, HF_HOME, REPO_ID, RESET_JUDGEMENT_ENV
from src.leaderboard.build_leaderboard import build_leadearboard_df, download_openbench, download_dataset
import huggingface_hub
# huggingface_hub.login(token=H4_TOKEN)

os.environ["GRADIO_ANALYTICS_ENABLED"] = "false"

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

# Start ephemeral Spaces on PRs (see config in README.md)
enable_space_ci()

# download_openbench()


def restart_space():
    API.restart_space(repo_id=REPO_ID)
    download_openbench()


def update_plot(selected_models):
    return create_plot(selected_models)


def build_demo():
    """
    Строит интерфейс лидерборда DeathMath
    """
    # Загружаем данные для лидерборда
    download_openbench()

    # Создаем интерфейс с настройками темы
    demo = gr.Blocks(
        title="DeathMath Leaderboard",
        css=custom_css,
        theme=gr.themes.Default(
            primary_hue="indigo",
            secondary_hue="purple",
        ),
    )

    # Получаем данные для лидерборда
    leaderboard_df = build_leadearboard_df()

    # Строим интерфейс
    with demo:
        # Заголовок и введение
        gr.HTML(TITLE)
        gr.Markdown(INTRODUCTION_TEXT, elem_classes="markdown-text")

        # Основные вкладки
        with gr.Tabs(elem_classes="tab-buttons"):
            # Вкладка лидерборда
            with gr.TabItem("🏅 Лидерборд", elem_id="llm-benchmark-tab-table", id=0):
                gr.Markdown("### Таблица результатов моделей DeathMath")

                # Таблица с результатами
                Leaderboard(
                    value=leaderboard_df,
                    datatype=[c.type for c in fields(AutoEvalColumn)],
                    select_columns=SelectColumns(
                        default_selection=[c.name for c in fields(AutoEvalColumn) if c.displayed_by_default],
                        cant_deselect=[c.name for c in fields(AutoEvalColumn) if c.never_hidden or c.dummy],
                        label="Выберите колонки для отображения:",
                    ),
                    search_columns=[
                        AutoEvalColumn.model.name,
                    ],
                )

            # Вкладка для отправки результатов
            with gr.TabItem("🚀 Отправить результаты", elem_id="submit-tab", id=1):
                with gr.Row():
                    gr.Markdown(LLM_BENCHMARKS_TEXT, elem_classes="markdown-text")

                with gr.Column():
                    model_name_textbox = gr.Textbox(label="Название модели")
                    file_output = gr.File(label="Перетащите JSON файл с результатами сюда", type="filepath")
                    uploaded_file = gr.State()
                    file_path = gr.State()

                    with gr.Row():
                        with gr.Column():
                            out = gr.Textbox("Здесь будет показан статус отправки")
                        with gr.Column():
                            login_button = gr.LoginButton(elem_id="oauth-button")

                    submit_button = gr.Button("Отправить результаты", elem_id="submit_button", variant="primary")

                    # Обработчики событий
                    file_output.upload(handle_file_upload, file_output, [uploaded_file, file_path])

                    submit_button.click(submit_file, [uploaded_file, file_path, model_name_textbox], [out])

            # Вкладка с аналитикой
            with gr.TabItem("📊 Аналитика", elem_id="analytics-tab", id=2):
                with gr.Column():
                    if len(leaderboard_df) > 0:
                        model_dropdown = gr.Dropdown(
                            choices=leaderboard_df["model"].tolist(),
                            label="Модели",
                            value=leaderboard_df["model"].tolist()[:5]
                            if len(leaderboard_df) >= 5
                            else leaderboard_df["model"].tolist(),
                            multiselect=True,
                            info="Выберите модели для сравнения",
                        )
                    else:
                        model_dropdown = gr.Dropdown(
                            choices=["example/model-1", "example/model-2"],
                            label="Модели",
                            value=["example/model-1", "example/model-2"],
                            multiselect=True,
                            info="Выберите модели для сравнения",
                        )

                    # Вкладки для разных типов визуализации
                    with gr.Tabs():
                        with gr.TabItem("Столбчатая диаграмма"):
                            bar_plot = gr.Plot(create_plot(model_dropdown.value))

                        with gr.TabItem("Радарная диаграмма"):
                            from src.radial.radial import create_radar_plot

                            radar_plot = gr.Plot(create_radar_plot(model_dropdown.value))

                    # Обновление графиков при изменении выбранных моделей
                    model_dropdown.change(fn=create_plot, inputs=[model_dropdown], outputs=[bar_plot])

                    model_dropdown.change(fn=create_radar_plot, inputs=[model_dropdown], outputs=[radar_plot])

    return demo


# print(os.system('cd src/gen && ../../.venv/bin/python gen_judgment.py'))
# print(os.system('cd src/gen/ && python show_result.py --output'))


def update_board():
    need_reset = os.environ.get(RESET_JUDGEMENT_ENV)
    logging.info("Updating the leaderboard: %s", need_reset)
    if need_reset != "1":
        return

    os.environ[RESET_JUDGEMENT_ENV] = "0"

    try:
        # Загружаем актуальные данные из репозитория
        download_dataset("Vikhrmodels/DeathMath-leaderboard-data", "m_data")
        logging.info("Successfully downloaded model evaluation data")

        # Собираем данные из всех файлов моделей
        data_list = []
        seen_models = set()  # Для отслеживания дубликатов

        for file in glob.glob("./m_data/model_data/external/*.json"):
            try:
                with open(file) as f:
                    data = json.load(f)

                    # Проверяем наличие необходимых полей
                    model_name = data.get("model_name", "")
                    if not model_name:
                        logging.error(f"Failed to parse {file}: 'model_name' not found")
                        continue

                    # Предотвращаем дублирование моделей
                    model_base_name = model_name.split("/")[-1].split("_v")[0]
                    if model_base_name in seen_models:
                        logging.info(f"Skipping duplicate model: {model_name}")
                        continue

                    seen_models.add(model_base_name)

                    # Добавляем модель в список
                    data_list.append(
                        {
                            "model_name": model_name,
                            "score": float(data.get("score", 0.0)),
                            "math_score": float(data.get("math_score", 0.0)),
                            "physics_score": float(data.get("physics_score", 0.0)),
                            "total_tokens": int(data.get("total_tokens", 0)),
                            "evaluation_time": float(data.get("evaluation_time", 0.0)),
                            "system_prompt": data.get(
                                "system_prompt",
                                "Вы - полезный помощник по математике и физике. Ответьте на русском языке.",
                            ),
                        }
                    )

            except Exception as e:
                logging.error(f"Failed to process file {file}: {e}")
                continue

        # Если есть данные, сортируем их по общему баллу и сохраняем
        if data_list:
            # Сортируем по общему баллу
            data_list.sort(key=lambda x: x["score"], reverse=True)

            # Сохраняем обновленный лидерборд
            with open("genned.json", "w", encoding="utf-8") as f:
                json.dump(data_list, f, ensure_ascii=False, indent=2)

            # Загружаем обновленный лидерборд в репозиторий
            API.upload_file(
                path_or_fileobj="genned.json",
                path_in_repo="leaderboard.json",
                repo_id="Vikhrmodels/DeathMath-leaderboard-metainfo",
                repo_type="dataset",
            )
            logging.info(f"Updated leaderboard with {len(data_list)} models")

            # Генерируем README с таблицей лидерборда
            update_readme(data_list)

    except Exception as e:
        logging.error(f"Error updating leaderboard: {e}")


def update_readme(data_list):
    """
    Генерирует README.md с таблицей лидерборда
    """
    try:
        import pandas as pd
        from datetime import datetime

        # Создаем DataFrame для удобного форматирования таблицы
        df = pd.DataFrame(data_list)

        # Форматируем числовые колонки
        for col in ["score", "math_score", "physics_score"]:
            if col in df.columns:
                df[col] = df[col].apply(lambda x: f"{x:.3f}")

        if "total_tokens" in df.columns:
            df["total_tokens"] = df["total_tokens"].apply(lambda x: f"{int(x):,}")

        if "evaluation_time" in df.columns:
            df["evaluation_time"] = df["evaluation_time"].apply(lambda x: f"{x:.1f}s")

        # Создаем содержимое README
        current_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        readme_content = f"""---
title: DeathMath Leaderboard
emoji: 🔢
colorFrom: indigo
colorTo: purple
sdk: gradio
sdk_version: "4.20.0"
app_file: app.py
pinned: false
hf_oauth: true
---
# DeathMath Leaderboard

DeathMath - это бенчмарк для оценки способности моделей решать сложные математические и физические задачи на русском языке.

## Текущий лидерборд

Последнее обновление: {current_date}

| Модель | Общий балл | Математика | Физика | Токены | Время оценки |
|--------|------------|------------|---------|---------|--------------|
"""

        # Добавляем строки таблицы
        for _, row in df.iterrows():
            model_name = row.get("model_name", "")
            readme_content += f"| {model_name} | {row.get('score', 'N/A')} | {row.get('math_score', 'N/A')} | {row.get('physics_score', 'N/A')} | {row.get('total_tokens', 'N/A')} | {row.get('evaluation_time', 'N/A')} |\n"

        readme_content += """
## Как принять участие в бенчмарке

Для участия в бенчмарке DeathMath:

1. Клонируйте репозиторий и запустите тесты вашей модели
2. Загрузите результаты через [HuggingFace Space](https://huggingface.co/spaces/Vikhrmodels/DeathMath-leaderboard)
3. Дождитесь проверки и добавления результатов в лидерборд

## Формат результатов

Результаты должны быть в формате JSON со следующей структурой:
```json
{
  "score": 0.586,
  "math_score": 0.8,
  "physics_score": 0.373,
  "total_tokens": 1394299,
  "evaluation_time": 4533.2,
  "system_prompt": "Вы - полезный помощник по математике и физике. Ответьте на русском языке."
}
```

## Лицензия

Бенчмарк распространяется под лицензией Apache 2.0
"""

        # Сохраняем README
        with open("README.md", "w", encoding="utf-8") as f:
            f.write(readme_content)

        # Загружаем README в репозиторий
        API.upload_file(
            path_or_fileobj="README.md",
            path_in_repo="README.md",
            repo_id="Vikhrmodels/DeathMath-leaderboard-metainfo",
            repo_type="dataset",
        )

        logging.info("Updated README with leaderboard table")

    except Exception as e:
        logging.error(f"Failed to update README: {e}")


def update_board_():
    """
    Инициализирующая версия функции обновления лидерборда.
    Удаляет все существующие данные перед первой загрузкой
    """
    try:
        # Очищаем каталоги перед загрузкой данных
        if os.path.exists("m_data"):
            # Удаляем все JSON-файлы моделей из старой версии Small Shlepa
            for old_file in glob.glob("./m_data/model_data/external/*.json"):
                try:
                    os.remove(old_file)
                    logging.info(f"Removed old file: {old_file}")
                except Exception as e:
                    logging.error(f"Error removing old file {old_file}: {e}")

        # Вызываем основную функцию обновления
        update_board()

    except Exception as e:
        logging.error(f"Error in update_board_(): {e}")


if __name__ == "__main__":
    os.environ[RESET_JUDGEMENT_ENV] = "1"

    scheduler = BackgroundScheduler()
    update_board_()
    scheduler.add_job(update_board, "interval", minutes=10)
    scheduler.start()

    demo_app = build_demo()
    demo_app.launch()