xinb
/
ip2region_private
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

utils is ready and with unit tests passed

This commit is contained in:
lionsoul2014 2025-10-30 11:37:19 +08:00
parent 6c653f85cd
commit d4d5525392
2 changed files with 146 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
binding/python/util_test.py
View File

@ -33,6 +33,23 @@ def test_version():
print("version_from_header(v4_header) -> ", util.version_from_header(v4_header))
print("version_from_header(v6_header) -> ", util.version_from_header(v6_header))
def test_verify():
# v4 xdb verify
try:
util.verify_from_file(xdb_v4_path)
except Exception as e:
print("failed to verify the xdb file `{}`: {}".format(xdb_v4_path, str(e)))
else:
print("xdb file `{}` verified".format(xdb_v4_path))
# v6 xdb verify
try:
util.verify_from_file(xdb_v6_path)
except Exception as e:
print("failed to verify the xdb file `{}`: {}".format(xdb_v6_path, str(e)))
else:
print("xdb file `{}` verified".format(xdb_v6_path))
def test_load_header():
v4_header = util.load_header_from_file(xdb_v4_path)
v6_header = util.load_header_from_file(xdb_v6_path)
@ -51,6 +68,37 @@ def test_load_content():
print("v4_content.length={}".format(len(v4_content)))
print("v6_content.length={}".format(len(v6_content)))
def test_parse_ip():
ip_list = [
"1.0.0.0", "58.251.30.115", "192.168.1.100", "126.255.32.255", "219.xx.xx.11",
"::", "::1", "fffe::", "2c0f:fff0::", "2c0f:fff0::1", "2a02:26f7:c409:4001::",
"2fff:ffff:ffff:ffff:ffff:ffff:ffff:ffff", "240e:982:e617:ffff:ffff:ffff:ffff:ffff", "::xx:ffff"
]
for ip in ip_list:
try :
ip_bytes = util.parse_ip(ip)
ip_string = util.ip_to_string(ip_bytes)
print("parse_ip({}) -> {{addr:{}, equal:{}}}".format(ip, ip_string, ip_string == ip))
except ValueError as e:
print("failed to parse ip `{}`: {}".format(ip, e))
def test_ip_compare():
ip_list = [
["1.0.0.0", "1.0.0.1", -1],
["192.168.1.101", "192.168.1.90", 1],
["219.133.111.87", "114.114.114.114", 1],
["2000::", "2000:ffff:ffff:ffff:ffff:ffff:ffff:ffff", -1],
["2001:4:112::", "2001:4:112:ffff:ffff:ffff:ffff:ffff", -1],
["ffff::", "2001:4:ffff:ffff:ffff:ffff:ffff:ffff", 1]
]
for ip_pair in ip_list:
ip1 = util.parse_ip(ip_pair[0])
ip2 = util.parse_ip(ip_pair[1])
cmp = util.ip_compare(ip1, ip2)
print("compare({}, {}) -> {} ? {}".format(util.ip_to_string(ip1), util.ip_to_string(ip2), cmp, cmp == ip_pair[2]))
if __name__ == "__main__":
# check and call the specified function
if len(sys.argv) < 2:

127
binding/python/xdb/util.py
View File

@ -6,6 +6,9 @@
# Author Leon<chenxin619315@gmail.com>
import io
import os
import ipaddress
from collections.abc import Callable
# global constants
XdbStructure20 = 2
@ -21,9 +24,6 @@ VectorIndexSize = 8
VectorIndexLength = 524288
class Header(object):
'''
header class
'''
def __init__(self, buff):
self.version = le_get_uint16(buff, 0)
self.indexPolicy = le_get_uint16(buff, 2)
@ -59,22 +59,55 @@ class Header(object):
)
# ---
# ip parse and convert functions
def parse_ip(ip_string: str):
try:
return ipaddress.ip_address(ip_string).packed
except:
raise ValueError("invalid ip address `{}`".format(ip_string))
def ip_to_string(ip_bytes: bytes):
if isinstance(ip_bytes, bytes):
return str(ipaddress.ip_address(ip_bytes))
else:
raise ValueError("invalid bytes ip `{}`".format(ip_bytes))
def ip_compare(ip1: bytes, ip2: bytes):
if ip1 > ip2:
return 1
elif ip1 < ip2:
return -1
else:
return 0
def ip_sub_compare(ip1: bytes, buff: bytes, offset: int):
ip2 = buff[offset:offset+len(ip1)]
if ip1 > ip2:
return 1
elif ip1 < ip2:
return -1
else:
return 0
# ---
# ip version class and functions
class Version(object):
'''
version class
'''
def __init__(self, id, name, byte_num, index_size, ip_compare_func):
def __init__(self, id, name, byte_num, index_size, ip_compare_func: Callable[[bytes, bytes, int], int]):
self.id = id
self.name = name
self.byte_num = byte_num
self.index_size = index_size
self.ip_compare_func = ip_compare_func
def ip_compare(self, ip1, ip2):
def ip_compare(self, ip1: bytes, ip2: bytes):
return self.ip_sub_compare(ip1, ip2, 0)
def ip_sub_compare(self, ip1, ip2, offset):
pass
def ip_sub_compare(self, ip1: bytes, buff: bytes, offset: int):
return self.ip_sub_compare(ip1, buff, offset)
def __str__(self):
return '{{"id": {}, "name": "{}", "bytes": {}, "index_size": {}}}'.format(
@ -84,10 +117,29 @@ class Version(object):
self.index_size
)
def _v4_sub_compare(ip1: bytes, buff: bytes, offset: int):
# ip1: Big endian byte order parsed from input
# ip2: Little endian byte order read from xdb index.
# @Note: to compatible with the old Litten endian index encode implementation.
j = offset + len(ip1) - 1
for i in range(len(ip1)):
i1 = ip1[i]
i2 = buff[j]
if i1 < i2:
return -1
if i1 > i2:
return 1
return 0
# ---
# IPv4 and IPv6 version constants
IPv4 = Version(XdbIPv4Id, "IPv4", 4, 14, None)
IPv6 = Version(XdbIPv6Id, "IPv6", 16, 38, None)
# 14 = 4 + 4 + 2 + 4
IPv4 = Version(XdbIPv4Id, "IPv4", 4, 14, _v4_sub_compare)
# 38 = 16 + 16 + 2 + 4
IPv6 = Version(XdbIPv6Id, "IPv6", 16, 38, ip_sub_compare)
def version_from_name(name):
u_name = name.upper()
@ -113,22 +165,6 @@ def version_from_header(header):
return None
# ---
# ip parse and convert functions
def parse_ip(ip_string):
pass
def ip_to_string(ip_bytes):
pass
def ip_compare(ip1, ip2):
pass
def ip_sub_compare(ip1, ip2, offset):
pass
# ---
# buffer decode functions
@ -195,4 +231,37 @@ def load_content_from_file(db_file):
handle = io.open(db_file, "rb")
c_buff = load_content(handle)
handle.close()
return c_buff
return c_buff
# ---
# Verify if the current Searcher could be used to search the specified xdb file.
# Why do we need this check ?
# The future features of the xdb impl may cause the current searcher not able to work properly.
#
# @Note: You Just need to check this ONCE when the service starts
# Or use another process (eg, A command) to check once Just to confirm the suitability.
def verify(handle):
header = load_header(handle)
# get the runtime ptr bytes
runtime_ptr_bytes = 0
if header.version == XdbStructure20:
runtime_ptr_bytes = 4
elif header.version == XdbStructure30:
runtime_ptr_bytes = header.runtimePtrBytes
else:
# Higher versions of the structure are usually incompatible.
raise ValueError("invalid structure version {}".format(header.version))
# 1, confirm the xdb file size
# to ensure that the maximum file pointer does not overflow
max_file_ptr = (1 << (runtime_ptr_bytes * 8)) - 1
__file_bytes = os.stat(handle.fileno()).st_size
# print("max_file_ptr: {}, file_bytes: {}".format(max_file_ptr, __file_bytes))
if __file_bytes > max_file_ptr:
raise Exception("xdb file exceeds the maximum supported bytes: {}".format(max_file_ptr))
def verify_from_file(db_file):
handle = io.open(db_file, "rb")
verify(handle)
handle.close()