Mayacontigo/.mise/tasks/dev.py

#!/usr/bin/env -S uv run --script
#MISE silent=true
#MISE description="Start the development environment for a project"
# /// script
# dependencies = ["rich"]
# ///
import argparse
import signal
import subprocess
import sys
import threading
import time
import json
from pathlib import Path
from typing import Optional, List, Dict

from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.live import Live
from rich.layout import Layout
from rich.status import Status
from rich.prompt import Prompt

console = Console()

# Configuration
DEFAULT_FRONTEND_CMD = ["pnpm", "run", "dev"]
DEFAULT_BACKEND_CMD = ["uv", "run", "uvicorn", "api.server:app", "--reload"]
DEFAULT_FRONTEND_PORT = 3000
DEFAULT_BACKEND_PORT = 8000

class ProcessManager:
    def __init__(self):
        self.processes: Dict[str, subprocess.Popen] = {}
        self.running = False
        self.logs: Dict[str, List[str]] = {"frontend": [], "backend": []}

    def start_process(self, name: str, cmd: List[str], cwd: Optional[str] = None) -> bool:
        """Start a process and track it."""
        try:
            console.print(f"[dim]Starting {name} in {cwd or 'current directory'}: {' '.join(cmd)}[/dim]")

            process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                bufsize=1,
                universal_newlines=True,
                cwd=cwd
            )
            self.processes[name] = process

            # Start thread to capture output
            thread = threading.Thread(
                target=self._capture_output,
                args=(name, process),
                daemon=True
            )
            thread.start()

            return True
        except Exception as e:
            console.print(f"[red]Failed to start {name}: {e}[/red]")
            return False

    def _capture_output(self, name: str, process: subprocess.Popen):
        """Capture process output in a separate thread."""
        while True:
            stdout = process.stdout
            assert stdout is not None
            output = stdout.readline()
            if output == '' and process.poll() is not None:
                break
            if output:
                # Keep only last 50 lines
                self.logs[name].append(output.strip())
                if len(self.logs[name]) > 50:
                    self.logs[name].pop(0)

    def stop_all(self):
        """Stop all processes."""
        self.running = False
        for name, process in self.processes.items():
            try:
                process.terminate()
                process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()
            except Exception:
                pass

    def is_running(self) -> bool:
        """Check if any process is still running."""
        return any(p.poll() is None for p in self.processes.values())

    def get_status(self) -> Dict[str, str]:
        """Get status of all processes."""
        status = {}
        for name, process in self.processes.items():
            if process.poll() is None:
                status[name] = "🟢 Running"
            else:
                status[name] = f"🔴 Stopped (exit code: {process.poll()})"
        return status

def find_project_root() -> Optional[Path]:
    """Find the project root directory by looking for package.json."""
    current_dir = Path(__file__).resolve().parent

    # Start from the script directory and go up
    while current_dir != current_dir.parent:
        if (current_dir / "package.json").exists():
            return current_dir
        current_dir = current_dir.parent

    return None

def get_available_apps(project_root: Path) -> List[str]:
    """Get list of available apps in the monorepo."""
    apps_dir = project_root / "apps"
    if not apps_dir.exists():
        return []

    apps = []
    for app_dir in apps_dir.iterdir():
        if app_dir.is_dir() and (app_dir / "package.json").exists():
            apps.append(app_dir.name)

    return sorted(apps)

def get_package_name(app_path: Path) -> str:
    """Get the package name from package.json."""
    try:
        with open(app_path / "package.json", 'r') as f:
            package_data = json.load(f)
            return package_data.get("name", "")
    except (FileNotFoundError, json.JSONDecodeError):
        return ""

def check_dependencies(project_root: Path, app_name: str | None = None) -> Dict[str, bool]:
    """Check if required dependencies are available."""
    deps = {}

    # Check pnpm global availability
    try:
        subprocess.run(["pnpm", "--version"], capture_output=True, check=True)
        deps["pnpm"] = True
    except (subprocess.CalledProcessError, FileNotFoundError):
        deps["pnpm"] = False

    # Check uv global availability
    try:
        subprocess.run(["uv", "--version"], capture_output=True, check=True)
        deps["uv"] = True
    except (subprocess.CalledProcessError, FileNotFoundError):
        deps["uv"] = False

    # Check project structure
    deps["package.json"] = (project_root / "package.json").exists()
    deps["apps directory"] = (project_root / "apps").exists()

    # If app is specified, check app-specific dependencies
    if app_name:
        app_path = project_root / "apps" / app_name

        # Check pnpm workspace install for specific app
        if deps["pnpm"] and (app_path / "package.json").exists():
            try:
                # Use pnpm list to check if the workspace package is valid
                subprocess.run(
                    ["pnpm", "list", "--filter", app_name, "--depth", "0"],
                    capture_output=True,
                    check=True,
                    cwd=str(project_root),
                    timeout=15
                )
                deps[f"pnpm install ({app_name})"] = True
            except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
                deps[f"pnpm install ({app_name})"] = False
        else:
            deps[f"pnpm install ({app_name})"] = False

        # Check uv sync for specific app
        if deps["uv"] and (app_path / "api").exists():
            try:
                # Check if we can sync the package (dry-run)
                subprocess.run(
                    ["uv", "sync", "--package", app_name, "--dry-run"],
                    capture_output=True,
                    check=True,
                    cwd=str(project_root),
                    timeout=15
                )
                deps[f"uv sync ({app_name})"] = True
            except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
                deps[f"uv sync ({app_name})"] = False
        else:
            deps[f"uv sync ({app_name})"] = False

    return deps

def display_dependency_status(deps: Dict[str, bool]) -> bool:
    """Display dependency status and return if all are satisfied."""
    table = Table(title="🔍 Dependency Check")
    table.add_column("Dependency", style="cyan")
    table.add_column("Status", style="bold")
    table.add_column("Description", style="dim")

    descriptions = {
        "pnpm": "Node.js package manager",
        "uv": "Python package manager",
        "package.json": "Root project configuration",
        "apps directory": "Apps directory for monorepo"
    }

    all_good = True
    for dep, available in deps.items():
        if available:
            status = "[green]✅ Available[/green]"
        else:
            status = "[red]❌ Missing[/red]"
            all_good = False

        # Generate description for app-specific dependencies
        description = descriptions.get(dep, "")
        if "pnpm install" in dep:
            description = "Frontend dependencies installation"
        elif "uv sync" in dep:
            description = "Backend dependencies synchronization"

        table.add_row(dep, status, description)

    console.print(table)
    return all_good

def display_available_apps(apps: List[str], project_root: Path, show_numbers: bool = False):
    """Display available apps."""
    if not apps:
        console.print("[yellow]No apps found in the monorepo.[/yellow]")
        return

    table = Table(title="📱 Available Apps")

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

    table.add_column("App Name", style="cyan")
    table.add_column("Frontend", style="green")
    table.add_column("Backend", style="blue")

    for i, app in enumerate(apps, 1):
        app_path = project_root / "apps" / app
        has_frontend = (app_path / "package.json").exists()
        has_backend = (app_path / "api").exists()

        frontend_status = "✅" if has_frontend else "❌"
        backend_status = "✅" if has_backend else "❌"

        if show_numbers:
            table.add_row(str(i), app, frontend_status, backend_status)
        else:
            table.add_row(app, frontend_status, backend_status)

    console.print(table)

def interactive_select_app(available_apps: List[str], project_root: Path) -> Optional[str]:
    """Interactively select an app from the available apps."""
    if not available_apps:
        console.print("[red]❌ No apps found in the monorepo.[/red]")
        return None

    console.print("\n[bold]Available apps in monorepo:[/bold]")
    display_available_apps(available_apps, project_root, show_numbers=True)

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

    console.print()

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

        selected_app = available_apps[int(choice) - 1]
        console.print(f"[green]✅ Selected app: {selected_app}[/green]")
        return selected_app

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

def create_live_dashboard(manager: ProcessManager, app_name: str) -> Layout:
    """Create a live dashboard layout."""
    layout = Layout()

    layout.split_column(
        Layout(name="header", size=3),
        Layout(name="body"),
        Layout(name="footer", size=3)
    )

    layout["body"].split_row(
        Layout(name="frontend"),
        Layout(name="backend")
    )

    return layout

def update_dashboard(layout: Layout, manager: ProcessManager, app_name: str):
    """Update the dashboard with current status."""
    # Header
    layout["header"].update(
        Panel(
            f"[bold blue]🚀 Development Environment - {app_name}[/bold blue]\n"
            "[dim]Press Ctrl+C to stop all services[/dim]",
            border_style="blue"
        )
    )

    # Status
    status = manager.get_status()

    # Frontend panel
    frontend_logs = "\n".join(manager.logs.get("frontend", [])[-10:])  # Last 10 lines
    frontend_status = status.get("frontend", "🔴 Not started")

    layout["frontend"].update(
        Panel(
            f"[bold]Status:[/bold] {frontend_status}\n\n"
            f"[dim]Recent output:[/dim]\n{frontend_logs}",
            title="🎨 Frontend (pnpm)",
            border_style="green" if "Running" in frontend_status else "red"
        )
    )

    # Backend panel
    backend_logs = "\n".join(manager.logs.get("backend", [])[-10:])  # Last 10 lines
    backend_status = status.get("backend", "🔴 Not started")

    layout["backend"].update(
        Panel(
            f"[bold]Status:[/bold] {backend_status}\n\n"
            f"[dim]Recent output:[/dim]\n{backend_logs}",
            title="🔧 Backend (uvicorn)",
            border_style="green" if "Running" in backend_status else "red"
        )
    )

    # Footer
    layout["footer"].update(
        Panel(
            f"[bold]Frontend:[/bold] http://localhost:{DEFAULT_FRONTEND_PORT} | "
            f"[bold]Backend:[/bold] http://localhost:{DEFAULT_BACKEND_PORT}",
            border_style="blue"
        )
    )

def start_development_environment(
    app_name: str,
    frontend_cmd: List[str],
    backend_cmd: List[str],
    show_dashboard: bool = True,
    project_root: Path | None = None
) -> bool:
    """Start the development environment."""
    manager = ProcessManager()

    # Set up signal handler for graceful shutdown
    def signal_handler(sig, frame):
        console.print(f"\n[yellow]Shutting down development environment for {app_name}...[/yellow]")
        manager.stop_all()
        sys.exit(0)

    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    assert project_root is not None, "project_root must be provided"

    # Check dependencies first
    console.print("[bold]Checking dependencies...[/bold]")
    deps = check_dependencies(project_root, app_name)

    if not display_dependency_status(deps):
        console.print("\n[red]❌ Some dependencies are missing. Please install them first.[/red]")
        return False

    console.print("\n[green]✅ All dependencies are available![/green]")

    # Set up paths
    app_path = project_root / "apps" / app_name
    frontend_cwd = str(app_path) if (app_path / "package.json").exists() else None
    backend_cwd = str(app_path) if (app_path / "api").exists() else None

    # Show startup configuration
    config_panel = Panel.fit(
        f"[bold]App:[/bold] {app_name}\n"
        f"[bold]Frontend:[/bold] {' '.join(frontend_cmd)} (cwd: {frontend_cwd or 'N/A'})\n"
        f"[bold]Backend:[/bold] {' '.join(backend_cmd)} (cwd: {backend_cwd or 'N/A'})\n"
        f"[bold]Frontend URL:[/bold] http://localhost:{DEFAULT_FRONTEND_PORT}\n"
        f"[bold]Backend URL:[/bold] http://localhost:{DEFAULT_BACKEND_PORT}",
        title="🚀 Development Configuration",
        border_style="green"
    )
    console.print(config_panel)

    # Start processes
    console.print("\n[bold]Starting development servers...[/bold]")

    # Start frontend if available
    if frontend_cwd:
        with Status("Starting frontend server...", spinner="dots"):
            if not manager.start_process("frontend", frontend_cmd, frontend_cwd):
                console.print("[red]Failed to start frontend[/red]")
                return False
            time.sleep(2)  # Give it a moment to start
    else:
        console.print("[yellow]No frontend found for this app[/yellow]")

    # Start backend if available
    if backend_cwd:
        with Status("Starting backend server...", spinner="dots"):
            if not manager.start_process("backend", backend_cmd, backend_cwd):
                console.print("[red]Failed to start backend[/red]")
                return False
            time.sleep(2)  # Give it a moment to start
    else:
        console.print("[yellow]No backend found for this app[/yellow]")

    if not manager.processes:
        console.print("[red]No services to start for this app[/red]")
        return False

    manager.running = True

    if show_dashboard:
        # Live dashboard
        layout = create_live_dashboard(manager, app_name)

        with Live(layout, refresh_per_second=2, screen=True):
            try:
                while manager.is_running():
                    update_dashboard(layout, manager, app_name)
                    time.sleep(0.5)
            except KeyboardInterrupt:
                pass
    else:
        # Simple mode - just wait
        try:
            console.print(f"\n[green]✅ Development environment started for {app_name}![/green]")
            console.print("[dim]Press Ctrl+C to stop all services[/dim]")

            while manager.is_running():
                time.sleep(1)
        except KeyboardInterrupt:
            pass

    # Cleanup
    console.print(f"\n[yellow]Stopping all services for {app_name}...[/yellow]")
    manager.stop_all()
    console.print("[green]✅ Development environment stopped.[/green]")

    return True

def main():
    # Find project root
    project_root = find_project_root()
    if not project_root:
        console.print("[red]❌ Could not find project root (package.json not found)[/red]")
        sys.exit(1)

    console.print(f"[dim]Project root: {project_root}[/dim]")

    # Get available apps
    available_apps = get_available_apps(project_root)

    parser = argparse.ArgumentParser(
        description="Start development environment with frontend and backend servers for a monorepo app",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=f"""
Available apps: {', '.join(available_apps) if available_apps else 'None found'}

Examples:
  python dev.py                                    # Interactive app selection
  python dev.py --app bursatil                    # Start bursatil app directly
  python dev.py --app ocp --no-dashboard          # Start ocp app without dashboard
  python dev.py --list-apps                       # List available apps
  python dev.py --app bursatil --frontend-cmd "pnpm run build" # Custom frontend command
        """
    )

    parser.add_argument(
        "--app",
        choices=available_apps,
        help="App to start (if not provided, you'll be prompted to select one)"
    )

    parser.add_argument(
        "--frontend-cmd",
        default=" ".join(DEFAULT_FRONTEND_CMD),
        help="Frontend command to run (default: pnpm run dev)"
    )

    parser.add_argument(
        "--backend-cmd",
        default=" ".join(DEFAULT_BACKEND_CMD),
        help="Backend command to run (default: uv run uvicorn api.server:app --reload)"
    )

    parser.add_argument(
        "--no-dashboard",
        action="store_true",
        help="Disable the live dashboard"
    )

    parser.add_argument(
        "--check-deps",
        action="store_true",
        help="Only check dependencies and exit"
    )

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

    args = parser.parse_args()

    # Handle --check-deps
    if args.check_deps:
        console.print("[bold]Checking dependencies...[/bold]")
        deps = check_dependencies(project_root)
        display_dependency_status(deps)
        return

    # Handle --list-apps
    if args.list_apps:
        console.print("[bold]Available apps in monorepo:[/bold]")
        display_available_apps(available_apps, project_root)
        return

    # Handle app selection
    if not args.app:
        if not available_apps:
            console.print("[red]❌ No apps found in the monorepo.[/red]")
            sys.exit(1)

        # Interactive selection
        selected_app = interactive_select_app(available_apps, project_root)
        if not selected_app:
            sys.exit(1)
        args.app = selected_app

    # Parse commands
    frontend_cmd = args.frontend_cmd.split()
    backend_cmd = args.backend_cmd.split()
    show_dashboard = not args.no_dashboard

    # Start development environment
    success = start_development_environment(
        args.app,
        frontend_cmd,
        backend_cmd,
        show_dashboard,
        project_root
    )
    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)