在纯真IP数据库（qqwry.dat）里查询IP地址归属地，Python 3版
特点：1、for Python 3.0+。
2、有两套实现供选择。有一个查找速度更快，但加载稍慢一点、占用内存稍多一点。
3、i3 3.6GHz，在Python 3.5上查询速度达10.2万次/秒。
4、仅import array和bisect这两个Python自带的模块。
5、经过较严格测试。
6、另提供一个从纯真网络(cz88.net)更新qqwry.dat的小工具，见github项目。
用法：
from qqwry import QQwry
q = QQwry()
q.load_file('qqwry.dat', loadindex=False)
q.lookup('8.8.8.8')
详细用法见qqwry.py文件里面的说明
解释load_file函数的loadindex参数：
loadindex=False（默认）
loadindex=True
说明
把整个文件读入内存。
把整个文件读入内存。
额外加载索引，把索引读入更快的数据结构。
加载速度
稍快
稍慢
进程占用内存
12.6 MB
17.7 MB
查询速度
3.9 万次/秒
10.2 万次/秒
备注
适合桌面程序
适合服务器程序
以上是在i3 3.6GHz, Win10, Python 3.5.0rc2 64bit，qqwry.dat 8.85MB时的数据
github上的项目：
https://github.com/animalize/qqwry-python3
源码：
# coding=utf-8
#
# for Python 3.0+
# 来自： https://github.com/animalize/qqwry-python3
# 版本：2015-10-23
#
# 用法：
# from qqwry import QQwry
# q = QQwry()# 生成对象
# q.load_file(filename, loadindex=False) # 加载qqwry.dat文件
# q.lookup('8.8.8.8')# 查询IP地址
#
# q.load_file(filename, loadindex=False)函数:
# 参数filename可以是qqwry.dat的文件名，也可以为bytes类型的文件内容
# 参数loadindex为False时，把qqwry.dat整体读入内存，不额外加载索引
# 参数loadindex为True时，把qqwry.dat整体读入内存，额外加载索引
# 后者比前者查找更快（3.9万次/秒，10.2万次/秒）
# 后者比前者占用更多内存（测试进程的内存分别为12.6MB，17.7MB）
# 后者比前者加载稍慢
# 以上是在i3 3.6GHz, Win10, Python 3.5.0rc2 64bit，qqwry.dat 8.85MB时的数据
# 函数成功返回True，失败返回False
#
# q.lookup('8.8.8.8')函数:
# 找到则返回一个含有两个字符串的元组：('国家', '省份')
# 没有找到结果返回一个None
#
# q.get_lastone()函数:
# 返回最后一条数据，最后一条通常为数据的版本号
# 没有数据则返回None
#
# q.is_loaded()函数:
# 是否已加载数据，返回True或False
#
# q.clear()函数:
# 清空已加载的qqwry.dat
# 再次调用load_file时不必执行q.clear()
import array
import bisect
__all__ = ('QQwry')
def int3(data, offset):
return data + (data << 8) + \
(data << 16)
def int4(data, offset):
return data + (data << 8) + \
(data << 16) + (data << 24)
class QQwry:
def __init__(self):
self.clear()
def clear(self):
self.idx1 = None
self.idx2 = None
self.idxo = None
self.data = None
self.index_begin = -1
self.index_end = -1
self.index_count = -1
self.__fun = None
def load_file(self, filename, loadindex=False):
self.clear()
if type(filename) == bytes:
self.data = buffer = filename
filename = 'memory data'
elif type(filename) == str:
# read file
try:
with open(filename, 'br') as f:
self.data = buffer = f.read()
except Exception as e:
print('打开、读取文件时出错：', e)
self.clean()
return False
if self.data == None:
print('%s load failed' % filename)
self.clear()
return False
else:
self.clean()
return False
if len(buffer) < 8:
print('%s load failed, file only %d bytes' %
(filename, len(buffer))
)
self.clear()
return False
# index range
index_begin = int4(buffer, 0)
index_end = int4(buffer, 4)
if index_begin > index_end or \
(index_end - index_begin) % 7 != 0 or \
index_end + 7 > len(buffer):
print('%s index error' % filename)
self.clear()
return False
self.index_begin = index_begin
self.index_end = index_end
self.index_count = (index_end - index_begin) // 7 + 1
if not loadindex:
print('%s %s bytes, %d segments. without index.' %
(filename, format(len(buffer),','), self.index_count)
)
self.__fun = self.__raw_search
return True
# load index
self.idx1 = array.array('L')
self.idx2 = array.array('L')
self.idxo = array.array('L')
try:
for i in range(self.index_count):
ip_begin = int4(buffer, index_begin + i*7)
offset = int3(buffer, index_begin + i*7 + 4)
# load ip_end
ip_end = int4(buffer, offset)
self.idx1.append(ip_begin)
self.idx2.append(ip_end)
self.idxo.append(offset+4)
except:
print('%s load index error' % filename)
self.clear()
return False
print('%s %s bytes, %d segments. with index.' %
(filename, format(len(buffer),','), len(self.idx1))
)
self.__fun = self.__index_search
return True
def __get_addr(self, offset):
# mode 0x01, full jump
mode = self.data
if mode == 1:
offset = int3(self.data, offset+1)
mode = self.data
# country
if mode == 2:
off1 = int3(self.data, offset+1)
c = self.data
offset += 4
else:
c = self.data
offset += len(c) + 1
# province
if self.data == 2:
offset = int3(self.data, offset+1)
p = self.data
return c.decode('gb18030', errors='replace'), \
p.decode('gb18030', errors='replace')
def lookup(self, ip_str):
try:
ip = sum(256**j*int(i) for j,i
in enumerate(ip_str.strip().split('.')[::-1]))
return self.__fun(ip)
except:
return None
def __raw_search(self, ip):
l = 0
r = self.index_count
while r - l > 1:
m = (l + r) // 2
offset = self.index_begin + m * 7
new_ip = int4(self.data, offset)
if ip < new_ip:
r = m
else:
l = m
offset = self.index_begin + 7 * l
ip_begin = int4(self.data, offset)
offset = int3(self.data, offset+4)
ip_end = int4(self.data, offset)
if ip_begin <= ip <= ip_end:
return self.__get_addr(offset+4)
return None
def __index_search(self, ip):
posi = bisect.bisect_right(self.idx1, ip) - 1
if posi >= 0 and self.idx1 <= ip <= self.idx2:
return self.__get_addr(self.idxo)
else:
return None
def is_loaded(self):
return self.__fun != None
def get_lastone(self):
try:
offset = int3(self.data, self.index_end+4)
return self.__get_addr(offset+4)
except:
return None
if __name__ == '__main__':
import sys
if len(sys.argv) > 1:
fn = 'qqwry.dat'
q = QQwry()
q.load_file(fn)
for ipstr in sys.argv:
s = q.lookup(ipstr)
print('%s\n%s' % (ipstr, s))
else:
print('请以查询ip作为参数运行')
页:
[1]