#!/usr/bin/env python3 """ Comprehensive Gradio interface for Dressify artifact management. Provides download, upload, and organization features for all system artifacts. """ import os import json import gradio as gr from typing import Dict, List, Any from utils.artifact_manager import create_artifact_manager def create_artifact_management_interface(): """Create the main artifact management interface.""" with gr.Blocks(title="Dressify Artifact Management", theme=gr.themes.Soft()) as interface: gr.Markdown("# 🎯 Dressify Artifact Management System") gr.Markdown("## 📦 Download, Upload, and Organize All System Artifacts") with gr.Tabs(): # Overview Tab with gr.Tab("📊 System Overview"): gr.Markdown("### 🚀 Complete System Status and Artifact Summary") with gr.Row(): refresh_overview = gr.Button("🔄 Refresh Overview", variant="primary") export_summary = gr.Button("📥 Export Summary JSON", variant="secondary") overview_display = gr.JSON(label="System Overview", value=get_system_overview()) refresh_overview.click( fn=get_system_overview, outputs=overview_display ) export_summary.click( fn=export_system_summary, outputs=gr.File(label="Download Summary") ) # Download Management Tab with gr.Tab("📥 Download Management"): gr.Markdown("### 🎁 Create Downloadable Packages") with gr.Row(): with gr.Column(scale=1): gr.Markdown("#### 📦 Package Types") package_type = gr.Dropdown( choices=[ "complete - Everything (splits + models + metadata + configs)", "splits_only - Dataset splits only (lightweight)", "models_only - Trained models only" ], value="splits_only - Dataset splits only (lightweight)", label="Package Type" ) create_package_btn = gr.Button("🚀 Create Package", variant="primary") package_status = gr.Textbox(label="Package Status", interactive=False) with gr.Column(scale=1): gr.Markdown("#### 📋 Available Packages") packages_list = gr.JSON(label="Created Packages", value=get_available_packages()) refresh_packages = gr.Button("🔄 Refresh Packages") create_package_btn.click( fn=create_download_package, inputs=[package_type], outputs=[package_status, packages_list] ) refresh_packages.click( fn=get_available_packages, outputs=packages_list ) # Individual Files Tab with gr.Tab("📁 Individual Files"): gr.Markdown("### 🔍 Browse and Download Individual Artifacts") with gr.Row(): refresh_files = gr.Button("🔄 Refresh Files", variant="primary") download_all_btn = gr.Button("📥 Download All as ZIP", variant="secondary") files_display = gr.JSON(label="Available Files", value=get_individual_files()) refresh_files.click( fn=get_individual_files, outputs=files_display ) download_all_btn.click( fn=download_all_files, outputs=gr.File(label="Download All Files") ) # Upload & Restore Tab with gr.Tab("📤 Upload & Restore"): gr.Markdown("### 🔄 Upload Pre-processed Artifacts") gr.Markdown("Upload previously downloaded packages to avoid reprocessing.") with gr.Row(): with gr.Column(scale=1): gr.Markdown("#### 📤 Upload Package") upload_package = gr.File( label="Upload Artifact Package (.tar.gz)", file_types=[".tar.gz", ".zip"] ) upload_btn = gr.Button("📤 Upload & Extract", variant="primary") upload_status = gr.Textbox(label="Upload Status", interactive=False) with gr.Column(scale=1): gr.Markdown("#### 📋 Restore Options") restore_splits = gr.Button("🔄 Restore Splits Only", variant="secondary") restore_models = gr.Button("🔄 Restore Models Only", variant="secondary") restore_all = gr.Button("🔄 Restore Everything", variant="secondary") upload_btn.click( fn=upload_and_extract_package, inputs=[upload_package], outputs=upload_status ) restore_splits.click( fn=restore_splits_only, outputs=gr.Textbox(label="Restore Status") ) restore_models.click( fn=restore_models_only, outputs=gr.Textbox(label="Restore Status") ) restore_all.click( fn=restore_everything, outputs=gr.Textbox(label="Restore Status") ) # Hugging Face Integration Tab with gr.Tab("🤗 HF Hub Integration"): gr.Markdown("### 🚀 Push Artifacts to Hugging Face Hub") gr.Markdown("Upload your artifacts to HF Hub for easy access and sharing.") with gr.Row(): with gr.Column(scale=1): gr.Markdown("#### 🔑 Authentication") hf_token = gr.Textbox( label="Hugging Face Token", placeholder="hf_...", type="password" ) hf_username = gr.Textbox( label="HF Username", placeholder="yourusername" ) with gr.Column(scale=1): gr.Markdown("#### 📤 Push Options") push_splits = gr.Button("📤 Push Splits to HF", variant="primary") push_models = gr.Button("📤 Push Models to HF", variant="primary") push_all = gr.Button("📤 Push Everything to HF", variant="primary") push_status = gr.Textbox(label="Push Status", interactive=False) push_splits.click( fn=push_splits_to_hf, inputs=[hf_token, hf_username], outputs=push_status ) push_models.click( fn=push_models_to_hf, inputs=[hf_token, hf_username], outputs=push_status ) push_all.click( fn=push_everything_to_hf, inputs=[hf_token, hf_username], outputs=push_status ) # Runtime Fetching Tab with gr.Tab("⚡ Runtime Fetching"): gr.Markdown("### 🔄 Fetch Artifacts at Runtime") gr.Markdown("Configure the system to fetch artifacts from HF Hub instead of reprocessing.") with gr.Row(): with gr.Column(scale=1): gr.Markdown("#### 🔗 HF Hub Sources") splits_repo = gr.Textbox( label="Splits Repository", placeholder="yourusername/dressify-splits", value="Stylique/dressify-splits" ) models_repo = gr.Textbox( label="Models Repository", placeholder="yourusername/dressify-models", value="Stylique/dressify-models" ) enable_runtime_fetch = gr.Checkbox( label="Enable Runtime Fetching", value=False ) with gr.Column(scale=1): gr.Markdown("#### 🚀 Fetch Actions") fetch_splits = gr.Button("🔄 Fetch Splits", variant="primary") fetch_models = gr.Button("🔄 Fetch Models", variant="primary") fetch_all = gr.Button("🔄 Fetch Everything", variant="primary") fetch_status = gr.Textbox(label="Fetch Status", interactive=False) fetch_splits.click( fn=fetch_splits_from_hf, inputs=[splits_repo], outputs=fetch_status ) fetch_models.click( fn=fetch_models_from_hf, inputs=[models_repo], outputs=fetch_status ) fetch_all.click( fn=fetch_everything_from_hf, inputs=[splits_repo, models_repo], outputs=fetch_status ) # Footer gr.Markdown("---") gr.Markdown("### 💡 Usage Instructions") gr.Markdown(""" 1. **System Overview**: Check what artifacts are available and their sizes 2. **Download Management**: Create packaged downloads for easy sharing 3. **Individual Files**: Browse and download specific artifacts 4. **Upload & Restore**: Upload previously downloaded packages 5. **HF Hub Integration**: Push artifacts to Hugging Face for sharing 6. **Runtime Fetching**: Configure automatic fetching from HF Hub """) gr.Markdown("### 🎯 Benefits") gr.Markdown(""" - **Save Time**: No more reprocessing expensive splits - **Save Resources**: Avoid re-downloading and re-extracting - **Easy Sharing**: Download packages and share with others - **HF Integration**: Push to Hub for community access - **Runtime Fetching**: Automatic artifact retrieval """) return interface # Helper functions for the interface def get_system_overview(): """Get comprehensive system overview.""" try: manager = create_artifact_manager() return manager.get_artifact_summary() except Exception as e: return {"error": str(e)} def export_system_summary(): """Export system summary as JSON file.""" try: manager = create_artifact_manager() summary = manager.get_artifact_summary() # Save to exports directory export_dir = os.getenv("EXPORT_DIR", "models/exports") os.makedirs(export_dir, exist_ok=True) summary_path = os.path.join(export_dir, "system_summary.json") with open(summary_path, 'w') as f: json.dump(summary, f, indent=2) return summary_path except Exception as e: return None def create_download_package(package_type: str): """Create a downloadable package.""" try: manager = create_artifact_manager() # Extract package type from the dropdown choice if "complete" in package_type: pkg_type = "complete" elif "splits_only" in package_type: pkg_type = "splits_only" elif "models_only" in package_type: pkg_type = "models_only" else: return f"❌ Invalid package type: {package_type}", get_available_packages() package_path = manager.create_download_package(pkg_type) package_name = os.path.basename(package_path) return f"✅ Package created: {package_name}", get_available_packages() except Exception as e: return f"❌ Failed to create package: {e}", get_available_packages() def get_available_packages(): """Get list of available packages.""" try: export_dir = os.getenv("EXPORT_DIR", "models/exports") packages = [] if os.path.exists(export_dir): for file in os.listdir(export_dir): if file.endswith((".tar.gz", ".zip")): file_path = os.path.join(export_dir, file) packages.append({ "name": file, "size_mb": round(os.path.getsize(file_path) / (1024 * 1024), 2), "path": file_path, "url": f"/files/{file}" }) return {"packages": packages} except Exception as e: return {"error": str(e)} def get_individual_files(): """Get list of individual downloadable files.""" try: manager = create_artifact_manager() files = manager.get_downloadable_files() # Group by category categorized = {} for file in files: category = file["category"] if category not in categorized: categorized[category] = [] categorized[category].append(file) return categorized except Exception as e: return {"error": str(e)} def download_all_files(): """Download all files as a ZIP archive.""" try: manager = create_artifact_manager() files = manager.get_downloadable_files() # Create ZIP with all files export_dir = os.getenv("EXPORT_DIR", "models/exports") os.makedirs(export_dir, exist_ok=True) zip_path = os.path.join(export_dir, "all_artifacts.zip") import zipfile with zipfile.ZipFile(zip_path, 'w') as zipf: for file in files: if os.path.exists(file["path"]): zipf.write(file["path"], file["name"]) return zip_path except Exception as e: return None # Placeholder functions for upload/restore features def upload_and_extract_package(upload_file): """Upload and extract a package.""" if upload_file is None: return "❌ No file uploaded" try: # This would implement actual upload and extraction logic return f"✅ Package uploaded: {upload_file.name}" except Exception as e: return f"❌ Upload failed: {e}" def restore_splits_only(): """Restore splits only.""" return "🔄 Splits restoration not yet implemented" def restore_models_only(): """Restore models only.""" return "🔄 Models restoration not yet implemented" def restore_everything(): """Restore everything.""" return "🔄 Full restoration not yet implemented" # Placeholder functions for HF Hub integration def push_splits_to_hf(token, username): """Push splits to HF Hub.""" if not token or not username: return "❌ Please provide HF token and username" return f"📤 Pushing splits to {username}/dressify-splits..." def push_models_to_hf(token, username): """Push models to HF Hub.""" if not token or not username: return "❌ Please provide HF token and username" return f"📤 Pushing models to {username}/dressify-models..." def push_everything_to_hf(token, username): """Push everything to HF Hub.""" if not token or not username: return "❌ Please provide HF token and username" return f"📤 Pushing everything to {username}/dressify..." # Placeholder functions for runtime fetching def fetch_splits_from_hf(repo): """Fetch splits from HF Hub.""" return f"🔄 Fetching splits from {repo}..." def fetch_models_from_hf(repo): """Fetch models from HF Hub.""" return f"🔄 Fetching models from {repo}..." def fetch_everything_from_hf(splits_repo, models_repo): """Fetch everything from HF Hub.""" return f"🔄 Fetching everything from {splits_repo} and {models_repo}..." if __name__ == "__main__": # Test the interface interface = create_artifact_management_interface() interface.launch()