You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在IP探测循环中临时移除异常IP并后续恢复?

问题

我有一份存储IP地址列表的Excel文档,当前会循环遍历这些IP执行ping检测。现在的逻辑是如果IP响应异常,会暂停整个循环30秒,我想要改成:把异常IP临时从循环中移除,继续遍历其他IP,等30秒后再重新检测这个异常IP,恢复正常就加回循环。

现有相关代码如下:

主循环代码

while True:
    get_ip() 

    # Ping each IP Address in the list
    for ip in ip_addr:
        response = os.popen(f"ping {ip} ").read()

        if "Request timed out." in response:
            print("Stops the loop for 30 seconds")
        ### 想要在这里把IP移出循环,继续遍历其他IP,30秒后再重新检测该IP ###
            time.sleep(30)
        
        else:
            print(ip, " is ONLINE")  # Prints IP and its status

Excel读取IP的函数

# Gets the WAN 1 IP address from excel
def get_ip():
    global ip_addr

    # Loads spreadsheet
    getExcelSpreadsheet()

    # Create list to append IPs to
    ip_list = []

    # Takes all the values in a column and puts them in a list 'ip_list'
    for row in ws.iter_rows(min_row=starting_row, min_col=2, max_row=ending_row, max_col=2):
        for cell in row:
            ip_list.append(cell.value)

    ip_addr = iter(ip_list)

生成带唯一标识的IP字典的代码

ip_dict = {}
for enum, row in enumerate(ws.iter_rows(min_row=2, min_col=2,max_row=4), 1):
    k = f'IP({enum})'
    v = row[0].value
    ip_dict[k] = v

输出示例:

{'IP(1)': '8.8.8.8', 'IP(2)': '10.1.40.1', 'IP(3)': '10.1.30.1'}

请问怎么整合这些内容实现需求?


解决方案

核心思路是维护两个列表分别管理正常检测和待恢复的IP,同时记录异常IP的下次检测时间,避免暂停整个循环:

  • active_ips:当前处于正常检测流程的IP列表
  • pending_ips:存储异常IP及其可重新检测的时间点,到时间后自动重试

完整代码实现

import os
import time
from datetime import datetime, timedelta

# 全局变量:管理IP状态
active_ips = []
pending_ips = []

def load_ips_from_excel():
    """从Excel加载IP列表,初始化或刷新检测列表时调用"""
    global active_ips
    # 保留原有加载表格的逻辑
    getExcelSpreadsheet()
    
    ip_list = []
    # 读取Excel指定列的IP,避免重复添加到待恢复或正常列表中
    existing_ips = [item[0] for item in pending_ips] + active_ips
    for row in ws.iter_rows(min_row=starting_row, min_col=2, max_row=ending_row, max_col=2):
        for cell in row:
            ip = cell.value
            if ip not in existing_ips:
                ip_list.append(ip)
    active_ips = ip_list

def ping_ip(ip):
    """执行单次ping检测,返回IP是否在线"""
    # Windows用-n 1,Linux/macOS替换成-c 1
    response = os.popen(f"ping {ip} -n 1").read()
    return "Request timed out." not in response

def main():
    # 初始化加载Excel中的IP
    load_ips_from_excel()
    
    while True:
        current_time = datetime.now()
        
        # 处理待恢复的IP:到检测时间则重试
        to_remove = []
        for idx, (ip, check_time) in enumerate(pending_ips):
            if current_time >= check_time:
                if ping_ip(ip):
                    print(f"{ip} 已恢复,重新加入检测列表")
                    active_ips.append(ip)
                    to_remove.append(idx)
                else:
                    # 仍异常,延长30秒后重试
                    pending_ips[idx] = (ip, current_time + timedelta(seconds=30))
                    print(f"{ip} 仍未恢复,30秒后再次检测")
        
        # 倒序删除已恢复的IP,避免索引错乱
        for idx in reversed(to_remove):
            del pending_ips[idx]
        
        # 遍历正常检测的IP列表(用copy避免遍历中修改列表导致的异常)
        for ip in active_ips.copy():
            if ping_ip(ip):
                print(f"{ip} is ONLINE")
            else:
                print(f"{ip} 响应异常,临时移出检测列表,30秒后重新检测")
                active_ips.remove(ip)
                pending_ips.append((ip, current_time + timedelta(seconds=30)))
        
        # 循环间隔,避免占用过高CPU
        time.sleep(1)

if __name__ == "__main__":
    main()

关键改动说明

  1. 取消全局迭代器:原代码用iter(ip_list)生成的迭代器只能遍历一次,改为直接操作列表,更灵活地添加/移除IP
  2. 拆分功能模块:将IP加载、ping检测拆为独立函数,代码更易维护
  3. 双列表状态管理:彻底解决异常IP阻塞整个循环的问题,其他IP可正常遍历检测
  4. 优化ping命令:仅发送1个数据包,加快检测速度,减少等待时间
  5. 支持Excel动态刷新(可选):如果Excel内容会更新,可在main循环中定时调用load_ips_from_excel(),例如每5分钟刷新一次:
    # 在main循环的current_time定义后添加
    if current_time.minute % 5 == 0 and current_time.second == 0:
        load_ips_from_excel()
        print("已刷新Excel中的IP列表")
    

整合IP唯一标识的方案

如果需要用你生成的IP唯一标识(比如区分Excel中重复的IP),只需修改IP存储格式为元组(唯一标识, IP),示例如下:

def load_ips_from_excel():
    global active_ips
    getExcelSpreadsheet()
    ip_list = []
    existing_items = [(item[0], item[1]) for item in pending_ips] + active_ips
    for enum, row in enumerate(ws.iter_rows(min_row=starting_row, min_col=2, max_row=ending_row, max_col=2), 1):
        ip_id = f'IP({enum})'
        ip = row[0].value
        if (ip_id, ip) not in existing_items:
            ip_list.append((ip_id, ip))
    active_ips = ip_list

def ping_ip(ip_item):
    ip_id, ip = ip_item
    response = os.popen(f"ping {ip} -n 1").read()
    return "Request timed out." not in response

后续的active_ips和pending_ips操作都对应调整为元组即可。


内容的提问来源于stack exchange,提问作者Student1860

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.29 23:32:26