вопрос по организации данных (возможно проще код-ревью попросить...)
Пишу сканер сети на доступные 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)! Еще больше меня смущает то, что в самой программе создается один экземпляр класса (но можно и без него!), а через него по командам (через внутренние скрытые методы) формируются экземпляры этого же класса и они как раз записываются в классовый словарь.
С одной стороны я полностью выделил в один класс код по формированию новых данных и их обработке, и даже очень доволен этим, с другой стороны как-то коряво смотрится... И таких классов хранителей у меня три подряд!
Порядок работы с классом такой:
- из основного кода даем команду проверить ip через класс
- класс пингует и если ответ получен - пролистывает свои данные в классовом словаре и если не находит - запускает скрытый метод для создания новых данных - создает еще один свой же экземпляр через cls().ххх
- вносит новую запись в свой словарь чтобы он не очистился из памяти "ip_2: ip_data_obj_2"
- затем дополняет данные этого экземпляра индивидуальными атрибутами (мас, os, ...)
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 и все!