createWorkerCommand.py

"""Generate and optionally run the Wrangler KV redirect command."""

from __future__ import annotations

import json
import os
from pathlib import Path
import shutil
import subprocess
from typing import NamedTuple

from rich.console import Console
from rich.panel import Panel
from rich.prompt import Confirm, Prompt
from rich.table import Table

console = Console()
KV_BINDING = "TOBEZDEV_COM_REDIRECTS"


class CommandPlan(NamedTuple):
    kind: str
    action: str
    commands: list[list[str]]
    details: list[tuple[str, str]]


def _resolve_wrangler_command() -> str:
    if os.name == "nt":
        local_wrangler = Path("node_modules") / ".bin" / "wrangler.cmd"
        if local_wrangler.exists():
            return str(local_wrangler)

        resolved = shutil.which("wrangler.cmd")
        if resolved:
            return resolved

    resolved = shutil.which("wrangler")
    if resolved:
        return resolved

    return "wrangler"


def _wrangler_base_args() -> list[str]:
    return [_resolve_wrangler_command(), "kv", "key"]


def _create_insert_kv_namespace_args(slug: str, url: str) -> list[str]:
    return [
        *_wrangler_base_args(),
        "put",
        f"--binding={KV_BINDING}",
        slug,
        url,
        "--preview",
        "false",
        "--remote",
    ]


def _create_list_kv_namespace_args() -> list[str]:
    return [
        *_wrangler_base_args(),
        "list",
        f"--binding={KV_BINDING}",
        "--preview",
        "false",
        "--remote",
    ]


def _create_delete_kv_namespace_args(slug: str) -> list[str]:
    return [
        *_wrangler_base_args(),
        "delete",
        slug,
        f"--binding={KV_BINDING}",
        "--preview",
        "false",
        "--remote",
    ]


def _create_get_kv_namespace_args(slug: str) -> list[str]:
    return [
        *_wrangler_base_args(),
        "get",
        slug,
        f"--binding={KV_BINDING}",
        "--text",
        "--preview",
        "false",
        "--remote",
    ]


def _display_command(args: list[str]) -> str:
    display_args = ["wrangler", *args[1:]] if len(args) > 1 else ["wrangler"]
    return subprocess.list2cmdline(display_args)


def _parse_aliases(raw_aliases: str, primary_slug: str) -> list[str]:
    aliases: list[str] = []
    seen = {primary_slug}
    for alias in raw_aliases.split(","):
        normalized = alias.strip()
        if not normalized or normalized in seen:
            continue
        seen.add(normalized)
        aliases.append(normalized)
    return aliases


def _render_header() -> None:
    console.print(
        Panel(
            "[bold cyan]Redirect Manager[/bold cyan]\n"
            f"Manage KV redirects stored in [bold]{KV_BINDING}[/bold].",
            border_style="cyan",
        )
    )


def _render_menu() -> None:
    table = Table(title="Main Menu", title_style="bold green")
    table.add_column("Option", style="bold")
    table.add_column("Action")
    table.add_row("1", "Add a new redirect entry")
    table.add_row("2", "List all redirect entries")
    table.add_row("3", "Delete an existing redirect entry")
    table.add_row("4", "Exit")
    console.print(table)


def _render_summary(plan: CommandPlan) -> None:
    table = Table(title=f"{plan.action} Preview", title_style="bold green")
    table.add_column("Field", style="bold")
    table.add_column("Value")
    for label, value in plan.details:
        table.add_row(label, value)
    for index, command in enumerate(plan.commands, start=1):
        table.add_row(f"Command {index}", _display_command(command))
    console.print(table)


def _run_subprocess(args: list[str], capture_output: bool = False) -> subprocess.CompletedProcess[str] | None:
    try:
        return subprocess.run(
            args,
            check=False,
            capture_output=capture_output,
            text=True,
        )
    except FileNotFoundError:
        console.print(
            "[bold red]Wrangler could not be found.[/bold red] Install it or run npm install in this project."
        )
        return None


def _render_redirect_rows(rows: list[tuple[str, str]]) -> None:
    table = Table(title="Redirect Entries", title_style="bold green")
    table.add_column("Key", style="bold")
    table.add_column("Value")
    for key, value in rows:
        table.add_row(key, value)
    console.print(table)


def _run_list_command(args: list[str]) -> int:
    console.print("[cyan]Loading redirect entries...[/cyan]")
    completed = _run_subprocess(args, capture_output=True)
    if completed is None:
        return 1

    if completed.returncode != 0:
        error_output = completed.stderr.strip() or completed.stdout.strip()
        if error_output:
            console.print(error_output)
        console.print(
            f"[bold red]Command failed with exit code {completed.returncode}.[/bold red]"
        )
        return completed.returncode

    try:
        keys = json.loads(completed.stdout)
    except json.JSONDecodeError:
        console.print("[bold red]Unable to parse Wrangler list output.[/bold red]")
        if completed.stdout.strip():
            console.print(completed.stdout.strip())
        return 1

    if not keys:
        console.print("[yellow]No redirect entries found.[/yellow]")
        return 0

    rows: list[tuple[str, str]] = []
    with console.status("Fetching redirect values..."):
        for item in keys:
            key = item.get("name", "")
            if not key:
                continue

            value_result = _run_subprocess(
                _create_get_kv_namespace_args(key),
                capture_output=True,
            )
            if value_result is None:
                return 1

            if value_result.returncode != 0:
                error_output = value_result.stderr.strip() or "Unable to read value"
                rows.append((key, f"[error] {error_output}"))
                continue

            rows.append((key, value_result.stdout.strip()))

    _render_redirect_rows(rows)
    console.print("[bold green]Command completed successfully.[/bold green]")
    return 0


def _run_command(plan: CommandPlan) -> int:
    if plan.kind == "list":
        return _run_list_command(plan.commands[0])

    if plan.kind == "add" and len(plan.commands) > 1:
        console.print(f"[cyan]Running {len(plan.commands)} Wrangler commands...[/cyan]")
    else:
        console.print("[cyan]Running Wrangler command...[/cyan]")

    for index, command in enumerate(plan.commands, start=1):
        if plan.kind == "add" and len(plan.commands) > 1:
            console.print(f"[dim]Creating entry {index} of {len(plan.commands)}[/dim]")

        completed = _run_subprocess(command)
        if completed is None:
            return 1

        if completed.returncode != 0:
            console.print(
                f"[bold red]Command failed with exit code {completed.returncode}.[/bold red]"
            )
            return completed.returncode

    console.print("[bold green]Command completed successfully.[/bold green]")
    return 0


def _build_add_plan() -> CommandPlan | None:
    slug = Prompt.ask("[bold]Slug[/bold]").strip()
    if not slug:
        console.print("[bold red]Slug is required.[/bold red]")
        return None

    alias_input = console.input("[bold]Aliases[/bold] (comma-separated, optional): ").strip()
    aliases = _parse_aliases(alias_input, slug)

    url = Prompt.ask("[bold]URL[/bold]").strip()
    if not url:
        console.print("[bold red]URL is required.[/bold red]")
        return None

    slugs = [slug, *aliases]
    commands = [_create_insert_kv_namespace_args(item, url) for item in slugs]

    return CommandPlan(
        kind="add",
        action="Add Redirect",
        commands=commands,
        details=[
            ("Primary slug", slug),
            ("Aliases", ", ".join(aliases) if aliases else "None"),
            ("Total entries", str(len(slugs))),
            ("URL", url),
            ("Binding", KV_BINDING),
        ],
    )


def _build_list_plan() -> CommandPlan:
    return CommandPlan(
        kind="list",
        action="List Redirects",
        commands=[_create_list_kv_namespace_args()],
        details=[("Binding", KV_BINDING), ("Scope", "Remote production namespace")],
    )


def _build_delete_plan() -> CommandPlan | None:
    slug = Prompt.ask("[bold]Slug to delete[/bold]").strip()
    if not slug:
        console.print("[bold red]Slug is required.[/bold red]")
        return None

    return CommandPlan(
        kind="delete",
        action="Delete Redirect",
        commands=[_create_delete_kv_namespace_args(slug)],
        details=[("Slug", slug), ("Binding", KV_BINDING)],
    )


def _choose_action() -> str:
    _render_menu()
    return Prompt.ask(
        "[bold]Choose an option[/bold]",
        choices=["1", "2", "3", "4"],
        default="1",
    )


def main() -> int:
    _render_header()

    while True:
        choice = _choose_action()
        if choice == "1":
            plan = _build_add_plan()
        elif choice == "2":
            plan = _build_list_plan()
        elif choice == "3":
            plan = _build_delete_plan()
        else:
            console.print("[yellow]Exited without making changes.[/yellow]")
            return 0

        if plan is None:
            continue

        _render_summary(plan)

        if not Confirm.ask("Run this command now?", default=False):
            console.print("[yellow]Skipped running command.[/yellow]")
        else:
            result = _run_command(plan)
            if result != 0:
                return result

        console.print()


if __name__ == "__main__":
    raise SystemExit(main())