0

вопрос по организации данных (возможно проще код-ревью попросить...)

Пишу сканер сети на доступные IP на Python. Сначала данные хранил в словарях типа

ip_found_dict = {ip_key:{mac_key:, hostname:, os:, ...}}

Начал расширять функционал и понял, что нужно переводить данные в классы - ок, перевел! Но я полностью отказался от хранения экземпляров в основном коде программы, тоесть идея такая, что я даю команду классу "проверь такой то адрес, сам все решай, я знать ничего не хочу, и храни все у себя", у себя - это в классовом словаре, который со временем наполняется данными

    ip_found_obj_dict = {ip_1: ip_data_obj_1, ip_2: ip_data_obj_2}

В итоге получился класс без конструктора и все методы классовые! это меня смущает (почти на каждом методе стоит декоратор @classmethod)! Еще больше меня смущает то, что в самой программе создается один экземпляр класса (но можно и без него!), а через него по командам (через внутренние скрытые методы) формируются экземпляры этого же класса и они как раз записываются в классовый словарь.

С одной стороны я полностью выделил в один класс код по формированию новых данных и их обработке, и даже очень доволен этим, с другой стороны как-то коряво смотрится... И таких классов хранителей у меня три подряд!

Порядок работы с классом такой:

  1. из основного кода даем команду проверить ip через класс
  2. класс пингует и если ответ получен - пролистывает свои данные в классовом словаре и если не находит - запускает скрытый метод для создания новых данных - создает еще один свой же экземпляр через cls().ххх
  3. вносит новую запись в свой словарь чтобы он не очистился из памяти "ip_2: ip_data_obj_2"
  4. затем дополняет данные этого экземпляра индивидуальными атрибутами (мас, os, ...)

полный код на GitHub

class Hosts():
mac_obj_dict = {}               # {mac: host_obj, }

UPDATE_LISTBOX = lambda: None
ip_found_list = []      # use list! if found 2 mac with same ip - ok! let be 2 items with same ip!!!
ip_last_scanned = None
ip_last_answered = None
flag_scan_manual_stop = False
count_ip_scanned = 0

# LIMITS
limit_ping_timewait_ms = 100  # BEST=100
limit_ping_thread = 300  # BEST=300
# even 1000 is OK! but use sleep(0.001) after ping! it will not break your net
# but it can overload you CPU!
# 300 is ok for my notebook ([email protected]/16Gb) even for unlimited ranges

# -----------------------------------------------------------
# INSTANCE
@contracts.contract(ip=ipaddress.IPv4Address, mac=str)
def _instance_add_if_not(self, ip, mac):
    # return instance new or existed!
    if mac not in Hosts.mac_obj_dict:
        with lock:
            Hosts.mac_obj_dict.update({mac: self})
            Hosts.ip_found_list.append(ip)

            self.mac = mac
            self.ip = ip

            self.active = True
            self.was_lost = False
            self.was_changed_ip = False
            self.hostname = None
            self.vendor = None
            self.os = None
            self.time_response = None

            self.count_lost = 0
            self.count_response = 0
            return self
    else:
        host_obj = Hosts.mac_obj_dict[mac]
        if host_obj.ip != ip:
            host_obj.was_changed_ip = ip
            Hosts.mac_obj_dict[mac].ip = ip    # need to update if host will change its IP!
        return host_obj

def _instance_del(self):
    Hosts.mac_obj_dict.pop(self.mac)
    Hosts.ip_found_list.remove(self.ip)
    Hosts._update_listbox()

@classmethod
@contracts.contract(mac=str)
def del_mac(cls, mac):
    cls.mac_obj_dict[mac]._instance_del()

@classmethod
@contracts.contract(ip=ipaddress.IPv4Address)
def del_ip(cls, ip):
    del_obj_list = []
    for obj in cls.mac_obj_dict.values():
        if obj.ip == ip:
            del_obj_list.append(obj)
    for obj in del_obj_list:
            obj._instance_del()

@classmethod
def clear_all(cls):
    cls.mac_obj_dict = {}
    cls.ip_found_list = []
    cls.ip_last_scanned = None
    cls.ip_last_answered = None
    cls.flag_scan_manual_stop = False
    cls.count_ip_scanned = 0
    cls._update_listbox()

@classmethod
def instance_get_from_text(cls, text):
    # attempt 1 -----------------
    # most correct finding
    for obj in cls.mac_obj_dict.values():
        if obj.mac not in (None, "") and obj.mac in text:
            return obj

    # attempt 2 -----------------
    # try auxiliary finding
    for key in cls.mac_obj_dict:
        if str(key) not in (None, "") and str(key) in text:
            return cls.mac_obj_dict[key]

    # attempt 3 -----------------
    return None

def _instance_print(self):
    for attr in dir(self):
        if not attr.startswith("_") and not callable(getattr(self, attr)):
            print(f"{attr}=[{getattr(self, attr)}]")

# -----------------------------------------------------------
# GENERATE DATA
@classmethod
@contracts.contract(ip_range=tuple)
def ping_range(cls, ip_range):
    ip_start = ipaddress.ip_address(str(ip_range[0]))
    ip_finish = ipaddress.ip_address(str(ip_range[-1]))

    ip_current = ip_start
    while ip_current <= ip_finish and not cls.flag_scan_manual_stop:
        if ip_current not in cls.ip_found_list:   # don't ping if found! it will ping at first in ping_found_hosts func!!!
            cls.ping_start_thread(ip_current)
        ip_current = ip_current + 1
    return

@classmethod
def ping_found_hosts(cls):
    for obj in cls.mac_obj_dict.values():
        cls.ping_start_thread(obj.ip)

@classmethod
@contracts.contract(ip=ipaddress.IPv4Address)
def ping_start_thread(cls, ip):
    thread_name_ping = "ping"
    if ip not in Adapters.ip_margin_set:
        while threading.active_count() > cls.limit_ping_thread:
            time.sleep(0.1)    # USE=0.01
        threading.Thread(target=cls._ping, args=(ip,), daemon=True, name=thread_name_ping).start()
    return

@classmethod
@contracts.contract(ip=ipaddress.IPv4Address)
def _ping(cls, ip):
    # DONT START DIRECTLY!!! USE ONLY THROUGH THREADING!
    cmd_list = ["ping", "-a", "-4", str(ip), "-n", "1", "-l", "0", "-w", str(cls.limit_ping_timewait_ms)]
    """
    -4 = ipv4
    -n = requests count
    -l = request load size
    -i = TTL 
        if add "-i 3" it will get all ghosts when ping ip from outOfHomeNet
        but if "-i 2" it will OK!!!))
    -w = waiting time
    """

    cls.ip_last_scanned = ip
    cls.count_ip_scanned += 1
    sp_ping = subprocess.Popen(cmd_list, text=True, stdout=subprocess.PIPE, encoding="cp866")
    sp_ping.wait()
    ping_readlines = sp_ping.stdout.readlines()
    time.sleep(0.001)   # very necessary =0.001 was good! maybe not need)

    if sp_ping.returncode != 0 and ip in cls.ip_found_list:
        cls._mark_nonactive_ip(ip)
        cls._update_listbox()
        return

    if sp_ping.returncode == 0:
        # ---------------------------------------------------------------------
        # get MAC = use first!!!
        mac = cls._get_mac(ip)

        if mac is None:     # don't pay attention if have not mac! just an accident(ghost)!
            return
        else:
            host_obj = cls()._instance_add_if_not(ip=ip, mac=mac)

        # ---------------------------------------------------------------------
        # get TIME_RESPONSE in ms
        mask = r'.*\sвремя\S(\S+)мс\s.*'
        match = False
        for line in ping_readlines:
            match = re.search(mask, line)
            if match:
                host_obj.time_response = match[1]
                break
        if not match:
            cls._mark_nonactive_ip(ip)
            cls._update_listbox()
            return

        # =====================================================================
        # NOW IT IS REAL POINT THAT IP IS REAL ACTIVE!
        print(f"***************hit=[{ip}]")
        host_obj.active = True
        host_obj.count_response += 1

        cls.ip_last_answered = ip
        cls._mark_nonactive_ip(ip=ip, mac_except=mac)

        # ---------------------------------------------------------------------
        # go out if exists - this code will execute if instance just start filling! first time!
        if host_obj.hostname is not None:
            cls._update_listbox()
            return

        # ---------------------------------------------------------------------
        # get HOSTNAME(+IP)
        if ip in Adapters.ip_localhost_set:
            host_obj.hostname = f"*{Adapters.hostname}*"
        else:
            mask = r'.*\s(\S+)\s\[(\S+)\]\s.*'
            match = False
            for line in ping_readlines:
                match = re.search(mask, line)
                if match:
                    host_obj.hostname = match[1]
                    break

            if not match:
                # some devises don't have hostname! and "ping -a" can't resolve it!
                host_obj.hostname = "NoNameDevice"

        # ---------------------------------------------------------------------
        # NMAP=get OS+VENDOR
        nmap_dict = cls._use_nmap(ip)
        host_obj.os = nmap_dict.get("os", None)
        host_obj.vendor = nmap_dict.get("vendor", None)

        # ---------------------------------------------------------------------
        # exit
        winsound.Beep(1000, 500)
        cls._update_listbox()
    return

# -----------------------------------------------------------
# AUXILIARY
@classmethod
def _update_listbox(cls):
    cls._sort_dict()
    cls.UPDATE_LISTBOX()

@classmethod
def _sort_dict(cls):
    the_dict = cls.mac_obj_dict
    sorted_dict_keys = sorted(the_dict, key=lambda key: the_dict.get(key).ip)
    sorted_dict = dict(zip(sorted_dict_keys, [the_dict[value] for value in sorted_dict_keys]))

    cls.mac_obj_dict = sorted_dict
    return

@classmethod
@contracts.contract(ip=ipaddress.IPv4Address, mac_except="None|str")
def _mark_nonactive_ip(cls, ip, mac_except=None):
    for obj in cls.mac_obj_dict.values():
        if obj.ip == ip and obj.mac != mac_except:
            obj.active = False
            obj.was_lost = True
            obj.time_response = "---"
            obj.count_lost += 1
    return

@classmethod
@contracts.contract(ip=ipaddress.IPv4Address, returns="None|str")
def _get_mac(cls, ip):
    # attempt 1 -----------------
    sp_mac = subprocess.Popen(f"arp -a {str(ip)}", text=True, stdout=subprocess.PIPE, encoding="cp866")
    arp_readlines = sp_mac.stdout.readlines()
    mask = r"[0-9a-fA-F]{2}(?:[:-][0-9a-fA-F]{2}){5}"
    for line in arp_readlines:
        match = re.search(mask, line)
        if match is not None:
            return match[0]

    # attempt 2 -----------------
    # if not returned before, try to find in adapters
    if ip in Adapters.ip_localhost_set:
        for adapter_obj in Adapters.name_obj_dict.values():
            if adapter_obj.ip == ip:
                return adapter_obj.mac

    # attempt 3 -----------------
    return None

@classmethod
@contracts.contract(ip=ipaddress.IPv4Address, returns=dict)
def _use_nmap(cls, ip):
    try:
        ip = str(ip)

        nm = nmap.PortScanner()
        nm.scan(ip, arguments='-O')

        hostname = nm[ip].get("hostnames", None)[0]["name"]     # BLANK value "" at embedded
        mac = nm[ip]["addresses"].get("mac", None)              # can't see KEY at localhost
        vendor = nm[ip].get("vendor", None).get(mac, None)      # can't see KEY at localhost
        os = nm[ip]["osmatch"][0]["name"]
        return {"hostname": hostname, "mac": mac, "vendor": vendor, "os": os}
    except:
        return {"vendor": "install Nmap.EXE", "os": "install Nmap.EXE"}

UPD: мои мысли

  1. первое и минимальное - выделить создаваемые данные во вложенный класс - в целом ОК, читаемость повышается, но поскольку данные я храню все равно в классовом словаре, то все методы остаются все равно классовыми. Тоесть решение из этого пункта приведет к тому, что в классе вообще все методы станут классовыми.
  2. можно вывести такие классы в отдельные модули и тогда все встает на свои места, классовые методы станут обычными функциями модуля, и классовый словарь станет обычным словарем модуля, и будет в модуле класс для непосредственно обьектов, НО эти классы взаимосвязаны между собой, тоесть вызов какого то метода может вызвать запрос на обновление данных в соседнем словаре и чтение его данных и это вроде затрудняет структуру.

Как итог: сделать решение 1 и все!

1 ответ 1

0

в общем так и получилось, добавил внутренние классы для непосредственных данных-обьектов, в целом доволен.

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.