muxtools.cli

@proxy_app.default
def proxy(*tokens: Annotated[str, Parameter(allow_leading_hyphen=True)]) -> None:
    """Run a managed binary without PATH fallback, honoring project or environment configuration."""
    try:
        options, command = getopt(tokens, "hp:", ["help", "package="])
    except GetoptError as exc:
        raise ValueError(str(exc)) from None
    package = None
    for option, value in options:
        if option in ("--help", "-h"):
            app.console.print(
                "Usage: muxtools proxy [--package PACKAGE] EXECUTABLE [ARGS...]\n\n"
                "Run a binary using project/environment configuration, or the latest global install.\n"
                "  -p, --package PACKAGE  Select an installed package provider.\n\n"
                "All arguments after EXECUTABLE are forwarded unchanged. No binaries are downloaded or resolved from PATH."
            )
            return
        if package is not None:
            raise ValueError("Provide --package only once")
        if not value:
            raise ValueError("--package requires a package name")
        package = value
    if not command:
        raise ValueError("Provide an executable: muxtools proxy [--package PACKAGE] EXECUTABLE [ARGS...]")
    name, *arguments = command
    try:
        executable = _resolve_executable(name, package=package, default_global=True, strict=True)
        if executable is None:
            raise FileNotFoundError(f"Managed executable {name!r} was not found; Install or add a package to your current project.")
        args = [executable, *arguments]
        if sys.platform == "win32":
            raise SystemExit(subprocess.call(args))
        os.execv(executable, args)
    except OSError as exc:
        app.error_console.print(Text(str(exc), style="red"))
        raise SystemExit(127 if isinstance(exc, FileNotFoundError) else 126) from None

@app.command
def init(
    *,
    standalone: Annotated[bool, Parameter(negative=False)] = False,
    pyproject: Annotated[bool, Parameter(negative=False)] = False,
    local: Annotated[bool, Parameter(name=["--local", "-l"], negative=False, help="Use binaries installed in this project.")] = False,
    global_: Annotated[bool, Parameter(name=["--global", "-g"], negative=False, help="Use binaries installed for this user.")] = False,
    system: Annotated[bool, Parameter(name=["--system", "-s"], negative=False, help="Use executables on PATH.")] = False,
) -> None:
    """Initialize project configuration. Prompts for a binary mode in a terminal; defaults to local in scripts."""
    if standalone and pyproject:
        raise ValueError("Choose only one of --standalone and --pyproject")
    if sum((local, global_, system)) > 1:
        raise ValueError("Choose only one of --local, --global, and --system")
    name = "muxtools.toml" if standalone else "pyproject.toml" if pyproject else _choose("Configuration file", ["pyproject.toml", "muxtools.toml"])
    if local or global_ or system:
        mode: BinaryMode = "local" if local else "global" if global_ else "system"
    elif sys.stdin.isatty():
        modes: dict[str, BinaryMode] = {"Local (project)": "local", "Global (user)": "global", "System (PATH)": "system"}
        mode = modes[_choose("Binary mode", list(modes))]
    else:
        mode = "local"
    config = init_config(Path(name), mode)
    _result("Configuration ready:", config.path, f"{config.mode} binaries")

@app.command(name="migrate-config")
def migrate_config_command(source: Path = Path("config.ini"), *, destination: Path) -> None:
    """Copy legacy INI settings into a selected TOML file."""
    _result("Migrated configuration:", migrate_config(source, destination).path)

@app.command(name=["video-meta", "vm"])
def video_meta(input: Path, output: Path | None = None) -> None:
    """Generate VideoMeta JSON from a video."""
    path = ensure_path_exists(input, None)
    parsed = ParsedFile.from_file(path, None, False)
    if not parsed.is_video_file:
        raise ValueError(f"{path.name!r} is not a video file")
    destination = output or Path.cwd() / f"{path.stem}_meta.json"
    get_timemeta_from_video(path, 0, destination, parsed, None)
    _result("VideoMeta written:", destination)

@binaries.command(
    help_epilogue=(
        "Examples:\n"
        "- `muxtools binaries add opusenc` — add the latest version.\n"
        "- `muxtools binaries add 'x265==4.2'` — pin a version.\n"
        "- `muxtools binaries add --offline opusenc` — use cached data.\n\n"
        "No specs: choose packages in a terminal."
    )
)
def add(*specs: str, offline: Annotated[bool, Parameter(help="Use the cached catalog and installed binaries.")] = False) -> None:
    """Declare and install packages in the current project."""
    config = discover_config()
    if config is None:
        raise ValueError("No project configuration found; run 'muxtools init'")
    catalog = manager.load_catalog(offline)
    selected = _specs(specs, catalog, offline)
    if not selected:
        app.console.print("[dim]No packages selected.[/dim]")
        return
    _, versions = manager.add(config, selected, offline, catalog)
    for name, version in versions.items():
        if config.mode == "system":
            _result(name, f"{version} (catalog)", "declared; system version not checked", kind="system")
        else:
            location = "locally" if config.mode == "local" else "globally"
            _result(
                name,
                version,
                f"{'already installed' if offline else 'installed'} {location}, added to project",
                kind="existing" if offline else "success",
            )

@binaries.command(
    help_epilogue=(
        "Examples:\n"
        "- `muxtools binaries install x265` — install without declaring.\n"
        "- `muxtools binaries install -g 'x265==4.2'` — install globally.\n"
        "- `muxtools binaries install --offline opusenc` — use an installed version.\n\n"
        "No specs: choose packages in a terminal."
    )
)
def install(
    *specs: str,
    local: Annotated[bool, Parameter(name=["--local", "-l"], help="Install in this project.")] = False,
    global_: Annotated[bool, Parameter(name=["--global", "-g"], help="Install for this user.")] = False,
    offline: Annotated[bool, Parameter(help="Use the cached catalog and installed binaries.")] = False,
) -> None:
    """Install packages without modifying project declarations."""
    if local and global_:
        raise ValueError("Choose only one of --local and --global")
    config = discover_config()
    scope = "local" if local else "global" if global_ else None
    mode = scope or (config.mode if config else "global")
    root = manager.scope_path(config, scope)
    catalog = manager.load_catalog(offline)
    selected = _specs(specs, catalog, offline)
    if not selected:
        app.console.print("[dim]No packages selected.[/dim]")
        return
    for raw in selected:
        spec = manager.parse_spec(raw)
        if offline:
            name = manager.resolve_name(spec.name, catalog)
            canonical = manager.Spec(name, spec.operator, spec.version)
            matches = manager.usable_installed(canonical, catalog["packages"][name], root)
            if not matches:
                raise ValueError(f"{name} is missing and cannot be installed offline")
            item = max(matches, key=lambda value: value["version_code"])
        else:
            item = manager.install(spec, catalog, root)
        location = "locally" if mode == "local" else "globally"
        _result(
            item["name"], item["version"], f"{'already installed' if offline else 'installed'} {location}", kind="existing" if offline else "success"
        )

@binaries.command(
    help_epilogue=(
        "Examples:\n"
        "- `muxtools binaries sync` — install missing declared packages.\n"
        "- `muxtools binaries sync --offline` — check installed packages without downloading."
    )
)
def sync(*, offline: Annotated[bool, Parameter(help="Check without downloading; require installed binaries.")] = False) -> None:
    """Ensure project binary declarations are available."""
    config = resolve_binary_config(discover_config())
    if config is None:
        raise ValueError("No project configuration found")
    if not config.packages:
        app.console.print("[dim]No packages declared.[/dim]")
        return
    results = manager.sync(config, offline)
    location = "locally" if config.mode == "local" else "globally"
    for result in results:
        if result.state == "installed":
            _result(result.name, result.version, f"installed {location}")
        elif result.state == "existing":
            _result(result.name, result.version, f"already installed {location}", kind="existing")
        else:
            detail = Text("available on PATH", style="cyan")
            if result.version_unchecked:
                detail.append(" (version not checked)", style="yellow")
            _result(result.name, detail=detail, kind="system")

@binaries.command(
    name="list",
    help_epilogue=(
        "Examples:\n"
        "- `muxtools binaries list` — show project binaries.\n"
        "- `muxtools binaries list -g` — show global binaries.\n"
        "`*` marks the version selected by this project."
    ),
)
def list_binaries(*, global_: Annotated[bool, Parameter(name=["--global", "-g"], help="List user-global binaries.")] = False) -> None:
    """Show installed versions and the version selected by active declarations."""
    config = resolve_binary_config(discover_config())
    if config is None and not global_:
        app.console.print("[yellow]No project configuration found.[/yellow]")
        return
    try:
        catalog = manager.load_catalog(offline=True)
    except ValueError:
        catalog = None
    if config and config.mode == "system" and not global_:
        app.console.print(Text("Project binaries (system)", style="bold cyan"))
        for raw in config.packages:
            spec = manager.parse_spec(raw)
            package = catalog["packages"].get(spec.name) if catalog else None
            provided = package.get("provides", []) if package else [spec.name]
            missing = [binary for binary in provided if not shutil.which(binary)]
            line = Text(f"{spec.name}: ", style="bold")
            line.append("missing " + ", ".join(missing) if missing else "available on PATH", style="yellow" if missing else "green")
            if spec.operator:
                line.append(" (version not checked)", style="dim")
            app.console.print(line)
        if not config.packages:
            app.console.print("[dim]No packages declared.[/dim]")
        return

    if global_:
        scope = "global"
    else:
        assert config is not None
        scope = config.mode
    root = manager.scope_path(config, scope)
    items = manager.installed(root)
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for item in items:
        grouped[item["name"]].append(item)
    selected: dict[str, str] = {}
    problems: dict[str, str] = {}
    if config and config.mode == scope:
        selected, problems = _project_versions(config, catalog, root, items)
    if global_:
        heading = "Global binaries"
    elif config is not None and config.project_config is None:
        heading = "Environment binaries (global)"
    else:
        heading = f"Project binaries ({scope})"
    app.console.print(Text(heading, style="bold cyan"))
    names = sorted(grouped.keys() | problems.keys() | selected.keys())
    if not names:
        app.console.print("[dim]No installed binaries.[/dim]")
        return
    for name in names:
        line = Text(f"{name}: ", style="bold")
        versions = sorted(grouped.get(name, []), key=lambda item: item["version_code"], reverse=True)
        for index, item in enumerate(versions):
            if index:
                line.append(", ")
            active = selected.get(name) == item["version"]
            line.append(item["version"], style="green" if active else None)
            if active:
                line.append("*", style="bold green")
        if name in problems:
            if versions:
                line.append(" ")
            line.append(f"({problems[name]})", style="yellow")
        app.console.print(line)
    if selected:
        source = "the environment configuration" if config and config.project_config is None else "the current project"
        app.console.print(f"[dim]* selected by {source}[/dim]")

@binaries.command(
    name=["remove", "rm"],
    help_epilogue=(
        "Examples:\n"
        "- `muxtools binaries rm x265` — remove all versions (and a local declaration).\n"
        "- `muxtools binaries rm 'x265<=4.1'` — remove matching versions.\n"
        "- `muxtools binaries rm '*'` — clean unused local versions.\n"
        "- `muxtools binaries rm -g -y x265` — remove global versions without prompting.\n\n"
        "No selectors: choose installed versions in a terminal."
    ),
)
def remove(
    *selectors: str,
    local: Annotated[bool, Parameter(name=["--local", "-l"], help="Remove from project storage.")] = False,
    global_: Annotated[bool, Parameter(name=["--global", "-g"], help="Remove from user-global storage.")] = False,
    yes: Annotated[bool, Parameter(name=["--yes", "-y"], negative=False, help="Skip the confirmation prompt.")] = False,
) -> None:
    """Remove installed versions; '*' cleans unused project-local versions."""
    if local and global_:
        raise ValueError("Choose only one of --local and --global")
    config = discover_config()
    scope = "local" if local else "global" if global_ else config.mode if config and config.mode != "system" else None
    if scope is None:
        if config and sys.stdin.isatty():
            scope = _choose("Remove from which scope?", ["local", "global"])
        else:
            raise ValueError("Choose --local or --global for binary removal")
    selected = list(selectors)
    if not selected:
        if not sys.stdin.isatty():
            raise ValueError("Provide package or version selectors")
        root = manager.scope_path(config, scope)
        items = manager.installed(root)
        if not items:
            app.console.print("[dim]No installed binaries.[/dim]")
            return
        active: dict[str, str] = {}
        if config and config.mode == scope:
            try:
                catalog = manager.load_catalog(offline=True)
            except ValueError:
                catalog = None
            active, _ = _project_versions(config, catalog, root, items)
        import questionary

        choices: list[Any] = []
        last_name = None
        for item in sorted(items, key=lambda item: (item["name"], -item["version_code"], item["version"])):
            if item["name"] != last_name:
                choices.append(questionary.Separator(f"── {item['name']} ──"))
                last_name = item["name"]
            label = f"{item['version']}{'  * selected' if active.get(item['name']) == item['version'] else ''}"
            choices.append(questionary.Choice(label, value=f"{item['name']}=={item['version']}"))
        selected = _choose_many("Select installed versions to remove", choices)
        if not selected:
            app.console.print("[dim]No binaries selected.[/dim]")
            return
    plan = manager.plan_removal(config, scope, selected)
    if not plan.items and not plan.declarations:
        app.console.print("[dim]No matching installed binaries or local declarations.[/dim]")
        return
    app.console.print(Text(f"Remove from {scope} binaries:", style="bold cyan"))
    for item in plan.items:
        app.console.print(f"  {item['name']} {item['version']}")
    for raw in plan.declarations:
        app.console.print(f"  Project declaration: {raw}")
    if not yes:
        if not sys.stdin.isatty():
            raise ValueError("Use -y/--yes to confirm removal in a noninteractive session")
        if _choose("Remove these binaries and declarations?", ["No", "Yes"]) != "Yes":
            app.console.print("[dim]Removal cancelled.[/dim]")
            return
    result = manager.apply_removal(plan)
    for item in result.removed:
        _result(item["name"], item["version"], f"removed {scope} install")
    for raw in result.declarations:
        _result(raw, detail="removed from project")
    if result.failed:
        for failure in result.failed:
            app.console.print(f"[red]Could not remove {failure.item['name']} {failure.item['version']}: {failure.error}[/red]")
        raise ValueError(f"Failed to remove {len(result.failed)} installed version(s)")

@binaries.default
def binaries_menu() -> None:
    """Choose a binary management action."""
    if not sys.stdin.isatty():
        app.help_print("binaries")
        return
    config = discover_config()
    actions: dict[str, Callable[[], None]]
    if config is None:
        actions = {
            "Initialize project": init,
            "Install globally": lambda: install(global_=True),
            "List global binaries": lambda: list_binaries(global_=True),
            "Remove global binaries": lambda: remove(global_=True),
        }
    else:

        def install_selected() -> None:
            if config.mode == "system":
                scope = _choose("Install scope", ["global", "local"])
                install(global_=scope == "global", local=scope == "local")
            else:
                install()

        actions = {
            "List project binaries": list_binaries,
            "List global binaries": lambda: list_binaries(global_=True),
            "Add package": add,
            "Install package": install_selected,
            "Sync project": sync,
            "Clean unused local binaries": lambda: remove("*", local=True),
            "Remove local binaries": lambda: remove(local=True),
            "Remove global binaries": lambda: remove(global_=True),
        }
    actions[_choose("Choose a binary action", list(actions))]()