在下周公子

  • 首页
  • 域名收藏
  • SEM搜索引擎营销
  • 操作系统
    • 服务器应用
  • 特色专题
    • 跑步运动
  • VPS主机推荐
  • 搜索
在下周公子
三十而立,学以致用
  1. 首页
  2. 编程之路
  3. 正文

在windows下基于Python的nginx控制脚本

2024年11月22日 606点热度 0人点赞 0条评论

在windows下有xampp这样的优秀测试开发环境,但是使用代理的时候,apache的引擎在代理方面是一个弱势。具体的应用场景是,我在开发vue3应用,但是是类手机界面,现在需要真机访问体验,使用node.js是没办法直接在手机访问到的,这个时候。使用nginx,配合代理配置文件,就能跨端直接体验内容。

下面是nginx的配置文件nginx.conf修改后的内容2

#user  nobody;
worker_processes  1;

#error_log  logs/error.log;
#error_log  logs/error.log  notice;
#error_log  logs/error.log  info;

#pid        logs/nginx.pid;

events {
    worker_connections  1024;
}

http {
    include       mime.types;
    default_type  application/octet-stream;

    sendfile        on;
    keepalive_timeout  65;

    server {
        listen       80;
        server_name  localhost;

        # 代理本地5173端口 例如vite端口
        location / {
            proxy_pass http://localhost:5173;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            
            # 跨域配置移到这里(非必要)
            add_header Access-Control-Allow-Origin *;
            add_header Access-Control-Allow-Methods 'GET, POST, OPTIONS, PUT, DELETE';
            add_header Access-Control-Allow-Headers 'DNT,X-Mx-ReqToken,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Authorization';
        }

        # 代理远程API 例如其它主机的后端服务
        location /api/ {
            proxy_pass https://localhost:9998/Workshop-MES/;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            
            # 跨域配置
            add_header Access-Control-Allow-Origin *;
            add_header Access-Control-Allow-Methods 'GET, POST, OPTIONS, PUT, DELETE';
            add_header Access-Control-Allow-Headers 'DNT,X-Mx-ReqToken,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Authorization';
            
            # 关闭SSL验证(开发环境)
            proxy_ssl_verify off;
        }

        # OPTIONS请求处理
        location = /options-request {
            add_header Access-Control-Allow-Origin *;
            add_header Access-Control-Allow-Methods 'GET, POST, OPTIONS, PUT, DELETE';
            add_header Access-Control-Allow-Headers 'DNT,X-Mx-ReqToken,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Authorization';
            add_header Access-Control-Max-Age 1728000;
            add_header Content-Type 'text/plain charset=UTF-8';
            add_header Content-Length 0;
            return 204;
        }

        # 错误页面配置
        error_page   500 502 503 504  /50x.html;
        location = /50x.html {
            root   html;
        }
    }
}

今天分享的是python控制nginx的操作。正常是会用cmd来运行 nginx.exe 但是不够优雅。这里使用python写了一个脚本。本人代码能力菜鸡,下方代码来自AI生成,非本人功劳,这里只能提供一个思路。

对了,代码启动需要安装一个库pip install psutil

import os
import sys
import time
import subprocess
import psutil
from pathlib import Path

# Nginx路径配置
NGINX_PATH = r"G:\xampp\nginx-1.26.2\nginx.exe"
def check_nginx_exists():
    return os.path.exists(NGINX_PATH)

def get_nginx_processes():
    """获取所有Nginx相关进程"""
    nginx_processes = []
    try:
        for proc in psutil.process_iter(['name', 'pid', 'ppid']):
            if proc.info['name'].lower() == 'nginx.exe':
                nginx_processes.append(proc)
    except Exception as e:
        print(f"\033[93m获取进程列表时出错: {str(e)}\033[0m")
    return nginx_processes


def is_nginx_running():
    """检查Nginx是否在运行"""
    return len(get_nginx_processes()) > 0


def get_nginx_master_process():
    """获取Nginx主进程"""
    processes = get_nginx_processes()
    for proc in processes:
        try:
            # Nginx主进程通常没有父进程或父进程不是nginx
            ppid = proc.ppid()
            parent = psutil.Process(ppid) if ppid else None
            if not parent or parent.name().lower() != 'nginx.exe':
                return proc
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return None


def kill_process_tree(pid, include_parent=True):
    """递归终止进程树"""
    try:
        parent = psutil.Process(pid)
        children = parent.children(recursive=True)

        # 先终止子进程
        for child in children:
            try:
                child.kill()
            except psutil.NoSuchProcess:
                pass

        # 等待子进程完全结束
        gone, alive = psutil.wait_procs(children, timeout=3)

        # 如果需要,终止父进程
        if include_parent:
            parent.kill()
            parent.wait(3)
    except psutil.NoSuchProcess:
        pass
def force_kill_nginx():
    """强制终止所有Nginx相关进程"""
    # 使用taskkill /F /T 命令(包含子进程树)
    try:
        subprocess.run(
            ["taskkill", "/F", "/T", "/IM", "nginx.exe"],
            shell=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=5
        )
    except:
        pass

    # 使用psutil查找并终止所有nginx进程
    for proc in psutil.process_iter(['name', 'pid', 'ppid']):
        try:
            if proc.info['name'].lower() == 'nginx.exe':
                kill_process_tree(proc.info['pid'])
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue

    # 最后再检查一遍并使用最底层的系统调用
    for proc in psutil.process_iter(['name', 'pid']):
        try:
            if proc.info['name'].lower() == 'nginx.exe':
                os.kill(proc.info['pid'], signal.SIGTERM)
                time.sleep(0.1)
                if psutil.pid_exists(proc.info['pid']):
                    os.kill(proc.info['pid'], signal.SIGKILL)
        except:
            continue


def stop_nginx():
    try:
        if not is_nginx_running():
            print("\033[93mNginx未在运行,无需停止!\033[0m")
            return

        print("\033[93m正在尝试优雅停止Nginx...\033[0m")
        try:
            # 先尝试正常停止
            subprocess.run([NGINX_PATH, "-s", "quit"],
                           shell=False,
                           stdout=subprocess.PIPE,
                           stderr=subprocess.PIPE,
                           timeout=5)
        except:
            pass

        # 等待一段时间看是否正常停止
        time.sleep(3)

        if is_nginx_running():
            print("\033[93m正在强制停止所有Nginx进程...\033[0m")
            # 使用增强的强制终止方法
            force_kill_nginx()
            time.sleep(2)

            # 再次尝试强制终止
            if is_nginx_running():
                print("\033[93m使用终极方式停止进程...\033[0m")
                force_kill_nginx()
                time.sleep(2)

        # 最终检查
        if not is_nginx_running():
            print("\033[92mNginx服务已成功停止!\033[0m")
        else:
            remaining_processes = get_nginx_processes()
            print("\033[91m警告:仍有Nginx进程无法停止!\033[0m")
            print("剩余进程:")
            for proc in remaining_processes:
                try:
                    print(f"PID: {proc.pid}, PPID: {proc.ppid()}")
                except:
                    continue
            print("\033[91m建议手动使用任务管理器终止这些进程。\033[0m")

    except Exception as e:
        print(f"\033[91m停止过程中发生错误: {str(e)}\033[0m")

def start_nginx():
    try:
        if is_nginx_running():
            print("\033[93mNginx已经在运行中!\033[0m")
            return

        nginx_dir = os.path.dirname(NGINX_PATH)
        current_dir = os.getcwd()
        os.chdir(nginx_dir)

        # 启动nginx
        process = subprocess.Popen(
            [NGINX_PATH],
            shell=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

        # 等待启动完成
        time.sleep(2)

        if is_nginx_running():
            print("\033[92mNginx服务已成功启动!\033[0m")
            # 打印进程信息
            processes = get_nginx_processes()
            print(f"\033[92m已启动 {len(processes)} 个Nginx进程:\033[0m")
            for proc in processes:
                try:
                    print(f"PID: {proc.pid}, PPID: {proc.ppid()}")
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    continue
        else:
            print("\033[91m启动Nginx服务失败!\033[0m")
            stdout, stderr = process.communicate(timeout=1)
            if stderr:
                print(f"错误信息: {stderr.decode()}")

        os.chdir(current_dir)
    except Exception as e:
        print(f"\033[91m启动过程中发生错误: {str(e)}\033[0m")

def check_ports():
    """检查Nginx常用端口占用情况"""
    ports_to_check = [80, 443]
    for port in ports_to_check:
        try:
            for conn in psutil.net_connections():
                if conn.laddr.port == port:
                    try:
                        process = psutil.Process(conn.pid)
                        print(f"\033[93m端口 {port} 被进程 {process.name()} (PID: {conn.pid}) 占用\033[0m")
                    except:
                        print(f"\033[93m端口 {port} 被占用 (PID: {conn.pid})\033[0m")
        except:
            continue
def show_status():
    processes = get_nginx_processes()
    if processes:
        print("\033[92mNginx状态: 运行中")
        print(f"运行中的Nginx进程数: {len(processes)}")
        for proc in processes:
            try:
                print(f"PID: {proc.pid}, PPID: {proc.ppid()}")
                try:
                    print(f"进程路径: {proc.exe()}")
                    print(f"启动命令: {' '.join(proc.cmdline())}")
                except:
                    pass
            except:
                continue
        print("\033[0m")
        check_ports()
    else:
        print("\033[91mNginx状态: 已停止\033[0m")

def clear_screen():
    os.system('cls' if os.name == 'nt' else 'clear')

def show_menu():
    clear_screen()
    print("\033[96m=================================")
    print("      Nginx服务控制面板")
    print("=================================")
    print(f"Nginx路径: {NGINX_PATH}")
    print("\n可用命令:")
    print("    run    - 启动Nginx")
    print("    stop   - 停止Nginx")
    print("    status - 检查状态")
    print("    clear  - 清屏")
    print("    exit   - 退出脚本")
    print("\n当前状态:\033[0m")
    show_status()

def main():
    # 检查Nginx是否存在
    if not check_nginx_exists():
        print(f"\033[91m错误: 找不到Nginx可执行文件 ({NGINX_PATH})\033[0m")
        print("请确认路径是否正确!")
        input("按Enter键退出...")
        sys.exit(1)

    show_menu()

    while True:
        print()
        choice = input("请输入命令 (run/stop/status/clear/exit): ").lower()

        if choice == "run":
            start_nginx()
        elif choice == "stop":
            stop_nginx()
        elif choice == "status":
            show_status()
        elif choice == "clear":
            show_menu()
        elif choice == "exit":
            if is_nginx_running():
                confirm = input("Nginx正在运行,是否停止?(y/n): ").lower()
                if confirm == 'y':
                    stop_nginx()
            print("\033[96m感谢使用,再见!\033[0m")
            time.sleep(1)
            break
        else:
            print("\033[91m未知命令,请使用 run/stop/status/clear/exit\033[0m")

if __name__ == "__main__":
    main()

优雅的解决一些问题。这里还可以扩展结合php等脚本。

标签: 暂无
最后更新:2025年3月17日

zhou

这个人很懒,什么都没留下

点赞
< 上一篇

COPYRIGHT © 2021 zhoubin.net. ALL RIGHTS RESERVED.

Theme Kratos Made By Seaton Jiang

浙ICP备09065757号-10