|
|
import os |
|
|
import gradio as gr |
|
|
from hf_hub_cto import Repo |
|
|
from hf_hub_cto.config import ( |
|
|
HF_TOKEN_FILE_PATH, |
|
|
MAX_FILE_SIZE, |
|
|
MAX_REPO_SIZE, |
|
|
REPO_TYPES, |
|
|
) |
|
|
from hf_hub_cto.repo_info import RepoInfo |
|
|
from hf_hub_cto.repo_utils import ( |
|
|
clone_repo, |
|
|
delete_repo, |
|
|
get_repo_info_df, |
|
|
get_repo_files, |
|
|
is_valid_file_size, |
|
|
repo_file_upload, |
|
|
repo_rename_file, |
|
|
sync_repo, |
|
|
) |
|
|
|
|
|
|
|
|
repo = Repo( |
|
|
repo_id=os.getenv("REPO_ID"), |
|
|
repo_type=os.getenv("REPO_TYPE"), |
|
|
token=os.getenv("HF_TOKEN"), |
|
|
local_dir=os.getenv("LOCAL_DIR"), |
|
|
) |
|
|
|
|
|
def get_session_info(session_id: str): |
|
|
user_info = gr.LoginButton.get_user_info(session_id) |
|
|
if not user_info: |
|
|
return None |
|
|
return user_info |
|
|
|
|
|
|
|
|
def set_repo_info(repo_id: str, repo_type: str, token: str): |
|
|
repo_id_info = RepoInfo(repo_id=repo_id, repo_type=repo_type) |
|
|
return gr.update(value=repo_id), gr.update(value=repo_type), gr.update(value=token), repo_id_info |
|
|
|
|
|
|
|
|
def get_all_repos_df(session_id: str): |
|
|
user_info = get_session_info(session_id) |
|
|
if not user_info: |
|
|
return gr.DataFrame(visible=False), gr.update(visible=True) |
|
|
|
|
|
df = get_repo_info_df(user_info["name"], token=os.getenv("HF_TOKEN")) |
|
|
df = df.sort_values(by="last_modified", ascending=False).reset_index(drop=True) |
|
|
return gr.DataFrame(value=df, visible=True), gr.update(visible=False) |
|
|
|
|
|
|
|
|
def handle_repo_selection( |
|
|
evt: gr.SelectData, |
|
|
repo_df: gr.DataFrame, |
|
|
session_id: str, |
|
|
): |
|
|
selected_row = repo_df.loc[evt.index[0]] |
|
|
repo_id = selected_row["repoId"] |
|
|
repo_type = selected_row["type"] |
|
|
return ( |
|
|
gr.Textbox(value=repo_id), |
|
|
gr.Textbox(value=repo_type), |
|
|
gr.Button(visible=True), |
|
|
gr.Button(visible=True), |
|
|
gr.Button(visible=True), |
|
|
) |
|
|
|
|
|
|
|
|
def handle_create_repo(repo_id: str, repo_type: str): |
|
|
try: |
|
|
if repo_type == "Model": |
|
|
Repo(repo_id, repo_type="model").create_repo() |
|
|
elif repo_type == "Dataset": |
|
|
Repo(repo_id, repo_type="dataset").create_repo() |
|
|
else: |
|
|
Repo(repo_id, repo_type="space").create_repo() |
|
|
return "Repo creado exitosamente.", None |
|
|
except Exception as e: |
|
|
return None, str(e) |
|
|
|
|
|
|
|
|
def handle_delete_repo(repo_id: str, repo_type: str, session_id: str): |
|
|
user_info = get_session_info(session_id) |
|
|
if not user_info: |
|
|
return "Error: usuario no autenticado." |
|
|
try: |
|
|
delete_repo( |
|
|
repo_id, |
|
|
repo_type, |
|
|
user_info["name"], |
|
|
os.getenv("HF_TOKEN"), |
|
|
) |
|
|
return "Repo eliminado exitosamente." |
|
|
except Exception as e: |
|
|
return f"Error al eliminar el repositorio: {str(e)}" |
|
|
|
|
|
|
|
|
def get_repo_contents(repo_id: str, repo_type: str): |
|
|
files = get_repo_files(repo_id=repo_id, repo_type=repo_type, token=os.getenv("HF_TOKEN")) |
|
|
return gr.DataFrame(value=files) |
|
|
|
|
|
|
|
|
def handle_clone_repo(repo_id: str, repo_type: str): |
|
|
try: |
|
|
clone_repo(repo_id, repo_type) |
|
|
return "Repositorio clonado exitosamente." |
|
|
except Exception as e: |
|
|
return f"Error al clonar el repositorio: {str(e)}" |
|
|
|
|
|
|
|
|
def handle_sync_repo(repo_id: str, repo_type: str): |
|
|
try: |
|
|
sync_repo(repo_id, repo_type) |
|
|
return "Repositorio sincronizado exitosamente." |
|
|
except Exception as e: |
|
|
return f"Error al sincronizar el repositorio: {str(e)}" |
|
|
|
|
|
|
|
|
def handle_file_upload( |
|
|
files: list, repo_id: str, repo_type: str, upload_folder: str |
|
|
): |
|
|
try: |
|
|
for file in files: |
|
|
if not is_valid_file_size(file.size, MAX_FILE_SIZE): |
|
|
raise ValueError( |
|
|
f"El archivo {file.name} excede el límite de tamaño de {MAX_FILE_SIZE / (1024 ** 2)} MB." |
|
|
) |
|
|
repo_file_upload( |
|
|
repo_id, repo_type, file.name, upload_folder, os.getenv("HF_TOKEN") |
|
|
) |
|
|
return "Archivos cargados exitosamente." |
|
|
except Exception as e: |
|
|
return f"Error al cargar archivos: {str(e)}" |
|
|
|
|
|
|
|
|
def handle_rename_file(repo_id: str, repo_type: str, old_path: str, new_path: str): |
|
|
try: |
|
|
repo_rename_file( |
|
|
repo_id, |
|
|
repo_type, |
|
|
old_path, |
|
|
new_path, |
|
|
os.getenv("HF_TOKEN"), |
|
|
) |
|
|
return "Archivo renombrado exitosamente." |
|
|
except Exception as e: |
|
|
return f"Error al renombrar el archivo: {str(e)}" |
|
|
|
|
|
|
|
|
|
|
|
with gr.Blocks(theme=gr.themes.Base()) as demo: |
|
|
|
|
|
pass |
|
|
|
|
|
demo.launch() |
|
|
|