python如何可视化管理nginx的网站支持网站新增修改暂停配置等可视化修改？

默认适配 Ubuntu/Debian（sites-available/sites-enabled），同时兼容 CentOS/RHEL（conf.d，文件后缀 .disabled 控制启停）。

功能：新增/修改站点、启用/暂停、nginx -t 测试、reload、查看健康状态

技术：Flask + Jinja2 模板；元数据 JSON 记录站点表单字段，便于编辑回显

安全：支持管理口令（ADMIN_TOKEN），默认仅监听 127.0.0.1，建议内网或 SSH 隧道访问

权限：需 root 运行（或 sudo 精准授权）

一、快速开始

 1) 安装依赖

Ubuntu/Debian

sudo apt update && sudo apt install -y python3-venv nginx

CentOS/RHEL

sudo dnf install -y python3 nginx

2) 放置项目（示例目录 /opt/nginx-admin）

mkdir -p /opt/nginx-admin && cd /opt/nginx-admin

把下面三个文件按路径保存：

app.py

templates/site.conf.j2

static/index.html

3) 创建虚拟环境并安装

python3 -m venv .venv

. .venv/bin/activate

pip install flask jinja2

4) 运行

可选安全口令：export ADMIN_TOKEN='请自己设置一串复杂口令'

启动：sudo .venv/bin/python app.py

访问：http://127.0.0.1:8080

若设置了 ADMIN_TOKEN，首次访问可在 URL 带 ?token=你的口令，或在页面顶部输入口令

二、后端 app.py说明

默认路径：Ubuntu/Debian 使用 /etc/nginx/sites-available 与 /etc/nginx/sites-enabled；CentOS/RHEL 使用 /etc/nginx/conf.d（本代码自动兼容）

想明确指定路径，可用环境变量覆写：

SITES_AVAILABLE, SITES_ENABLED

口令：环境变量 ADMIN_TOKEN；如果不设置，则不启用鉴权（仅限内网/隧道环境）

保存为 app.py

import os, shutil, subprocess, tempfile, json
from pathlib import Path
from functools import wraps
from flask import Flask, request, jsonify, send_from_directory
from jinja2 import Environment, FileSystemLoader

app = Flask(__name__, static_folder="static")

# 可用环境变量覆写
NGINX_BIN = shutil.which("nginx") or "/usr/sbin/nginx"
SYSTEMCTL = shutil.which("systemctl") or "/usr/bin/systemctl"
SITES_AVAILABLE = Path(os.getenv("SITES_AVAILABLE", "/etc/nginx/sites-available"))
SITES_ENABLED = Path(os.getenv("SITES_ENABLED", "/etc/nginx/sites-enabled"))
COMBINED_DIR = SITES_AVAILABLE.resolve() == SITES_ENABLED.resolve()  # CentOS/RHEL 常为 True
CONF_EXT = ".conf"
DISABLED_EXT = ".disabled"
META_DIR = SITES_AVAILABLE / ".meta"
TEMPLATE_DIR = Path("templates")
env = Environment(loader=FileSystemLoader(str(TEMPLATE_DIR)))
tpl = env.get_template("site.conf.j2")

ADMIN_TOKEN = os.getenv("ADMIN_TOKEN")  # 建议设置

def require_auth(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        if not ADMIN_TOKEN:
            return fn(*args, **kwargs)
        token = (
            (request.headers.get("Authorization") or "").replace("Bearer ", "", 1)
            or request.headers.get("X-Admin-Token")
            or request.args.get("token")
        )
        if token != ADMIN_TOKEN:
            return jsonify({"ok": False, "msg": "Unauthorized"}), 401
        return fn(*args, **kwargs)
    return wrapper

def run(cmd):
    p = subprocess.run(cmd, text=True, capture_output=True)
    return p.returncode == 0, (p.stdout + "\n" + p.stderr).strip()

def atomic_write(path: Path, content: str):
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile("w", delete=False, dir=str(path.parent)) as tf:
        tf.write(content)
        tmp = tf.name
    os.replace(tmp, path)

def nginx_test():
    return run([NGINX_BIN, "-t"])

def nginx_reload():
    ok, out = run([SYSTEMCTL, "reload", "nginx"])
    if not ok:
        ok2, out2 = run([NGINX_BIN, "-s", "reload"])
        out = out + "\n" + out2
        return ok2, out
    return ok, out

def conf_available_path(name: str) -> Path:
    return SITES_AVAILABLE / f"{name}{CONF_EXT}"

def conf_enabled_symlink(name: str) -> Path:
    return SITES_ENABLED / f"{name}{CONF_EXT}"

def conf_disabled_path(name: str) -> Path:
    return SITES_AVAILABLE / f"{name}{CONF_EXT}{DISABLED_EXT}"

def meta_path(name: str) -> Path:
    return META_DIR / f"{name}.json"

def is_enabled(name: str) -> bool:
    if COMBINED_DIR:
        return conf_available_path(name).exists()
    else:
        return conf_enabled_symlink(name).exists()

def enable_site(name: str):
    if COMBINED_DIR:
        # CentOS/RHEL: 用 .disabled 后缀控制启停
        disabled = conf_disabled_path(name)
        active = conf_available_path(name)
        if disabled.exists():
            os.replace(disabled, active)
        # 如果两者都不存在，不在此创建文件；写入时创建
    else:
        src = conf_available_path(name)
        dst = conf_enabled_symlink(name)
        if not dst.exists():
            dst.symlink_to(src.resolve())

def disable_site(name: str):
    if COMBINED_DIR:
        active = conf_available_path(name)
        disabled = conf_disabled_path(name)
        if active.exists():
            os.replace(active, disabled)
    else:
        dst = conf_enabled_symlink(name)
        if dst.exists():
            dst.unlink()

def render_conf(payload: dict) -> str:
    allowed = {"server_name", "root", "proxy_pass", "enable_ssl", "ssl_cert", "ssl_key", "https_redirect"}
    data = {k: v for k, v in (payload or {}).items() if k in allowed}
    if not data.get("server_name"):
        raise ValueError("server_name is required")
    # root 与 proxy_pass 二选一，优先 proxy_pass
    if data.get("proxy_pass") and data.get("root"):
        data.pop("root", None)
    return tpl.render(**data)

def save_meta(name: str, payload: dict):
    META_DIR.mkdir(parents=True, exist_ok=True)
    atomic_write(meta_path(name), json.dumps(payload, ensure_ascii=False, indent=2))

def load_meta(name: str) -> dict:
    p = meta_path(name)
    if p.exists():
        return json.loads(p.read_text())
    return {}

def create_or_update_site(payload: dict, enable=True):
    server_name = payload.get("server_name")
    if not server_name:
        return False, "server_name is required"

    # 记录当前启用状态
    was_enabled = is_enabled(server_name)

    # 渲染配置 & 选择写入路径
    conf_text = render_conf(payload)
    if COMBINED_DIR:
        target_path = conf_available_path(server_name) if (enable or was_enabled) else conf_disabled_path(server_name)
    else:
        target_path = conf_available_path(server_name)

    # 备份
    backup_path = None
    if target_path.exists():
        backup_path = target_path.with_suffix(target_path.suffix + ".bak")
        shutil.copy2(target_path, backup_path)

    # 写入配置与元数据
    atomic_write(target_path, conf_text)
    save_meta(server_name, payload)

    # 根据状态启用/保持状态
    if enable or was_enabled:
        try:
            enable_site(server_name)
            ok, out = nginx_test()
            if not ok:
                # 回滚
                if backup_path and backup_path.exists():
                    shutil.copy2(backup_path, target_path)
                disable_site(server_name)
                return False, f"nginx -t failed:\n{out}"
            ok2, out2 = nginx_reload()
            if not ok2:
                return False, f"nginx reload failed:\n{out2}"
            return True, "applied and reloaded"
        except Exception as e:
            disable_site(server_name)
            if backup_path and backup_path.exists():
                shutil.copy2(backup_path, target_path)
            return False, str(e)
    else:
        return True, "saved (not enabled)"

def list_all_sites():
    items = {}
    if COMBINED_DIR:
        # 识别 .conf 和 .conf.disabled
        for p in SITES_AVAILABLE.glob(f"*{CONF_EXT}"):
            items[p.stem] = True
        for p in SITES_AVAILABLE.glob(f"*{CONF_EXT}{DISABLED_EXT}"):
            name = p.name.replace(CONF_EXT + DISABLED_EXT, "")
            items[name] = items.get(name, False) and True or False
    else:
        for p in SITES_AVAILABLE.glob(f"*{CONF_EXT}"):
            name = p.stem
            items[name] = conf_enabled_symlink(name).exists()
    return [{"server_name": k, "enabled": v} for k, v in sorted(items.items(), key=lambda x: x[0])]

# ---------- Routes ----------
@app.get("/")
def index():
    return send_from_directory(app.static_folder, "index.html")

@app.get("/health")
@require_auth
def health():
    ok, out = nginx_test()
    return jsonify({"nginx_ok": ok, "output": out})

@app.post("/nginx/reload")
@require_auth
def reload_api():
    ok, out = nginx_reload()
    return (jsonify({"ok": ok, "msg": out}), 200 if ok else 400)

@app.get("/sites")
@require_auth
def sites_list():
    return jsonify(list_all_sites())

@app.get("/sites/<server_name>")
@require_auth
def site_get(server_name):
    enabl...

点击查看剩余70%