feat(main): 增加新功能并优化现有功能

- 添加了新的命令行参数和功能:
  - --nginx-reload: 重新加载 Nginx 配置
  - --docker-compose-restart: 重启 Docker Compose 服务
  - --docker-ps: 列出运行中的 Docker 容器
- 重构了 main 函数,使其支持更灵活的参数配置
- 优化了配置文件读取和解析逻辑
- 更新了 .gitignore 文件,添加了 requirements.txt
- 修正了拼写错误:INNER_NGINX_OUTPATH -> inner-nginx.conf
This commit is contained in:
ZZY 2025-02-01 14:47:38 +08:00
parent 3b9dac3b7f
commit a4ebe33fca
6 changed files with 222 additions and 79 deletions

3
.gitignore vendored
View File

@ -3,4 +3,5 @@
!src/* !src/*
!.gitignore !.gitignore
!default* !default*
!main.py !main.py
!requirements.txt

View File

@ -2,5 +2,5 @@ DOCKER_INIPATH = docker.ini
NGINX_INIPATH = nginx.ini NGINX_INIPATH = nginx.ini
DOCKER_COMPOSE_OUTPATH = docker-compose.yml DOCKER_COMPOSE_OUTPATH = docker-compose.yml
INNER_NGINX_OUTPATH = innner-nginx.conf INNER_NGINX_OUTPATH = inner-nginx.conf
OUTER_NGINX_OUTPATH = outer-nginx.conf OUTER_NGINX_OUTPATH = outer-nginx.conf

102
main.py
View File

@ -6,63 +6,111 @@ import os
import subprocess import subprocess
from pathlib import Path from pathlib import Path
from dotenv import load_dotenv from dotenv import load_dotenv
from jinja2_render import render_dataclass from src.jinja2_render import render_dataclass
from src.config_reader import nginx_config, docker_config, docker2nginx from src.config_reader import nginx_config, docker_config, docker2nginx
load_dotenv() load_dotenv()
def run_command(command): def run_command(command):
"run a command and use stdout"
try: try:
result = subprocess.run(command, shell=True, check=True, result = subprocess.run(command, shell=True, check=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout=subprocess.PIPE, stderr=subprocess.PIPE)
print(result.stdout.decode()) print(result.stdout.decode())
return True
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
print(f"Command failed with error: {e.stderr.decode()}") print(f"Command failed with error: {e.stderr.decode()}")
return False
def docker_compose_action(): def docker_compose_action(config_path, output_path):
config_path = Path(os.getenv('DOCKER_CONFIG_PATH', 'docker.ini')) "读取并渲染配置文件"
output_path = Path(os.getenv('DOCKER_COMPOSE_OUTPATH', 'docker-compose.conf'))
(auto_config, other) = docker_config(config_path) (auto_config, other) = docker_config(config_path)
render_dataclass(auto_config, Path('./src/docker-compose.j2'), output_path, other) render_dataclass(auto_config, Path('./src/docker-compose.j2'), output_path, other=other)
def inner_nginx_action():
config_path = Path(os.getenv('DOCKER_CONFIG_PATH', 'docker.ini'))
output_path = Path(os.getenv('INNER_NGINX_OUTPATH', 'inner-nginx.conf'))
def inner_nginx_action(config_path, output_path):
"读取并渲染配置文件"
(auto_config, other) = docker2nginx(config_path) (auto_config, other) = docker2nginx(config_path)
render_dataclass(auto_config, Path('./src/nginx-conf.j2'), output_path, other) render_dataclass(auto_config, Path('./src/nginx-conf.j2'), output_path, other=other)
def outer_nginx_action():
config_path = Path(os.getenv('NGINX_CONFIG_PATH', 'docker.ini'))
output_path = Path(os.getenv('OUTER_NGINX_OUTPATH', 'outer-nginx.conf'))
def outer_nginx_action(config_path, output_path):
"读取并渲染配置文件"
(auto_config, other) = nginx_config(config_path) (auto_config, other) = nginx_config(config_path)
render_dataclass(auto_config, Path('./src/nginx-conf.j2'), output_path, other) render_dataclass(auto_config, Path('./src/nginx-conf.j2'), output_path, other=other)
def nginx_reload():
"""reload nginx config"""
res = True
if res:
res = run_command("docker exec -it nginx nginx -t")
if res:
res = run_command("docker exec -it nginx nginx -s reload")
def docker_compose_restart():
"docker compose restart"
run_command("docker compose restart")
def docker_ps():
"list running docker containers by my format"
cmd = "docker ps --format "\
"\"table {{.Names}}\t{{.Image}}\t{{.Command}}\t{{.CreatedAt}}\t{{.Status}}\""
run_command(cmd)
def main(): def main():
"main function"
parser = argparse.ArgumentParser(description="Automate Docker and Nginx configuration.") parser = argparse.ArgumentParser(description="Automate Docker and Nginx configuration.")
parser.add_argument('-a', '--all', action='store_true', default=True,
help='Perform all actions')
parser.add_argument('-c', '--docker-compose', action='store_true', parser.add_argument('-c', '--docker-compose', action='store_true',
help='Generate Docker Compose') help='Generate Docker Compose')
parser.add_argument('-i', '--inner-nginx', action='store_true', parser.add_argument('-i', '--inner-nginx', action='store_true',
help='Generate Inner Nginx (docker network)') help='Generate Inner Nginx (docker network)')
parser.add_argument('-o', '--outer-nginx', action='store_true', parser.add_argument('-o', '--outer-nginx', action='store_true',
help='Generate Outer Nginx (host machine)') help='Generate Outer Nginx (host machine)')
parser.add_argument('-r', '--nginx-reload', action='store_true',
help='Reload Nginx configuration')
parser.add_argument('-s', '--docker-compose-restart', action='store_true',
help='Restart Docker Compose services')
parser.add_argument('-p', '--docker-ps', action='store_true',
help='List running Docker containers')
parser.add_argument('--docker-config-path', type=Path,
default=Path(os.getenv('DOCKER_INIPATH', 'docker.ini')),
help='Path to the Docker INI configuration file')
parser.add_argument('--docker-output-path', type=Path,
default=Path(os.getenv('DOCKER_COMPOSE_OUTPATH', 'docker-compose.conf')),
help='Path to the output Docker Compose configuration file')
parser.add_argument('--inner-config-path', type=Path,
default=Path(os.getenv('DOCKER_INIPATH', 'docker.ini')),
help='Path to the Docker INI configuration file for inner Nginx')
parser.add_argument('--inner-output-path', type=Path,
default=Path(os.getenv('INNER_NGINX_OUTPATH', 'inner-nginx.conf')),
help='Path to the output Inner Nginx configuration file')
parser.add_argument('--outer-config-path', type=Path,
default=Path(os.getenv('NGINX_INIPATH', 'nginx.ini')),
help='Path to the Nginx INI configuration file for outer Nginx')
parser.add_argument('--outer-output-path', type=Path,
default=Path(os.getenv('OUTER_NGINX_OUTPATH', 'outer-nginx.conf')),
help='Path to the output Outer Nginx configuration file')
args = parser.parse_args() args = parser.parse_args()
if args.docker_compose: if args.all:
docker_compose_action() docker_compose_action(args.docker_config_path, args.docker_output_path)
elif args.inner_nginx: inner_nginx_action(args.inner_config_path, args.inner_output_path)
inner_nginx_action() outer_nginx_action(args.outer_config_path, args.outer_output_path)
elif args.outer_nginx: nginx_reload()
outer_nginx_action()
else: else:
# Default behavior: perform all actions if args.docker_compose:
docker_compose_action() docker_compose_action(args.docker_config_path, args.docker_output_path)
inner_nginx_action() if args.inner_nginx:
outer_nginx_action() inner_nginx_action(args.inner_config_path, args.inner_output_path)
if args.outer_nginx:
outer_nginx_action(args.outer_config_path, args.outer_output_path)
if args.nginx_reload:
nginx_reload()
if args.docker_compose_restart:
docker_compose_restart()
if args.docker_ps:
docker_ps()
if __name__ == "__main__": if __name__ == "__main__":
main() main()

2
requirements.txt Normal file
View File

@ -0,0 +1,2 @@
python-dotenv
jinja2

View File

@ -1,20 +1,68 @@
"""
config_reader.py
"""
import configparser import configparser
from dataclasses import dataclass from dataclasses import dataclass
from typing import Optional from typing import Optional, List, Tuple
from pathlib import Path from pathlib import Path
def read_config(file_path: Path) -> configparser.ConfigParser: def read_config(file_path: Path) -> configparser.ConfigParser:
"""读取INI配置文件""" """
读取并解析指定路径的 INI 配置文件
参数:
file_path (Path): 配置文件的路径
返回:
configparser.ConfigParser: 解析后的配置对象
"""
config = configparser.ConfigParser() config = configparser.ConfigParser()
config.read(file_path, encoding='utf-8') config.read(file_path, encoding='utf-8')
return config return config
def config_get(config: configparser.ConfigParser, section, key): def config_get(config: configparser.ConfigParser, section: str, key: str,
"""从配置文件中获取值,支持默认值""" default: str = None) -> str:
return config.get(section, key, fallback=config.get(config.default_section, key, fallback='')) """
从配置文件中获取指定键的值支持默认值
如果在指定的 section 中找不到键则尝试从默认 section 中获取
如果最终值为None则会抛出异常
参数:
config (configparser.ConfigParser): 配置对象
section (str): 要查询的 section 名称
key (str): 要查询的键名称
default (str): 最终配置文件没有时硬编码的默认值
返回:
str: 查询到的值或默认值
"""
res = config.get(section, key, fallback=
config.get(config.default_section, key, fallback=default))
# if is_raise and res is None:
# raise ValueError(f"Key '{key}' not found in section '{section}' or "\
# "default section or default value.")
return res
@dataclass @dataclass
class NginxConfig: class NginxConfig:
"""
表示一个 Nginx 配置项的数据类
属性:
name (str): 配置项名称默认为 "unknown"
host (str): 主机地址默认为 "localhost"
port (int): 端口号默认为 80
output_path (Optional[str]): 输出路径默认为 None
tcp_nopush (str): TCP 推送选项默认为 "off"
sendfile (str): Sendfile 选项默认为 "off"
client_max_body_size (str): 客户端最大请求体大小默认为 "100m"
proxy_connect_timeout (str): 代理连接超时时间默认为 "60s"
proxy_send_timeout (str): 代理发送超时时间默认为 "60s"
proxy_read_timeout (str): 代理读取超时时间默认为 "60s"
web_socket_proxy (bool): 是否启用 WebSocket 代理默认为 False
"""
name: str = "unknown" name: str = "unknown"
host: str = "localhost" host: str = "localhost"
port: int = 80 port: int = 80
@ -27,70 +75,109 @@ class NginxConfig:
proxy_read_timeout: str = "60s" proxy_read_timeout: str = "60s"
web_socket_proxy: bool = False web_socket_proxy: bool = False
def nginx_config(file_path: Path) -> tuple[list[NginxConfig], dict[str, str]]: def nginx_config(file_path: Path) -> Tuple[List[NginxConfig], dict]:
"""读取INI配置文件并解析为NginxConfig列表""" """
读取 INI 配置文件并解析为 NginxConfig 对象列表
参数:
file_path (Path): 配置文件的路径
返回:
Tuple[List[NginxConfig], dict]: 包含 NginxConfig 对象的列表和其他相关信息的字典
"""
config = read_config(file_path) config = read_config(file_path)
configs = [] configs = []
g_section = config.default_section g_section = config.default_section
def _config_get(name): def _config_get(name, default = None):
return config_get(config, g_section, name) return config_get(config, g_section, name, default)
for section in config.sections(): for section in config.sections():
g_section = section g_section = section
configs.append(NginxConfig( configs.append(NginxConfig(
name = section, name=section,
host = _config_get('host'), host=_config_get('host'),
port = _config_get('port'), port=int(_config_get('port')), # 将端口转换为整数类型
web_socket_proxy = _config_get('web_socket_proxy') == 'True', web_socket_proxy=_config_get('web_socket_proxy') == 'True',
client_max_body_size=_config_get('client_max_body_size',
NginxConfig.client_max_body_size)
)) ))
return (configs, {}) return (configs, {})
@dataclass @dataclass
class DockerComposeConfig: class DockerComposeConfig:
name: str """
表示一个 Docker Compose 配置项的数据类
属性:
name (str): 服务名称
container_name (str): 容器名称
image (str): 使用的镜像
ports (Optional[list[str]]): 映射的端口列表默认为 None
volumes (Optional[list[str]]): 挂载的卷列表默认为 None
environment (Optional[list[str]]): 环境变量列表默认为 None
extra_hosts (Optional[list[str]]): 额外的主机映射列表默认为 None
export (Optional[str]): 导出的端口默认为 None
network (Optional[str]): 所属网络默认为 None
"""
name: str
container_name: str container_name: str
image: str image: str
ports: Optional[list[str]] = None ports: Optional[List[str]] = None
volumes: Optional[list[str]] = None volumes: Optional[List[str]] = None
environment: Optional[list[str]] = None environment: Optional[List[str]] = None
extra_hosts: Optional[list[str]] = None extra_hosts: Optional[List[str]] = None
export: Optional[str] = None
network: Optional[str] = None
export: Optional[str] = None def docker_config(file_path: Path) -> Tuple[List[DockerComposeConfig], dict]:
network: Optional[str] = None """
读取 INI 配置文件并解析为 DockerComposeConfig 对象列表
def docker_config(file_path: Path) -> tuple[list[DockerComposeConfig], dict[str, str]]: 参数:
"""读取INI配置文件并解析为DockerComposeConfig列表""" file_path (Path): 配置文件的路径
返回:
Tuple[List[DockerComposeConfig], dict]: 包含 DockerComposeConfig 对象的列表和其他相关信息的字典
"""
config = read_config(file_path) config = read_config(file_path)
configs = [] configs = []
g_section = '' g_section = ''
def _config_get(name): def _config_get(name, default = None):
return config_get(config, g_section, name) return config_get(config, g_section, name, default)
def _get_item(name) -> List[str]:
"""将逗号分隔的字符串转换为列表,并去除空格和空字符串。"""
return [i.strip() for i in _config_get(name, '').split(',') if i.strip()]
def _get_item(name) -> list[str]:
return [i.strip() for i in _config_get(name).split(',') if i.strip()]
network = set() network = set()
for section in config.sections(): for section in config.sections():
g_section = section g_section = section
configs.append(DockerComposeConfig( configs.append(DockerComposeConfig(
name = section, name=section,
container_name = _config_get('container_name'), container_name=_config_get('container_name'),
image = _config_get('image'), image=_config_get('image'),
volumes = _get_item('volumes'), volumes=_get_item('volumes'),
ports = _get_item('ports'), ports=_get_item('ports'),
environment = _get_item('environment'), environment=_get_item('environment'),
extra_hosts = _get_item('extra_hosts'), extra_hosts=_get_item('extra_hosts'),
export = _config_get('export'), export=_config_get('export'),
network = _config_get('network'), network=_config_get('network'),
)) ))
network.add(_config_get('network')) network.add(_config_get('network'))
return (configs, {'networks': list(network)}) return (configs, {'networks': list(network)})
def docker2nginx(file_path: Path) -> tuple[list[NginxConfig], dict[str, str]]: def docker2nginx(file_path: Path) -> Tuple[List[NginxConfig], dict]:
"将 docker-compose 服务转换为 nginx 代理" """
"""读取INI配置文件并解析为DockerComposeConfig列表""" Docker Compose 服务配置转换为 Nginx 代理配置
参数:
file_path (Path): 配置文件的路径
返回:
Tuple[List[NginxConfig], dict]: 包含 NginxConfig 对象的列表和其他相关信息的字典
"""
config = read_config(file_path) config = read_config(file_path)
configs = [] configs = []
@ -98,18 +185,17 @@ def docker2nginx(file_path: Path) -> tuple[list[NginxConfig], dict[str, str]]:
def _config_get(name): def _config_get(name):
return config_get(config, g_section, name) return config_get(config, g_section, name)
def _get_item(name) -> list[str]: def _get_item(name) -> List[str]:
"""将逗号分隔的字符串转换为列表,并去除空格和空字符串。"""
return [i.strip() for i in _config_get(name).split(',') if i.strip()] return [i.strip() for i in _config_get(name).split(',') if i.strip()]
network = set()
for section in config.sections(): for section in config.sections():
if (section == 'nginx'): if section == 'nginx':
continue continue
g_section = section g_section = section
configs.append(NginxConfig( configs.append(NginxConfig(
name = config.get(section, 'nginx_name', fallback=section), name=config.get(section, 'nginx_name', fallback=section),
host = _config_get('container_name'), host=_config_get('container_name'),
port = _config_get('export'), port=_config_get('export'),
)) ))
network.add(_config_get('network')) return (configs, {})
return (configs, {'networks': list(network)})

View File

@ -1,8 +1,12 @@
"""
jinja2_render.py
"""
from pathlib import Path from pathlib import Path
from dataclasses import asdict, dataclass from dataclasses import asdict, dataclass
from jinja2 import Environment, FileSystemLoader, Template from jinja2 import Environment, FileSystemLoader, Template
def render_dataclass(configs: list[dataclass], tmp_path: Path, out_path: Path, other = None) -> str: def render_dataclass(configs: list[dataclass], tmp_path: Path, out_path: Path,
verbose :bool = True, other = None) -> str:
"from dataclasses to jinja template" "from dataclasses to jinja template"
env = Environment(loader=FileSystemLoader(str(tmp_path.parent), env = Environment(loader=FileSystemLoader(str(tmp_path.parent),
encoding='utf-8')) encoding='utf-8'))
@ -10,4 +14,6 @@ def render_dataclass(configs: list[dataclass], tmp_path: Path, out_path: Path, o
data = [asdict(config) for config in configs] data = [asdict(config) for config in configs]
res = template.render({'autoconfigs': data}, **other) res = template.render({'autoconfigs': data}, **other)
out_path.write_text(res, encoding='utf-8') out_path.write_text(res, encoding='utf-8')
if verbose:
print(f"{out_path.absolute()} is rendered")
return res return res