alan/bird_list_ip
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Добавлена полная поддержка выгрузки ipv6 по аналогии с ipv4. По умолчанию ipv6 не выгружается, см. конфигурацию. Изменена структура списка выгрузки. Параметры конфигурации вынесены глобально для каждой выгрузки, со значениями по умолчанию и являются не обязательными.

Browse Source
This commit is contained in:
Александр
2025-11-21 17:44:10 +10:00
parent a4eee42ec6
commit 87bba5279a
3 changed files with 254 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

199
include/net_tree.py
View File

@@ -1,51 +1,127 @@
#
# Unified IPv4/IPv6 Network Aggregator
#
BIG_MASK = (1 << 32) - 1
def detect_ip_version(ip_int: int):
"""Определяем IPv4 или IPv6 по величине числа"""
if ip_int <= 0xFFFFFFFF:
return 4, 32
else:
return 6, 128
def getMaskByMaskSize(mask_size):
return BIG_MASK ^ ((1 << (32 - mask_size)) - 1)
def getIpVolumeByMaskSize(mask_size):
return 1 << (32 - mask_size)
def get_mask_by_mask_size(mask_size, total_bits):
return ((1 << total_bits) - 1) ^ ((1 << (total_bits - mask_size)) - 1)
def get_ip_volume(mask_size, total_bits):
return 1 << (total_bits - mask_size)
def int_to_ipv4(n):
return ".".join(str((n >> (24 - 8*i)) & 0xFF) for i in range(4))
def int_to_ipv6(n):
# разбор на 8 блоков по 16 бит
blocks = [(n >> (112 - 16*i)) & 0xFFFF for i in range(8)]
# убираем ведущие нули (схлопывание "::")
best_start = -1
best_len = 0
cur_start = -1
cur_len = 0
for i in range(8):
if blocks[i] == 0:
if cur_start == -1:
cur_start = i
cur_len = 1
else:
cur_len += 1
else:
if cur_len > best_len:
best_len = cur_len
best_start = cur_start
cur_start = -1
cur_len = 0
if cur_len > best_len:
best_len = cur_len
best_start = cur_start
if best_len > 1:
new_blocks = []
i = 0
while i < 8:
if i == best_start:
new_blocks.append("")
i += best_len
else:
new_blocks.append(format(blocks[i], "x"))
i += 1
res = ":".join(new_blocks)
# иногда получается ":::" → исправим
while ":::" in res:
res = res.replace(":::", "::")
return res
else:
return ":".join(format(b, "x") for b in blocks)
def int_to_ip(n, version):
return int_to_ipv4(n) if version == 4 else int_to_ipv6(n)
class Net:
__slots__ = ['mask_size', 'net', 'mask', 'ip_volume']
__slots__ = ['mask_size', 'net', 'mask', 'ip_volume', 'version', 'total_bits']
def __init__(self, net: int, mask_size: int):
# Определяем IPv4/IPv6
self.version, self.total_bits = detect_ip_version(net)
self.mask_size = mask_size
self.net = net & getMaskByMaskSize(mask_size)
self.mask = getMaskByMaskSize(self.mask_size)
self.ip_volume = getIpVolumeByMaskSize(mask_size)
def hasSubnet(self, Net: 'Net'):
if Net.mask_size <= self.mask_size: return 0
return self.net == Net.net & self.mask
self.mask = get_mask_by_mask_size(mask_size, self.total_bits)
self.net = net & self.mask
self.ip_volume = get_ip_volume(mask_size, self.total_bits)
def isSameNet(self, Net: 'Net'):
return (Net.mask_size == self.mask_size) and (Net.net == self.net)
def hasSubnet(self, other: 'Net'):
if other.version != self.version:
return 0
if other.mask_size <= self.mask_size:
return 0
return self.net == (other.net & self.mask)
def getCommonNet(self, OtherNet: 'Net', min_mask_size: int):
def isSameNet(self, other: 'Net'):
return (
self.version == other.version and
self.mask_size == other.mask_size and
self.net == other.net
)
def getCommonNet(self, other: 'Net', min_mask_size: int):
if self.version != other.version:
return 0
if self.mask_size <= min_mask_size: return 0
if OtherNet.mask_size <= min_mask_size: return 0
for mask_size in range(min(self.mask_size, OtherNet.mask_size) - 1, min_mask_size - 1, -1):
mask = getMaskByMaskSize(mask_size)
if (self.net & mask) == (OtherNet.net & mask):
return Net(self.net, mask_size)
if other.mask_size <= min_mask_size: return 0
upper = min(self.mask_size, other.mask_size) - 1
for mask_size in range(upper, min_mask_size - 1, -1):
mask = get_mask_by_mask_size(mask_size, self.total_bits)
if (self.net & mask) == (other.net & mask):
return Net(self.net, mask_size)
return 0
def getAsString(self, fmt='{addr}/{masklen}'):
net = self.net
mask = self.mask
addrbytes = []
maskbytes = []
for i in range(4):
addrbytes.append(str(net % 256))
maskbytes.append(str(mask % 256))
net = net >> 8
mask = mask >> 8
return fmt.format(addr='.'.join(reversed(addrbytes)), mask='.'.join(reversed(maskbytes)), masklen=self.mask_size)
return fmt.format(
addr=int_to_ip(self.net, self.version),
masklen=self.mask_size
)
class Node:
__slots__ = ['net', 'child1', 'child2', 'is_real_net', 'real_ip_volume', 'real_ip_records_count', 'weight', 'max_child_weight', 'added_fake_ip_volume']
__slots__ = ['net', 'child1', 'child2', 'is_real_net', 'real_ip_volume',
'real_ip_records_count', 'weight', 'max_child_weight', 'added_fake_ip_volume']
def __init__(self, net: Net, is_real_net: int):
self.net = net
@@ -79,47 +155,23 @@ class Node:
if Child and Child.addSubnet(NewNode):
return 1
if self.child1:
CommonNet = self.child1.net.getCommonNet(NewNode.net, self.net.mask_size + 1)
if CommonNet:
CommonNode = Node(CommonNet, 0)
CommonNode.addSubnet(NewNode)
CommonNode.addSubnet(self.child1)
self.child1 = CommonNode
return 1
if self.child2:
CommonNet = self.child2.net.getCommonNet(NewNode.net, self.net.mask_size + 1)
if CommonNet:
CommonNode = Node(CommonNet, 0)
CommonNode.addSubnet(NewNode)
CommonNode.addSubnet(self.child2)
self.child2 = CommonNode
return 1
for child_attr in ('child1', 'child2'):
Child = getattr(self, child_attr)
if Child:
CommonNet = Child.net.getCommonNet(NewNode.net, self.net.mask_size + 1)
if CommonNet:
CommonNode = Node(CommonNet, 0)
CommonNode.addSubnet(NewNode)
CommonNode.addSubnet(Child)
setattr(self, child_attr, CommonNode)
return 1
if not self.child1:
self.child1 = NewNode
else:
self.child2 = NewNode
return 1
def printTree(self, level):
prefix = ''
for i in range(level):
prefix = prefix + ' '
if self.is_real_net: sign = '*'
elif self.weight == 0: sign = '.'
else: sign = ''
print(prefix + self.net.getAsString() + ' ' + str(self.real_ip_records_count))
if self.child1:
self.child1.printTree(level + 1)
if self.child2:
self.child2.printTree(level + 1)
def finishTreeFirst(self):
if self.is_real_net:
self.real_ip_volume = self.net.ip_volume
@@ -139,7 +191,6 @@ class Node:
self.recalcWeight()
def collapse(self, min_weight, max_net_delta):
# trying to collapse self
if self.weight >= min_weight:
self.weight = 0
self.max_child_weight = 0
@@ -150,6 +201,7 @@ class Node:
net_delta = 0
fake_ip_delta = 0
self.max_child_weight = 0
for Child in (self.child1, self.child2):
if Child:
if net_delta < max_net_delta and min_weight <= max(Child.weight, Child.max_child_weight):
@@ -163,7 +215,6 @@ class Node:
self.real_ip_records_count -= net_delta
self.recalcWeight()
# trying to collapse self
if self.weight >= min_weight:
self.weight = 0
self.max_child_weight = 0
@@ -178,14 +229,6 @@ class Node:
delta, fake_ip_volume = self.collapse(self.max_child_weight, required_net_delta)
required_net_delta -= delta
def printCollapsedTree(self, fmt='{addr}/{masklen}'):
if self.is_real_net or self.weight == 0:
print(self.net.getAsString(fmt))
else:
for Child in (self.child1, self.child2):
if Child:
Child.printCollapsedTree(fmt)
def returnCollapsedTree(self, fmt='{addr}/{masklen}'):
if self.is_real_net or self.weight == 0:
return self.net.getAsString(fmt) + "\n"
@@ -198,8 +241,8 @@ class Node:
def recalcWeight(self):
fake_ip_delta = self.net.ip_volume - self.real_ip_volume - self.added_fake_ip_volume
if fake_ip_delta:
self.weight = (self.real_ip_records_count - 1) / fake_ip_delta
if fake_ip_delta > 0:
self.weight = (self.real_ip_records_count - 1) / (fake_ip_delta ** 0.5)
else:
self.weight = float('Inf')
@@ -209,5 +252,5 @@ class Node:
res = 0
for Child in (self.child1, self.child2):
if Child:
res = res + Child.getNotRealIpCount()
res += Child.getNotRealIpCount()
return res