纯真IP地址库setup.exe程序提取qqwry.dat文件并解析txt纯文本

一. qqwry.dat文件提取

setup.exe是Windows运行程序，如果有Windows程序的话，直接点击安装，然后从安装目录里面获取 qqwry.dat 文件即可。假如我没有Windows系统怎么办，可以安装innoextract命令行工具提取 qqwry.dat。

下载地址 https://update.cz88.net/help?id=free，需要关注微信公众号，然后获取最新的下载地址，下载后会得到类似 su9Izakq-2023-08-23.zip 的压缩包，解压后就有 setup.exe 文件了。

• 安装工具 innoextract

# macos
brew install innoextract

# centos
sudo yum install innoextract

# ubuntu
sudo apt-get install innoextract

• 解压setup.ext

$ innoextract setup.exe
Extracting "纯真IP地址数据库" - setup data version 6.1.0 (unicode)
 - "app/ip.exe"
 - "app/qqwry.dat"
 - "app/纯真网络.url"
 - "app/License.txt"
 - "app/说明.txt"
Done.

二. qqwry.dat文件转换为txt

• qqwry.dat 的格式如下

+----------+
|  文件头  |  (8字节)
+----------+
|  记录区  | （不定长）
+----------+
|  索引区  | （大小由文件头决定）
+----------+

文件头：4字节开始索引偏移值+4字节结尾索引偏移值

记录区： 每条IP记录格式 ==> IP地址[国家信息][地区信息]

索引区： 每条索引记录格式 ==> 4字节起始IP地址 + 3字节指向IP记录的偏移值

• 使用Python3解析qqwry.dat的代码如下
• 输入文件 qqwry.dat，输出文件 qqwry.txt

import socket
import codecs
import mmap
from struct import pack, unpack


class QQWry:

    def __init__(self, path):
        self.path = path
        self.db = None
        self.open_db()
        self.idx_start, self.idx_end = self._read_idx()
        self.total = (self.idx_end - self.idx_start) / 7 + 1

    @staticmethod
    def decode_str(old):
        try:
            return old.decode('gbk')
        except:
            print("decode string error: " + old)
            return 'Invalid'

    def open_db(self):
        self.db = open(self.path, 'rb')
        self.db = mmap.mmap(self.db.fileno(), 0, access=1)

    def _read_idx(self):
        self.db.seek(0)
        start = unpack('I', self.db.read(4))[0]
        end = unpack('I', self.db.read(4))[0]
        return start, end

    def version(self):
        ip_end_offset = self.read_offset(self.idx_end + 4)
        a_raw, b_raw = self.read_record(ip_end_offset + 4)
        return self.decode_str(a_raw + b_raw)

    def read_offset(self, off, seek=True):
        if seek:
            self.db.seek(off)
        buf = self.db.read(3)
        return unpack('I', buf + b'\0')[0]

    def read_record(self, offset):
        self.db.seek(offset)
        flag = ord(self.db.read(1))
        if flag == 1:
            buf = self.db.read(3)
            a_offset = unpack('I', buf + b'\0')[0]
            a_raw = self.read_string(a_offset)
            a_flag = self.get_flag(a_offset)
            if a_flag == 2:
                b_raw = self.read_string(a_offset + 4)
            else:
                b_raw = self.read_string(a_offset + len(a_raw) + 1)
        elif flag == 2:
            buf = self.db.read(3)
            a_offset = unpack('I', buf + b'\0')[0]
            a_raw = self.read_string(a_offset)
            b_raw = self.read_string(offset + 4)
        else:
            a_raw = self.read_string(offset)
            b_raw = self.read_string(offset + len(a_raw) + 1)
        return a_raw, b_raw

    def read_ip(self, off, seek=True):
        if seek:
            self.db.seek(off)
        buf = self.db.read(4)
        return unpack('I', buf)[0]

    def read_string(self, offset):
        if offset == 0:
            return 'N/A1'
        flag = self.get_flag(offset)
        if flag == 0:
            return 'N/A2'
        elif flag == 2:
            offset = self.read_offset(offset + 1)
            return self.read_string(offset)
        self.db.seek(offset)
        raw_string = b''
        while True:
            x = self.db.read(1)
            if x == b'\0':
                break
            raw_string += x
        return raw_string

    def get_flag(self, offset):
        self.db.seek(offset)
        c = self.db.read(1)
        if not c:
            return 0
        return ord(c)

    def output(self, output_file):
        fp = codecs.open(output_file, 'w', 'utf8')
        fp.write(self.version() + '\n')
        idx = self.idx_start
        while idx <= self.idx_end:
            ip_int = self.read_ip(idx)
            ip_start = socket.inet_ntoa(pack('!I', ip_int))
            ip_end_offset = self.read_offset(idx + 4)
            ip_int = self.read_ip(ip_end_offset)
            ip_end = socket.inet_ntoa(pack('!I', ip_int))
            a_raw, b_raw = self.read_record(ip_end_offset + 4)
            a_info = self.decode_str(a_raw)
            b_info = self.decode_str(b_raw)
            fp.write(u'%15s\t%15s\t%s,%s\n' % (ip_start, ip_end, a_info, b_info))
            idx += 7
        fp.close()


if __name__ == '__main__':
    qqwry = QQWry("./qqwry.dat")
    qqwry.output("./qqwry.txt")

三. 使用在线IP查询

我使用纯真IP库、geoip2、asn IP段归属，部署了一个在线查询IP归属地的服务，你可以通过它查询指定IP地址的归属地信息，如果不指定任何IP地址，会返回你自己的公网IP地址以及这个IP地址的详细归属地。

访问博主的在线工具网站查询IP归属地：https://goldtools.cn/network/iplocation

也可以直接请求API获取数据

# 查询指定IP地址的归属地
$ curl ip.goldtools.cn/1.51.160.0
{
  "ip" : "1.51.160.0",
  "city" : "山东省潍坊市",
  "region" : "教育网/潍坊学院"
}

# 查询自己的IP地址，以及归属地
$ curl ip.goldtools.cn
{
  "ip" : "43.230.68.120",
  "city" : "北京市",
  "region" : "联通"
}

• 参考项目 https://github.com/gwind/ylinux/