mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-10-09 15:06:39 +00:00
MacRimi
1f9d2acef5
The installation record of an OCI application travels with its container: a copy inside the container and another in /etc/pve, written together with the one kept on the host. A container restored on a newly installed Proxmox, restored with another ID or moved to another node of a cluster is recognised and registered again, with its private network, hookscript, Rclone mount, host firewall rule and NVIDIA runtime. The Monitor offers the same recovery from the Updates tab. AMD GPUs are offered by generation. A GPU the ROCm image supports takes the profile as it is; one of a supported family (Radeon 680M, 780M) is an experimental option that asks for confirmation and is never proposed; an older one is not offered. The GPU is checked with a real inference before the installation accepts it. Recreate changes what runs recognition in an installed Immich, between the CPU and a GPU of the host. Updates: - A failed update that is restored and checked removes its temporary container and the disks of the failed attempt. - Every container volume is part of the backups, so Jellyfin, Plex and Hugo update with their default installation. - An image published with a Docker-format manifest is recognised by its layers and build time and updates. - The Proxmox notes of a multi-container application link to its LAN address.
702 lines
35 KiB
Python
702 lines
35 KiB
Python
"""OCI manager Apps: catalog menus, installation flow and maintenance commands."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import json
|
|
import re
|
|
import sys
|
|
import textwrap
|
|
import unicodedata
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from . import console, images
|
|
from .catalog import Catalog
|
|
from .converter import ConversionError
|
|
from .github_source import SourceError
|
|
from .i18n import N_, source_text, translate
|
|
from .installer import (
|
|
ADVANCED_MODE,
|
|
DEFAULT_MODE,
|
|
InstallError,
|
|
build_deployment,
|
|
build_rclone_mount_deployment,
|
|
redacted,
|
|
run_remote_install,
|
|
run_remote_rclone_mount,
|
|
)
|
|
from .ui import APP_TITLE, UserCancelled, interactive_ui
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
DISPLAY_ARCHITECTURES = ("amd64", "arm64")
|
|
PUBLISHERS = {"linuxserver.io": "LinuxServer", "official": N_("Official image")}
|
|
STACK_LABELS = {
|
|
"server": N_("Server"),
|
|
"application": N_("Application"),
|
|
"machine_learning": "Machine learning",
|
|
"database": "PostgreSQL",
|
|
"valkey": "Valkey",
|
|
"cache": "Redis",
|
|
"paperless": "Paperless-ngx",
|
|
"broker": "Valkey",
|
|
"tandoor": "Tandoor",
|
|
}
|
|
|
|
|
|
def _display_architectures(item: dict[str, Any]) -> str:
|
|
supported = [a for a in DISPLAY_ARCHITECTURES if a in item.get("architectures", [])]
|
|
return "/".join(supported) or "?"
|
|
|
|
|
|
def publisher(item: dict[str, Any]) -> str:
|
|
provider = str(item.get("provider") or "")
|
|
app_id = str(item.get("id") or "").casefold()
|
|
# A project that publishes its own image (immich, frigate, nextcloud...) is its official image.
|
|
if provider == "official" or (provider and app_id.startswith(provider.casefold())):
|
|
return translate(PUBLISHERS["official"])
|
|
return PUBLISHERS.get(provider, provider or "?")
|
|
|
|
|
|
def is_tested(item: dict[str, Any]) -> bool:
|
|
"""Tested for real, in the ProxMenux lab or by the community."""
|
|
return (item.get("template_status") == "laboratory-validated"
|
|
or bool((item.get("community_tested") or {}).get("by")))
|
|
|
|
|
|
MENU_SIZE = (22, 75, 15)
|
|
MULTI_LABEL = r"\Z4MULTI\Zn"
|
|
TESTED_LABEL = r"\Z2✓\Zn"
|
|
|
|
|
|
def _installable(applications: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
return [item for item in applications if not item.get("hidden") and item.get("automatic_install_candidate")]
|
|
|
|
|
|
NAME_WIDTH, ARCH_WIDTH, SOURCE_WIDTH = 26, 12, 14
|
|
ROW_WIDTH = 72
|
|
|
|
|
|
def _app_menu(applications: list[dict[str, Any]], keep_order: bool = False
|
|
) -> tuple[list[tuple[str, str]], dict[str, dict[str, Any]], str]:
|
|
"""Numbered rows in the Helper Scripts layout (name, architecture, source,
|
|
verified) and the column header aligned with them."""
|
|
options: list[tuple[str, str]] = []
|
|
index: dict[str, dict[str, Any]] = {}
|
|
ordered = applications if keep_order else sorted(
|
|
applications, key=lambda entry: str(entry.get("title") or entry["id"]).casefold())
|
|
for number, item in enumerate(ordered, 1):
|
|
title = str(item.get("title") or item["id"])
|
|
name = " ".join("".join(ch for ch in title if unicodedata.category(ch) != "So").split())
|
|
if item.get("multi_container"):
|
|
name = name[:NAME_WIDTH - 6]
|
|
cell = f"{name} {MULTI_LABEL}" + " " * (NAME_WIDTH - len(name) - 6)
|
|
else:
|
|
cell = f"{name[:NAME_WIDTH]:<{NAME_WIDTH}}"
|
|
row = f"{cell} {_display_architectures(item):<{ARCH_WIDTH}} {publisher(item)[:SOURCE_WIDTH]:<{SOURCE_WIDTH}}"
|
|
if is_tested(item):
|
|
row += f" {TESTED_LABEL}"
|
|
# Rows as wide as the list: dialog centers narrower lists, which would move them off the header.
|
|
options.append((str(number), f"{row:<{ROW_WIDTH}}"))
|
|
index[str(number)] = item
|
|
# dialog draws the rows after the list border and the tag column.
|
|
offset = " " * (len(str(len(ordered))) + 3)
|
|
header = (f"{offset}{translate('Name')[:NAME_WIDTH]:<{NAME_WIDTH}} "
|
|
f"{translate('Architecture')[:ARCH_WIDTH]:<{ARCH_WIDTH}} "
|
|
f"{translate('Source')[:SOURCE_WIDTH]:<{SOURCE_WIDTH}} {translate('Verified')}")
|
|
return options, index, header
|
|
|
|
|
|
def _yes_no(value: Any) -> str:
|
|
return translate("yes") if value else translate("no")
|
|
|
|
|
|
# ---------------------------------------------------------------- summaries
|
|
|
|
DETAIL_WIDTH = 92
|
|
SERVICE_KINDS = (("postgres", "PostgreSQL"), ("valkey", "Valkey"), ("redis", "Redis"), ("mariadb", "MariaDB"),
|
|
("mysql", "MySQL"), ("mongo", "MongoDB"), ("meilisearch", "Meilisearch"))
|
|
|
|
|
|
def _image_label(reference: str) -> str:
|
|
return str(reference or "").split("@", 1)[0]
|
|
|
|
|
|
def _service_kind(service: dict[str, Any]) -> str:
|
|
image = str(service.get("image") or "").casefold()
|
|
name = str(service.get("name") or "").casefold()
|
|
if service.get("is_main"):
|
|
return translate("Application")
|
|
if "machine-learning" in name or "machine-learning" in image:
|
|
return "Machine learning"
|
|
for key, label in SERVICE_KINDS:
|
|
if key in image:
|
|
return label
|
|
return translate("Service")
|
|
|
|
|
|
def _recommended_storage(volume: dict[str, Any]) -> str:
|
|
"""What the installation uses for this path, and the alternative when the
|
|
user can choose; the recommended one comes first."""
|
|
choices = set(volume.get("installation_choice", []))
|
|
default = volume.get("default")
|
|
if {"managed-volume", "host-bind"} <= choices:
|
|
both = f"{translate('Container volume')} / {translate('Host directory')}"
|
|
return f"{translate('Optional')}: {both}" if default == "skip" else both
|
|
if default == "skip":
|
|
return translate("Optional, not mounted by default")
|
|
return translate("Host system path") if default == "host-bind" else translate("Container volume")
|
|
|
|
|
|
def _app_detail_text(catalog: Catalog, item: dict[str, Any], template: dict[str, Any]) -> str:
|
|
ui = template["catalog_ui"]
|
|
contract = template["container_contract"]
|
|
profile = template.get("proxmox", {}).get("installer_profile", {})
|
|
lines = [rf"\Zb{source_text(ui.get('title')) or item['id']}\Zn", ""]
|
|
# The one-line tagline, not the full upstream description: it is what the
|
|
# reader needs to know what this is, it survives translation without
|
|
# drifting, and it goes through translate() like every other string.
|
|
description = translate(source_text(ui.get("tagline")) or source_text(ui.get("description")))
|
|
if description:
|
|
wrapped = textwrap.wrap(description, DETAIL_WIDTH)
|
|
if len(wrapped) > 8:
|
|
wrapped = wrapped[:8]
|
|
wrapped[-1] = wrapped[-1].rstrip(" .,;") + "…"
|
|
lines += wrapped + [""]
|
|
|
|
def row(label: str, value: str) -> None:
|
|
lines.append(f"{label + ':':<17} {value}")
|
|
|
|
row(translate("Source"), publisher(item))
|
|
# Only a real test is shown; OCI manager Apps is labelled beta as a whole.
|
|
if is_tested(item):
|
|
row(translate("Status"), translate("Verified"))
|
|
row(translate("Architectures"), _display_architectures(ui))
|
|
endpoints = template.get("first_run", {}).get("endpoints", [])
|
|
if endpoints:
|
|
row(translate("Web access"), ", ".join(
|
|
f"{e.get('scheme', 'http')}://<IP>:{e.get('port')}{e.get('path') or '/'}" for e in endpoints))
|
|
|
|
if profile.get("stack_driver") == "arr-suite":
|
|
row(translate("Type"), translate("Application suite: one independent LXC per selected application"))
|
|
defaults = set(profile.get("default_applications") or ("prowlarr", "sonarr", "radarr", "qbittorrent"))
|
|
lines += ["", translate("Applications you can choose:")]
|
|
for app_id in profile.get("applications") or []:
|
|
try:
|
|
child = catalog.compose(app_id)
|
|
image = child["container_contract"]["image"]["reference"]
|
|
name = source_text(child["catalog_ui"]["title"]) or app_id
|
|
except (ConversionError, OSError, ValueError, KeyError):
|
|
image, name = "", app_id
|
|
mark = f" ({translate('selected by default')})" if app_id in defaults else ""
|
|
lines.append(f" {name + mark:<34} {_image_label(image)}")
|
|
lines.append(f" {translate('Media server') + ':':<34} Jellyfin, Plex, Emby {translate('or none')}")
|
|
elif item.get("multi_container"):
|
|
services = template.get("compose_stack", {}).get("services", [])
|
|
row(translate("Type"), translate("Multi-container application (experimental)"))
|
|
lines += ["", translate("Containers that will be created (one LXC per service, on a private network):")]
|
|
for service in services:
|
|
lines.append(f" {_service_kind(service):<20} {_image_label(service.get('image'))}")
|
|
else:
|
|
row(translate("Image"), _image_label(contract["image"]["reference"]))
|
|
volumes = contract.get("volumes", [])
|
|
if volumes:
|
|
width = max(28, *(len(volume["container_path"]) + 2 for volume in volumes))
|
|
lines += ["", f"{translate('Persistent data:'):<{width + 2}} {translate('Recommended')}"]
|
|
for volume in volumes:
|
|
lines.append(f" {volume['container_path']:<{width}} {_recommended_storage(volume)}")
|
|
if any({"managed-volume", "host-bind"} <= set(volume.get("installation_choice", []))
|
|
for volume in volumes):
|
|
lines.append(translate("The installation asks which one to use for these paths."))
|
|
|
|
hardware = profile.get("hardware_acceleration", {}).get("profiles", [])
|
|
if hardware:
|
|
lines += ["", translate("Hardware acceleration options:")]
|
|
lines += [f" {' '.join(translate(p.get('label', p['id'])).split())}" for p in hardware]
|
|
security = template.get("proxmox", {}).get("security_profile", {})
|
|
if security.get("requires_privileged_lxc"):
|
|
lines += ["", f"{translate('Security') + ':':<17} {translate('needs a privileged LXC')}"]
|
|
elif security.get("requires_relaxed_confinement"):
|
|
lines += ["", f"{translate('Security') + ':':<17} {translate('needs a relaxed AppArmor or seccomp profile')}"]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _deployment_summary_text(template: dict[str, Any], deployment: dict[str, Any]) -> str:
|
|
plan = redacted(deployment)
|
|
title = source_text(template["catalog_ui"]["title"]) or template["id"]
|
|
lines = [title, ""]
|
|
|
|
def row(label: str, value: Any) -> None:
|
|
lines.append(f"{label + ':':<16} {value}")
|
|
|
|
on = translate("on")
|
|
if plan.get("suite_arr"):
|
|
lines.append(translate("Independent applications, without a main container."))
|
|
else:
|
|
row(translate("Image"), template["container_contract"]["image"]["reference"])
|
|
|
|
if plan.get("deployment_kind"):
|
|
base_vmid = plan.get("base_vmid")
|
|
row(translate("Stack"), plan.get("stack_name", title))
|
|
row(translate("Base VMID"), base_vmid if base_vmid is not None else translate("next free block"))
|
|
services = template.get("compose_stack", {}).get("services", [])
|
|
if plan.get("deployment_kind") == "generic-multi-lxc-stack":
|
|
services = [dict(s, vmid_offset=s["offset"]) for s in plan["services"]]
|
|
for service in sorted(services, key=lambda item: item.get("vmid_offset", 0)):
|
|
offset = service.get("vmid_offset", 0)
|
|
vmid = base_vmid + offset if base_vmid is not None else f"base+{offset}"
|
|
lines.append(f" - {service['name']}: CT {vmid}")
|
|
lines.append("")
|
|
row("Rootfs", plan.get("rootfs_storage", "-"))
|
|
row(translate("Image cache"), plan.get("template_storage", "-"))
|
|
if plan.get("database_storage"):
|
|
row("PostgreSQL", f"{plan.get('database_size_gb', '-')} GB {on} {plan['database_storage']}")
|
|
for service in plan.get("services", []):
|
|
child = service["deployment"]
|
|
lines.append(f" {service['name']}: {child['resources']['cores']} CPU, "
|
|
f"{child['resources']['memory_mb']} MB RAM")
|
|
for mount in child["mounts"]:
|
|
target = (f"{mount['size_gb']} GB {on} {mount['source']}"
|
|
if mount["type"] == "managed-volume" else mount["source"])
|
|
lines.append(f" {mount['container_path']} → {target}")
|
|
for key, label in (("media", "Library"), ("application", "Application data"), ("transfer", "Consume/export")):
|
|
storage = plan.get(key)
|
|
if isinstance(storage, dict) and "mode" in storage:
|
|
if storage["mode"] == "host-bind":
|
|
row(translate(label), f"{translate('host directory')} {storage.get('host_path', '-')}")
|
|
else:
|
|
row(translate(label), f"{storage.get('size_gb', '-')} GB {on} {storage.get('storage', '-')}")
|
|
if isinstance(plan.get("resources"), dict):
|
|
row(translate("Resources"), f"{plan['resources']['cores']} CPU, {plan['resources']['memory_mb']} MB RAM")
|
|
if plan.get("extra_mounts"):
|
|
lines.append(f"{translate('Extra paths') + ':':<16}")
|
|
for mount in plan["extra_mounts"]:
|
|
target = (f"{translate('host directory')} {mount['source']}" if mount["type"] == "host-bind"
|
|
else f"{translate('Container volume')} {mount.get('size_gb', '-')} GB {on} {mount['source']}")
|
|
if mount.get("read_only"):
|
|
target += f" ({translate('read-only')})"
|
|
lines.append(f" {mount['container_path']} → {target}")
|
|
if plan.get("extra_devices"):
|
|
row(translate("Devices"), ", ".join(device["host_path"] for device in plan["extra_devices"]))
|
|
else:
|
|
row(translate("Container"), f"CT {plan.get('vmid') or translate('next free')} · {plan.get('hostname', '-')}")
|
|
|
|
resources = plan.get("resources", {})
|
|
if resources:
|
|
row(translate("Resources"), f"{resources.get('cores', '-')} CPU · {resources.get('memory_mb', '-')} MB RAM · "
|
|
f"{resources.get('swap_mb', '-')} MB swap")
|
|
rootfs = plan.get("rootfs")
|
|
if rootfs:
|
|
row(translate("Storage"), f"rootfs {rootfs['size_gb']} GB {on} {rootfs['storage']} · "
|
|
f"{translate('image cache on')} {plan.get('template_storage', '-')}")
|
|
mounts = plan.get("mounts", [])
|
|
if mounts:
|
|
lines.append(f"{translate('Data') + ':':<16}")
|
|
for mount in mounts:
|
|
if mount["type"] == "host-bind":
|
|
target = f"{translate('host directory')} {mount['source']}"
|
|
else:
|
|
target = f"{translate('Container volume')} {mount.get('size_gb', '-')} GB {on} {mount['source']}"
|
|
if mount.get("backup"):
|
|
target += f" ({translate('in backups')})"
|
|
if mount.get("read_only"):
|
|
target += f" ({translate('read-only')})"
|
|
lines.append(f" {mount['container_path']} → {target}")
|
|
network = plan.get("network", {})
|
|
if plan.get("host_monitor"):
|
|
row(translate("Network"), translate("IP address and firewall of the host"))
|
|
firewall = plan.get("host_firewall")
|
|
if firewall:
|
|
row(translate("Host firewall"),
|
|
translate("allow TCP {port} from {subnet} via {bridge}").format(
|
|
port=firewall["port"], subnet=firewall["source"], bridge=firewall["bridge"]
|
|
))
|
|
else:
|
|
row(translate("Host firewall"), translate("not changed"))
|
|
elif "frontend_bridge" in network:
|
|
addresses = [network[key] for key in ("frontend_ipv4", "machine_learning_frontend_ipv4") if network.get(key)]
|
|
addresses += [service["frontend_ipv4"] for service in plan.get("services", []) if service.get("frontend_ipv4")]
|
|
static = [address for address in addresses if address != "dhcp"]
|
|
row(translate("Network"), f"{network['frontend_bridge']} · {', '.join(static) if static else 'DHCP'}"
|
|
+ (f" · gw {network['frontend_gateway']}" if network.get("frontend_gateway") else "")
|
|
+ f" · {translate('private network assigned automatically')}")
|
|
elif network:
|
|
ipv4 = network.get("ipv4", "dhcp")
|
|
row(translate("Network"), f"{network.get('bridge', '-')} · {'DHCP' if ipv4 == 'dhcp' else ipv4}"
|
|
+ (f" · gw {network['gateway']}" if network.get("gateway") else ""))
|
|
devices = plan.get("devices", [])
|
|
if plan.get("hardware_profile") or devices:
|
|
profiles = (template.get("proxmox", {}).get("installer_profile", {})
|
|
.get("hardware_acceleration", {}).get("profiles", []))
|
|
selected = next((p for p in profiles if p["id"] == plan.get("hardware_profile")), None)
|
|
profile_label = (translate(selected["label"]).split(" ")[0] if selected
|
|
else plan.get("hardware_profile") or translate("custom"))
|
|
row(translate("Acceleration"),
|
|
", ".join([profile_label, *[d.get("host_path") or d.get("kind", "-") for d in devices]]))
|
|
security = plan.get("security")
|
|
if security:
|
|
row(translate("Security"),
|
|
translate("unprivileged LXC") if security.get("unprivileged") else translate("privileged LXC"))
|
|
environment = plan.get("environment", [])
|
|
if environment:
|
|
row(translate("Variables"), ", ".join(f"{item['name']}={item['value']}" for item in environment))
|
|
if plan.get("suite_arr"):
|
|
lines += ["", *[translate(note) for note in plan.get("completion_notes", [])]]
|
|
lines.append("")
|
|
row(translate("Start"), f"{translate('when finished')}: {_yes_no(plan.get('start_after_create'))} · "
|
|
f"{translate('with Proxmox')}: {_yes_no(plan.get('onboot'))}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _print_installation_summary(result: dict[str, Any], images_removed: str | None = None) -> None:
|
|
console.msg_title(translate("Installation completed"))
|
|
if result.get("suite_arr"):
|
|
console.msg_ok(translate("Independent LXC applications installed"))
|
|
else:
|
|
console.msg_ok(f"CT {result['vmid']} · {translate('IP address')}: "
|
|
f"{result.get('ip') or translate('not available yet')}")
|
|
for name, vmid in (result.get("stack_vmids") or {}).items():
|
|
console.msg_ok(f"{translate(STACK_LABELS.get(name, name))}: CT {vmid}")
|
|
for item in result.get("urls") or []:
|
|
console.msg_ok(f"{translate(item['label'])}: {item['url']}")
|
|
for item in result.get("credentials") or []:
|
|
console.msg_info2(translate(item["label"]))
|
|
username = str(item["username"])
|
|
console.msg_ok(f"{translate('User')}: {translate(username) if ' ' in username else username}")
|
|
if item.get("password") is not None:
|
|
console.msg_ok(f"{translate('Password')}: {item['password'] or translate('(empty)')}")
|
|
else:
|
|
console.msg_warn(translate("The password could not be retrieved automatically"))
|
|
if item.get("change_required"):
|
|
console.msg_warn(translate("Change it after the first login."))
|
|
if images_removed:
|
|
console.msg_ok(images_removed)
|
|
for note in result.get("completion_notes") or []:
|
|
console.msg_note(translate(note))
|
|
if result.get("log"):
|
|
console.msg_note(f"{translate('Installation log:')} {result['log']}")
|
|
print()
|
|
console.msg_note(translate("OCI manager Apps is a beta: if something does not work as expected, "
|
|
"please report it on GitHub with the application name."))
|
|
|
|
|
|
# ---------------------------------------------------------------- menus
|
|
|
|
def _install(catalog: Catalog, ui, item: dict[str, Any], mode: str) -> None:
|
|
install_template(ui, catalog.compose(item["id"]), item["id"], mode)
|
|
|
|
|
|
def install_template(ui, template: dict[str, Any], identifier: str, mode: str) -> dict[str, Any] | None:
|
|
"""Configures and installs one template, from the catalog or written from a
|
|
definition the user gave."""
|
|
from .ui import BacktrackUI, RestartWizard
|
|
wizard = BacktrackUI(ui)
|
|
try:
|
|
while True:
|
|
candidate = copy.deepcopy(template)
|
|
try:
|
|
deployment = build_deployment(candidate, wizard, mode)
|
|
approved = wizard.review(_deployment_summary_text(candidate, deployment),
|
|
translate("Installation summary"),
|
|
question=translate("Install with this configuration?"))
|
|
break
|
|
except RestartWizard:
|
|
wizard.restart()
|
|
except (ValueError, InstallError) as error:
|
|
# A mistyped value asks that question again instead of
|
|
# sending the user back to the start of the wizard.
|
|
if not wizard.retry_last(error):
|
|
raise
|
|
finally:
|
|
wizard.close()
|
|
if not approved:
|
|
return None
|
|
template = candidate
|
|
console.show_logo()
|
|
console.msg_title(f"{source_text(template['catalog_ui']['title']) or identifier} · {APP_TITLE}")
|
|
try:
|
|
result = run_remote_install(PROJECT_ROOT, template, deployment, "auto")
|
|
except InstallError as exc:
|
|
console.msg_error(str(exc))
|
|
console.wait_for_enter(translate("Press Enter to return to the menu..."))
|
|
return None
|
|
if result:
|
|
from .management import carry_records
|
|
carry_records(PROJECT_ROOT)
|
|
vmids = {int(v) for v in [result.get("vmid"), *(result.get("stack_vmids") or {}).values()] if v}
|
|
_, removed = images.offer_removal(ui, sorted(vmids))
|
|
_print_installation_summary(result, removed)
|
|
console.wait_for_enter(translate("Press Enter to return to the menu..."))
|
|
return result
|
|
|
|
|
|
def _rclone_mount(catalog: Catalog, ui) -> None:
|
|
template = catalog.compose("rclone")
|
|
deployment = build_rclone_mount_deployment(template, ui)
|
|
console.show_logo()
|
|
console.msg_title(translate("Rclone mount"))
|
|
result = run_remote_rclone_mount(PROJECT_ROOT, template, deployment, "auto")
|
|
if result:
|
|
from .management import carry_records
|
|
carry_records(PROJECT_ROOT)
|
|
console.msg_ok(f"Remote: {result['remote']}:")
|
|
console.msg_ok(f"{translate('Read/write')}: {result['read_write_path']}")
|
|
console.msg_ok(f"{translate('Read-only')}: {result['read_only_path']}")
|
|
console.wait_for_enter(translate("Press Enter to return to the menu..."))
|
|
|
|
|
|
def _app_detail(catalog: Catalog, ui, item: dict[str, Any]) -> None:
|
|
template = catalog.compose(item["id"])
|
|
actions = []
|
|
actions += [("default", translate("Install with default settings")),
|
|
("advanced", translate("Install with advanced settings"))]
|
|
if item["id"] == "rclone":
|
|
actions.append(("mount", translate("Enable a mount on an existing Rclone OCI container")))
|
|
numbered = {str(number): action for number, (action, _) in enumerate(actions, 1)}
|
|
options = [(str(number), label) for number, (_, label) in enumerate(actions, 1)]
|
|
title = source_text(template["catalog_ui"]["title"]) or item["id"]
|
|
selection = ui.detail_menu(_app_detail_text(catalog, item, template), options, "1", title=title)
|
|
action = numbered.get(selection or "")
|
|
if action is None:
|
|
return
|
|
if action == "mount":
|
|
_rclone_mount(catalog, ui)
|
|
return
|
|
_install(catalog, ui, item, DEFAULT_MODE if action == "default" else ADVANCED_MODE)
|
|
|
|
|
|
def _app_list(catalog: Catalog, ui, applications: list[dict[str, Any]], title: str,
|
|
keep_order: bool = False) -> None:
|
|
options, index, header = _app_menu(applications, keep_order)
|
|
selection = None
|
|
while True:
|
|
selection = ui.choose(header, options, selection, title=title, size=MENU_SIZE, colors=True)
|
|
if selection is None:
|
|
return
|
|
if selection in index:
|
|
_app_detail(catalog, ui, index[selection])
|
|
|
|
|
|
def _search(catalog: Catalog, ui, applications: list[dict[str, Any]]) -> None:
|
|
query = ui.ask(translate("Name or part of the description of the application"), required=False).strip()
|
|
if not query:
|
|
return
|
|
folded = query.casefold()
|
|
|
|
def rank(item: dict[str, Any]) -> int | None:
|
|
names = (item["id"].casefold(), str(item.get("title") or "").casefold())
|
|
if folded in names:
|
|
return 0
|
|
if any(name.startswith(folded) for name in names):
|
|
return 1
|
|
if any(folded in name for name in names):
|
|
return 2
|
|
return 3 if folded in str(item.get("description") or "").casefold() else None
|
|
|
|
ranked = sorted(((rank(item), str(item.get("title") or item["id"]).casefold(), item) for item in applications
|
|
if rank(item) is not None), key=lambda entry: entry[:2])
|
|
matches = [item for _, _, item in ranked]
|
|
if not matches:
|
|
ui.message(f"{translate('No applications match')}: '{query}'")
|
|
return
|
|
_app_list(catalog, ui, matches, f"{translate('Search results for:')} '{query}' ({len(matches)})",
|
|
keep_order=True)
|
|
|
|
|
|
def interactive(catalog: Catalog) -> int:
|
|
ui = interactive_ui()
|
|
if not catalog.index_path.exists():
|
|
ui.message(translate("The OCI catalog is not installed. Update ProxMenux and try again."))
|
|
return 1
|
|
applications = _installable(catalog.load_index()["applications"])
|
|
categories: dict[str, list[dict[str, Any]]] = {}
|
|
labels: dict[str, str] = {}
|
|
for item in applications:
|
|
key = item.get("category") or "misc"
|
|
categories.setdefault(key, []).append(item)
|
|
labels[key] = item.get("category_label") or key
|
|
order = sorted(categories, key=lambda key: (key == "misc", translate(labels[key]).casefold()))
|
|
category_by_index = {str(number): key for number, key in enumerate(order, 1)}
|
|
options = [
|
|
("search", translate("Search applications")),
|
|
("all", f"{translate('All applications'):<35} ({len(applications):>3})"),
|
|
("manage", translate("Manage installed OCI applications")),
|
|
("custom", translate("Install an image that is not in the catalog")),
|
|
("", ""),
|
|
] + [(number, f"{translate(labels[key]):<35} ({len(categories[key]):>3})")
|
|
for number, key in category_by_index.items()]
|
|
selection = "search"
|
|
while True:
|
|
selection = ui.choose(translate("Select a category or search for applications:"), options, selection,
|
|
size=MENU_SIZE)
|
|
if selection is None:
|
|
return 0
|
|
try:
|
|
if selection == "search":
|
|
_search(catalog, ui, applications)
|
|
elif selection == "all":
|
|
_app_list(catalog, ui, applications,
|
|
f"{translate('All applications')} ({len(applications)})")
|
|
elif selection == "manage":
|
|
from .management import interactive_management
|
|
interactive_management(PROJECT_ROOT, ui)
|
|
elif selection == "custom":
|
|
from .custom import explore
|
|
explore(ui)
|
|
elif selection in category_by_index:
|
|
key = category_by_index[selection]
|
|
_app_list(catalog, ui, categories[key], translate(labels[key]))
|
|
except UserCancelled:
|
|
continue
|
|
except (ConversionError, InstallError, OSError, ValueError) as exc:
|
|
console.stop_spinner()
|
|
ui.message(f"{translate('The operation could not be completed')}:\n\n{exc}")
|
|
|
|
|
|
# ---------------------------------------------------------------- maintenance CLI
|
|
|
|
def _template_summary_text(template: dict[str, Any]) -> str:
|
|
contract = template["container_contract"]
|
|
lines = [
|
|
source_text(template["catalog_ui"]["title"]),
|
|
f"Image: {contract['image']['reference']}",
|
|
f"Status: {template['status']}",
|
|
"Ports: " + (", ".join(f"{p['container_port']}/{p['protocol']}" for p in contract["ports"]) or "none"),
|
|
"Volumes: " + (", ".join(v["container_path"] for v in contract["volumes"]) or "none"),
|
|
]
|
|
blockers = template["compatibility"]["untranslated_blockers"]
|
|
if blockers:
|
|
lines.append(f"Blockers: {', '.join(blockers)}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(prog="oci_manager_apps.sh",
|
|
description="ProxMenux OCI manager Apps (beta)")
|
|
subparsers = parser.add_subparsers(dest="command")
|
|
subparsers.add_parser("sync", help="Refresh the index from the image sources")
|
|
list_parser = subparsers.add_parser("list", help="List applications")
|
|
list_parser.add_argument("--filter", default="")
|
|
generate_parser = subparsers.add_parser("generate", help="Regenerate the template of one application")
|
|
generate_parser.add_argument("app")
|
|
generate_all_parser = subparsers.add_parser("generate-all", help="Regenerate the catalog templates")
|
|
generate_all_parser.add_argument("--provider", choices=["all", "linuxserver.io", "imported", "curated"],
|
|
default="all")
|
|
subparsers.add_parser("apply-icons", help="Resolve catalog icons against the jsdelivr icon sets")
|
|
show_parser = subparsers.add_parser("show", help="Show the summary of a template")
|
|
show_parser.add_argument("app")
|
|
install_parser = subparsers.add_parser("install", help="Configure and install an application")
|
|
install_parser.add_argument("app")
|
|
install_parser.add_argument("--host", default="auto", help="'auto'/'local', or root@IP for development")
|
|
install_parser.add_argument("--advanced", action="store_true")
|
|
install_parser.add_argument("--dry-run", action="store_true")
|
|
rclone_parser = subparsers.add_parser("rclone-mount", help="Enable a mount on an installed Rclone OCI")
|
|
rclone_parser.add_argument("--host", default="auto")
|
|
rclone_parser.add_argument("--dry-run", action="store_true")
|
|
subparsers.add_parser("recover", help="Register again the OCI applications restored from a backup")
|
|
manage_parser = subparsers.add_parser("manage", help="Update or recreate one installed OCI instance")
|
|
manage_parser.add_argument("vmid", type=int)
|
|
manage_parser.add_argument("--action", choices=("update", "recreate"), required=True)
|
|
manage_parser.add_argument("--keep-backup", metavar="STORAGE",
|
|
help="Keep the backup taken before the update in this Proxmox storage")
|
|
manage_parser.add_argument("--unattended", action="store_true",
|
|
help="Scheduled run: no questions; stops where a person has to decide")
|
|
manage_parser.add_argument("--acknowledge-external-data", action="store_true",
|
|
help="Host directories are not reverted by the backup (confirmed beforehand)")
|
|
manage_parser.add_argument("--min-image-age-days", type=int, default=0)
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
catalog = Catalog(PROJECT_ROOT)
|
|
try:
|
|
if not args.command:
|
|
return interactive(catalog)
|
|
if args.command == "sync":
|
|
payload = catalog.sync_index()
|
|
print(f"Index updated: {len(payload['applications'])} candidate applications")
|
|
return 0
|
|
if args.command == "list":
|
|
query = args.filter.casefold()
|
|
for item in catalog.load_index()["applications"]:
|
|
if not item.get("hidden", False) and query in item["id"].casefold():
|
|
candidate = item.get("automatic_install_candidate")
|
|
status = ("installable" if candidate else "pending adaptation" if candidate is False
|
|
else item.get("template_status") or "not generated")
|
|
print(f"{item['id']:<30} {_display_architectures(item):<12} {publisher(item):<14} {status}")
|
|
return 0
|
|
if args.command == "generate":
|
|
path, template = catalog.generate(args.app)
|
|
print(f"{_template_summary_text(template)}\n\n{path}")
|
|
return 0
|
|
if args.command == "generate-all":
|
|
def progress(current: int, total: int, app_id: str, outcome: str) -> None:
|
|
marker = "OK" if outcome == "ok" else "ERROR"
|
|
print(f"\r[{current:3}/{total}] {marker:<5} {app_id:<32}", end="", flush=True)
|
|
|
|
report = catalog.generate_all(progress=progress, provider=args.provider)
|
|
print(f"\nGenerated: {report['generated_count']}; failed: {report['failed_count']}; "
|
|
f"family: {args.provider}")
|
|
return 0 if not report["failed"] else 2
|
|
if args.command == "apply-icons":
|
|
report = catalog.apply_icons()
|
|
print(f"Icons resolved: {report['resolved']}; cleared: {report['cleared']}; "
|
|
f"unchanged: {report['unchanged']}; with a theme variant: {report['themed']}")
|
|
return 0
|
|
if args.command == "show":
|
|
print(_template_summary_text(catalog.compose(args.app)))
|
|
return 0
|
|
if args.command == "install":
|
|
template = catalog.compose(args.app)
|
|
if not template["compatibility"]["automatic_install_candidate"]:
|
|
raise InstallError("Installation blocked: " + ", ".join(template["compatibility"]["untranslated_blockers"]))
|
|
deployment = build_deployment(template, None, ADVANCED_MODE if args.advanced else DEFAULT_MODE)
|
|
print(_deployment_summary_text(template, deployment))
|
|
if not args.dry_run and input("\nInstall? [y/N]: ").strip().casefold() not in {"y", "yes"}:
|
|
print("Installation cancelled.")
|
|
return 0
|
|
result = run_remote_install(PROJECT_ROOT, template, deployment, args.host, args.dry_run)
|
|
if result:
|
|
if not args.dry_run:
|
|
from .management import carry_records
|
|
carry_records(PROJECT_ROOT)
|
|
_print_installation_summary(result)
|
|
return 0
|
|
if args.command == "recover":
|
|
from .management import direct_recovery
|
|
return direct_recovery(PROJECT_ROOT)
|
|
if args.command == "manage":
|
|
from .management import direct_management
|
|
lifecycle_args = []
|
|
if args.keep_backup:
|
|
if not re.fullmatch(r"[A-Za-z0-9._-]{1,64}", args.keep_backup):
|
|
raise InstallError(translate("Invalid storage name"))
|
|
lifecycle_args += ["--keep-backup", args.keep_backup]
|
|
if args.acknowledge_external_data:
|
|
lifecycle_args.append("--acknowledge-external-data")
|
|
return direct_management(PROJECT_ROOT, args.vmid, args.action, lifecycle_args,
|
|
args.unattended, max(0, args.min_image_age_days))
|
|
if args.command == "rclone-mount":
|
|
template = catalog.compose("rclone")
|
|
deployment = build_rclone_mount_deployment(template)
|
|
print(json.dumps(deployment, ensure_ascii=False, indent=2))
|
|
result = run_remote_rclone_mount(PROJECT_ROOT, template, deployment, args.host, args.dry_run)
|
|
if result:
|
|
print(f"Read/write: {result['read_write_path']}\nRead-only: {result['read_only_path']}")
|
|
return 0
|
|
return 1
|
|
except (KeyboardInterrupt, UserCancelled):
|
|
console.stop_spinner()
|
|
print(f"\n{translate('Operation cancelled.')}")
|
|
return 130
|
|
except (ConversionError, SourceError, InstallError, OSError, ValueError) as exc:
|
|
console.stop_spinner()
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|