主页 / 问题 / 1605097
Accepted
Young Sideways
Asked:2025-01-19 14:47:41 +0000 UTC2025-01-19 14:47:41 +0000 UTC

从 python 脚本执行 ping 命令

  • 772

最后的问题：是否可以用 python 更简洁地编写这个，而不失去容错能力，避免可怕的 bash 正则表达式？
首先，我想说我已经用 Python 编写了大约一周的时间（在此之前，只使用 C/C++）。
我需要ping从我的 python 脚本调用一个命令，不仅有关地址可用性的答案很重要，而且ping.需要注意的是，该脚本将在无法从 pip 传递包的机器上运行。

首先，我以一种相当简单的方式做到了 - 我为输出编写了一个解析器函数subprocess.run，它看起来像这样

SYSTEM_PING_ITERATIONS: int       = 1
SYSTEM_CMD_PING       : str       = "ping -4 -c{0} {1} | tail -n2"

def run_system_cmd(cmd: str) -> Tuple[List[str], int]:
    ret = subprocess.run(cmd,
                         shell=True,
                         executable='/bin/bash',
                         stdout=subprocess.PIPE,
                         stderr=subprocess.STDOUT
    )
    output = ret.stdout.decode('utf-8', errors='ignore')
    out_split: List[str] = output.split('\n')
    return (out_split[:-1] if out_split[-1] == '' else out_split, ret.returncode)


def process_ping(interations: int, address: Union[str, List[int]]) -> Dict[str, Union[int, float]]:
    address = address if isinstance(address, str) else '.'.join(map(str, address))

    from_cmd = run_system_cmd(str.format(SYSTEM_CMD_PING, interations, address))
    if from_cmd[1] != 0:
        return {}
    
    output = from_cmd[0]
    if not output or len(output) != 2:
        return {}
    
    suffixes: Dict[str, float] = {
        "ns": 1000000.0,
        "us": 1000.0,
        "ms": 1.0,
        "s" : 0.001
    }
    result: Dict[str, Union[int, float]] = {
        "transmitted": 0  ,
        "received"   : 0  ,
        "loss"       : 0.0,
        "total"      : 0.0,
        "min"        : 0.0,
        "max"        : 0.0,
        "avg"        : 0.0
    }

    data1 = output[0].split(' ')
    if len(data1) < 10:
        return {}
    result['transmitted'] = int(data1[0])
    result['received']    = int(data1[3])
    result['loss']        = float(re.sub('\\D', '', data1[5]))
    result['total']       = float(re.sub('\\D', '', data1[9])) * suffixes[re.sub(r'[^A-Za-z]', '', data1[9])]

    # do not parse further, if destination unreachable
    if result['loss'] == 100.0:
        return result
    
    data2 = output[1].split(' ')[-2:]
    splitted = data2[0].split('/')
    result['min'] = float(splitted[0]) * suffixes[data2[1]]
    result['max'] = float(splitted[2]) * suffixes[data2[1]]
    result['avg'] = float(splitted[1]) * suffixes[data2[1]]

    return result

对于如此简单的任务需要相当多的代码......

在对 python 缺乏足够了解的情况下，我想到的下一件事是更改 bash 命令：

ping -c{0} {1} | tail -n2 | awk -F'[ |,]' '/packets transmitted/ {transmitted=$1; received=$5; loss=$8; time=$13} /rtt min\/avg\/max/ {split($4, rtt, "/"); min=rtt[1]; avg=rtt[2]; max=rtt[3]} END {print transmitted, received, loss, time, min, avg, max, $5}'

对于可访问和不可访问的地址，其输出分别如下：

1 1 0% 0ms 36.477 36.477 36.477 ms

1 0 100% 0ms

这比原始版本更容易解析和检查。

所以现在的问题是：是否可以用 python 更简洁地编写这个，而不失去容错能力，避免可怕的 bash 正则表达式？

python

1 个回答

  1. Best Answer

    恕我直言，您的一行解决方案看起来非常可读

    真正的 Python 风格是冗长、冗长、绝不单行的。因此，是否值得将文本页面围起来以替换一行脚本？

    #!/usr/bin/env python3
"ping"
import argparse
from subprocess import CompletedProcess, run, PIPE
from itertools import dropwhile
from typing import Iterable, Tuple


def run_ping(address: str, count: int = 4, interval: float = 0.1, args:Iterable = []) -> CompletedProcess:
    '''Запускает ping и ждёт завершения процесса'''
    return run(['ping', '-i', str(interval), '-c', str(count), '-q', address, *args],
               stdout=PIPE,
               stderr=PIPE,
               check=False,
               text=True,
               encoding='ascii',
               env={'LANG': 'C'},
               )


def get_ping_stats_text(ping_proc: CompletedProcess) -> Tuple[str | None, str | None]:
    '''Возвращает две последние строки из вывода ping.

    Если строк меньше двух, возвращает (None, None).'''
    lines = dropwhile(
        lambda line: not line.startswith('---'),
        ping_proc.stdout.splitlines())
    # lines состоит из строки '--- ping statistics ---' и двух строк с данными
    lines = list(lines)
    if len(lines) < 3:
        return None, None
    return lines[1:]


TRANSMITTED = 'packets transmitted'
RECEIVED = 'received'
LOST = 'packet loss'
TIME = 'time'
ERRORS = 'errors'

def parse_packet_data(line: str) -> dict:
    '''Возвращает набор пар с ключами 'packets transmitted', 'received', 'packet loss', 'time'.'''
    return dict(
        map(
            # в выводе ping сначала идёт значение, затем ключ
            # кроме случая с 'time', там наоборот
            lambda p: (p[1], p[0]) if p[0] != 'time' else (p[0], p[1]),
            map(
                # разделить пару на первое слово и остальной текст
                lambda _str: _str.strip().split(' ', 1),
                # Разбить строку на пары, разделённые запятыми
                line.split(',')
                )
            )
        )

def parse_ping_rtt(line: str) -> Tuple[list, str]:
    '''Возвращает список времён RTT в мс.'''
    if ' = ' not in line:
        return None, None
    times_str, unit = line.split(' = ')[1].split()
    times = times_str.split('/')
    return times, unit

def get_ping_stats(ping_proc: CompletedProcess) -> dict|None:
    '''Возвращает словарь с данными о пинге.'''
    first, second = get_ping_stats_text(ping_proc)
    if first is None:
        return None
    stats = parse_packet_data(first)
    times, unit = parse_ping_rtt(second)
    return {
        'packets': stats,
        'times': times ,
        'unit': unit,
    }

def flatten_ping_stats(stats: dict) -> list:
    '''Возвращает список значений из словаря stats.'''
    result = [
        stats['packets'][TRANSMITTED],
        stats['packets'][RECEIVED],
        stats['packets'][LOST],
        stats['packets'][TIME],
    ]
    if stats['times']:
        result.extend(stats['times'])
        result.append(stats['unit'])
    return result

def main():
    '''Вызывает ping и выдаёт сокращённые результаты.'''
    arg_parser = argparse.ArgumentParser(
        description='Ping a host and print the results')
    arg_parser.add_argument(
        'address',
        help='The address to ping')
    arg_parser.add_argument(
        '-c', '--count',
        type=int,
        default=10,
        help='The number of packets to send')
    arg_parser.add_argument(
        '-i', '--interval',
        type=float,
        default=0.1,
        help='The interval between packets')
    arg_parser.add_argument(
        'extra_args',
        nargs=argparse.REMAINDER,
        help='Extra arguments to pass to ping')
    
    args = arg_parser.parse_args()
    address = args.address
    
    ping_proc = run_ping(address, args.count, args.interval, args.extra_args)
    stats = get_ping_stats(ping_proc)
    if stats is None:
        print('No data')
        return
    print(' '.join(flatten_ping_stats(stats)))
    
if __name__ == '__main__':
    main()
    • 2

相关问题