查询程序页面示例：
一.架构概述：
服务器系统环境是centos的，客户端是windows，除图形界面是用python写的封装成exe在windows上直接打开使用，其他的服务都是搭建在centos上。整个系统实现了通过systemd后台自动启动，用户在图形界面输入IP后，客户端发起请求到后端API，后端调用封装的数据库查询方法获取资产信息，经过格式化后返回并在界面动态生成可调列宽的结构化表格进行展示
数据库存储：自建数据库通过Python脚本导入台账基础信息，提供给其他服务查询。
后端API服务(server_api.py)：基于Flask框架提供RESTful接口，实现对MySQL数据库的操作，包括物理服务器和虚拟机信息的查询。
数据库连接（db_handler.py）：封装MySQL连接与查询逻辑，确保数据库连接的可靠性和操作的效率。
图形界面（Tkinter应用）：用户友好型桌面程序，支持IP输入、查询请求，并以结构化表格动态展示查询结果。
二.数据库存储
如何自建数据库，在此处不多赘述，自行查询资料搭建，此处提供两种不同类型主机的导入脚本，可酌情修改，主要还是字段方面的差异。
完成数据库的部署后，在当前目录新建个文件做环境变量，后续很多脚本要引用
注：我所有的脚本路径都在/mysql/excel 下，这个可以自行调整
配置文件（config.py）
import os
from dotenv import load_dotenvload_dotenv() # 从.env文件加载环境变量DB_CONFIG = {'host': os.getenv('DB_HOST', 'localhost'),'user': os.getenv('DB_USER', 'tmpuser'),'password': os.getenv('DB_PASSWORD', 'pswd1234567'),'database': os.getenv('DB_NAME', 'server_inventory')
}API_KEY = os.getenv('API_KEY', 'keybaidu123')接下来就需要进行数据库导入台账信息了。
虚拟机示例表格
导入脚本（引用了DB的环境变量）import_vm_excel_to_db.py：
import pandas as pd
import pymysql
import sys
import os
import logging
import config# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s [%(levelname)s] %(message)s',handlers=[logging.StreamHandler(),logging.FileHandler('import_virtual_machines.log', encoding='utf-8')]
)def ip_str_to_int(ip_str):"""将点分十进制IP转换成整数"""if pd.isnull(ip_str) or ip_str == '':return Nonetry:parts = [int(p) for p in ip_str.strip().split('.')]if len(parts) == 4 and all(0 <= p <= 255 for p in parts):return (parts[0]<<24) + (parts[1]<<16) + (parts[2]<<8) + parts[3]else:return Noneexcept:return Noneclass DatabaseHandler:"""数据库操作封装类"""def __init__(self):db_conf = config.DB_CONFIGself.conn = pymysql.connect(host=db_conf['host'],port=3306,user=db_conf['user'],password=db_conf['password'],database=db_conf['database'],charset='utf8mb4')self.cursor = self.conn.cursor()def execute(self, query, params):self.cursor.execute(query, params)def commit(self):self.conn.commit()def close(self):self.cursor.close()self.conn.close()def main(excel_file_path):# 1. 读取Exceltry:df = pd.read_excel(excel_file_path, engine='openpyxl')except Exception as e:logging.error(f"读取Excel失败：{e}")return False# 2. 定义列名映射（根据你的Excel列重写）column_mapping = {'业务IP': 'business_ip','名称': 'vm_name','分类': 'category','服务器类型': 'server_type','使用状态': 'status','业务系统': 'business_system','应用': 'application','环境': 'environment','服务': 'service','责任人A': 'responsible_person_a','责任人B': 'responsible_person_b','系统分级': 'system_level','宿主机': 'host_machine_ip','置备的空间': 'provisioned_space','客户机操作系统': 'guest_os','存储配置': 'storage_config','内存大小': 'memory','CPU': 'cpu','网卡': 'network_cards','DNS 名称': 'dns_name','UUID': 'uuid'}# 3. 重命名列，使得列名与映射一致df.rename(columns=column_mapping, inplace=True)# 4. 处理空值：将NaN和空字符串均转成Nonedf = df.where(pd.notnull(df), None)df = df.replace('', None)# 5. 转换IP地址为整数if 'business_ip' in df.columns:df['business_ip_int'] = df['business_ip'].apply(ip_str_to_int)if 'host_machine_ip' in df.columns:df['host_machine_ip_int'] = df['host_machine_ip'].apply(ip_str_to_int)# 6. 连接数据库db = DatabaseHandler()# 7. SQL 插入语句insert_query = """INSERT INTO virtual_machines (business_ip, vm_name, category, server_type, status,business_system, application, environment, service,responsible_person_a, responsible_person_b, system_level,host_machine_ip, provisioned_space, guest_os, storage_config,memory, cpu, network_cards, dns_name, uuid) VALUES (%s, %s, %s, %s, %s,%s, %s, %s, %s,%s, %s, %s,%s, %s, %s, %s,%s, %s, %s, %s, %s)"""total_rows = len(df)success_count = 0for index, row in df.iterrows():# 转换nan为None（已在前面处理，但保险起见）row = row.replace({float('nan'): None})# 构建参数params = (row.get('business_ip_int'),row.get('vm_name'),row.get('category'),row.get('server_type'),row.get('status'),row.get('business_system'),row.get('application'),row.get('environment'),row.get('service'),row.get('responsible_person_a'),row.get('responsible_person_b'),row.get('system_level'),row.get('host_machine_ip_int'),row.get('provisioned_space'),row.get('guest_os'),row.get('storage_config'),row.get('memory'),row.get('cpu'),row.get('network_cards'),row.get('dns_name'),row.get('uuid'))# 业务IP为空跳过if params[0] is None:logging.warning(f"第{index+2}行的业务IP为空或无效，已跳过")continuetry:db.execute(insert_query, params)success_count += 1except Exception as e:logging.error(f"第{index+2}行插入失败：{e}")try:db.commit()except Exception as e:logging.error(f"提交事务失败：{e}")finally:db.close()logging.info(f"导入完成：成功导入 {success_count}/{total_rows} 条记录")return success_count == total_rowsif __name__ == '__main__':if len(sys.argv) != 2:print("用法：python3 import_virtual_machines.py <Excel文件路径>")sys.exit(1)excel_path = sys.argv[1]if not os.path.isfile(excel_path):print(f"文件不存在：{excel_path}")sys.exit(1)result = main(excel_path)sys.exit(0 if result else 1)自行把虚拟机的xlsx上传到centos服务器上，然后切换到脚本和要导入的excel目录，执行以下命令：
python3 ./import_vm_excel_to_db.py 虚拟机.xlsx
物理机机示例表格
导入脚本（数据库链接信息在脚本中）import_excel_to_db.py：
import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy.engine import URL
import sys
import socket
import struct# 将 IP 地址字符串转换为整数
def ip_to_int(ip):return struct.unpack("!I", socket.inet_aton(ip))[0]def import_physical_servers(excel_file):try:# 读取Excel文件df = pd.read_excel(excel_file)# 检查必要列是否存在required_columns = ['业务IP', '服务器类型']for col in required_columns:if col not in df.columns:raise ValueError(f"缺少必要列: {col}")# 重命名列名以匹配数据库字段column_mapping = {'机柜号': 'cabinet_number','机架位': 'rack_position','类别缩写': 'category_abbreviation','资产类型': 'asset_type','资产编码': 'asset_code','服务器类型': 'server_type','设备名称': 'device_name','设备型号': 'device_model','设备CPU': 'device_cpu','设备内存': 'device_memory','设备存储1': 'storage1','设备存储2': 'storage2','设备存储3': 'storage3','系统RAID': 'system_raid','数据RAID': 'data_raid','配置类型': 'configuration_type','序列号': 'serial_number','价值': 'value','采购时间': 'purchase_date','维保期': 'warranty_period','业务IP': 'business_ip','管理IP': 'management_ip','责任人A': 'responsible_person_a','责任人B': 'responsible_person_b','系统分级': 'system_level','业务系统': 'business_system','系统类别': 'system_category'}df.rename(columns=column_mapping, inplace=True)# 将 IP 地址转换为整数df['business_ip'] = df['business_ip'].apply(ip_to_int)df['management_ip'] = df['management_ip'].apply(ip_to_int)# 创建连接URL（安全处理密码中的特殊字符）connection_url = URL.create(drivername="mysql+pymysql",username="tmpuser",password="123123123",host="localhost",database="server_inventory",query={"charset": "utf8mb4"})# 连接数据库engine = create_engine(connection_url)# 导入数据df.to_sql('physical_servers', engine, if_exists='append', index=False)print(f"成功导入 {len(df)} 条物理服务器记录")except Exception as e:print(f"导入失败: {str(e)}")sys.exit(1)if __name__ == "__main__":if len(sys.argv) != 2:print("用法: python import_excel_to_db.py <Excel文件路径>")sys.exit(1)excel_file = sys.argv[1]import_physical_servers(excel_file)
自行把物理机机的xlsx上传到centos服务器上，然后切换到脚本和要导入的excel目录，执行以下命令：
python3 ./import_excel_to_db.py 物理机.xlsx
三.后端API服务设计（server_api.py）
后端API主要提供两个功能接口：
单点查询接口：/api/query
自动优先查询物理服务器，再查询虚拟机，返回详细信息
关联查询接口：/api/query/related
根据指定IP，查询所有关联的物理机和虚拟机列表
作用与流程
功能：GUI客户端通过HTTP POST请求调用该API，发送请求后，接受客户端HTTP请求（包括IP和类型参数），后端通过db_handler.py提供的方法访问数据库，查询数据库中对应的服务器信息（物理机或虚拟机），然后以JSON格式返回，在返回结果前，将数据库中的Decimal类型转换为浮点数，并将存储的二进制IP转换为易读的字符串，出错或未查到数据时，返回相应提示。
server_api.py（引用了DB的环境变量）脚本内容：
from flask import Flask, request, jsonify
from db_handler import DatabaseHandler
import config
import logging
import decimalapp = Flask(__name__)
db = DatabaseHandler(config.DB_CONFIG)logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)def convert_decimal(obj):if isinstance(obj, dict):return {k: convert_decimal(v) for k, v in obj.items()}elif isinstance(obj, list):return [convert_decimal(i) for i in obj]elif isinstance(obj, decimal.Decimal):return float(obj)else:return obj@app.route('/api/query', methods=['POST'])
def query_server():try:data = request.jsonip = data.get('ip', '').strip()server_type = data.get('type', '').strip().lower() # 'physical' or 'virtual'if not ip:return jsonify({"error": "IP地址不能为空"}), 400logger.info(f"正在查询IP: {ip}, 类型: {server_type if server_type else '自动检测'}")if not db.ensure_connection():return jsonify({"error": "数据库连接失败"}), 500result = Noneif not server_type or server_type == 'physical':# 先查询物理机try:server = db.get_server_by_ip(ip)if server:server = convert_decimal(server)server["business_ip"] = server.get("business_ip_str")server["management_ip"] = server.get("management_ip_str")server["server_type"] = "physical"result = serverexcept Exception as e:logger.exception(f"查询物理机时发生异常：{str(e)}")if not result and (not server_type or server_type == 'virtual'):# 如果没有找到物理机或明确要求查询虚拟机try:vm = db.get_vm_by_ip(ip)if vm:vm = convert_decimal(vm)vm["business_ip"] = vm.get("business_ip_str")vm["server_type"] = "virtual"result = vmexcept Exception as e:logger.exception(f"查询虚拟机时发生异常：{str(e)}")if result:return jsonify(result)else:logger.warning(f"未找到IP为 {ip} 的服务器")return jsonify({"error": "未找到匹配的服务器", "searched_ip": ip}), 404except Exception as e:logger.exception(f"处理请求时异常：{str(e)}")return jsonify({"error": "服务器内部错误", "details": str(e)}), 500@app.route('/api/query/related', methods=['POST'])
def query_related_servers():try:data = request.jsonip = data.get('ip', '').strip()if not ip:return jsonify({"error": "IP地址不能为空"}), 400logger.info(f"正在查询关联服务器IP: {ip}")if not db.ensure_connection():return jsonify({"error": "数据库连接失败"}), 500result = {"physical_servers": [],"virtual_machines": []}try:# 查询物理机physical_servers = db.get_physical_servers_by_ip(ip)if physical_servers:for server in physical_servers:server = convert_decimal(server)server["business_ip"] = server.get("business_ip_str")server["management_ip"] = server.get("management_ip_str")server["server_type"] = "physical"result["physical_servers"] = physical_servers# 查询虚拟机virtual_machines = db.get_virtual_machines_by_ip(ip)if virtual_machines:for vm in virtual_machines:vm = convert_decimal(vm)vm["business_ip"] = vm.get("business_ip_str")vm["server_type"] = "virtual"result["virtual_machines"] = virtual_machinesreturn jsonify(result)except Exception as e:logger.exception(f"查询关联服务器时发生异常：{str(e)}")return jsonify({"error": "查询数据库时出错", "details": str(e)}), 500except Exception as e:logger.exception(f"处理请求时异常：{str(e)}")return jsonify({"error": "服务器内部错误", "details": str(e)}), 500if __name__ == '__main__':if db.ensure_connection():app.run(host='0.0.0.0', port=5000, debug=False)else:logger.error("服务启动失败: 无法连接数据库")
四.数据库封装（db_handler.py）
作用与逻辑
功能：封装MySQL的连接操作和查询逻辑。
作用：
管理数据库连接的创建和重连
提供查询物理机、虚拟机信息的方法
避免在业务层重复编写SQL和连接管理代码
主要方法：
connect(): 建立数据库连接
ensure_connection(): 保证连接有效，无效会自动重建
get_server_by_ip(ip): 查询某个IP对应的物理机
get_vm_by_ip(ip): 查询虚拟机
get_physical_servers_by_ip(ip): 查询所有关联物理机
get_virtual_machines_by_ip(ip): 查询所有关联虚拟机
db_handler.py脚本内容：
import pymysql
from pymysql.cursors import DictCursor
import logginglogger = logging.getLogger(__name__)class DatabaseHandler:def __init__(self, config):self.config = configself.connection = Nonedef connect(self):try:self.connection = pymysql.connect(cursorclass=DictCursor,**self.config)logger.info("数据库连接成功")return Trueexcept pymysql.Error as err:logger.error(f"数据库连接失败: {err}")return Falsedef ensure_connection(self):try:if not self.connection or not self.connection.open:if not self.connect():raise ConnectionError("数据库重连失败")self.connection.ping(reconnect=True)return Trueexcept pymysql.Error as err:logger.error(f"连接验证失败: {err}")return Falsedef get_server_by_ip(self, ip):try:self.ensure_connection()with self.connection.cursor() as cursor:sql = """SELECT *,INET_NTOA(business_ip) AS business_ip_str,INET_NTOA(management_ip) AS management_ip_str FROM physical_servers WHERE business_ip = INET_ATON(%s) OR management_ip = INET_ATON(%s)LIMIT 1"""cursor.execute(sql, (ip, ip))result = cursor.fetchone()if result:result.pop('business_ip', None)result.pop('management_ip', None)return resultexcept pymysql.Error as err:logger.error(f"查询物理机失败: {err}")raisedef get_vm_by_ip(self, ip):try:self.ensure_connection()with self.connection.cursor() as cursor:sql = """SELECT *,INET_NTOA(business_ip) AS business_ip_str,INET_NTOA(host_machine_ip) AS host_machine_ip_str FROM virtual_machines WHERE business_ip = INET_ATON(%s)LIMIT 1"""cursor.execute(sql, (ip,))result = cursor.fetchone()if result:result.pop('business_ip', None)result.pop('host_machine_ip', None)return resultexcept pymysql.Error as err:logger.error(f"查询虚拟机失败: {err}")raisedef get_physical_servers_by_ip(self, ip):try:self.ensure_connection()with self.connection.cursor() as cursor:sql = """SELECT *,INET_NTOA(business_ip) AS business_ip_str,INET_NTOA(management_ip) AS management_ip_str FROM physical_servers WHERE business_ip = INET_ATON(%s) OR management_ip = INET_ATON(%s)"""cursor.execute(sql, (ip, ip))return cursor.fetchall()except pymysql.Error as err:logger.error(f"查询物理机列表失败: {err}")raisedef get_virtual_machines_by_ip(self, ip):try:self.ensure_connection()with self.connection.cursor() as cursor:sql = """SELECT *,INET_NTOA(business_ip) AS business_ip_str,INET_NTOA(host_machine_ip) AS host_machine_ip_str FROM virtual_machines WHERE business_ip = INET_ATON(%s)OR host_machine_ip = INET_ATON(%s)"""cursor.execute(sql, (ip, ip))return cursor.fetchall()except pymysql.Error as err:logger.error(f"查询虚拟机列表失败: {err}")raisedef close(self):if self.connection and self.connection.open:self.connection.close()logger.info("数据库连接已关闭")此前编写的所有脚本最好都是放在一个目录，我的是/mysql/excel下，切换到此目录，接下来我们给脚本赋权，可以尝试执行一下各脚本，如果提示缺少模块，可以自行pip3安装，确认都没问题了，我们开始配置系统服务。
cd /mysql/excel
chmod +x *.py
vi /etc/systemd/system/server_inventory.service
内容如下：
[Unit]
Description=Server Inventory API Service
After=network.target[Service]
User=root
WorkingDirectory=/mysql/excel
ExecStart=/usr/bin/python3 /mysql/excel/server_api.py
Restart=always
RestartSec=5
EnvironmentFile=/mysql/excel/.env# 日志配置
StandardOutput=append:/mysql/excel/server_inventory.log
StandardError=append:/mysql/excel/server_inventory_error.log[Install]
WantedBy=multi-user.target配置服务：
sudo systemctl daemon-reload
sudo systemctl enable server_inventory.service
sudo systemctl start server_inventory.service
sudo systemctl status server_inventory.service
服务运行正常。
五.桌面GUI客户端（Tkinter脚本）
前面基本的查询API接口服务已经部署好了，接下来需要编制windows客户端使用的图形化查询小程序。
为了实现简便、直观的操作界面，采用Python的Tkinter库开发桌面应用程序。界面主要包括：
**输入区域：**用户输入目标IP地址
**查询按钮：**触发后端请求
结果显示区域：
默认用Text控件显示原始返回数据
查询成功后，隐藏文本框，用结构化Treeview表格展示详细字段，支持列宽拖动调整
**底部状态栏：**实时显示操作状态和提示信息
脚本中192.168.1.1是服务器的IP，self.api_key = “keybaidu123”，这个需要和最开始的那个config.py的API_KEY是一样的。
脚本内容：
import tkinter as tk
from tkinter import ttk, messagebox, simpledialog
import requests# 字段中文别名映射
FIELD_ALIASES = {# 物理服务器字段"id": "编号","cabinet_number": "机柜编号","rack_position": "机架位置","category_abbreviation": "类别简称","asset_type": "资产类型","asset_code": "资产编码","server_type": "服务器类型","device_name": "设备名称","device_model": "设备型号","device_cpu": "CPU信息","device_memory": "内存信息","storage1": "存储1","storage2": "存储2","storage3": "存储3","system_raid": "系统RAID","data_raid": "数据RAID","configuration_type": "配置类型","serial_number": "序列号","value": "值","purchase_date": "购买日期","warranty_period": "保修期","business_ip": "业务IP","management_ip": "管理IP","responsible_person_a": "责任人A","responsible_person_b": "责任人B","system_level": "系统级别","business_system": "业务系统","system_category": "系统类别",# 虚拟机字段"id_vm": "编号","business_ip": "业务IP","vm_name": "虚拟机名称","category": "类别","server_type": "服务器类型","status": "状态","business_system": "业务系统","application": "应用","environment": "环境","service": "服务","responsible_person_a": "责任人A","responsible_person_b": "责任人B","system_level": "系统级别","host_machine_ip": "宿主机IP","provisioned_space": "已分配空间","guest_os": "操作系统","storage_config": "存储配置GB","memory": "内存","cpu": "CPU","network_cards": "网卡信息","dns_name": "DNS名","uuid": "UUID","created_at": "创建时间","updated_at": "更新时间",
}class ServerInventoryApp:def __init__(self, root):self.root = rootself.root.title("ServerMG_V2.0")self.root.geometry("800x700") # 固定窗口大小self.status_var = tk.StringVar(value="准备就绪")self._init_ui()self.api_url = "http://192.168.1.1:5000/api/query"#self.manage_url = "http://192.168.1.1:5001/api/manage"self.api_key = "keybaidu123"self.tree = Nonedef _init_ui(self):# 【3】状态栏： 固定在窗口底部ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W).pack(side=tk.BOTTOM, fill=tk.X, padx=10, pady=2)# 【4】顶部输入区域：服务器IP地址输入 + 查询按钮top_frame = ttk.Frame(self.root, height=50)top_frame.pack(fill=tk.X, padx=20, pady=10)top_frame.pack_propagate(False)ttk.Label(top_frame, text="服务器IP地址:").place(x=10, y=10)self.ip_entry = ttk.Entry(top_frame, width=30)self.ip_entry.place(x=100, y=10)ttk.Button(top_frame, text="查询", command=self.query_server).place(x=330, y=8)# 【5】结果区域：固定大小，不随内容变化# 这里定义宽度和高度（单位：像素），以确保大小稳定self.result_frame = ttk.Frame(self.root, width=1000, height=600) # 【5a】定义固定区域（数字大小调整请修改）self.result_frame.pack(padx=20, pady=10) # 【5b】只用pack布局，没有fill或expand，确保区域尺寸固定# 【6】用Text控件显示内容（查询结果默认显示），后续会用Treeview替换self.result_text = tk.Text(self.result_frame, wrap=tk.NONE)self.result_text.place(x=0, y=0, width=980, height=600) # 【6a】位置和大小（可调整）# 【备注】：可调整宽度（比如960）和高度（600），确保空间足够。def set_status(self, msg):self.status_var.set(msg)self.root.after(5000, lambda: self.status_var.set("准备就绪"))def query_server(self):ip = self.ip_entry.get().strip()if not ip:messagebox.showerror("错误", "请输入IP地址")returnself.set_status("正在查询...")self.result_text.delete(1.0, tk.END)try:response = requests.post(self.api_url, json={"ip": ip}, timeout=5)data = response.json()if "error" in data:self.result_text.insert(tk.END, f"错误: {data['error']}\n")self.set_status("查询失败")else:self.display_server_info(data)self.set_status("查询成功")except requests.RequestException as e:self.result_text.insert(tk.END, f"请求异常: {str(e)}\n")self.set_status("请求失败")def display_server_info(self, data):# 【8】如果存在Treeview，则销毁，用于放置新数据if hasattr(self, 'tree') and self.tree:self.tree.destroy()self.tree = None# 【9】隐藏文本框，不显示self.result_text.place_forget()# 【10】创建可自调宽度的Treeviewcolumns = ("名称", "值")# columns = ("名称", "数据库字段", "值")self.tree = ttk.Treeview(self.result_frame, columns=columns, show='headings', height=25)for col in columns:# 【11】每列宽度可调，因此不要设置固定宽度，也可以设置一定的默认宽度# 让列支持拖动调整宽度self.tree.heading(col, text=col)self.tree.column(col, width=300, anchor=tk.W, stretch=True) # 【11a】 stretch=True:列宽可伸缩# 如果希望一开始宽一些，可调节宽度比如150像素# 【12】将数据插入for key, value in data.items():alias = FIELD_ALIASES.get(key, key)value_str = "" if value is None else str(value)# self.tree.insert('', tk.END, values=(alias, key, value_str))# 如果不需要数据库字段信息，可以只插入两列self.tree.insert('', tk.END, values=(alias, value_str))# 【13】放置Treeviewself.tree.pack(fill=tk.BOTH, expand=False)if __name__ == "__main__":root = tk.Tk()app = ServerInventoryApp(root)root.mainloop()
封装成exe命令如下：
注：在此处，这个脚本名为ServerMG_V2.0.py，他在D:\test\下
pyinstaller.exe --onefile --name=ServerMG_V2.0 --windowed D:\test\ServerMG_V2.0.py
至此，整个系统搭建完成，由于文章是后面补写的，可能很多地方写得不是很全，有问题可以留言评论或私聊我，后续会接着这个写一个台账编辑录入的程序。
——对象的生命周期)
