python中，如何实时获取PING命令的结果

绝地灬酷狼 2023-02-10 05:25 414阅读 0赞

ping命令在ping通或者不通的时候，大概是下面两个情况

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d5bWFpc3ls_size_16_color_FFFFFF_t_70]  
技术栈使用了 Flask框架和redis

## 数据处理 ##

# 存储ping数据的redis 1号库
    redis_sh = StrictRedis(host='127.0.0.1', port=6379, decode_responses=True, db=1)
    
    def flush_sh_redis():
    	# 清空redis 1号库
        redis_sh.flushdb()
    
    def sh(command, ip):
        p = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        not_ping = "PING %s (%s) 56(84) bytes of data." % (ip, ip)
        flag = 0  # 自定义的返回值--避免ping失败无数据返回
        num = 1  # 每一行ping结果标识 （key）
        redis_sh.set("sh1", not_ping + "\n")
        row = 1
        for line in iter(p.stdout.readline, b''):
            line = line.rstrip().decode('utf8')
    
            if flag == 0:  # ping操作有结果 第一次进入
                redis_sh.set("sh1", line + "\n") # 将自定义的返回值覆盖
                flag = 1 
                num += 1  # key 值加一
                row = 2
            else:
                if row == 2:
                    redis_sh.append("sh1", line + "\n") # 这里使用append的原因是，实际在使用ping命令时，【PING 1.1.1.1 (1.1.1.1) 56(84) bytes of data.】【64 bytes from 1.1.1.1: icmp_seq=2 ttl=54 time=185 ms】这两行是同时出现的，为了效果更真实，在sh1里面保存了两条返回值
    
                    row = 1
                else:
                    redis_sh.set("sh" + str(num), line + "\n")
                    print(num, line)
                    num += 1
    
        redis_sh.set("sh" + str(num), "over") # over表示当前ping操作结束
    
    
    def get_sh(num):
        result = redis_sh.get("sh" + num)
        return result

## 数据获取 ##

@diagnosis.route("/ping", methods=["POST"])
    def ping():
        # 执行ping命令
        flush_sh_redis()  # 每次请求ping时,清空指定redis某个库
        ip = request.form.get("ip")
        if not ip:
            return jsonify(status=4004, msg="ip参数不能为空")
    	
    	# 中文校验 
        a = re.match(r"^[\x00-\xff]+$", ip)
        if not a:
            sysconfig_log.info("ping测试失败")
            return jsonify(status=4004, msg='ping命令不能包含中文')
        command = "ping -c 5 -w 10 %s" % ip
        try:
           # 起线程执行ping命令
            task = threading.Thread(target=sh, args=(command, ip))
            task.start()
        except Exception as e:
            print(e)
            return jsonify(status=4004, msg='ping命令执行失败')
        return jsonify(status=0, msg="开始进行ping测试")
    
    
    @diagnosis.route("/get_ping_result", methods=["GET"])
    def get_ping_result():
       # 获取ping结果
       # num参数，表示前端第几次获取数据，num从1开始，前端在调用时，如果当前num值没有获取到数据，那么num值不变，继续调用该接口，获取当前num值得数据，直到拿到数据为止
        num = request.args.get("num")
        num = str(num)
    
        result = get_sh(num)
        if result == "over":
            return jsonify(status=2, msg="over")
        if not result:
            return jsonify(status=1, msg="暂无数据")  # 请求的数据暂时没有生成，继续使用当前num值发出请求
    
        result = result.split("\n")
        result.pop(-1)
        data = { 
            "data_list": result,
            "flag": True
        }
        return jsonify(data=data, status=0, msg="ping测试成功")

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d5bWFpc3ls_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200520155955756.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d5bWFpc3ls,size_16,color_FFFFFF,t_70