Python获取当前在线设备ip和mac地址

获取局域网所在的网段

with os.popen("ipconfig /all") as res:
    for line in res:
        line = line.strip()
        if line.startswith("IPv4"):
            ipv4 = map(int, re.findall("(\d+)\.(\d+)\.(\d+)\.(\d+)", line)[0])
        elif line.startswith("子网掩码"):
            mask = map(int, re.findall("(\d+)\.(\d+)\.(\d+)\.(\d+)", line)[0])
            break
net_segment = ".".join([str(i & j) for i, j in zip(ipv4, mask)]).strip(".0")
net_segment

结果：

'192.168.3'

当前我们实际只会使用最后一个位置作为网段，并不需要考虑子网掩码，所以可以简化代码：

with os.popen("ipconfig /all") as res:
    for line in res:
        line = line.strip()
        if line.startswith("IPv4"):
            net_segment = re.findall("(\d+\.\d+\.\d+)\.\d+", line)[0]
            break
net_segment

完整代码：

import os
import re
import time
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED

import pandas as pd


def get_net_segment():
    with os.popen("arp -a") as res:
        for line in res:
            line = line.strip()
            if line.startswith("接口"):
                net_segment = re.findall(
                    "(\d+\.\d+\.\d+)\.\d+", line)[0]
                break
    return net_segment


def ping_net_segment_all(net_segment):
    # for i in range(1, 255):
    #     os.system(f"ping -w 1 -n 1 {net_segment}.{i}")
    with ThreadPoolExecutor(max_workers=4) as executor:
        for i in range(1, 255):
            executor.submit(os.popen, f"ping -w 1 -n 1 {net_segment}.{i}")


def get_arp_ip_mac():
    header = None
    with os.popen("arp -a") as res:
        for line in res:
            line = line.strip()
            if not line or line.startswith("接口"):
                continue
            if header is None:
                header = re.split(" {2,}", line.strip())
                break
        df = pd.read_csv(res, sep=" {2,}",
                         names=header, header=0, engine='python')
    return df


def ping_ip_list(ips, max_workers=4):
    print("正在扫描在线列表")
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_tasks = []
        for ip in ips:
            future_tasks.append(executor.submit(os.popen, f"ping -w 1 -n 1 {ip}"))
        wait(future_tasks, return_when=ALL_COMPLETED)


if __name__ == '__main__':
    # 是否进行初始扫描
    init_search = False
    if init_search:
        print("正在扫描当前网段所有ip，预计耗时1分钟....")
        ping_net_segment_all(get_net_segment())

    last = None
    while 1:
        df = get_arp_ip_mac()
        df = df.loc[df.类型 == "动态", ["Internet 地址", "物理地址"]]
        if last is None:
            print("当前在线的设备：")
            print(df)
        else:
            online = df.loc[~df.物理地址.isin(last.物理地址)]
            if online.shape[0] > 0:
                print("新上线设备：")
                print(online)
            offline = last[~last.物理地址.isin(df.物理地址)]
            if offline.shape[0] > 0:
                print("刚下线设备：")
                print(offline)
        time.sleep(5)
        ping_ip_list(df["Internet 地址"].values)
        last = df

结果如下：

当前在线的设备：
     Internet 地址               物理地址
0    192.168.3.3  3c-7c-3f-83-e2-7c
1   192.168.3.10  3c-7c-3f-80-08-1b
2   192.168.3.25  f0-2f-74-82-15-7e
3   192.168.3.26  f0-2f-74-82-15-a2
4   192.168.3.28  f0-2f-74-82-15-38
5   192.168.3.29  f0-2f-74-82-15-d0
6   192.168.3.32  f0-2f-74-82-15-3b
7   192.168.3.33  f0-2f-74-82-15-56
8   192.168.3.39  a8-5e-45-16-79-99
9  192.168.3.225  30-24-a9-5a-eb-82
新上线设备：
    Internet 地址               物理地址
9  192.168.3.52  3c-7c-3f-c2-cd-cb
刚下线设备：
    Internet 地址               物理地址
9  192.168.3.52  3c-7c-3f-c2-cd-cb

大功告成了...