Files
shell-gen/generators/__init__.py
T
Your Name bee37565e1 fix(bash_header): PKG_INSTALL string → function + export DEBIAN_FRONTEND
The previous definition assigned a multi-word string to PKG_INSTALL:
  PKG_INSTALL="DEBIAN_FRONTEND=noninteractive apt-get install -y"
Downstream generators expand it as '$PKG_INSTALL pkg…', which bash
parses as 'DEBIAN_FRONTEND=noninteractive apt-get install -y pkg…'
— the leading assignment is treated as a command name, producing
'DEBIAN_FRONTEND=noninteractive: command not found' (e.g. docker.sh
line 40). The DEBIAN_FRONTEND variable was also never set, so apt
could fall into interactive mode for tzdata / debconf prompts.

Fix: turn PKG_INSTALL into a shell function and export
DEBIAN_FRONTEND separately. All ~50 call sites keep the existing
$PKG_INSTALL pkg… syntax unchanged.

Verified: bash -n OK; mock Ubuntu 22.04 run exits 0 with no
'command not found'; mock apt-get receives the correct args.
All 44 generators still render successfully.
2026-08-07 13:14:34 +08:00

213 lines
6.1 KiB
Python

"""
Generator registry — every generator module imports @register and a Generator class.
The class is responsible for rendering a shell script given a params dict.
"""
from typing import Dict, List, Optional
import shlex
_REGISTRY: Dict[str, "Generator"] = {}
def register(gen: "Generator"):
if gen.id in _REGISTRY:
raise ValueError(f"Duplicate generator id: {gen.id}")
_REGISTRY[gen.id] = gen
return gen
def get_generator(gid: str) -> Optional["Generator"]:
return _REGISTRY.get(gid)
def list_generators() -> Dict[str, "Generator"]:
return dict(_REGISTRY)
def get_categories() -> List[str]:
seen = []
for g in _REGISTRY.values():
if g.category not in seen:
seen.append(g.category)
return seen
# Human-readable labels and FontAwesome-style emoji icons
category_labels = {
"middleware": "中间件服务",
"databases": "数据库",
"runtimes": "开发语言运行时",
"systemd": "Systemd 服务",
"system": "系统工具",
"network": "网络工具",
"monitoring": "监控告警",
"security": "安全加固",
}
category_icons = {
"middleware": "🧩",
"databases": "🗄️",
"runtimes": "⚙️",
"systemd": "🛠️",
"system": "🖥️",
"network": "🌐",
"monitoring": "📊",
"security": "🔒",
}
# ----- Field / Param descriptors for the UI -----
class Field:
"""One form field rendered in the web UI."""
def __init__(self, name, label, type="text", default="", placeholder="",
help="", options=None, required=False, min_=None, max_=None,
step=None, pattern=None, group=None):
self.name = name
self.label = label
self.type = type # text|number|select|checkbox|textarea|password|portlist
self.default = default
self.placeholder = placeholder
self.help = help
self.options = options or []
self.required = required
self.min = min_
self.max = max_
self.step = step
self.pattern = pattern
self.group = group # for grouping in UI
def to_dict(self):
return {
"name": self.name, "label": self.label, "type": self.type,
"default": self.default, "placeholder": self.placeholder,
"help": self.help, "options": self.options, "required": self.required,
"min": self.min, "max": self.max, "step": self.step, "pattern": self.pattern,
"group": self.group,
}
# ----- Generator base class -----
class Generator:
id: str = ""
title: str = ""
category: str = "middleware"
description: str = ""
icon: str = "📦"
tags: List[str] = []
fields: List[Field] = []
warnings: List[str] = []
post_steps: List[str] = []
verify_steps: List[str] = []
os_support: List[str] = ["ubuntu", "debian", "centos", "rhel"]
dangerous: bool = False
def render(self, params: dict) -> str:
raise NotImplementedError
def summary(self) -> dict:
return {
"id": self.id, "title": self.title, "category": self.category,
"description": self.description, "icon": self.icon, "tags": self.tags,
"os": self.os_support, "dangerous": self.dangerous,
"fields": [f.to_dict() for f in self.fields],
}
# ----------------------------- Helpers -----------------------------
BASH_HEADER = """#!/usr/bin/env bash
# Generated by shell-gen (https://github.com/yourname/shell-gen)
# Generator: {title}
# Generated at: {ts}
# OS: {os_hint}
set -euo pipefail
SCRIPT_DIR="$(cd -- "$(dirname -- "${{BASH_SOURCE[0]}}")" &>/dev/null && pwd)"
LOG_PREFIX="[{title}]"
log() {{ echo "${{LOG_PREFIX}} $*" ; }}
warn() {{ echo "${{LOG_PREFIX}} WARN: $*" >&2 ; }}
die() {{ echo "${{LOG_PREFIX}} ERROR: $*" >&2 ; exit 1 ; }}
[ "$(id -u)" -eq 0 ] || die "This script must be run as root. Re-run with sudo."
# ---- detect distro ----
. /etc/os-release 2>/dev/null || true
DISTRO_ID="${{ID:-unknown}}"
DISTRO_LIKE="${{ID_LIKE:-}}"
case "$DISTRO_ID" in
ubuntu|debian)
PKG="apt-get"
export DEBIAN_FRONTEND=noninteractive
PKG_INSTALL() {{ DEBIAN_FRONTEND=noninteractive apt-get install -y "$@"; }}
;;
centos|rhel|rocky|almalinux|ol)
PKG="yum"
PKG_INSTALL() {{ yum install -y "$@"; }}
;;
fedora)
PKG="dnf"
PKG_INSTALL() {{ dnf install -y "$@"; }}
;;
*)
if echo "$DISTRO_LIKE" | grep -qiE "rhel|centos|fedora"; then
PKG="yum"
PKG_INSTALL() {{ yum install -y "$@"; }}
elif echo "$DISTRO_LIKE" | grep -qi "debian"; then
PKG="apt-get"
export DEBIAN_FRONTEND=noninteractive
PKG_INSTALL() {{ DEBIAN_FRONTEND=noninteractive apt-get install -y "$@"; }}
else
die "Unsupported distribution: $DISTRO_ID"
fi
;;
esac
log "Detected distro: $DISTRO_ID (package manager: $PKG)"
"""
def bash_header(title: str) -> str:
return BASH_HEADER.format(
title=title,
ts=datetime_now_str(),
os_hint="Ubuntu / Debian / CentOS / RHEL / Rocky / Alma",
)
def datetime_now_str() -> str:
import datetime as _dt
return _dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def quote(value) -> str:
"""Bash-safe single-quote escape."""
if value is None:
return "''"
s = str(value)
if s == "":
return "''"
return "'" + s.replace("'", "'\\''") + "'"
def bool_str(value, default=False) -> bool:
if isinstance(value, bool):
return value
if value is None:
return default
return str(value).lower() in ("1", "true", "yes", "on", "y", "t")
def yes(value) -> str:
return "yes" if bool_str(value) else "no"
def port_list(value, default=None) -> str:
"""Normalize port list input: '80,443' or '80 443' -> '80 443'."""
if value is None or value == "":
value = default or ""
s = str(value).replace(",", " ").replace(";", " ")
parts = [p.strip() for p in s.split() if p.strip()]
return " ".join(parts)