submit.py

from pathlib import Path
import tempfile
from typing import BinaryIO
import json

import gradio as gr
from datetime import datetime, timezone
import uuid

from constants import API, SUBMISSIONS_REPO, REGISTRATION_CODE
from validation import validate_csv_file, validate_username


def upload_submission(
    file_path: Path,
    user_state,
    submission_type: str,
    model_name: str,
    model_description: str,
    anonymous: bool = False,
):
    timestamp = datetime.now(timezone.utc).isoformat()
    date = datetime.now(timezone.utc).date().isoformat()
    submission_id = str(uuid.uuid4())

    with file_path.open("rb") as f_in:
        file_content = f_in.read().decode("utf-8")

        validate_csv_file(file_content, submission_type)

        # write to dataset
        filename = f"{user_state}/{date}_{model_name}_{submission_id}.json"  # Writing to a bit more structured name
        # Note: This may be represented as Parquet in the future, so the schema needs to stay consistent
        record = {
            "submission_id": submission_id,
            "submission_filename": filename,
            "submission_time": timestamp,
            "evaluated": False,
            "user": user_state,
            "model_name": model_name,
            "model_description": model_description,
            "csv_content": file_content,
            "dataset": submission_type,
            "anonymous": anonymous,
        }
        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as tmp:
            json.dump(record, tmp, indent=2)
            tmp.flush()
            tmp_name = tmp.name

        API.upload_file(
            path_or_fileobj=tmp_name,
            path_in_repo=filename,
            repo_id=SUBMISSIONS_REPO,
            repo_type="dataset",
            commit_message=f"Add submission for {user_state} at {timestamp}",
        )
        Path(tmp_name).unlink()


def make_submission(
    submitted_file: BinaryIO,
    user_state,
    submission_type: str = "GDPa1",
    model_name: str = "",
    model_description: str = "",
    anonymous: bool = False,
    registration_code: str = "",
    # profile: gr.OAuthProfile | None = None,
):

    # if profile:
    #    user_state = profile.name
    # user_state = user_state
    validate_username(user_state)

    model_name = model_name.strip()
    model_description = model_description.strip()

    if not model_name:
        raise gr.Error("Please provide a model name.")
    if not model_description:
        model_description = ""
        # raise gr.Error("Please provide a model description.") # Not mandatory anymore
    if str(registration_code).strip().upper() != REGISTRATION_CODE:
        raise gr.Error(
            "Invalid registration code. Please register on the <a href='https://datapoints.ginkgo.bio/ai-competitions/2025-abdev-competition'>Competition Registration page</a> or email <a href='mailto:[email protected]'>[email protected]</a>."
        )
    if submitted_file is None:
        raise gr.Error("Please upload a CSV file before submitting.")

    file_path = submitted_file.name

    if not file_path:
        raise gr.Error("Uploaded file object does not have a valid file path.")

    path_obj = Path(file_path)

    if path_obj.suffix.lower() != ".csv":
        raise gr.Error("File must be a CSV file. Please upload a .csv file.")

    upload_submission(
        file_path=path_obj,
        user_state=user_state,
        submission_type=submission_type,
        model_name=model_name,
        model_description=model_description,
        anonymous=anonymous,
    )

    return "✅ Your submission has been received! Your results should appear on the leaderboard within a minute."