DJ-Goanna-Coding/oppo-node

1"""2HuggingFace Space Synchronization Utilities3 4This module handles synchronization between GitHub repositories and HuggingFace Spaces.5"""6 7import os8import shutil9from pathlib import Path10from typing import List, Dict, Any, Optional11from huggingface_hub import HfApi, Repository, create_repo12from dotenv import load_dotenv13import yaml14 15load_dotenv()16 17 18class HFSpaceSync:19    """20    Manages synchronization between local files and HuggingFace Spaces.21    """22 23    def __init__(self, config_path: str = "config.yaml"):24        """25        Initialize HuggingFace Space synchronization.26 27        Args:28            config_path: Path to configuration file29        """30        self.config = self._load_config(config_path)31        self.hf_token = os.getenv("HF_TOKEN")32        self.hf_username = os.getenv("HF_USERNAME")33        self.api = HfApi(token=self.hf_token)34 35        if not self.hf_token:36            raise ValueError("HF_TOKEN not found in environment variables")37 38    def _load_config(self, config_path: str) -> Dict[str, Any]:39        """Load configuration from YAML file."""40        try:41            with open(config_path, 'r') as f:42                return yaml.safe_load(f)43        except FileNotFoundError:44            print(f"Config file {config_path} not found, using defaults")45            return {}46 47    def create_space(self, space_name: str, space_type: str = "streamlit",48                     private: bool = False) -> str:49        """50        Create a new HuggingFace Space.51 52        Args:53            space_name: Name of the space to create54            space_type: Type of space (streamlit, gradio, static)55            private: Whether the space should be private56 57        Returns:58            Space repository ID59        """60        try:61            repo_id = f"{self.hf_username}/{space_name}"62            create_repo(63                repo_id=repo_id,64                token=self.hf_token,65                repo_type="space",66                space_sdk=space_type,67                private=private68            )69            print(f"Created space: {repo_id}")70            return repo_id71        except Exception as e:72            if "already exists" in str(e).lower():73                print(f"Space {repo_id} already exists")74                return repo_id75            else:76                raise77 78    def upload_files(self, space_name: str, local_path: str,79                     repo_path: str = "", commit_message: str = "Update files") -> None:80        """81        Upload files to a HuggingFace Space.82 83        Args:84            space_name: Name of the space85            local_path: Local file or directory path86            repo_path: Path in the repository (default: root)87            commit_message: Commit message88        """89        repo_id = f"{self.hf_username}/{space_name}"90 91        try:92            if os.path.isfile(local_path):93                self.api.upload_file(94                    path_or_fileobj=local_path,95                    path_in_repo=repo_path or os.path.basename(local_path),96                    repo_id=repo_id,97                    repo_type="space",98                    commit_message=commit_message99                )100                print(f"Uploaded {local_path} to {repo_id}")101            elif os.path.isdir(local_path):102                self.api.upload_folder(103                    folder_path=local_path,104                    path_in_repo=repo_path,105                    repo_id=repo_id,106                    repo_type="space",107                    commit_message=commit_message108                )109                print(f"Uploaded folder {local_path} to {repo_id}")110        except Exception as e:111            print(f"Error uploading to {repo_id}: {e}")112            raise113 114    def sync_directory(self, space_name: str, local_dir: str = ".",115                       exclude_patterns: List[str] = None) -> Dict[str, Any]:116        """117        Synchronize a local directory with a HuggingFace Space.118 119        Args:120            space_name: Name of the space121            local_dir: Local directory to sync122            exclude_patterns: Patterns to exclude from sync123 124        Returns:125            Dictionary with sync results126        """127        if exclude_patterns is None:128            exclude_patterns = self.config.get('sync', {}).get('exclude_patterns', [])129 130        repo_id = f"{self.hf_username}/{space_name}"131 132        # Ensure space exists133        try:134            self.api.repo_info(repo_id=repo_id, repo_type="space")135        except Exception:136            print(f"Space {repo_id} doesn't exist, creating...")137            self.create_space(space_name)138 139        # Upload directory140        uploaded_files = []141        skipped_files = []142 143        for root, dirs, files in os.walk(local_dir):144            # Filter directories145            dirs[:] = [d for d in dirs if not any(pattern in d for pattern in exclude_patterns)]146 147            for file in files:148                file_path = os.path.join(root, file)149 150                # Check if file should be excluded151                if any(pattern in file_path for pattern in exclude_patterns):152                    skipped_files.append(file_path)153                    continue154 155                try:156                    rel_path = os.path.relpath(file_path, local_dir)157                    self.api.upload_file(158                        path_or_fileobj=file_path,159                        path_in_repo=rel_path,160                        repo_id=repo_id,161                        repo_type="space",162                        commit_message=f"Sync: {rel_path}"163                    )164                    uploaded_files.append(rel_path)165                    print(f"Synced: {rel_path}")166                except Exception as e:167                    print(f"Error syncing {file_path}: {e}")168                    skipped_files.append(file_path)169 170        return {171            "space": repo_id,172            "uploaded": len(uploaded_files),173            "skipped": len(skipped_files),174            "uploaded_files": uploaded_files,175            "skipped_files": skipped_files176        }177 178    def list_spaces(self) -> List[Dict[str, Any]]:179        """180        List all spaces for the authenticated user.181 182        Returns:183            List of space information dictionaries184        """185        spaces = self.api.list_models(author=self.hf_username, filter="space")186        return [{"id": space.id, "private": space.private} for space in spaces]187 188    def download_space(self, space_name: str, local_dir: str) -> str:189        """190        Download a space to a local directory.191 192        Args:193            space_name: Name of the space194            local_dir: Local directory to download to195 196        Returns:197            Path to downloaded directory198        """199        repo_id = f"{self.hf_username}/{space_name}"200 201        from huggingface_hub import snapshot_download202 203        snapshot_download(204            repo_id=repo_id,205            repo_type="space",206            local_dir=local_dir,207            token=self.hf_token208        )209 210        print(f"Downloaded {repo_id} to {local_dir}")211        return local_dir212 213 214if __name__ == "__main__":215    # Example usage216    sync = HFSpaceSync()217    print("Available spaces:")218    for space in sync.list_spaces():219        print(f"  - {space['id']}")220