""" Hugging Face Hub Uploader Menggunakan library resmi huggingface_hub untuk upload file/folder ke dataset """ import os import logging from pathlib import Path from huggingface_hub import login, upload_folder, upload_file as hf_upload_file from huggingface_hub.utils import RepositoryNotFoundError, HfHubHTTPError logger = logging.getLogger(__name__) def authenticate(token: str) -> None: """ Authenticate dengan Hugging Face Hub Args: token: Hugging Face API token """ try: login(token=token) logger.info('✅ Authenticated with Hugging Face') except Exception as e: logger.error(f'Authentication failed: {str(e)}') raise Exception('Gagal authenticate dengan Hugging Face Hub') def upload_file( file_path: str, file_name: str, repo_id: str, token: str, subfolder: str = '' ) -> dict: """ Upload single file ke Hugging Face dataset Args: file_path: Path file lokal yang akan di-upload file_name: Nama file di repository repo_id: ID repository (format: username/repo-name) token: Hugging Face API token subfolder: Subfolder dalam repository (opsional) Returns: dict: Info hasil upload dengan keys: url, file_path, repo_id, status """ try: # Validasi file if not os.path.exists(file_path): raise FileNotFoundError(f'File tidak ditemukan: {file_path}') # Setup path di repository dengan prefix /u/ if subfolder: path_in_repo = f'u/{subfolder}/{file_name}' else: path_in_repo = f'u/{file_name}' logger.info(f'Uploading {file_name} to {repo_id}/{path_in_repo}...') # Authenticate authenticate(token) # Upload file menggunakan library resmi upload_info = hf_upload_file( path_or_fileobj=file_path, path_in_repo=path_in_repo, repo_id=repo_id, repo_type='dataset', commit_message=f'Upload {file_name}' ) # Build URL url = f'https://huggingface.co/datasets/{repo_id}/blob/main/{path_in_repo}' logger.info(f'✅ File uploaded successfully: {url}') return { 'url': url, 'file_path': path_in_repo, 'repo_id': repo_id, 'path_prefix': '/u/', 'status': 'uploaded' } except RepositoryNotFoundError: raise Exception(f'Repository tidak ditemukan: {repo_id}') except HfHubHTTPError as e: if '401' in str(e): raise Exception('Token tidak valid atau expired') elif '403' in str(e): raise Exception('Akses ditolak ke repository') else: raise Exception(f'Hugging Face API Error: {str(e)}') except Exception as error: logger.error(f'Upload error: {str(error)}') raise def upload_folder_contents( folder_path: str, repo_id: str, token: str, subfolder: str = '' ) -> dict: """ Upload entire folder ke Hugging Face dataset Args: folder_path: Path folder lokal repo_id: ID repository token: Hugging Face API token subfolder: Subfolder dalam repository (opsional) Returns: dict: Info hasil upload """ try: # Validasi folder if not os.path.isdir(folder_path): raise FileNotFoundError(f'Folder tidak ditemukan: {folder_path}') # Setup path di repository dengan prefix /u/ if subfolder: path_in_repo = f'u/{subfolder}' else: path_in_repo = 'u/' logger.info(f'Uploading folder {folder_path} to {repo_id}/{path_in_repo}...') # Authenticate authenticate(token) # Upload folder menggunakan library resmi upload_info = upload_folder( folder_path=folder_path, repo_id=repo_id, repo_type='dataset', path_in_repo=path_in_repo, commit_message=f'Upload folder contents' ) # Build URL url = f'https://huggingface.co/datasets/{repo_id}/tree/main/{path_in_repo}' logger.info(f'✅ Folder uploaded successfully: {url}') return { 'url': url, 'folder_path': path_in_repo, 'repo_id': repo_id, 'path_prefix': '/u/', 'status': 'uploaded', 'upload_info': str(upload_info) } except RepositoryNotFoundError: raise Exception(f'Repository tidak ditemukan: {repo_id}') except HfHubHTTPError as e: if '401' in str(e): raise Exception('Token tidak valid atau expired') elif '403' in str(e): raise Exception('Akses ditolak ke repository') else: raise Exception(f'Hugging Face API Error: {str(e)}') except Exception as error: logger.error(f'Upload error: {str(error)}') raise