uploads.py

from email.utils import parseaddr
from huggingface_hub import HfApi
import io
import os
import base64
import pandas as pd
from utils import DEFAULT_COLUMNS, DEFAULT_METRICS, LEADERBOARD_PATH


api = HfApi()
TOKEN = os.environ.get("TOKEN", None)
YEAR_VERSION = "2024"


def format_error(msg):
    return f"<p style='color: red; font-size: 20px; text-align: center;'>{msg}</p>"


def format_warning(msg):
    return f"<p style='color: orange; font-size: 20px; text-align: center;'>{msg}</p>"


def format_log(msg):
    return f"<p style='color: green; font-size: 20px; text-align: center;'>{msg}</p>"


def add_new_eval(
    corpus: str,
    organization: str,
    mail: str,
    fpath: str,
):
    for input in [corpus, organization, mail, fpath]:
        if not input:
            return format_warning("Please fill all the fields.")
    
    if organization == 'Baseline':
        return format_warning("Your organization name cannot be Baseline.")

    _, parsed_mail = parseaddr(mail)
    if '@' not in parsed_mail:
        return format_warning("Please provide a valid email adress.")

    # load the file 
    io_path = f"submission/{corpus}.csv"
    df = pd.read_csv(io_path)
    df_new = pd.read_csv(fpath)

    for col in DEFAULT_METRICS:
        if col not in df_new.columns:
            return format_warning(f"Missing column in the submitted file: {col}")

    df_new['organization'] = organization
    df_new['id'] = base64.b64encode(os.urandom(6)).decode('ascii')
    df_new = df_new[DEFAULT_COLUMNS]

    df = pd.concat([df, df_new]).reset_index(drop=True)
    buffer = io.BytesIO()
    df.to_csv(buffer, index=False)  # Write the DataFrame to a buffer in CSV format
    buffer.seek(0)  # Rewind the buffer to the beginning

    api.delete_file(
        repo_id = LEADERBOARD_PATH,
        path_in_repo = io_path,
        token = TOKEN,
        repo_type='space'
    )

    api.upload_file(
        repo_id = LEADERBOARD_PATH,
        path_in_repo = io_path,
        path_or_fileobj = buffer,
        token = TOKEN,
        repo_type = 'space',
    )

    return format_log(f"Submitted to {corpus} by {organization} successfully.\nPlease refresh the leaderboard, and wait a bit to see the score displayed")