Mayacontigo/.mise/tasks/container/push.py

#!/usr/bin/env -S uv run --script
#MISE silent=true
#MISE description="Push the container to Azure"
# /// script
# dependencies = ["rich"]
# ///
import argparse
import subprocess
import sys
from typing import Optional

from rich.console import Console
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.prompt import Prompt, Confirm
from rich.table import Table

console = Console()

# Azure Container Registry configuration
AZURE_REGISTRY = "iaservicecontainers.azurecr.io"
IMAGE_PREFIX = "mayacontigo/"


def run_docker_command(command: list[str], description: str) -> bool:
    """Run a docker command with rich progress indication."""
    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        console=console,
        transient=True,
    ) as progress:
        task = progress.add_task(description, total=None)

        try:
            subprocess.run(
                command,
                capture_output=True,
                text=True,
                check=True
            )
            progress.update(task, description=f"✅ {description}")
            return True
        except subprocess.CalledProcessError as e:
            progress.update(task, description=f"❌ {description}")
            console.print(f"[red]Error running command: {' '.join(command)}[/red]")
            console.print(f"[red]Error output: {e.stderr}[/red]")
            return False
        except FileNotFoundError:
            progress.update(task, description=f"❌ {description}")
            console.print("[red]Docker command not found. Please ensure Docker is installed and in your PATH.[/red]")
            return False


def check_docker_image_exists(image_name: str) -> bool:
    """Check if a Docker image exists locally."""
    try:
        result = subprocess.run(
            ["docker", "images", "-q", image_name],
            capture_output=True,
            text=True,
            check=True
        )
        return bool(result.stdout.strip())
    except subprocess.CalledProcessError:
        return False


def get_available_images() -> list[str]:
    """Get list of available Docker images filtered by prefix."""
    try:
        result = subprocess.run(
            ["docker", "images", "--format", "{{.Repository}}:{{.Tag}}"],
            capture_output=True,
            text=True,
            check=True
        )
        images = [line.strip() for line in result.stdout.split('\n') if line.strip()]
        # Filter images to only include those with the specified prefix
        filtered_images = [img for img in images if not img.startswith('<none>') and img.startswith(IMAGE_PREFIX)]
        return filtered_images
    except subprocess.CalledProcessError:
        return []


def display_available_images(images: list[str], show_numbers: bool = False) -> None:
    """Display available Docker images in a nice table."""
    if not images:
        console.print(f"[yellow]No Docker images found locally with prefix '{IMAGE_PREFIX}'.[/yellow]")
        return

    table = Table(title=f"Available Docker Images (prefix: {IMAGE_PREFIX})")

    if show_numbers:
        table.add_column("No.", style="bold yellow", width=4)

    table.add_column("Image", style="cyan")
    table.add_column("Registry Status", style="magenta")

    for i, image in enumerate(images[:10], 1):  # Show first 10 images
        registry_status = "🏷️  Ready to tag" if not image.startswith(AZURE_REGISTRY) else "🚀 Already tagged"

        if show_numbers:
            table.add_row(str(i), image, registry_status)
        else:
            table.add_row(image, registry_status)

    if len(images) > 10:
        remaining_text = f"[dim]and {len(images) - 10} more[/dim]"
        if show_numbers:
            table.add_row("...", "...", remaining_text)
        else:
            table.add_row("...", remaining_text)

    console.print(table)


def interactive_mode() -> Optional[str]:
    """Run in interactive mode to select an image."""
    console.print(Panel.fit(
        "[bold blue]🐳 Docker Image Push to Azure[/bold blue]\n"
        "[dim]Interactive mode - Let's select an image to push[/dim]",
        border_style="blue"
    ))

    # Get available images
    console.print("\n[bold]Fetching available Docker images...[/bold]")
    images = get_available_images()

    if not images:
        console.print(f"[red]No Docker images found with prefix '{IMAGE_PREFIX}'. Please build an image first.[/red]")
        return None

    display_available_images(images, show_numbers=True)

    # Create choices with numbers
    choices = [str(i) for i in range(1, len(images) + 1)]

    console.print()

    try:
        choice = Prompt.ask(
            "Select an image to push",
            choices=choices,
            default="1"
        )

        selected_image = images[int(choice) - 1]
        console.print(f"[green]✅ Selected image: {selected_image}[/green]")
        return selected_image

    except (ValueError, IndexError, KeyboardInterrupt):
        console.print("[yellow]Selection cancelled.[/yellow]")
        return None


def push_image(image_name: str, force: bool = False) -> bool:
    """Push a Docker image to Azure Container Registry."""
    # Validate image name
    if not image_name:
        console.print("[red]Error: Image name cannot be empty[/red]")
        return False

    # Check if image has the required prefix
    if not image_name.startswith(IMAGE_PREFIX):
        console.print(f"[red]Error: Image '{image_name}' does not have the required prefix '{IMAGE_PREFIX}'[/red]")
        return False

    # Check if image exists locally
    if not check_docker_image_exists(image_name):
        console.print(f"[red]Error: Image '{image_name}' not found locally[/red]")
        console.print(f"[yellow]Tip: Run 'docker images' to see available images with prefix '{IMAGE_PREFIX}'[/yellow]")
        return False

    # Prepare Azure registry image name (remove the prefix since it will be added by the registry)
    image_name_without_prefix = image_name[len(IMAGE_PREFIX):]
    azure_image_name = f"{AZURE_REGISTRY}/{image_name_without_prefix}"

    # Show what we're about to do
    info_panel = Panel.fit(
        f"[bold]Image:[/bold] {image_name}\n"
        f"[bold]Target:[/bold] {azure_image_name}\n"
        f"[bold]Registry:[/bold] {AZURE_REGISTRY}",
        title="🚀 Push Configuration",
        border_style="green"
    )
    console.print(info_panel)

    # Confirm action unless force is specified
    if not force:
        if not Confirm.ask(f"\nProceed with pushing [cyan]{image_name}[/cyan] to Azure?"):
            console.print("[yellow]Operation cancelled.[/yellow]")
            return False

    console.print("\n[bold]Starting push process...[/bold]")

    # Tag the image
    if not run_docker_command(
        ["docker", "tag", image_name, azure_image_name],
        f"Tagging image as {azure_image_name}"
    ):
        return False

    # Push the image
    if not run_docker_command(
        ["docker", "push", azure_image_name],
        f"Pushing {azure_image_name} to registry"
    ):
        return False

    # Success message
    success_panel = Panel.fit(
        f"[bold green]✅ Successfully pushed![/bold green]\n"
        f"[dim]Image: {azure_image_name}[/dim]",
        title="🎉 Success",
        border_style="green"
    )
    console.print(success_panel)
    return True


def main():
    parser = argparse.ArgumentParser(
        description="Push Docker images to Azure Container Registry",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=f"""
Examples:
  python push.py {IMAGE_PREFIX}my-app:latest          # Push specific image
  python push.py --interactive                        # Interactive mode
  python push.py {IMAGE_PREFIX}my-app:latest --force  # Skip confirmation
  python push.py --list                              # List available images
        """
    )

    parser.add_argument(
        "image_name",
        nargs="?",
        help="Name of the Docker image to push"
    )

    parser.add_argument(
        "-i", "--interactive",
        action="store_true",
        help="Run in interactive mode"
    )

    parser.add_argument(
        "-f", "--force",
        action="store_true",
        help="Skip confirmation prompts"
    )

    parser.add_argument(
        "--list",
        action="store_true",
        help="List available Docker images and exit"
    )

    args = parser.parse_args()

    # Handle --list option
    if args.list:
        console.print("[bold]Available Docker Images:[/bold]")
        images = get_available_images()
        display_available_images(images)
        return

    # Determine image name
    image_name = args.image_name

    # Interactive mode or no image provided
    if args.interactive or not image_name:
        image_name = interactive_mode()
        if not image_name:
            sys.exit(1)

    # Push the image
    success = push_image(image_name, args.force)
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        console.print("\n[yellow]Operation cancelled by user.[/yellow]")
        sys.exit(1)
    except Exception as e:
        console.print(f"[red]Unexpected error: {e}[/red]")
        sys.exit(1)