enhanced-mcp-tools/examples/demo_archive_operations.py

#!/usr/bin/env python3
"""
Demo script showing archive operations in action
This demonstrates the comprehensive archive support for tar, tgz, bz2, xz, and zip formats.
"""

import asyncio
import tempfile
from pathlib import Path

from enhanced_mcp.mcp_server import MCPToolServer


async def demo_archive_operations():
    """Demonstrate the archive operations functionality"""

    print("🗃️  Enhanced MCP Tools - Archive Operations Demo")
    print("=" * 60)

    # Create server instance
    server = MCPToolServer("Archive Demo Server")

    # Access the archive operations directly
    archive_ops = server.archive

    # Create test environment
    with tempfile.TemporaryDirectory() as temp_dir:
        temp_path = Path(temp_dir)
        print(f"📁 Working in: {temp_path}")

        # Create sample project structure
        project_dir = temp_path / "sample_project"
        project_dir.mkdir()

        # Add some files
        (project_dir / "README.md").write_text(
            """# Sample Project
        
This is a sample project for testing archive operations.

## Features
- Comprehensive format support
- Security-focused extraction
- Progress reporting
"""
        )

        (project_dir / "main.py").write_text(
            """#!/usr/bin/env python3
def main():
    print("Hello from the archived project!")
    
if __name__ == "__main__":
    main()
"""
        )

        # Create subdirectory with config
        config_dir = project_dir / "config"
        config_dir.mkdir()
        (config_dir / "settings.json").write_text('{"debug": true, "version": "1.0.0"}')

        docs_dir = project_dir / "docs"
        docs_dir.mkdir()
        (docs_dir / "api.md").write_text("# API Documentation\n\nAPI endpoints and usage examples.")

        print(f"📋 Created sample project with {len(list(project_dir.rglob('*')))} items")

        # Test different archive formats
        formats = [
            ("tar.gz", "Standard gzipped tar"),
            ("tar.bz2", "Bzip2 compressed tar"),
            ("tar.xz", "XZ compressed tar"),
            ("zip", "Standard ZIP format"),
        ]

        for fmt, description in formats:
            print(f"\n📦 Testing {description} ({fmt})")

            # Create archive
            archive_path = temp_path / f"project_backup.{fmt}"

            result = await archive_ops.create_archive(
                source_paths=[str(project_dir)],
                output_path=str(archive_path),
                format=fmt,
                exclude_patterns=["*.pyc", "__pycache__", ".git"],
                compression_level=6,
            )

            print(f"   ✅ Created: {result['files_count']} files")
            print(f"   📊 Original: {result['total_size_bytes']} bytes")
            print(f"   📊 Compressed: {result['compressed_size_bytes']} bytes")
            print(f"   📊 Ratio: {result['compression_ratio_percent']:.1f}%")

            # List contents
            list_result = await archive_ops.list_archive(
                archive_path=str(archive_path), detailed=False
            )

            print(f"   📋 Archive contains {list_result['total_files']} items:")
            for item in list_result["contents"][:3]:
                print(f"      {item['type']}: {item['name']}")
            if len(list_result["contents"]) > 3:
                print(f"      ... and {len(list_result['contents']) - 3} more")

            # Test extraction
            extract_dir = temp_path / f"extracted_{fmt.replace('.', '_')}"
            extract_result = await archive_ops.extract_archive(
                archive_path=str(archive_path), destination=str(extract_dir), overwrite=True
            )

            print(f"   📤 Extracted {extract_result['files_extracted']} files")

            # Verify extraction
            extracted_readme = extract_dir / "sample_project" / "README.md"
            if extracted_readme.exists():
                print("   ✅ Verification: README.md extracted successfully")
            else:
                print("   ❌ Verification failed")

        # Test individual file compression
        print("\n🗜️  Testing individual file compression")
        test_file = project_dir / "README.md"

        algorithms = ["gzip", "bzip2", "xz"]
        for algorithm in algorithms:
            result = await archive_ops.compress_file(
                file_path=str(test_file),
                algorithm=algorithm,
                compression_level=6,
                keep_original=True,
            )

            print(
                f"   {algorithm.upper()}: {result['original_size_bytes']} → "
                f"{result['compressed_size_bytes']} bytes "
                f"({result['compression_ratio_percent']:.1f}% reduction)"
            )

        print("\n🎉 Archive operations demo completed successfully!")
        print("📝 All formats supported: tar, tar.gz, tgz, tar.bz2, tar.xz, zip")
        print("🔧 Features include: compression levels, exclusion patterns, security checks")
        print("🚀 Ready for production use with uv and MCP!")


if __name__ == "__main__":
    asyncio.run(demo_archive_operations())