当前位置:首页>python>Python窗口启动windows下nginx

Python窗口启动windows下nginx

  • 2026-09-08 02:49:04
Python窗口启动windows下nginx
import osimport socketimport reimport tkinter as tkfrom tkinter import ttk, messagebox, filedialog, scrolledtext# ===================== 自动获取当前脚本所在目录(和 nginx.exe 同目录) =====================BASE_DIR = os.path.dirname(os.path.abspath(__file__))NGINX_ROOT = BASE_DIR# ======================================================================================NGINX_CONF = os.path.join(NGINX_ROOT, "conf", "nginx.conf")VHOST_DIR = os.path.join(NGINX_ROOT, "conf", "vhost")if not os.path.exists(VHOST_DIR):    os.makedirs(VHOST_DIR)class NginxSiteManager(tk.Tk):    def __init__(self):        super().__init__()        self.title("Nginx 站点管理器 - 增强版")        self.geometry("768x400")        self.tab_control = ttk.Notebook(self)        self.tab_control.pack(fill=tk.BOTH, expand=1, padx=10, pady=10)        # === 顶部按钮栏:添加站点、启动、停止、重启、测试、保存、主配置 ===        self.frame_btn = tk.Frame(self)        self.frame_btn.pack(fill=tk.X, padx=10, pady=5)        tk.Button(self.frame_btn, text="➕ 添加站点", command=self.add_tab).pack(side=tk.LEFT, padx=4)        tk.Button(self.frame_btn, text="▶️ 启动 Nginx", command=self.start_nginx).pack(side=tk.LEFT, padx=4)        tk.Button(self.frame_btn, text="⏹️ 停止 Nginx", command=self.stop_nginx).pack(side=tk.LEFT, padx=4)        tk.Button(self.frame_btn, text="🔄 重启 Nginx", command=self.reload_nginx).pack(side=tk.LEFT, padx=4)        tk.Button(self.frame_btn, text="✅ 保存所有站点", command=self.save_all).pack(side=tk.LEFT, padx=4)        tk.Button(self.frame_btn, text="🔧 测试配置", command=self.test_nginx).pack(side=tk.LEFT, padx=4)        tk.Button(self.frame_btn, text="📄 编辑主配置文件", command=self.open_main_config).pack(side=tk.LEFT, padx=4)        self.tabs = []        # ===================== 【关键改动】启动时自动加载已有站点 =====================        self.load_existing_sites()        # 如果没有任何站点,自动创建一个        if len(self.tabs) == 0:            self.add_tab()    def load_existing_sites(self):        """启动时自动加载 vhost 目录下的所有站点配置"""        if not os.path.exists(VHOST_DIR):            return        for fname in os.listdir(VHOST_DIR):            if not fname.endswith(".conf"):                continue            fpath = os.path.join(VHOST_DIR, fname)            try:                with open(fpath, "r", encoding="utf-8") as f:                    content = f.read()                # 解析域名                server_name = re.search(r"server_name\s+([^\s;]+)", content)                domain = server_name.group(1) if server_name else fname.replace(".conf", "")                # 解析端口                listen = re.search(r"listen\s+(\d+)", content)                port = listen.group(1) if listen else "80"                # 解析根目录                root = re.search(r"root\s+([^\s;]+)", content)                root_path = root.group(1) if root else os.path.join(NGINX_ROOT, "html")                # 解析 index                index = re.search(r"index\s+([^\n;]+)", content)                index_str = index.group(1) if index else "index.html index.htm index.php"                # 解析 autoindex                autoindex = "autoindex on" in content                # 解析 SSL                is_ssl = "ssl_certificate" in content                ssl_crt = ""                ssl_key = ""                if is_ssl:                    crt_match = re.search(r"ssl_certificate\s+([^\s;]+)", content)                    key_match = re.search(r"ssl_certificate_key\s+([^\s;]+)", content)                    ssl_crt = crt_match.group(1) if crt_match else ""                    ssl_key = key_match.group(1) if key_match else ""                # 创建选项卡                tab = ttk.Frame(self.tab_control)                self.tab_control.add(tab, text=domain)                self.tabs.append(tab)                self.create_tab_ui(tab, len(self.tabs), domain, port, root_path, autoindex, index_str, is_ssl, ssl_crt, ssl_key)            except Exception as e:                print(f"加载 {fname} 失败:{e}")    def add_tab(self):        tab = ttk.Frame(self.tab_control)        site_num = len(self.tabs) + 1        self.tab_control.add(tab, text=f"站点 {site_num}")        self.tabs.append(tab)        self.create_tab_ui(tab, site_num)    def create_tab_ui(self, tab, idx, domain_val="", port_val="", root_val="", autoindex_val=False, index_val="", is_ssl_val=False, ssl_crt_val="", ssl_key_val=""):        # 域名        tk.Label(tab, text="站点域名:").grid(row=0, column=0, sticky="w", padx=10, pady=6)        domain = tk.Entry(tab, width=30)        domain.grid(row=0, column=1, padx=10, pady=6)        domain.insert(0, domain_val or f"site{idx}.local")        # 端口 + 检测        tk.Label(tab, text="监听端口:").grid(row=1, column=0, sticky="w", padx=10, pady=6)        port_frame = tk.Frame(tab)        port_frame.grid(row=1, column=1, sticky="w", padx=10, pady=6)        port = tk.Entry(port_frame, width=15)        port.pack(side=tk.LEFT)        port.insert(0, port_val or "80")        tk.Button(port_frame, text="检测端口", command=lambda: self.check_port(port.get())).pack(side=tk.LEFT, padx=5)        # 根目录 + 浏览        tk.Label(tab, text="网站根目录:").grid(row=2, column=0, sticky="w", padx=10, pady=6)        root_frame = tk.Frame(tab)        root_frame.grid(row=2, column=1, sticky="w", padx=10, pady=6)        root_path = tk.Entry(root_frame, width=50)        root_path.pack(side=tk.LEFT)        default_root = root_val or os.path.join(NGINX_ROOT, "html")        root_path.insert(0, default_root)        tk.Button(root_frame, text="浏览", command=lambda: self.select_root(root_path)).pack(side=tk.LEFT, padx=5)        # 目录浏览 + index 文件        autoindex = tk.BooleanVar(value=autoindex_val)        tk.Checkbutton(tab, text="开启目录浏览", variable=autoindex).grid(row=3, column=1, sticky="w", padx=10)        tk.Label(tab, text="首页文件(index):").grid(row=4, column=0, sticky="w", padx=10, pady=6)        index_entry = tk.Entry(tab, width=40)        index_entry.grid(row=4, column=1, sticky="w", padx=10, pady=6)        index_entry.insert(0, index_val or "index.html index.htm index.php")        # 独立配置        is_vhost = tk.BooleanVar(value=True)        ck_vhost = tk.Checkbutton(tab, text="生成独立配置文件", variable=is_vhost)        ck_vhost.grid(row=5, column=1, sticky="w", padx=10, pady=4)        # 编辑独立配置按钮        edit_vhost_btn = tk.Button(tab, text="✏️ 编辑此站点配置", command=lambda t=tab: self.open_vhost_config(t))        edit_vhost_btn.grid(row=5, column=2, padx=10)        # HTTPS        is_ssl = tk.BooleanVar(value=is_ssl_val)        ck_ssl = tk.Checkbutton(tab, text="启用 HTTPS", variable=is_ssl,                                command=lambda: self.toggle_ssl(ssl_frame, is_ssl))        ck_ssl.grid(row=6, column=1, sticky="w", padx=10, pady=4)        ssl_frame = tk.Frame(tab)        ssl_frame.grid(row=7, column=0, columnspan=3, sticky="w", padx=10, pady=4)        tk.Label(ssl_frame, text="证书 crt:").grid(row=0, column=0, padx=5)        ssl_crt = tk.Entry(ssl_frame, width=45)        ssl_crt.grid(row=0, column=1, padx=5)        ssl_crt.insert(0, ssl_crt_val or "conf/ssl/site.crt")        tk.Label(ssl_frame, text="证书 key:").grid(row=1, column=0, padx=5)        ssl_key = tk.Entry(ssl_frame, width=45)        ssl_key.grid(row=1, column=1, padx=5)        ssl_key.insert(0, ssl_key_val or "conf/ssl/site.key")        if not is_ssl_val:            ssl_frame.grid_remove()        tab.data = {            "domain": domain, "port": port, "root": root_path,            "is_vhost": is_vhost, "is_ssl": is_ssl,            "ssl_crt": ssl_crt, "ssl_key": ssl_key,            "ssl_frame": ssl_frame,            "autoindex": autoindex,            "index_entry": index_entry,            "edit_btn": edit_vhost_btn        }    # === 端口检测 ===    def check_port(self, port_str):        try:            port = int(port_str)            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)            s.settimeout(0.5)            res = s.connect_ex(("127.0.0.1", port))            s.close()            if res == 0:                messagebox.showerror("检测", f"端口 {port} 已占用")            else:                messagebox.showinfo("检测", f"端口 {port} 可用")        except:            messagebox.showwarning("提示", "端口必须是数字")    # === 选择目录 ===    def select_root(self, entry):        p = filedialog.askdirectory(title="选择网站根目录")        if p:            entry.delete(0, tk.END)            entry.insert(0, p)    def toggle_ssl(self, frame, var):        frame.grid() if var.get() else frame.grid_remove()    # === 编辑主配置 ===    def open_main_config(self):        win = tk.Toplevel(self)        win.title("编辑主配置文件 nginx.conf")        win.geometry("800x600")        txt = scrolledtext.ScrolledText(win, width=100, height=35)        txt.pack(fill=tk.BOTH, expand=1, padx=10, pady=10)        with open(NGINX_CONF, "r", encoding="utf-8") as f:            txt.insert(tk.END, f.read())        def save():            with open(NGINX_CONF, "w", encoding="utf-8") as f:                f.write(txt.get(1.0, tk.END))            messagebox.showinfo("成功", "主配置已保存")        tk.Button(win, text="💾 保存配置", command=save).pack(pady=5)    # === 编辑独立站点配置 ===    def open_vhost_config(self, tab):        d = tab.data        domain = d["domain"].get().strip()        if not domain:            messagebox.showwarning("提示", "请先填写域名")            return        fname = f"{domain}.conf".replace(":", "_").replace("*", "_")        fpath = os.path.join(VHOST_DIR, fname)        if not os.path.exists(fpath):            messagebox.showwarning("提示", "配置文件不存在,请先保存站点")            return        win = tk.Toplevel(self)        win.title(f"编辑站点配置:{fname}")        win.geometry("750x550")        txt = scrolledtext.ScrolledText(win)        txt.pack(fill=tk.BOTH, expand=1, padx=10, pady=10)        with open(fpath, "r", encoding="utf-8") as f:            txt.insert(tk.END, f.read())        def save_vhost():            with open(fpath, "w", encoding="utf-8") as f:                f.write(txt.get(1.0, tk.END))            messagebox.showinfo("成功", "站点配置已保存")        tk.Button(win, text="💾 保存此配置", command=save_vhost).pack(pady=5)    # === 生成配置 ===    def gen_server(self, domain, port, root, is_ssl, crt, key, autoindex, index_list):        ai = "autoindex on;" if autoindex else "autoindex off;"        index = f"index {index_list};"        if is_ssl:            return f"""server {{    listen {port} ssl;    server_name {domain};    root {root};    {index}    {ai}    ssl_certificate {crt};    ssl_certificate_key {key};    ssl_session_cache shared:SSL:1m;    ssl_session_timeout 5m;    ssl_ciphers HIGH:!aNULL:!MD5;    ssl_prefer_server_ciphers on;}}"""        else:            return f"""server {{    listen {port};    server_name {domain};    root {root};    {index}    {ai}}}"""    def save_all(self):        main_conf = self.get_main_conf_template()        includes = []        for tab in self.tabs:            d = tab.data            domain = d["domain"].get().strip()            port = d["port"].get().strip()            root = d["root"].get().strip()            is_vhost = d["is_vhost"].get()            is_ssl = d["is_ssl"].get()            crt = d["ssl_crt"].get().strip()            key = d["ssl_key"].get().strip()            ai = d["autoindex"].get()            idx = d["index_entry"].get().strip()            if not all([domain, port, root]):                messagebox.showwarning("提示", "站点信息不完整")                return            conf = self.gen_server(domain, port, root, is_ssl, crt, key, ai, idx)            if is_vhost:                fn = f"{domain}.conf".replace(":", "_").replace("*", "_")                fp = os.path.join(VHOST_DIR, fn)                with open(fp, "w", encoding="utf-8") as f:                    f.write(conf)                includes.append(f"include vhost/{fn};")            else:                main_conf += "\n" + conf        final = main_conf + "\n" + "\n".join(includes) + "\n}"        with open(NGINX_CONF, "w", encoding="utf-8") as f:            f.write(final)        messagebox.showinfo("成功", "✅ 所有站点配置已保存!")    # === Nginx 控制 ===    def start_nginx(self):        os.chdir(NGINX_ROOT)        os.system("start nginx")        messagebox.showinfo("操作", "▶️ Nginx 已启动")    def stop_nginx(self):        os.chdir(NGINX_ROOT)        os.system("nginx -s stop")        messagebox.showinfo("操作", "⏹️ Nginx 已停止")    def reload_nginx(self):        os.chdir(NGINX_ROOT)        os.system("nginx -s quit")        os.system("start nginx")        messagebox.showinfo("操作", "🔄 Nginx 已重启")    def test_nginx(self):        os.chdir(NGINX_ROOT)        code = os.system("nginx -t")        if code == 0:            messagebox.showinfo("测试", "✅ 配置语法正确")        else:            messagebox.showerror("错误", "❌ 配置错误")    def get_main_conf_template(self):        return """worker_processes  1;events {    worker_connections  1024;}http {    include       mime.types;    default_type  application/octet-stream;    sendfile        on;    keepalive_timeout  65;"""if __name__ == "__main__":    app = NginxSiteManager()    app.mainloop()

最新文章

随机文章