如何为列表中每个IP分配PASS/FAIL状态并写入MariaDB？

核心修改点 #

• 调整echo_ping函数，直接返回PASS或FAIL状态，同时保留原提示输出
• 补充完整数据库连接逻辑，采用参数化查询规避SQL注入风险
• 修正原SQL语句的语法错误（多余逗号）
• 建立IP与数据库列的对应关系，确保结果准确写入目标字段

完整实现代码 #

import subprocess
import mysql.connector
from mysql.connector import Error

class Ping:
    def echo_ping(self, echo_ip):
        # 执行Windows环境下的ping命令
        ping_reply = subprocess.run(
            ["ping", "-n", "2", echo_ip],
            stderr=subprocess.PIPE,
            stdout=subprocess.PIPE,
            text=True  # 直接返回字符串，无需手动转换
        )
        print(".", end='')
        
        # 判断IP可达性，返回PASS/FAIL状态
        if ping_reply.returncode == 0:
            if "unreachable" in ping_reply.stdout:
                print(f"\n* No response from device {echo_ip}")
                return "FAIL"
            else:
                print(f"\n* OK response from device {echo_ip}")
                return "PASS"
        elif ping_reply.returncode == 1:
            print(f"\n* No response from device {echo_ip}")
            return "FAIL"
        else:
            # 处理其他异常情况（如参数错误）
            print(f"\n* Error pinging device {echo_ip}")
            return "FAIL"

    def ping_all_va(self):
        # 定义IP与对应数据库列的映射关系
        ip_port_mapping = [
            ("10.10.10.2", "Ethernet_Port_1"),
            ("192.168.11.1", "Ethernet_Port_2"),
            ("192.168.12.2", "E_port_3"),
            ("192.168.13.3", "Ethernet_Port_4"),
            ("192.168.14.4", "Ethernet_Port_5")
        ]
        
        # 收集所有IP的检测结果
        results = {}
        for ip, _ in ip_port_mapping:
            status = self.echo_ping(ip)
            results[ip] = status
        
        # 连接MariaDB并写入数据
        conn = None
        try:
            # 初始化数据库连接（替换为你的实际配置）
            conn = mysql.connector.connect(
                host='你的数据库地址',
                database='你的数据库名',
                user='你的用户名',
                password='你的密码'
            )
            
            if conn.is_connected():
                cursor = conn.cursor()
                # 构造参数化INSERT语句，避免SQL注入
                insert_query = """
                INSERT INTO db_ServerEcho(
                    Ethernet_Port_1, Ethernet_Port_2, E_port_3, 
                    Ethernet_Port_4, Ethernet_Port_5
                ) VALUES (%s, %s, %s, %s, %s)
                """
                # 按数据库列顺序提取结果值
                values = [
                    results["10.10.10.2"],
                    results["192.168.11.1"],
                    results["192.168.12.2"],
                    results["192.168.13.3"],
                    results["192.168.14.4"]
                ]
                cursor.execute(insert_query, values)
                conn.commit()
                print(f"\n{cursor.rowcount} 行数据已插入数据库")
                
        except Error as e:
            print(f"\n数据库操作错误: {e}")
        finally:
            # 确保数据库连接关闭
            if conn and conn.is_connected():
                cursor.close()
                conn.close()
                print("数据库连接已关闭")

if __name__ == "__main__":
    Ping().ping_all_va()

关键说明 #

1. Ping函数优化：
   • 添加text=True参数，让subprocess直接返回字符串，省去手动类型转换步骤
   • 明确返回PASS/FAIL状态，便于后续数据库写入逻辑处理
2. 数据库操作规范：
   • 替换代码中你的数据库地址等占位符为实际数据库配置
   • 使用参数化查询%s，杜绝字符串拼接引发的SQL注入风险
   • 加入异常捕获和资源释放逻辑，提升代码健壮性
3. 映射关系维护：
   • 通过列表元组存储IP与列名的对应关系，后续新增或修改IP时只需调整该列表即可

内容的提问来源于stack exchange，提问作者Asram