如何批量逐个Ping多IP并避免重复代码，将结果存入MariaDB？

核心思路 #

1. 将单个IP的Ping检测逻辑封装为独立函数，彻底消除重复判断代码
2. 用列表存储所有待检测IP，通过循环逐个调用Ping函数实现串行检测
3. 集成MariaDB操作逻辑，将每个IP的检测结果实时存入数据库

步骤实现 #

1. 封装可复用的Ping检测函数 #

把IP连通性判断逻辑抽成单独函数，输入IP地址直接返回PASS/FAIL结果：

import subprocess

def check_ip_connectivity(ip):
    # Windows系统用-n指定Ping次数，Linux/macOS需替换为-c
    ping_result = subprocess.run(
        ["ping", "-n", "2", ip],
        stderr=subprocess.PIPE,
        stdout=subprocess.PIPE,
        text=True  # 直接返回字符串，无需手动转码
    )
    # 连通性判断逻辑
    if ping_result.returncode == 0:
        return "FAIL" if "unreachable" in ping_result.stdout else "PASS"
    else:
        return "FAIL"

2. 批量遍历待检测IP #

将所有需要检测的IP存入列表，通过循环逐个执行Ping检测：

# 待检测IP列表，可根据需求随时修改
target_ips = [
    "10.10.10.2",
    "192.168.11.10",
    "192.168.12.10",
    "192.168.13.10",
    "192.168.14.10",
    "192.168.100.2"
]

# 逐个检测并收集结果
ip_results = []
for ip in target_ips:
    status = check_ip_connectivity(ip)
    print(f"{ip}: {status}")
    ip_results.append((ip, status))

3. 将结果存入MariaDB #

使用mysql-connector-python库连接数据库，先创建存储表（若不存在），再批量插入或更新结果：
首先安装依赖：

pip install mysql-connector-python

数据库操作代码：

import mysql.connector
from mysql.connector import Error

def save_results_to_db(ip_results, db_config):
    try:
        # 建立数据库连接
        connection = mysql.connector.connect(**db_config)
        if connection.is_connected():
            cursor = connection.cursor()
            # 创建结果存储表
            create_table_query = """
            CREATE TABLE IF NOT EXISTS ip_connectivity (
                ip_address VARCHAR(15) PRIMARY KEY,
                status VARCHAR(4) NOT NULL,
                check_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
            """
            cursor.execute(create_table_query)
            
            # 插入或更新结果（重复IP自动覆盖最新状态）
            upsert_query = """
            INSERT INTO ip_connectivity (ip_address, status)
            VALUES (%s, %s)
            ON DUPLICATE KEY UPDATE status = %s, check_time = CURRENT_TIMESTAMP
            """
            # 批量执行写入
            for ip, status in ip_results:
                cursor.execute(upsert_query, (ip, status, status))
            connection.commit()
            print("所有检测结果已成功存入数据库")
            
    except Error as e:
        print(f"数据库操作错误: {e}")
    finally:
        # 关闭数据库连接
        if connection.is_connected():
            cursor.close()
            connection.close()

4. 整合完整类代码 #

将以上逻辑整合到类中，替换原有pingPreflash方法：

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

class IPChecker:
    def __init__(self):
        self.reachable = 'PASS'
        self.unreachable = 'FAIL'
        # 待检测IP列表
        self.target_ips = [
            "10.10.10.2",
            "192.168.11.10",
            "192.168.12.10",
            "192.168.13.10",
            "192.168.14.10",
            "192.168.100.2"
        ]
        # 数据库配置，需根据实际环境修改
        self.db_config = {
            'host': 'localhost',
            'database': 'network_monitor',
            'user': 'root',
            'password': 'your_db_password'
        }

    def check_ip_connectivity(self, ip):
        ping_result = subprocess.run(
            ["ping", "-n", "2", ip],
            stderr=subprocess.PIPE,
            stdout=subprocess.PIPE,
            text=True
        )
        if ping_result.returncode == 0:
            return self.unreachable if "unreachable" in ping_result.stdout else self.reachable
        else:
            return self.unreachable

    def save_results_to_db(self, ip_results):
        try:
            connection = mysql.connector.connect(**self.db_config)
            if connection.is_connected():
                cursor = connection.cursor()
                # 创建存储表
                create_table = """
                CREATE TABLE IF NOT EXISTS ip_connectivity (
                    ip_address VARCHAR(15) PRIMARY KEY,
                    status VARCHAR(4) NOT NULL,
                    check_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
                """
                cursor.execute(create_table)
                # 插入/更新结果
                upsert_query = """
                INSERT INTO ip_connectivity (ip_address, status)
                VALUES (%s, %s)
                ON DUPLICATE KEY UPDATE status = %s, check_time = CURRENT_TIMESTAMP
                """
                for ip, status in ip_results:
                    cursor.execute(upsert_query, (ip, status, status))
                connection.commit()
                print("结果已同步至数据库")
        except Error as e:
            print(f"数据库错误: {e}")
        finally:
            if connection.is_connected():
                cursor.close()
                connection.close()

    def pingPreflash(self):
        ip_results = []
        # 逐个Ping检测
        for ip in self.target_ips:
            status = self.check_ip_connectivity(ip)
            print(f"{ip}: {status}")
            ip_results.append((ip, status))
        # 存入数据库
        self.save_results_to_db(ip_results)

# 运行示例
if __name__ == "__main__":
    checker = IPChecker()
    checker.pingPreflash()

注意事项 #

• 若运行在Linux/macOS系统，需将ping命令的-n参数替换为-c
• 确保MariaDB服务正常运行，且配置的数据库账号拥有读写权限
• ON DUPLICATE KEY UPDATE用于处理重复检测同一IP的场景，自动更新最新状态和检测时间

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