Upload: app(2).py

app(2).py

@@ -0,0 +1,501 @@
+import gradio as gr
+import os
+import logging
+import asyncio
+import tempfile
+import git
+import shutil
+from pathlib import Path
+from typing import Tuple, Optional
+from huggingface_hub import login, HfApi, create_repo, upload_file
+from urllib.parse import urlparse
+from functools import partial
+
+# Setup logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+class HuggingFaceManager:
+    def __init__(self):
+        self.api = HfApi()
+        self.token: Optional[str] = None
+        self.temp_dir = Path(tempfile.mkdtemp())
+        
+    def cleanup(self):
+        """Clean up temporary directory"""
+        if self.temp_dir.exists():
+            shutil.rmtree(self.temp_dir, ignore_errors=True)
+            
+    def validate_repo_name(self, repo_name: str) -> bool:
+        """Validate repository name format"""
+        if not repo_name or '/' not in repo_name:
+            raise ValueError("Repository name must be in format 'username/repo-name'")
+        username, repo = repo_name.split('/', 1)
+        return all(name.strip() for name in [username, repo])
+
+    def validate_url(self, url: str) -> bool:
+        """Validate URL format"""
+        try:
+            result = urlparse(url)
+            return all([result.scheme, result.netloc])
+        except Exception:
+            return False
+
+    async def login_and_validate(self, token: str) -> str:
+        """Login to Hugging Face and validate token"""
+        if not token.strip():
+            return "Error: Token cannot be empty"
+        
+        try:
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(None, login, token)
+            await loop.run_in_executor(None, self.api.whoami)
+            
+            self.token = token
+            return "Login successful!"
+        except Exception as e:
+            logger.error(f"Login failed: {e}")
+            return f"Login failed: {str(e)}"
+
+    async def download_from_url(
+        self, 
+        url: str, 
+        download_type: str,
+        progress: Optional[gr.Progress] = None
+    ) -> Tuple[str, Optional[str]]:
+        """Download content from URL using wget or git"""
+        if not self.validate_url(url):
+            return "Invalid URL format!", None
+
+        output_path = self.temp_dir / "downloaded_content"
+        if output_path.exists():
+            shutil.rmtree(output_path)
+        output_path.mkdir(parents=True)
+
+        try:
+            if download_type == "wget":
+                if progress:
+                    progress(0, desc="Downloading with wget...")
+                
+                process = await asyncio.create_subprocess_exec(
+                    'wget', '-P', str(output_path), url,
+                    stdout=asyncio.subprocess.PIPE,
+                    stderr=asyncio.subprocess.PIPE
+                )
+                
+                try:
+                    stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=300)
+                    if process.returncode != 0:
+                        raise Exception(f"wget failed: {stderr.decode()}")
+                except asyncio.TimeoutError:
+                    raise Exception("Download timed out after 5 minutes")
+
+            elif download_type == "git":
+                if progress:
+                    progress(0, desc="Cloning git repository...")
+                
+                await asyncio.get_event_loop().run_in_executor(
+                    None,
+                    partial(git.Repo.clone_from, url, str(output_path), depth=1)
+                )
+
+            if progress:
+                progress(1, desc="Download complete")
+            
+            return "Download successful!", str(output_path)
+
+        except Exception as e:
+            logger.error(f"Download failed: {e}")
+            if output_path.exists():
+                shutil.rmtree(output_path)
+            return f"Download failed: {str(e)}", None
+
+    async def create_repository(
+        self, 
+        repo_name: str, 
+        repo_type: str, 
+        is_private: bool
+    ) -> str:
+        """Create a new repository on Hugging Face"""
+        if not self.token:
+            return "Please login first!"
+
+        try:
+            self.validate_repo_name(repo_name)
+            
+            await asyncio.get_event_loop().run_in_executor(
+                None,
+                partial(
+                    create_repo,
+                    repo_name,
+                    private=is_private,
+                    repo_type=repo_type,
+                    exist_ok=True
+                )
+            )
+            
+            return f"Repository '{repo_name}' created successfully!"
+        except Exception as e:
+            logger.error(f"Failed to create repository: {e}")
+            return f"Failed to create repository: {str(e)}"
+
+    async def upload_file_worker(
+        self, 
+        file_path: Path, 
+        relative_path: Path, 
+        repo_name: str, 
+        repo_type: str,
+        progress_callback=None
+    ) -> Optional[str]:
+        """Upload a single file to repository"""
+        try:
+            await asyncio.get_event_loop().run_in_executor(
+                None,
+                partial(
+                    upload_file,
+                    path_or_fileobj=str(file_path),
+                    path_in_repo=str(relative_path),
+                    repo_id=repo_name,
+                    repo_type=repo_type,
+                    commit_message=f"Upload: {relative_path}"
+                )
+            )
+            if progress_callback:
+                progress_callback()
+            return str(relative_path)
+        except Exception as e:
+            logger.error(f"Error uploading {relative_path}: {e}")
+            return None
+
+    async def upload_folder(
+        self, 
+        folder_path: str, 
+        repo_name: str, 
+        repo_type: str,
+        progress: Optional[gr.Progress] = None
+    ) -> str:
+        """Upload entire folder to repository"""
+        if not self.token:
+            return "Please login first!"
+
+        folder_path = Path(folder_path)
+        if not folder_path.exists():
+            return f"Folder {folder_path} does not exist!"
+
+        try:
+            self.validate_repo_name(repo_name)
+            
+            # Collect files for upload
+            files_to_upload = []
+            for root, dirs, files in os.walk(folder_path):
+                if '.git' in dirs:
+                    dirs.remove('.git')
+                
+                for file in files:
+                    file_path = Path(root) / file
+                    relative_path = file_path.relative_to(folder_path)
+                    files_to_upload.append((file_path, relative_path))
+
+            if not files_to_upload:
+                return "No files found to upload"
+
+            if progress:
+                progress(0, desc=f"Uploading {len(files_to_upload)} files...")
+
+            # Track progress
+            uploaded_count = 0
+            def update_progress():
+                nonlocal uploaded_count
+                uploaded_count += 1
+                if progress:
+                    progress(uploaded_count / len(files_to_upload))
+
+            # Upload files in parallel
+            tasks = [
+                self.upload_file_worker(
+                    file_path, 
+                    relative_path, 
+                    repo_name, 
+                    repo_type, 
+                    update_progress
+                )
+                for file_path, relative_path in files_to_upload
+            ]
+            
+            results = await asyncio.gather(*tasks, return_exceptions=True)
+            
+            successful = sum(1 for r in results if r is not None)
+            failed = len(files_to_upload) - successful
+            
+            return f"Uploaded {successful} files successfully. Failed: {failed}"
+
+        except Exception as e:
+            logger.error(f"Upload failed: {e}")
+            return f"Upload failed: {str(e)}"
+
+    async def delete_files(
+        self, 
+        folder_path: str, 
+        repo_name: str, 
+        repo_type: str, 
+        dry_run: bool = True,
+        progress: Optional[gr.Progress] = None
+    ) -> str:
+        """Delete files from repository based on local folder structure"""
+        if not self.token:
+            return "Please login first!"
+
+        try:
+            self.validate_repo_name(repo_name)
+            
+            folder_path = Path(folder_path)
+            if not folder_path.exists():
+                return f"Folder {folder_path} does not exist!"
+
+            # Get repository files
+            existing_files = await asyncio.get_event_loop().run_in_executor(
+                None,
+                partial(self.api.list_repo_files, repo_id=repo_name, repo_type=repo_type)
+            )
+
+            # Find files to delete
+            files_to_delete = []
+            for root, dirs, files in os.walk(folder_path):
+                if '.git' in dirs:
+                    dirs.remove('.git')
+                
+                for file in files:
+                    file_path = Path(root) / file
+                    relative_path = str(file_path.relative_to(folder_path))
+                    if relative_path in existing_files:
+                        files_to_delete.append(relative_path)
+
+            if not files_to_delete:
+                return "No matching files found to delete"
+
+            if dry_run:
+                return f"Dry run: Would delete {len(files_to_delete)} files"
+
+            # Perform deletion
+            if progress:
+                progress(0, desc=f"Deleting {len(files_to_delete)} files...")
+            
+            deleted_count = 0
+            for file_path in files_to_delete:
+                try:
+                    await asyncio.get_event_loop().run_in_executor(
+                        None,
+                        partial(
+                            self.api.delete_file,
+                            repo_id=repo_name,
+                            path_in_repo=file_path,
+                            repo_type=repo_type
+                        )
+                    )
+                    deleted_count += 1
+                    if progress:
+                        progress(deleted_count / len(files_to_delete))
+                except Exception as e:
+                    logger.error(f"Error deleting {file_path}: {e}")
+
+            return f"Successfully deleted {deleted_count} out of {len(files_to_delete)} files"
+
+        except Exception as e:
+            logger.error(f"Delete operation failed: {e}")
+            return f"Delete operation failed: {str(e)}"
+
+def create_interface() -> gr.Blocks:
+    """Create Gradio interface"""
+    manager = HuggingFaceManager()
+    
+    with gr.Blocks(title="Hugging Face Repository Manager") as app:
+        gr.Markdown("# Hugging Face Repository Manager")
+        
+        # Login Tab
+        with gr.Tab("Login"):
+            token_input = gr.Textbox(
+                label="Hugging Face Token",
+                type="password",
+                placeholder="Enter your Hugging Face token"
+            )
+            login_btn = gr.Button("Login", variant="primary")
+            login_output = gr.Textbox(label="Login Status", interactive=False)
+            
+            login_btn.click(
+                fn=lambda x: asyncio.run(manager.login_and_validate(x)),
+                inputs=[token_input],
+                outputs=[login_output]
+            )
+        
+        # Download Tab
+        with gr.Tab("Download"):
+            download_url = gr.Textbox(
+                label="URL",
+                placeholder="Enter URL to download"
+            )
+            download_type = gr.Radio(
+                choices=["wget", "git"],
+                label="Download Type",
+                value="wget"
+            )
+            download_btn = gr.Button("Download", variant="primary")
+            download_output = gr.Textbox(label="Status", interactive=False)
+            download_path = gr.Textbox(
+                label="Download Path",
+                interactive=False,
+                visible=False
+            )
+            
+            download_btn.click(
+                fn=lambda x, y: asyncio.run(manager.download_from_url(x, y)),
+                inputs=[download_url, download_type],
+                outputs=[download_output, download_path]
+            )
+        
+        # Create Repository Tab
+        with gr.Tab("Create Repository"):
+            repo_name = gr.Textbox(
+                label="Repository Name",
+                placeholder="username/repo-name"
+            )
+            repo_type = gr.Radio(
+                choices=["model", "dataset", "space"],
+                label="Repository Type",
+                value="model"
+            )
+            is_private = gr.Checkbox(label="Private Repository", value=True)
+            create_btn = gr.Button("Create Repository", variant="primary")
+            create_output = gr.Textbox(label="Status", interactive=False)
+            
+            create_btn.click(
+                fn=lambda x, y, z: asyncio.run(manager.create_repository(x, y, z)),
+                inputs=[repo_name, repo_type, is_private],
+                outputs=[create_output]
+            )
+        
+        # Upload Tab
+        with gr.Tab("Upload"):
+            with gr.Row():
+                upload_folder_path = gr.Textbox(
+                    label="Local Folder Path",
+                    placeholder="Path to local folder"
+                )
+                use_downloaded = gr.Checkbox(
+                    label="Use Downloaded Content",
+                    value=False
+                )
+            
+            upload_repo_name = gr.Textbox(
+                label="Repository Name",
+                placeholder="username/repo-name"
+            )
+            upload_repo_type = gr.Radio(
+                choices=["model", "dataset", "space"],
+                label="Repository Type",
+                value="model"
+            )
+            upload_btn = gr.Button("Upload Files", variant="primary")
+            upload_output = gr.Textbox(label="Status", interactive=False)
+            
+            def prepare_upload_path(folder_path, use_downloaded, downloaded_path):
+                return downloaded_path if use_downloaded and downloaded_path else folder_path
+            
+            upload_btn.click(
+                fn=lambda *args: asyncio.run(
+                    manager.upload_folder(
+                        prepare_upload_path(args[0], args[1], args[2]),
+                        args[3],
+                        args[4]
+                    )
+                ),
+                inputs=[
+                    upload_folder_path,
+                    use_downloaded,
+                    download_path,
+                    upload_repo_name,
+                    upload_repo_type
+                ],
+                outputs=[upload_output]
+            )
+        
+        # Delete Tab
+        with gr.Tab("Delete"):
+            delete_folder_path = gr.Textbox(
+                label="Local Folder Path",
+                placeholder="Path to local folder for reference"
+            )
+            delete_repo_name = gr.Textbox(
+                label="Repository Name",
+                placeholder="username/repo-name"
+            )
+            delete_repo_type = gr.Radio(
+                choices=["model", "dataset", "space"],
+                label="Repository Type",
+                value="model"
+            )
+            dry_run = gr.Checkbox(
+                label="Dry Run",
+                value=True,
+                info="Preview changes without making them"
+            )
+            delete_btn = gr.Button("Delete Files", variant="secondary")
+            delete_output = gr.Textbox(label="Status", interactive=False)
+            
+            delete_btn.click(
+                fn=lambda *args: asyncio.run(manager.delete_files(*args)),
+                inputs=[
+                    delete_folder_path,
+                    delete_repo_name,
+                    delete_repo_type,
+                    dry_run
+                ],
+                outputs=[delete_output]
+            )
+
+        # Error handling
+        gr.Error()
+        
+        # Footer information
+        gr.Markdown("""
+        ### Information
+        - Get your token from [Hugging Face Settings](https://huggingface.co/settings/tokens)
+        - Repository names must be in format: username/repository-name
+        - For files larger than 5GB, use Git LFS
+        - Repository types:
+          - Model: For ML models and weights
+          - Dataset: For datasets and data files
+          - Space: For demos and applications
+        """)
+
+    return app
+
+def main():
+    """Application entry point"""
+    try:
+        logger.info("Starting Hugging Face Repository Manager")
+        app = create_interface()
+        
+        # Configure and launch the application
+        app.queue()
+        app.launch(
+            server_name="0.0.0.0",
+            server_port=7860,
+            share=True,
+            max_threads=2,
+            # Security configurations
+            auth=None,
+            ssl_keyfile=None,
+            ssl_certfile=None,
+            ssl_verify=True
+        )
+    except Exception as e:
+        logger.error(f"Application failed to start: {e}")
+        raise
+    finally:
+        logger.info("Shutting down Hugging Face Repository Manager")
+
+if __name__ == "__main__":
+    main()