Python ping检测脚本仅写入最后一条结果到文件的问题排查

• 文件打开模式错误：ping函数内每次打开结果文件都用"w"覆盖写入模式，每次执行都会清空文件原有内容，最终只会保留最后一次写入的IP检测结果。
• 多线程写入无保护：如果调整并发线程数大于1，多个线程同时操作同一个文件会出现写入冲突，导致内容错乱、结果丢失。
• 状态判断逻辑错误：原代码if "Received = 1" and "Approximate" in response:的写法，实际仅判断了"Approximate"是否存在于返回结果中，"Received = 1"作为非空字符串会被直接判定为真，检测结果不准。
• 变量未定义风险：延迟提取逻辑没有默认值，一旦ping返回格式和预期不符，latencyfin变量会在写入时触发未定义报错。
• 线程调度逻辑缺陷：原循环仅对最后一个创建的线程执行join()等待，调大并发数时主线程可能提前退出，部分检测任务没跑完就被终止。

import subprocess
import threading
import time

timestr = time.strftime("%Y-%m-%d %H%M%S")
timesec = time.strftime("%Y-%m-%d %H:%M:%S")
# 初始化线程锁，避免多线程写文件冲突
file_lock = threading.Lock()
# 启动时初始化结果文件，清空旧内容并写入表头
with open(f"results{timestr}.txt", "w", encoding="utf-8") as f:
    f.write("host,status,latency(ms),check_time\n")

def ping(host):
    # 跳过IP列表中的空行
    if not host.strip():
        return
    latencyfin = 0
    p = subprocess.Popen(
        ["ping", host, "-n", "1"],
        shell=True,
        universal_newlines=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    response = p.communicate()[0]
    for i in response.split("\n"):
        para = i.split("=")
        try:
            if para[0].strip() == "Minimum":
                latency = para[3].strip()
                # 提取延迟数字，兼容不同返回格式
                latency_num = ''.join([c for c in latency if c.isdigit()])
                latencyfin = int(latency_num) if latency_num else 0
        except:
            pass

    # 修正判断逻辑，两个条件同时满足才判定为ping成功
    if "Received = 1" in response and "Approximate" in response:
        print(f"UP {host} Ping Successful")
        write_content = f"{host},UP,{latencyfin},{timesec}\n"
    else:
        print(f"Down {host} Ping Unsuccessful")
        write_content = f"{host},Down,0,{timesec}\n"
    # 加锁写入，避免多线程下内容错乱
    with file_lock:
        with open(f"results{timestr}.txt", "a", encoding="utf-8") as results_file:
            results_file.write(write_content)

if __name__ == "__main__":
    with open(r'input.txt', "r", encoding="utf-8") as server_list_file:
        hosts = server_list_file.read()
        hosts_list = hosts.split('\n')
    num_threads = 5  # 可按需调整并发检测的线程数
    threads = []
    for host in hosts_list:
        t = threading.Thread(target=ping, args=(host,))
        t.start()
        threads.append(t)
        # 达到并发上限时等待现有线程跑完再创建新任务
        if len(threads) >= num_threads:
            for t in threads:
                t.join()
            threads = []
    # 等待最后一批剩余线程执行完成
    for t in threads:
        t.join()

• 拆分文件操作逻辑：程序启动时用"w"模式初始化结果文件、写入表头，后续所有检测结果都用"a"追加模式写入，不会覆盖历史内容。
• 新增线程锁机制：多线程场景下写文件会自动排队，不会出现内容穿插、丢行问题，支持自由调整并发线程数。
• 修正状态判断逻辑：两个检测条件都明确校验是否存在于ping返回内容中，避免误判主机状态。
• 补全异常兜底：延迟参数默认值设为0，提取延迟时自动过滤非数字字符，不会出现变量未定义的报错。
• 重构线程调度：所有创建的检测线程都会被等待执行完成，不会出现主线程提前退出、检测任务中断的问题。
• 增加鲁棒性处理：自动跳过IP列表里的空行，指定文件读写编码，捕获ping命令的错误输出，适配更多运行场景。

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