data/dataset_manager.py

from huggingface_hub import HfApi
import gradio as gr
import pandas as pd
import os
import hashlib
from datasets import Dataset, concatenate_datasets, load_dataset
from huggingface_hub import login
import json
import uuid
from datetime import datetime, UTC
import torch
from huggingface_hub import snapshot_download
import numpy as np

from config.settings import *
from data.contexts import CONTEXTS

HF_TOKEN = os.environ.get("MyJulySecretToken")  # store your token as a secret in Spaces
login(HF_TOKEN)



api = HfApi(token = HF_TOKEN)

def load_hf_dataset():
    """Load existing HF Dataset or create empty one if not exists."""
    try:
        ds = load_dataset(HF_DATASET_NAME, split="train")
    except:
        # Dataset does not exist yet
        df = pd.DataFrame(columns=["user_id", "gender", "audio_file", "score"])
        ds = Dataset.from_pandas(df)
        ds.push_to_hub(HF_DATASET_NAME, private=True)
    return ds

def upload_generated_audio(
        audio_path,
        model_name,
        context,
        comparison_id
):
    audio_name = (
        f"{model_name}_{context}_{comparison_id}.wav"
    )

    api.upload_file(
        path_or_fileobj = audio_path,
        path_in_repo = f"data/{audio_name}",
        repo_id = HF_DATASET_NAME,
        repo_type = "dataset"
    )

    return audio_name

def upload_admin_audio(local_path: str, audio_name: str, audio_id: str) -> str:
    """Upload an audio file to the Hugging Face dataset repository."""
    repo_path = f"admin_audio/{audio_id}_{audio_name}"
    with open(local_path, "rb") as f:
        api.upload_file(
            path_or_fileobj = f,
            path_in_repo = repo_path,
            repo_id = HF_DATASET_NAME,
            repo_type = "dataset"
        )
    return repo_path

def export_full_dataset() -> str:
    import tempfile, zipfile, glob

    tmp_dir = tempfile.mkdtemp()
    repo_dir = snapshot_download(
        repo_id = HF_DATASET_NAME,
        repo_type = "dataset",
        token = HF_TOKEN,
        local_dir = tmp_dir
    )

    #Fusionner tous les JSONs annotation en un seul CSV
    json_files = glob.glob(os.path.join(repo_dir, "annotations", "*.json"))
    records = []
    for path in json_files:
        with open(path, "r", encoding = "utf-8") as f:
            records.append(json.load(f))
    df = pd.DataFrame(records)
    csv_path = os.path.join(tmp_dir, "annotations.csv")
    df.to_csv(csv_path, index = False)

    #Emballer CSV + les fichiers audio dans un ZIP
    zip_path = os.path.join(tmp_dir, "dataset_export.zip")
    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
        zf.write(csv_path, "annotations.csv")
        for audio in glob.glob(os.path.join(repo_dir, "data", "*.wav")):
            zf.write(audio, os.path.join("audio", os.path.basename(audio)))

    return zip_path

def export_admin_audio_annotations() -> str:
    """Fusionne uniquement les annotations issues du mode audio_file en un CSV"""
    import tempfile, glob

    tmp_dir = tempfile.mkdtemp()
    repo_dir = snapshot_download(
        repo_id = HF_DATASET_NAME,
        repo_type = "dataset",
        token = HF_TOKEN,
        local_dir = tmp_dir,
        allow_patterns = ["annotations_*/*.json"]
    )

    json_files = glob.glob(os.path.join(repo_dir, "annotations_*", "*.json"))
    records = []
    for path in json_files:
        with open(path, "r", encoding = "utf-8") as f:
            records.append(json.load(f))

    df = pd.DataFrame(records)
    csv_path = os.path.join(tmp_dir, "admin_audio_annotations.csv")
    df.to_csv(csv_path, index = False)
    return csv_path