Delete app(3).py

app(3).py

@@ -1,646 +0,0 @@
-import gradio as gr
-import os
-import logging
-import asyncio
-import tempfile
-import git
-import shutil
-import zipfile
-import secrets
-import string
-from pathlib import Path
-from typing import Tuple, Optional, Dict
-from huggingface_hub import login, HfApi, create_repo, upload_file
-from urllib.parse import urlparse
-from functools import partial
-from datetime import datetime
-
-# Setup logging
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s'
-)
-logger = logging.getLogger(__name__)
-
-class ZipHandler:
-    @staticmethod
-    def generate_password(length: int = 16) -> str:
-        """Generate a random password for ZIP encryption"""
-        alphabet = string.ascii_letters + string.digits + string.punctuation
-        return ''.join(secrets.choice(alphabet) for _ in range(length))
-
-    @staticmethod
-    async def create_zip(
-        source_path: Path,
-        use_password: bool = False,
-        password: Optional[str] = None,
-        progress_callback: Optional[callable] = None
-    ) -> Tuple[Path, Optional[str]]:
-        """Create a ZIP file from the source path"""
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-        zip_path = Path(tempfile.gettempdir()) / f"upload_{timestamp}.zip"
-        
-        if use_password and not password:
-            password = ZipHandler.generate_password()
-        
-        try:
-            total_files = sum(1 for _ in Path(source_path).rglob('*') if _.is_file())
-            processed_files = 0
-            
-            with zipfile.ZipFile(
-                zip_path,
-                'w',
-                zipfile.ZIP_DEFLATED,
-                compresslevel=9
-            ) as zipf:
-                for root, _, files in os.walk(source_path):
-                    for file in files:
-                        file_path = Path(root) / file
-                        arc_name = str(file_path.relative_to(source_path))
-                        
-                        if use_password:
-                            zipf.write(
-                                file_path,
-                                arc_name,
-                                zipfile.ZIP_DEFLATED,
-                                pwd=password.encode() if password else None
-                            )
-                        else:
-                            zipf.write(file_path, arc_name)
-                        
-                        processed_files += 1
-                        if progress_callback:
-                            progress_callback(processed_files / total_files)
-
-            return zip_path, password
-        except Exception as e:
-            logger.error(f"ZIP creation failed: {e}")
-            if zip_path.exists():
-                zip_path.unlink()
-            raise
-
-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,
-        use_zip: bool = False,
-        use_password: bool = False,
-        custom_password: Optional[str] = None,
-        progress: Optional[gr.Progress] = None
-    ) -> Dict[str, str]:
-        """Upload entire folder to repository with optional ZIP compression"""
-        if not self.token:
-            return {"status": "Please login first!", "password": None}
-
-        folder_path = Path(folder_path)
-        if not folder_path.exists():
-            return {"status": f"Folder {folder_path} does not exist!", "password": None}
-
-        try:
-            self.validate_repo_name(repo_name)
-            
-            if use_zip:
-                if progress:
-                    progress(0, desc="Creating ZIP archive...")
-                
-                zip_path, password = await ZipHandler.create_zip(
-                    folder_path,
-                    use_password,
-                    custom_password,
-                    lambda p: progress(p * 0.5) if progress else None
-                )
-                
-                if progress:
-                    progress(0.5, desc="Uploading ZIP file...")
-                
-                # Upload single ZIP file
-                await self.upload_file_worker(
-                    zip_path,
-                    zip_path.name,
-                    repo_name,
-                    repo_type,
-                    lambda: progress(1.0) if progress else None
-                )
-                
-                # Clean up ZIP file
-                zip_path.unlink()
-                
-                status = "Upload completed successfully!"
-                if password:
-                    status += f"\nZIP password: {password}"
-                return {"status": status, "password": password}
-            
-            else:
-                # Original folder upload logic
-                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 {"status": "No files found to upload", "password": None}
-
-                if progress:
-                    progress(0, desc=f"Uploading {len(files_to_upload)} files...")
-
-                uploaded_count = 0
-                def update_progress():
-                    nonlocal uploaded_count
-                    uploaded_count += 1
-                    if progress:
-                        progress(uploaded_count / len(files_to_upload))
-
-                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 {
-                    "status": f"Uploaded {successful} files successfully. Failed: {failed}",
-                    "password": None
-                }
-
-        except Exception as e:
-            logger.error(f"Upload failed: {e}")
-            return {"status": f"Upload failed: {str(e)}", "password": None}
-
-    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.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"
-            )
-            
-            with gr.Row():
-                use_zip = gr.Checkbox(label="Create ZIP Archive", value=False)
-                use_password = gr.Checkbox(
-                    label="Password Protect ZIP",
-                    value=False,
-                    interactive=True
-                )
-            
-            with gr.Row():
-                custom_password = gr.Textbox(
-                    label="Custom Password (optional)",
-                    placeholder="Leave empty for auto-generated password",
-                    visible=False
-                )
-            
-            # Show/hide password input based on checkbox
-            use_password.change(
-                fn=lambda x: gr.update(visible=x),
-                inputs=[use_password],
-                outputs=[custom_password]
-            )
-            
-            upload_btn = gr.Button("Upload Files", variant="primary")
-            upload_status = gr.Textbox(label="Status", interactive=False)
-            password_output = gr.Textbox(
-                label="ZIP Password",
-                interactive=False,
-                visible=False
-            )
-            
-            def prepare_upload_path(folder_path, use_downloaded, downloaded_path):
-                return downloaded_path if use_downloaded and downloaded_path else folder_path
-            
-            def process_upload_result(result):
-                """Process the upload result and split it into status and password"""
-                if isinstance(result, dict):
-                    return result["status"], result.get("password", "")
-                return str(result), ""
-            
-            def update_password_visibility(status, password):
-                """Update password field visibility based on whether a password exists"""
-                return gr.update(visible=bool(password))
-            
-            upload_btn.click(
-                fn=lambda *args: asyncio.run(
-                    manager.upload_folder(
-                        prepare_upload_path(args[0], args[1], args[2]),
-                        args[3],
-                        args[4],
-                        args[5],
-                        args[6],
-                        args[7]
-                    )
-                ),
-                inputs=[
-                    upload_folder_path,
-                    use_downloaded,
-                    download_path,
-                    upload_repo_name,
-                    upload_repo_type,
-                    use_zip,
-                    use_password,
-                    custom_password
-                ],
-                outputs=[upload_status, password_output]
-            ).then(
-                fn=update_password_visibility,
-                inputs=[upload_status, password_output],
-                outputs=[password_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()