py/sunhpc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add init pxe cmd

Browse Source
This commit is contained in:
Qichao
2026-05-05 02:12:33 +08:00
parent 9ec1cf4fdd
commit 2cc70998d1
24 changed files with 30461 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
lib/sunhpc/sunhpc/commands/__init__.py
View File

@@ -8,10 +8,13 @@ import types
import syslog
import inspect
import argparse
import subprocess
import sunhpc.util
import sunhpc.utils
from io import StringIO
from typing import Dict
from pathlib import Path
from typing import Dict, Optional, Tuple, List, Union
from sunhpc.logs import Logs
from sunhpc.disks import DiskInfo
@@ -19,6 +22,9 @@ from sunhpc.configs import Config
from sunhpc.network import Network
from sunhpc.memory import MemoryInfo
from sunhpc.output import Output
from sunhpc.firewalld import FWManager
from sunhpc.utils.shells import ShellExecutor
from xml.sax import saxutils
from xml.sax import handler
@@ -335,12 +341,15 @@ class Command:
self.mem = MemoryInfo()
self.disk = DiskInfo()
self.fmt = Output()
self.sh = ShellExecutor()
self.fw = FWManager('sunhpc')
self.os = os.uname()[0].lower()
self.arch = os.uname()[4]
self.user = pwd.getpwuid(os.getuid()).pw_name
self.parser = argparse.ArgumentParser()
self.parserArgs = []
self.subcommand = 'Sub Command'
self.subdesc = None
self.epilog = None
@@ -392,6 +401,77 @@ class Command:
else:
return 'no'
def rsync_copy(
self,
src: str,
dst: str,
options: str = '-ah --info=progress2'
) -> bool:
"""
简化版本的rsync拷贝、专注于进度显示
Args:
src: 源路径
dst: 目标路径
Returns:
bool: 是否成功
"""
options = options.split()
cmd = ["rsync"]
cmd.extend(options)
cmd.append(src)
cmd.append(dst)
self.log.info(f"执行命令: {' '.join(cmd)}")
try:
# 使用Popen实时显示输出
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1
)
# 实时打印输出
for line in process.stdout:
# 去除末尾换行符
line = line.rstrip('\n\r')
if line:
# 计算需要覆盖的长度（使用退格符或直接回车）
# 方法1：使用 \r 回车覆盖（推荐）
print(f'\r{line}', end='', flush=True)
last_line = line
else:
# 空行，换行
print()
# 输出完成后换行，避免下一行输出覆盖进度信息
if last_line:
print()
return_code = process.wait()
if return_code == 0:
self.log.info("\n✓ 拷贝成功完成")
return True
else:
self.log.error(f"\n✗ 拷贝失败，返回码: {return_code}")
return False
except Exception as e:
print(f"✗ 执行失败: {e}")
return False
def is_remote_path(self, path: str) -> bool:
"""判断是否为远程路径"""
return '@' in path and ':' in path and not path.startswith(':')
def dtons(self, data: dict) -> types.SimpleNamespace:
'''递归将字典转换为 SimpleNamespace'''
if isinstance(data, dict):
@@ -544,6 +624,13 @@ class Command:
else:
return ' '.join(plist)
def output_dict(self, data={}):
if not data: return
max_key_len = max(len(str(key)) for key in data.keys())
for key, value in data.items():
print(f"{str(key):<{max_key_len}} : {value}")
def command(self, command, args=[]):
'''Import and run a Sunhpc command.
Returns and output string.
@@ -599,13 +686,14 @@ class Command:
rlist = []
for (key, default) in dlist:
if 'key' in params:
if key in params:
rlist.append(params[key])
else:
rlist.append(default)
return rlist
def fillPositionalArgs(self, names, params=None, args=None):
def fillPostArgs(self, names, params=None, args=None):
'''
The helper function will allow named parameters
to be used in lieu of positional arguments
@@ -632,7 +720,7 @@ class Command:
Returns:
remaining args, Filled parameters
Example::
hostlist, iface, mac = self.fillPositionalArgs(
hostlist, iface, mac = self.fillPostArgs(
('iface','mac'), params, args)
'''
@@ -690,9 +778,14 @@ class Command:
plist = [] # arguments
nparams = 0
flagpattern = re.compile(r'^-[a-zA-Z0-9-_+]+=')
flagpattern = re.compile(r'^[a-zA-Z0-9-_+]+=')
for arg in args:
# 检查 - 或 -- 开头
if arg.startswith('--') or arg.startswith('-'):
self.parserArgs.append(arg)
continue
tokens = arg.split()
if tokens[0] == 'select':
plist.append(arg)

177
lib/sunhpc/sunhpc/commands/init/config/__init__.py
View File

@@ -12,106 +12,107 @@ class Command(sunhpc.commands.init.command):
提供一个内网的接口、例如: eth1
</arg>
<arg type='string' name='mnt'>
提供一个系统ISO文件、或者挂载路径、例如: /mnt/cdrom
</arg>
<param type='string' name='iface'>
提供一个内网的接口、例如: eth1
</param>
<param type='string' name='mnt'>
提供一个系统ISO文件、或者挂载路径、例如: /mnt/cdrom
</param>
<arguments>
-h --help 显示帮助信息
-d --debug 开启调试模式
-v --verbose 显示详细信息
</arguments>
<example cmd='init config eth1 /mnt/cdrom'>
使用eth1网络接口、和 /mnt/cdrom 挂载路径
<example cmd='init config eth1 '>
使用eth1网络接口
</example>
<example cmd='init config iface=eth1 mnt=/mnt/cdrom'>
使用eth1网络接口、和 /mnt/cdrom 挂载路径
<example cmd='init config iface=eth1'>
使用eth1网络接口
</example>
"""
def run(self, params, args):
self.parser_cmd_init()
self.parser.add_argument('-i', '--iface', action='store', help='Interface name')
self.parser.add_argument('-d', '--debug', action='store_true', help='Debug mode')
self.parser.add_argument('-f', '--file', action='store_true', help='Configuration file')
(args, iface) = self.fillPostArgs(('iface',))
self.parser.add_argument(
'iface',
help='Network interface (required, e.g., eth0, eth1)'
)
if not iface:
self.abort('must supply an interface')
self.parser.add_argument(
'mnt',
nargs='?',
default='/mnt/cdrom',
help='Mount point (optional, default: /mnt/cdrom)'
)
self.net.setiface(iface)
parsed_params = self.parser_cmd_args(args)
ipaddr = self.net.getip()
if ipaddr is None:
self.abort('interface "%s" has no ip address' % iface)
self.log.info('File "/opt/sunhpc/lib/sunhpc/sunhpc/commands/init/config/__init__.py", line 63, in run File "/opt/sunhpc/lib/sunhpc/sunhpc/commands/init/config/__init__.py", line 63, in run')
#print (parsed_params)
iface = parsed_params.get('iface', 'eth0')
'''
config = Config()
config.iface = 'eth1'
config.mnt = '/mnt/cdrom'
config.network = '1.1.1.1'
config.netmask = '255.255.255.0'
config.gateway = '1.1.1.1'
config.dhcp = Config()
config.dhcp.enabled = True
config.dhcp.server = '1.1.1.1'
config.dhcp.options = ['domain-name', 'domain-name-servers']
config.http = Config()
config.http.enabled = True
config.http.server = '1.1.1.1'
config.api = Config()
config.api.enabled = True
config.api.server = '1.1.1.1'
config.save('/tmp/sunhpc.yaml')
self.net.setiface('eth0')
print (f'Interface: {self.net.iface}')
print (f'IP : {self.net.getip()}')
print (f'Network: {self.net.getnetwork()}')
print (f'Netmask: {self.net.getnetmask()}')
print (f'Gateway: {self.net.getgateway()}')
print (f'MAC : {self.net.getmac()}')
print (f'CIDR : {self.net.getcidr()}')
print (f'IPv6 : {self.net.getipv6()}')
rx, tx = self.net.get_transfer_human()
print (f'RX : {rx}')
print (f'TX : {tx}')
#print (self.disk.get_disk_info('sda'))
#print (self.mem.get_memory_summary())
#print (self.mem.get_memory_info())
#self.fmt.add_header(['Interface', 'IP', 'Network', 'Netmask', 'Gateway', 'MAC', 'CIDR', 'IPv6'])
d1 = {
'name', 'alice',
'age', 18,
'city', 'beijing',
'occupation', 'student'
config_file = self.conf.getConfFile()
config_dict = {
'iface': iface,
'ipaddr': ipaddr,
'netmask': self.net.getnetmask(),
'network': self.net.getnetwork(),
'mac': self.net.getmac(),
'cidr': self.net.getcidr(),
'dnsmasq': {
'enabled': True,
'config': '/etc/dnsmasq.d/sunhpc.conf',
'server': self.net.getip(),
'start_ip': '.'.join(ipaddr.split('.')[:-1]) + '.100',
'end_ip': '.'.join(ipaddr.split('.')[:-1]) + '.200',
'dns': '223.5.5.5',
'systemd_name': 'dnsmasq',
},
'http': {
'enabled': True,
'wwwroot': '/var/www/html',
'config': '/etc/httpd/conf.d/sunhpc.conf',
'httpaddr': 'http://%s' % ipaddr,
'systemd_name': 'httpd',
},
'tftp': {
'enabled': False,
'tftpboot': '/srv/pxelinux/tftpboot',
'systemd_name': 'tftp-server',
},
'nfs': {
'enabled': True,
'paths': [
{
'path': '/home',
'options': 'rw,sync'
},
{
'path': '/opt',
'options': 'rw,sync,no_root_squash'
},
],
'systemd_name': 'nfs-server',
},
'ssh': {
'key_types': [
'ssh-rsa',
'ssh-dss',
'ssh-ed25519',
]
},
'firewall': {
'enabled': True,
'backend': 'firewalld',
},
'osinfo': {
'basename': 'sunhpc',
'os': 'rocky',
'version': '9.7',
},
'pxe':{
'bios': 'boot/undionly.kpxe',
'uefi': 'boot/bootx64.efi',
'ipxe': 'boot/ipxe.efi',
'ksfile': 'scripts/rhel.ks',
'ipxeboot': 'scripts/ipxe.sh',
'repo': 'sunhpc/rockylinux/9.7',
'initrd': 'images/pxeboot/initrd.img',
'vmlinuz': 'images/pxeboot/vmlinuz',
},
'paths': {
'bin': '/opt/sunhpc/bin',
'etc': '/opt/sunhpc/etc',
'lib': '/opt/sunhpc/lib',
'var': '/opt/sunhpc/var',
'pkgs': '/srv/repos/pkgs',
'repo': '/srv/repos/rockylinux/9.7',
}
}
self.fmt.dict_output(d1)
'''
self.conf.save(config_dict=config_dict)

264
lib/sunhpc/sunhpc/commands/init/pxe/__init__.py Normal file
View File

@@ -0,0 +1,264 @@
import os
import re
import sys
import textwrap
import sunhpc.commands
from sunhpc.configs import Config
from sunhpc.paths import ipxe_path, scripts_path
class Command(sunhpc.commands.init.command):
"""
这个命令是初始化系统Pxe引导配置服务.
<arg type='string' name='ISOFile'>
提供一个系统ISO挂载点、例如: /mnt/cdrom
或者提供一个ISO文件路径、例如: /path/to/linux.iso
或者提供一个ISO文件URL、例如: http://example.com/linux.iso
</arg>
<example cmd='init pxe /mnt/cdrom'>
初始化Pxe引导配置服务
</example>
"""
def run(self, params, args):
(args, mnt_src) = self.fillPostArgs(('mntPath', ))
if not mnt_src:
mnt_src = '/mnt/cdrom'
if not os.path.exists(mnt_src):
self.log.error(f"挂载点不存在: {mnt_src}")
self.abort("请挂载一下受支持的Linux系统ISO文件.")
# sunhpc
base_nm = self.conf.osinfo.basename
# /srv/repos/sunhpc
mnt_dst = os.path.join('/srv/repos', base_nm)
iface = self.conf.iface
ipaddr = self.conf.ipaddr
treeinfo = os.path.join(mnt_src, '.treeinfo')
if not os.path.exists(treeinfo):
self.log.error(f"不是一个有效或不受支持的Linux系统ISO挂载路径: {mnt_src}")
self.abort("重新挂载一下受支持的Linux系统ISO文件.")
info = {}
with open(treeinfo, 'r') as f:
for i in f.readlines():
tmp = i.strip().split('=', 1)
if len(tmp) == 2:
info[tmp[0].strip()] = tmp[1].strip()
osname = info.get('family', 'unknown').split()[0].lower()
if osname not in ['rocky', 'rhel', 'centos']:
self.log.error(f"不支持的系统类型: {osname}")
self.abort("请提供一个受支持的Linux系统ISO文件. 只支持Rocky, RHEL, CentOS系统.")
kernel = info.get('kernel', None)
if kernel is None:
self.log.error(f"没有找到内核文件: {kernel}")
self.abort("请提供一个支持 Pxe 的 vmlinuz 文件位置.")
initrd = info.get('initrd', None)
if initrd is None:
self.log.error(f"没有找到初始化文件: {initrd}")
self.abort("请提供一个支持 Pxe 的 initrd 文件位置.")
osvers = info.get('version', 'unknown')
http_head = self.conf.http.httpaddr
http_root = f"{http_head}/{base_nm}/{osname}/{osvers}"
http_ksfile = f"{http_root}/{base_nm}/{self.conf.pxe.ksfile}"
http_init = f"{http_root}/{base_nm}/{self.conf.pxe.initfile}"
lrepo_src = os.path.join(mnt_dst, osname, osvers)
if not os.path.exists(lrepo_src):
os.makedirs(lrepo_src)
# 复制 ISO 文件到目标目录
if not mnt_src.endswith('/'):
mnt_src = mnt_src + os.sep
if not lrepo_src.endswith('/'):
lrepo_src = lrepo_src + os.sep
self.rsync_copy(mnt_src, lrepo_src)
# 拷贝scripts目录到目标目录.
scripts_dst = mnt_dst + os.sep
self.rsync_copy(scripts_path, scripts_dst)
# 安装 httpd、dnsmasq 服务
#result = self.sh.execute("dnf install -y httpd dnsmasq")
wwwroot = self.conf.http.wwwroot
tftpboot = self.conf.tftp.tftpboot
vmlinuz = os.path.join(lrepo_src, kernel)
initrd = os.path.join(lrepo_src, initrd)
hostname = self.sh.execute("hostname", quiet=True).output().strip()
# 创建软连接
self.sh.execute(f"ln -s {mnt_dst} {wwwroot}", quiet=True)
# 创建tftpboot目录
pxe_boot_src = ipxe_path + os.sep
pxe_boot_dst = os.path.join(tftpboot, 'boot') + os.sep
self.sh.execute(f"mkdir -p {pxe_boot_dst}", quiet=True)
# 拷贝pxe引导文件到tftpboot目录
self.rsync_copy(pxe_boot_src, pxe_boot_dst)
# 配置 httpd 服务
httpd_conf = self.conf.http.config
self.create_httpd(hostname, base_nm, wwwroot, httpd_conf)
# 配置 dnsmasq 服务
start_ip = self.conf.dnsmasq.start_ip
end_ip = self.conf.dnsmasq.end_ip
addr_dns = self.conf.dnsmasq.dns
bios = self.conf.pxe.bios
ipxe = self.conf.pxe.ipxe
ipxe_sh = self.conf.pxe.ipxeboot
ipxe_http = f"{http_head}/{base_nm}/{ipxe_sh}"
dnsmasq_conf = self.conf.dnsmasq.config
self.create_dnsmasq(iface,
ipaddr, tftpboot, dnsmasq_conf,
start_ip, end_ip, addr_dns, bios, ipxe, ipxe_http)
# kickstart
ksfile_src = self.conf.pxe.ksfile
ksfile_dst = os.path.join(scripts_dst, ksfile_src)
self.create_ksfile(hostname, osvers, http_root, http_init, ksfile_dst)
# 配置 firewalld 服务
self.fw.load()
self.fw.add_port('80', 'tcp')
self.fw.add_port('443', 'tcp')
self.fw.save()
def create_ksfile(self, hostname, osvers, http_root, http_init, ksfile_dst):
ksfile_cnf = f"""
# Kickstart file for {hostname}
# Created by sunhpc
# Version: {osvers}
graphical
timezone Asia/Shanghai --utc
keyboard --xlayouts="us"
lang en_US.UTF-8
selinux --disabled
url --url='{http_root}'
network --bootproto=dhcp --device=link --ipv6=auto --activate
# autopart --type=lvm
# autopart --type=plain
# autopart --nohome
# autopart --noswap
%include /tmp/diskinfo
%packages
@^minimal-environment
@development
@standard
vim
wget
curl
autofs
nfs-utils
nfs4-acl-tools
sssd-nfs-idmap
%end
rootpw --plaintext "admin_b101"
user --name=dell --plaintext --password="admin_b101" --gecos="dell"
reboot
%pre --interpreter=/bin/bash
if [ -d /sys/firmware/efi ]; then
cat > /tmp/diskinfo <<EOF
clearpart --all --initlabel
part /boot/efi --fstype=efi --size=600 --fsoptions="umask=0077,shortname=uefi"
part /boot --fstype=xfs --size=1024
part pv.01 --size=1 --grow
volgroup vg_sys pv.01
logvol / --fstype=xfs --name=root --vgname=vg_sys --size=51200
logvol swap --fstype=swap --name=swap --vgname=vg_sys --size=4096
logvol /data --fstype=xfs --name=data --vgname=vg_sys --size=1 --grow
EOF
else
cat > /tmp/diskinfo <<EOF
clearpart --all --initlabel
part /boot --fstype=xfs --size=1024 --asprimary
part pv.01 --size=1 --grow
volgroup vg_sys pv.01
logvol / --fstype=xfs --name=root --vgname=vg_sys --size=51200
logvol swap --fstype=swap --name=swap --vgname=vg_sys --size=4096
logvol /data --fstype=xfs --name=data --vgname=vg_sys --size=1 --grow
EOF
fi
%end
%post
curl -s -o /tmp/init.sh {http_init}
chmod +x /tmp/init.sh
/tmp/init.sh
%end
%addon com_redhat_kdump --disable
%end
"""
with open(ksfile_dst, 'w') as f:
f.write(textwrap.dedent(ksfile_cnf))
def create_dnsmasq(self,
iface, ipaddr, tftpboot, dnsmasq_conf,
start_ip, end_ip, addr_dns, bios, ipxe, ipxe_http):
dnsmasq_tmpl = f"""
interface={iface}
bind-interfaces
port=0
enable-tftp
tftp-root={tftpboot}
# Gateway:3, DNS:6
dhcp-range={start_ip},{end_ip},12h
dhcp-option=3,{ipaddr}
dhcp-option=6,{addr_dns}
dhcp-match=set:ipxe,175
dhcp-match=set:bios,option:client-arch,0
dhcp-match=set:uefi,option:client-arch,7
dhcp-match=set:uefi,option:client-arch,9
dhcp-match=set:ipxe,option:user-class,ipxe
dhcp-boot=tag:bios,tag:!ipxe,{bios}
dhcp-boot=tag:uefi,tag:!ipxe,{ipxe}
dhcp-boot=tag:ipxe,{ipxe_http}
"""
with open(dnsmasq_conf, 'w') as f:
f.write(textwrap.dedent(dnsmasq_tmpl))
def create_httpd(self, hostname, base_nm, wwwroot, httpd_conf):
httpd_tmpl = f"""
Alias /{base_nm} {wwwroot}/{base_nm}
UseCanonicalName Off
ServerName {hostname}
<Directory "{wwwroot}/{base_nm}">
Options Indexes FollowSymLinks
AllowOverride None
Require all granted
EnableSendfile on
</Directory>
"""
with open(httpd_conf, 'w') as f:
f.write(textwrap.dedent(httpd_tmpl))

18
lib/sunhpc/sunhpc/configs.py
View File

@@ -31,6 +31,9 @@ class Config:
self._build_attributes()
self._initialized = True
def getConfFile(self):
return self._config_path
def _build_attributes(self):
"""构建属性访问"""
self._build_from_dict(self._data)
@@ -98,17 +101,26 @@ class Config:
# 如果属性不存在，返回 None（可选行为）
return None
def save(self, path: Optional[str] = None):
def save(self, path: Optional[str] = None, config_dict: Optional[Dict] = None):
"""保存配置到文件"""
save_path = path or self._config_path
if not save_path:
raise ValueError("未指定保存路径")
data = self.to_dict()
if config_dict:
data = config_dict
# 确保目录存在
Path(save_path).parent.mkdir(parents=True, exist_ok=True)
with open(save_path, 'w', encoding='utf-8') as f:
yaml.dump(self.to_dict(), f, default_flow_style=False, indent=2, allow_unicode=True)
yaml.dump(data, f,
indent=4,
sort_keys=False,
allow_unicode=True,
default_flow_style=False,
)
print(f"配置已保存到: {save_path}")

328
lib/sunhpc/sunhpc/firewalld.py Normal file
View File

@@ -0,0 +1,328 @@
"""
Firewalld 自定义服务管理工具
用于创建和修改 /etc/firewalld/services/ 目录下的自定义服务 XML 文件
"""
import os
from pathlib import Path
from xml.dom import minidom
import xml.etree.ElementTree as ET
from sunhpc.logs import Logs
slog = Logs(name="Firewd", show_time=True)
class FWManager:
"""Firewalld 服务文件管理器"""
# firewalld 服务目录路径
SYSTEM_SERVICES_DIR = Path('/usr/lib/firewalld/services')
CUSTOM_SERVICES_DIR = Path('/etc/firewalld/services')
def __init__(self, service_name):
"""
初始化服务管理器
Args:
service_name: 服务名称（不含 .xml 后缀）
"""
self.service_name = service_name
self.service_file = self.CUSTOM_SERVICES_DIR / f"{service_name}.xml"
self.xml_tree = None
self.root = None
def _ensure_custom_dir(self):
"""确保自定义服务目录存在"""
self.CUSTOM_SERVICES_DIR.mkdir(parents=True, exist_ok=True)
def _create_xml_structure(self):
"""创建基本的 XML 结构"""
# 创建根元素
self.root = ET.Element('service')
# 添加 short 元素
short = ET.SubElement(self.root, 'short')
short.text = self.service_name.capitalize()
# 添加 description 元素
description = ET.SubElement(self.root, 'description')
description.text = f"Custom service: {self.service_name}"
# 添加 port 元素（稍后添加）
# 添加 protocol 元素（稍后添加）
self.xml_tree = ET.ElementTree(self.root)
def _prettify_xml(self, elem):
"""
格式化 XML 输出
Args:
elem: XML 元素
Returns:
str: 格式化后的 XML 字符串
"""
rough_string = ET.tostring(elem, 'utf-8')
reparsed = minidom.parseString(rough_string)
return reparsed.toprettyxml(indent=' ')
def _save_xml(self):
"""保存 XML 文件到自定义服务目录"""
self._ensure_custom_dir()
# 格式化 XML
xml_str = self._prettify_xml(self.root)
# 写入文件
with open(self.service_file, 'w', encoding='utf-8') as f:
# 移除多余的 XML 声明（firewalld 不需要）
if xml_str.startswith('<?xml'):
xml_str = xml_str.split('\n', 1)[1]
f.write(xml_str)
# 设置正确的权限（644）
os.chmod(self.service_file, 0o644)
def load(self):
"""
加载现有服务文件或创建新文件
Returns:
bool: 是否成功加载/创建
"""
if self.service_file.exists():
# 加载现有文件
try:
self.xml_tree = ET.parse(self.service_file)
self.root = self.xml_tree.getroot()
slog.info(f"✓ 已加载现有服务: {self.service_file}")
return True
except ET.ParseError as e:
slog.error(f"✗ 解析 XML 文件失败: {e}")
return False
else:
# 创建新文件
slog.info(f"服务不存在，将创建新服务: {self.service_name}")
self._create_xml_structure()
return True
def set_short_description(self, short_text):
"""
设置服务简短描述
Args:
short_text: 简短描述文本
"""
short_elem = self.root.find('short')
if short_elem is None:
short_elem = ET.SubElement(self.root, 'short')
short_elem.text = short_text
slog.info(f"✓ 设置简短描述: {short_text}")
def set_description(self, desc_text):
"""
设置服务详细描述
Args:
desc_text: 详细描述文本
"""
desc_elem = self.root.find('description')
if desc_elem is None:
desc_elem = ET.SubElement(self.root, 'description')
desc_elem.text = desc_text
slog.info(f"✓ 设置详细描述: {desc_text}")
def add_port(self, port, protocol='tcp'):
"""
添加端口规则
Args:
port: 端口号或端口范围（例如：'8080' 或 '3000-3010'）
protocol: 协议类型（tcp/udp/sctp/dccp）
"""
# 查找是否已存在相同的端口规则
for port_elem in self.root.findall('port'):
if port_elem.get('port') == str(port) and port_elem.get('protocol') == protocol:
slog.info(f"⚠ 跳过,已存在, 端口 {port}/{protocol}")
return
# 添加新端口
port_elem = ET.SubElement(self.root, 'port')
port_elem.set('port', str(port))
port_elem.set('protocol', protocol)
slog.info(f"✓ 添加端口: {port}/{protocol}")
def remove_port(self, port, protocol='tcp'):
"""
移除端口规则
Args:
port: 端口号或端口范围
protocol: 协议类型
"""
for port_elem in self.root.findall('port'):
if port_elem.get('port') == str(port) and port_elem.get('protocol') == protocol:
self.root.remove(port_elem)
slog.info(f"✓ 移除端口: {port}/{protocol}")
return
slog.info(f"⚠ 未找到端口: {port}/{protocol}")
def add_protocol(self, protocol_name):
"""
添加协议规则（IP 协议，不是传输层协议）
Args:
protocol_name: 协议名称（如 ipv6-icmp, icmp, igmp 等）
"""
for proto_elem in self.root.findall('protocol'):
if proto_elem.get('name') == protocol_name:
slog.info(f"⚠ 协议 {protocol_name} 已存在，跳过")
return
proto_elem = ET.SubElement(self.root, 'protocol')
proto_elem.set('name', protocol_name)
slog.info(f"✓ 添加协议: {protocol_name}")
def remove_protocol(self, protocol_name):
"""
移除协议规则
Args:
protocol_name: 协议名称
"""
for proto_elem in self.root.findall('protocol'):
if proto_elem.get('name') == protocol_name:
self.root.remove(proto_elem)
slog.info(f"✓ 移除协议: {protocol_name}")
return
slog.info(f"⚠ 未找到协议: {protocol_name}")
def add_module(self, module_name):
"""
添加内核模块依赖
Args:
module_name: 模块名称（如 nf_conntrack_ftp）
"""
for module_elem in self.root.findall('module'):
if module_elem.get('name') == module_name:
slog.info(f"⚠ 模块 {module_name} 已存在，跳过")
return
module_elem = ET.SubElement(self.root, 'module')
module_elem.set('name', module_name)
slog.info(f"✓ 添加模块: {module_name}")
def remove_module(self, module_name):
"""
移除内核模块依赖
Args:
module_name: 模块名称
"""
for module_elem in self.root.findall('module'):
if module_elem.get('name') == module_name:
self.root.remove(module_elem)
slog.info(f"✓ 移除模块: {module_name}")
return
slog.info(f"⚠ 未找到模块: {module_name}")
def get_ports(self):
"""获取所有端口配置"""
ports = []
for port_elem in self.root.findall('port'):
ports.append({
'port': port_elem.get('port'),
'protocol': port_elem.get('protocol')
})
return ports
def get_protocols(self):
"""获取所有协议配置"""
protocols = []
for proto_elem in self.root.findall('protocol'):
protocols.append(proto_elem.get('name'))
return protocols
def display_info(self):
"""显示服务信息"""
slog.info("\n" + "="*50)
slog.info(f"服务名称: {self.service_name}")
slog.info(f"配置文件: {self.service_file}")
short = self.root.find('short')
if short is not None and short.text:
slog.info(f"简短描述: {short.text}")
desc = self.root.find('description')
if desc is not None and desc.text:
slog.info(f"详细描述: {desc.text}")
ports = self.get_ports()
if ports:
slog.info("\n端口规则:")
for port in ports:
slog.info(f" - {port['port']}/{port['protocol']}")
protocols = self.get_protocols()
if protocols:
print("\n协议规则:")
for proto in protocols:
print(f" - {proto}")
modules = [m.get('name') for m in self.root.findall('module')]
if modules:
slog.info("\n内核模块:")
for module in modules:
slog.info(f" - {module}")
slog.info("="*50 + "\n")
def save(self):
"""保存修改"""
self._save_xml()
def main():
"""示例用法"""
# 创建或修改名为 'myapp' 的服务
manager = FWManager('myapp')
# 加载或创建服务文件
if not manager.load():
return
# 设置描述信息
manager.set_short_description('My Application Service')
manager.set_description('Custom service for my web application with API and admin ports')
# 添加端口
manager.add_port(8080, 'tcp') # Web 服务端口
manager.add_port(8443, 'tcp') # HTTPS 替代端口
manager.add_port(9090, 'tcp') # API 端口
manager.add_port(3000, 'tcp') # 管理界面端口
manager.add_port(5000-5010, 'tcp') # 端口范围示例
manager.add_port(9000, 'udp') # UDP 端口
# 添加协议（如果需要）
# manager.add_protocol('icmp')
# 添加内核模块（如果需要，例如 FTP 辅助模块）
# manager.add_module('nf_conntrack_ftp')
# 显示当前配置
manager.display_info()
# 保存修改
manager.save()
# 示例：移除端口
# manager.remove_port(5000-5010, 'tcp')
# manager.save()
slog.info("\n✅ 操作完成！")
slog.info("提示：")
slog.info("1. 运行 'sudo firewall-cmd --reload' 重新加载防火墙")
slog.info("2. 使用 'sudo firewall-cmd --get-services' 查看所有服务")
slog.info("3. 使用 'sudo firewall-cmd --info-service=myapp' 查看服务详情")
slog.info(f"4. 使用 'sudo firewall-cmd --add-service={manager.service_name}' 启用服务")
if __name__ == '__main__':
main()

10
lib/sunhpc/sunhpc/main.py
View File

@@ -5,6 +5,7 @@ import yaml
import sunhpc
import signal
import syslog
import traceback
import sunhpc.util
import sunhpc.error as error
@@ -70,7 +71,6 @@ def _main(argv=None):
command.isRootUser() or command.isApacheUser()):
os.system('sudo %s' % ' '.join(sys.argv))
else:
#import sunhpc
try:
command.runWrapper(name, args[i:])
text = command.getText()
@@ -81,8 +81,10 @@ def _main(argv=None):
print (text)
print ()
except CommandError as e:
print ('Error: ', e)
print (command.usage())
print ('Error: ', e)
#slog.info(command.usage())
#slog.error(str(e))
return 1
config_file = '/etc/sunhpc/sunhpc.yaml'
@@ -107,9 +109,9 @@ def main(argv=None):
tty.error("Keyboard interrupt.")
return signal.SIGINT.value
except AttributeError:
except AttributeError as e:
sys.stderr.write("\n")
tty.error("Attribute error.")
slog.error(str(e))
return 1
except SystemExit as e:

14
lib/sunhpc/sunhpc/paths.py
View File

@@ -22,8 +22,20 @@ sunhpc_root = prefix
#: bin directory in the sunhpc prefix
bin_path = os.path.join(prefix, "bin")
#: var directory in the sunhpc
var_path = os.path.join(prefix, "var")
#: pkgs directory in the sunhpc
pkgs_path = os.path.join(var_path, "pkgs")
#: ipxe directory in the sunhpc
ipxe_path = os.path.join(pkgs_path, "ipxe")
#: scripts directory in the sunhpc
scripts_path = os.path.join(pkgs_path, "scripts")
#: The sunhpc script itself
sunhpc_script = os.path.join(bin_path, "sunhpc")
sunhpc_command = os.path.join(bin_path, "sunhpc")
#: The sbang script in the sunhpc installation
sbang_script = os.path.join(bin_path, "sbang")

611
lib/sunhpc/sunhpc/utils/shells.py Normal file
View File

@@ -0,0 +1,611 @@
import sys
import time
import shlex
import threading
import subprocess
from collections import deque
from typing import Optional, Union, List, Dict, Any, Callable
class ShellResult:
"""封装命令执行结果的类，支持布尔判断、输出获取、错误获取和返回码获取"""
def __init__(self,
returncode: int,
stdout: str,
stderr: str,
metadata: Optional[Dict[str, Any]] = None):
self._returncode = returncode
self._stdout = stdout
self._stderr = stderr
self._metadata = metadata or {} # 存储额外信息，如包名、版本等
def __bool__(self):
"""支持 if result: 直接判断命令是否成功"""
return self._returncode == 0
def output(self) -> str:
"""获取命令的标准输出"""
return self._stdout
def error(self) -> str:
"""获取命令的标准错误"""
return self._stderr
def code(self) -> int:
"""获取命令的返回码"""
return self._returncode
def output_lines(self) -> List[str]:
"""返回输出行列表，过滤空行"""
return [line for line in self._stdout.splitlines() if line]
def error_lines(self) -> List[str]:
return [line for line in self._stderr.splitlines() if line]
def get_metadata(self, key: str, default=None):
"""获取元数据"""
return self._metadata.get(key, default)
def set_metadata(self, key: str, value: Any):
"""设置元数据"""
self._metadata[key] = value
class ShellExecutor:
"""
Shell命令执行器基类,提供通用的命令执行功能
子类可重写方法以实现特定命令的定制行为
"""
def __init__(self,
show_animation: bool = True,
animation_chars: str = '|/-\\',
show_last_line: bool = True,
max_display_lines: int = 1
):
"""
初始化执行器
参数:
show_animation : 是否显示动态动画
animation_chars : 动画字符集
show_last_line : 是否显示最后一行
max_display_lines : 最大显示行数，超过则省略号显示
"""
self.show_animation = show_animation
self.animation_chars = animation_chars
self.show_last_line = show_last_line
self.max_display_lines = max_display_lines
self._line_count = 0
self._last_lines = deque(maxlen=max_display_lines)
def pre_execute_hook(self, cmd: Union[str, List[str]]) -> Union[str, List[str]]:
"""
执行前的钩子函数，可以修改命令或做预处理
子类可重写此方法
"""
return cmd
def post_execute_hook(self, result: ShellResult, cmd: Union[str, List[str]]) -> ShellResult:
"""
执行后的钩子函数，可以处理结果、提取信息等
子类可重写此方法
"""
return result
def get_display_message(self, cmd: Union[str, List[str]]) -> str:
"""
获取执行时显示的消息
子类可重写以显示不同的提示信息
"""
return "Running..."
def get_completion_message(self, cmd: Union[str, List[str]], result: ShellResult) -> str:
"""
获取完成时显示的消息
子类可重写以显示不同的完成信息
"""
if self._line_count > 0:
return f"Completed! (lines: {self._line_count})"
else:
return "Completed!"
def get_realtime_info(self, line: str) -> str:
"""
获取实时显示的信息（可被子类重写以显示不同的信息格式）
参数:
line: 最新读取的一行输出
返回:
要显示的附加信息字符串
"""
# 默认显示最后一行内容（截断过长内容）
if self.show_last_line and line:
# 限制行长度，避免显示过长
max_length = 80
if len(line) > max_length:
line = line[:max_length-3] + "..."
return f" [{line}]"
return ""
def update_realtime_output(self, line: str):
"""
更新实时输出信息（可被子类重写以处理特殊输出）
参数:
line: 最新读取的一行输出
"""
self._last_lines.append(line.rstrip('\n\r'))
def parse_output_for_metadata(self,
result: ShellResult,
cmd: Union[str, List[str]]) -> Dict[str, Any]:
"""
解析输出提取元数据
子类可重写此方法
"""
return {}
def _RRread_output_thread(self, process):
"""读取输出的线程函数"""
for line in iter(process.stdout.readline, ''):
if line:
self._line_count += 1
process.stdout.close()
def _read_output_thread(self, process, is_stderr=False):
"""读取输出的线程函数"""
if is_stderr:
# 错误输出只读取，不显示在动画中
for line in iter(process.stderr.readline, ''):
if line:
pass # 可以在这里处理stderr
process.stderr.close()
else:
# 标准输出需要计数和显示
for line in iter(process.stdout.readline, ''):
if line:
self._line_count += 1
# 更新实时输出
self.update_realtime_output(line)
# 可以添加一个回调，让子类实时处理每一行
self.on_output_line(line)
process.stdout.close()
def _display_animation(self, process):
"""显示动画的函数"""
if not self.show_animation:
return
idx = 0
while process.poll() is None:
spin_char = self.animation_chars[idx % len(self.animation_chars)]
if self._line_count > 0:
display = f"\r{spin_char} {self.get_display_message(process.args)} (lines: {self._line_count})"
else:
display = f"\r{spin_char} {self.get_display_message(process.args)}"
sys.stdout.write(display)
sys.stdout.flush()
idx += 1
time.sleep(0.1)
# 清除当前行并显示完成信息
completion_msg = self.get_completion_message(process.args, ShellResult(0, "", ""))
sys.stdout.write(f"\r✓ {completion_msg} \n")
sys.stdout.flush()
def on_output_line(self, line: str):
"""
当有新的输出行时的回调函数（可被子类重写）
参数:
line: 新输出的行
"""
pass
def execute(self,
cmd: Union[str, List[str]],
t_cmd: Optional[Union[str, List[str]]] = None,
quiet: bool = False,
timeout: Optional[int] = None,
show_animation: bool = True,
show_real_time: bool = True,
) -> ShellResult:
"""
执行shell命令
参数:
cmd : 要执行的字符串或列表
t_cmd : 可选，执行完成后的测试命令
timeout : 超时时间（秒）
show_real_time : 是否显示实时输出(仅当show_animation为True时有效)
返回:
ShellResult 实例
"""
# 执行预处理钩子
processed_cmd = self.pre_execute_hook(cmd)
# 转换命令为列表形式
if isinstance(processed_cmd, str):
cmd_list = shlex.split(processed_cmd)
else:
cmd_list = processed_cmd
# 重置计数器
self._line_count = 0
self._last_lines.clear()
# 启动子进程
process = subprocess.Popen(
cmd_list,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1
)
# 用于存储输出
stdout_lines = []
stderr_lines = []
# 创建读取线程
def collect_stdout():
for line in iter(process.stdout.readline, ''):
if line:
stdout_lines.append(line)
self._line_count += 1
self.update_realtime_output(line)
self.on_output_line(line)
process.stdout.close()
def collect_stderr():
for line in iter(process.stderr.readline, ''):
if line:
stderr_lines.append(line)
process.stderr.close()
stdout_thread = threading.Thread(target=collect_stdout)
stderr_thread = threading.Thread(target=collect_stderr)
stdout_thread.daemon = True
stderr_thread.daemon = True
stdout_thread.start()
stderr_thread.start()
# 显示动画
if self.show_animation:
idx = 0
last_display = ""
while process.poll() is None:
spin_char = self.animation_chars[idx % len(self.animation_chars)]
# 构建显示内容
if self._line_count > 0:
base_display = f"{spin_char} {self.get_display_message(cmd_list)} (lines: {self._line_count})"
# 添加实时输出信息
if show_real_time and self._last_lines:
# 显示最后一行内容
last_line = list(self._last_lines)[-1]
realtime_info = self.get_realtime_info(last_line)
display = f"\r{base_display}{realtime_info}"
else:
display = f"\r{base_display}"
else:
display = f"\r{spin_char} {self.get_display_message(cmd_list)}"
# 只在内容变化时更新，减少闪烁
if display != last_display or self._line_count != last_line_count:
sys.stdout.write(f'\r{display}\033[K')
sys.stdout.flush()
last_display = display
last_line_count = self._line_count
idx += 1
time.sleep(0.1)
# 等待线程结束
stdout_thread.join(timeout=1)
stderr_thread.join(timeout=1)
# 清除当前行
self._clear_line()
# 显示完成信息
if not quiet:
completion_msg = self.get_completion_message(cmd_list, ShellResult(0, "", ""))
# 清除当前行并显示完成信息
sys.stdout.write(f"\r✓ {completion_msg}\n")
sys.stdout.flush()
else:
# 不显示动画时，直接等待进程结束
process.wait()
stdout_thread.join()
stderr_thread.join()
# 合并输出
stdout = ''.join(stdout_lines)
stderr = ''.join(stderr_lines)
# 如果有测试命令，执行测试命令
if t_cmd is not None:
if isinstance(t_cmd, str):
test_cmd_list = shlex.split(t_cmd)
else:
test_cmd_list = t_cmd
test_process = subprocess.run(
test_cmd_list,
input=stdout,
capture_output=True,
text=True
)
final_returncode = 0 if test_process.returncode == 0 else 1
final_stderr = stderr + test_process.stderr
else:
final_returncode = process.returncode
final_stderr = stderr
# 创建结果对象
result = ShellResult(final_returncode, stdout, final_stderr)
# 解析元数据（如果子类需要）
metadata = self.parse_output_for_metadata(result, processed_cmd)
for key, value in metadata.items():
result.set_metadata(key, value)
# 执行后置钩子
result = self.post_execute_hook(result, processed_cmd)
return result
def _clear_line(self):
"""彻底清除当前行"""
# 先移动到行首，然后清除到行尾
sys.stdout.write('\r\033[K') # \033[K 清除从光标到行尾的内容
sys.stdout.flush()
class DnfExecutor(ShellExecutor):
"""针对dnf命令的专用执行器"""
def get_display_message(self, cmd: Union[str, List[str]]) -> str:
"""显示dnf特定消息"""
cmd_str = ' '.join(cmd) if isinstance(cmd, list) else cmd
if 'install' in cmd_str:
return "Installing packages..."
elif 'remove' in cmd_str or 'erase' in cmd_str:
return "Removing packages..."
elif 'update' in cmd_str:
return "Updating packages..."
elif 'search' in cmd_str:
return "Searching packages..."
else:
return "DNF Running..."
def parse_output_for_metadata(self, result: ShellResult, cmd: Union[str, List[str]]) -> Dict[str, Any]:
"""从dnf输出中提取包信息"""
metadata = {}
cmd_str = ' '.join(cmd) if isinstance(cmd, list) else cmd
# 提取安装的包名
if 'install' in cmd_str or 'remove' in cmd_str:
packages = []
for line in result.output_lines():
if 'Installing:' in line or 'Removing:' in line:
# 提取包名
parts = line.split()
for part in parts:
if '.x86_64' in part or '.noarch' in part:
packages.append(part)
if packages:
metadata['packages'] = packages
metadata['package_count'] = len(packages)
return metadata
def post_execute_hook(self, result: ShellResult, cmd: Union[str, List[str]]) -> ShellResult:
"""dnf执行后的额外处理"""
if result and 'install' in str(cmd):
# 如果安装成功，提取安装的包列表
packages = result.get_metadata('packeds', [])
if packages:
result.set_metadata('installed_packages', packages)
return result
class RpmExecutor(ShellExecutor):
"""针对rpm命令的专用执行器"""
def __init__(self, query_format: Optional[str] = None, **kwargs):
"""
初始化rpm执行器
参数:
query_format: 查询格式，如 "%{NAME} %{VERSION}"
"""
super().__init__(**kwargs)
self.query_format = query_format
def pre_execute_hook(self, cmd: Union[str, List[str]]) -> Union[str, List[str]]:
"""预处理rpm命令"""
if isinstance(cmd, str):
cmd_parts = shlex.split(cmd)
else:
cmd_parts = cmd.copy()
# 如果是查询命令且指定了格式，添加--qf参数
if self.query_format and len(cmd_parts) > 1 and cmd_parts[0] == 'rpm':
if cmd_parts[1] in ['-q', '--query', '-qa', '--query', '-qi']:
# 插入格式参数
if '--qf' not in cmd_parts and '--queryformat' not in cmd_parts:
cmd_parts.insert(2, '--qf')
cmd_parts.insert(3, self.query_format)
return cmd_parts
def get_display_message(self, cmd: Union[str, List[str]]) -> str:
"""显示rpm特定消息"""
cmd_str = ' '.join(cmd) if isinstance(cmd, list) else cmd
if '-i' in cmd_str and 'install' in cmd_str:
return "Installing RPM packages..."
elif '-e' in cmd_str:
return "Erasing RPM packages..."
elif '-q' in cmd_str:
return "Querying RPM database..."
elif '-U' in cmd_str or '-F' in cmd_str:
return "Upgrading RPM packages..."
else:
return "RPM Running..."
def parse_output_for_metadata(self, result: ShellResult, cmd: Union[str, List[str]]) -> Dict[str, Any]:
"""从rpm输出中提取包信息"""
metadata = {}
cmd_str = ' '.join(cmd) if isinstance(cmd, list) else cmd
# 提取查询结果的包名和版本
if '-q' in cmd_str and result and result.output():
packages = []
for line in result.output_lines():
if line.strip():
parts = line.split()
if len(parts) >= 2:
packages.append({
'name': parts[0],
'version': parts[1] if len(parts) > 1 else ''
})
if packages:
metadata['packages'] = packages
metadata['package_count'] = len(packages)
return metadata
class SystemdExecutor(ShellExecutor):
"""针对systemctl命令的专用执行器"""
def get_display_message(self, cmd: Union[str, List[str]]) -> str:
"""显示systemctl特定消息"""
cmd_str = ' '.join(cmd) if isinstance(cmd, list) else cmd
if 'start' in cmd_str:
return "Starting service..."
elif 'stop' in cmd_str:
return "Stopping service..."
elif 'restart' in cmd_str:
return "Restarting service..."
elif 'status' in cmd_str:
return "Checking service status..."
elif 'enable' in cmd_str:
return "Enabling service..."
elif 'disable' in cmd_str:
return "Disabling service..."
else:
return "Systemctl Running..."
def parse_output_for_metadata(self, result: ShellResult, cmd: Union[str, List[str]]) -> Dict[str, Any]:
"""从systemctl输出中提取服务状态"""
metadata = {}
cmd_str = ' '.join(cmd) if isinstance(cmd, list) else cmd
# 提取服务状态
if 'status' in cmd_str and result:
for line in result.output_lines():
if 'Active:' in line:
status = line.split('Active:')[1].strip()
metadata['service_status'] = status
if 'active' in status:
metadata['is_active'] = True
else:
metadata['is_active'] = False
return metadata
# ========== 使用示例 ==========
if __name__ == "__main__":
# 示例1：使用基础执行器
print("=== 基础执行器 ===")
executor = ShellExecutor()
result = executor.execute("ls -la")
print(f"命令成功: {bool(result)}")
print(f"输出行数: {len(result.output_lines())}")
print(f"前3行输出:\n{chr(10).join(result.output_lines()[:3])}")
print("-" * 50)
# 示例2：使用dnf执行器（如果有dnf命令）
print("=== DNF执行器 ===")
dnf = DnfExecutor()
# 搜索命令示例（不会实际修改系统）
result = dnf.execute("dnf search python3", show_animation=False) # 测试时关闭动画
print(f"搜索完成: {bool(result)}")
if result.get_metadata('package_count', 0) > 0:
print(f"找到 {result.get_metadata('package_count')} 个包")
# 示例3：使用rpm执行器
print("=== RPM执行器 ===")
rpm = RpmExecutor(query_format="%{NAME} %{VERSION}")
# 查询已安装的包（前5个）
result = rpm.execute("rpm -qa | head -5", show_animation=False)
print(f"查询完成: {bool(result)}")
print("前5个包:")
for line in result.output_lines()[:5]:
print(f" {line}")
# 示例4：使用systemd执行器
print("=== Systemd执行器 ===")
systemd = SystemdExecutor()
result = systemd.execute("systemctl status sshd", show_animation=False)
print(f"服务状态: {result.get_metadata('service_status', 'unknown')}")
print(f"服务活跃: {result.get_metadata('is_active', False)}")
# 示例5：带测试命令的示例
print("=== 带测试命令 ===")
executor = ShellExecutor()
result = executor.execute("ls -la", t_cmd="grep README")
print(f"包含README文件: {bool(result)}")
if not result:
print(f"未找到README文件")
# 示例6：如何扩展自己的专用执行器
print("=== 自定义扩展示例 ===")
class GitExecutor(ShellExecutor):
"""Git命令专用执行器"""
def get_display_message(self, cmd: Union[str, List[str]]) -> str:
cmd_str = ' '.join(cmd) if isinstance(cmd, list) else cmd
if 'status' in cmd_str:
return "Checking git status..."
elif 'pull' in cmd_str:
return "Pulling from remote..."
elif 'push' in cmd_str:
return "Pushing to remote..."
elif 'commit' in cmd_str:
return "Committing changes..."
else:
return "Git Running..."
def parse_output_for_metadata(self, result: ShellResult, cmd: Union[str, List[str]]) -> Dict[str, Any]:
"""提取git信息"""
metadata = {}
cmd_str = ' '.join(cmd) if isinstance(cmd, list) else cmd
if 'status' in cmd_str:
# 统计未跟踪、修改、新增的文件
untracked = 0
modified = 0
for line in result.output_lines():
if 'Untracked files:' in line:
untracked = 1
elif 'modified:' in line:
modified += 1
metadata['untracked'] = untracked > 0
metadata['modified_count'] = modified
return metadata
git = GitExecutor()
result = git.execute("git status", show_animation=False)
if result:
print(f"有未跟踪文件: {result.get_metadata('untracked', False)}")
print(f"修改文件数: {result.get_metadata('modified_count', 0)}")