python开发ip2region 离线IP库地址文件

一、项目简介

ip2region - 离线的ip地址查询库,ip到地区的映射库,提供二进制,B树,内存搜索三种查询算法,查询速度非常快。
支持Java,PHP,C,Python,Nodejs,Golang,C#等语言,本文以Python为例

下载地址:https://github.com/lionsoul2014/ip2region

项目测试代码:本例中ip2region.db与testSearcher.py在同一个目录

D:\code\ip2region\binding\python>python testSearcher.py ip2region.db
initializing b-tree...
+----------------------------------+
| ip2region test program |
| Author: chenxin619315@gmail.com. |
| Type 'quit' to exit program |
+----------------------------------+
ip2region>> 45.15.11.150
0|中国|0|香港|0|0 in 98.978027 millseconds

ip2region>> 206.189.50.204
14|新加坡|0|0|0|0 in 0.000000 millseconds
ip2region>> 106.13.211.178
2140|中国|0|广东省|广州市|电信 in 0.000000 millseconds
ip2region>>

二、详细分析

python开发ip2region 离线IP库地址文件

 

 

python 示例代码:testSearcher.py、ip2Region.py、benchmark.py

python开发ip2region 离线IP库地址文件

 

 

 

离线代码库ip2region.db在data目录下。

执行test.py示例如下,生成结果文件result.csv。

python开发ip2region 离线IP库地址文件

result.csv

python开发ip2region 离线IP库地址文件

 

 输入文件是ip.txt。

python开发ip2region 离线IP库地址文件

三、代码test.py

import time
from ip2Region import Ip2Region
import csv

def testSearch(ip_s):
    dbFile = "ip2region.db"
    searcher = Ip2Region(dbFile)
    try:
        print("开始检测:",(ip_s))
        sTime = time.time() * 1000
        data = searcher.binarySearch(ip_s)
        # elif algorithm == "memory":
        #     data = searcher.memorySearch(line)
        # else:
        #     data = searcher.btreeSearch(line)
        eTime = time.time() * 1000
        ip_info = ("%s|%s|%s" % (ip_s, data["city_id"], data["region"].decode('utf-8')))
        print("检测完成:" + ip_info)
        return ip_info

    except Exception as e:
        print("[Error]: %s" % e)
    searcher.close()


def all_in():

    c = open("result.csv","w", newline="")
    writer=csv.writer(c)
    title = ['源ip', '国家', '省市','运营商']
    writer.writerow(title)
    with open('ip.txt' ,'r') as file:
        for line in file.readlines():
            ip = line.strip()
            try:
                data = testSearch(ip)
                ct = data.split('|')[2].strip()
                pv = data.split('|')[4].strip()
                city = data.split('|')[5].strip()
                yys = data.split('|')[6].strip()
                if ct == "0":
                    print("地址库中未找到对应的IP归属地,请更新地址库或者确定ip准确性!")
                    writer.writerow([ip, "/", "/","/"])
                else:
                    if pv == "0":
                        print("省市查询为空!")
                        writer.writerow([ip, ct, "/","/"])
                    else:
                        if city == "0":
                            writer.writerow([ip, ct, pv + "-" + "///","/"])
                        else:
                            if yys == "0":
                                writer.writerow([ip, ct, pv + "-" + city,"/"])
                            else:
                                writer.writerow([ip, ct, pv + "-" + city,yys])
            except Exception as e:
                print("[Error]: %s" % e)
                writer.writerow([ip,"检测异常,请手动检测!"])
    c.close()

if __name__ == "__main__":
    all_in()

 

上一篇:postgresql的CopyManager流式数据入库


下一篇:自定义日志类