如何在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()
关键改动说明
- 取消全局迭代器:原代码用
iter(ip_list)生成的迭代器只能遍历一次,改为直接操作列表,更灵活地添加/移除IP - 拆分功能模块:将IP加载、ping检测拆为独立函数,代码更易维护
- 双列表状态管理:彻底解决异常IP阻塞整个循环的问题,其他IP可正常遍历检测
- 优化ping命令:仅发送1个数据包,加快检测速度,减少等待时间
- 支持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
相关产品推荐
相关产品推荐

