IP离线库与网站集成

#-*-coding:utf-8-*-
importcsv
importsqlite3
fromfunctoolsimportlru_cache
#=================数据导入=================
defimport_ip_data(csv_path,db_path='ip_database.db'):
"""CSV数据导入SQLite数据库示例"""
conn=sqlite3.connect(db_path)
c=conn.cursor()
#创建数据表
c.execute('''CREATETABLEIFNOTEXISTSip_ranges
(ip_startINT,ip_endINT,
countryTEXT,provinceTEXT,cityTEXT,
ispTEXT)''')
#读取CSV并插入数据
withopen(csv_path,'r',encoding='utf-8')asf:
reader=csv.reader(f)
forrowinreader:
#假设CSV格式：起始IP,结束IP,国家,省份,城市,运营商
#实际需根据离线库具体格式调整
ip_start=ip_to_int(row[0])
ip_end=ip_to_int(row[1])
c.execute('INSERTINTOip_rangesVALUES(?,?,?,?,?,?)',
(ip_start,ip_end,row[2],row[3],row[4],row[5]))
#创建查询索引
c.execute('CREATEINDEXidx_rangeONip_ranges(ip_start,ip_end)')
conn.commit()
conn.close()
defip_to_int(ip_str):
"""将点分十进制IP转换为整数"""
segments=list(map(int,ip_str.split('.')))
return(segments[0]<<24)+(segments[1]<<16)+(segments[2]<<8)+segments[3]
#=================接口开发=================
classIPQuery:
def__init__(self,db_path='ip_database.db'):
self.conn=sqlite3.connect(db_path)
@lru_cache(maxsize=1024)#缓存常用查询结果
defquery(self,ip):
"""IP查询接口"""
ip_num=ip_to_int(ip)
cursor=self.conn.cursor()
cursor.execute('''
SELECTcountry,province,city,isp
FROMip_ranges
WHERE?BETWEENip_startANDip_end
LIMIT1
''',(ip_num,))
returncursor.fetchone()or('Unknown','Unknown','Unknown','Unknown')
#=================功能调用=================
if__name__=='__main__':
#数据导入（首次运行时执行）
#import_ip_data('ip_data.csv')
#初始化查询器
ip_query=IPQuery()
#模拟用户登录场景
defuser_login(username,ip):
location=ip_query.query(ip)
print(f"[登录日志]用户{username}从{location[2]}地区登录，IP:{ip}")
#后续业务逻辑...
#模拟访问日志处理
access_log=[
('user1','202.106.0.20'),
('visitor','140.205.220.96'),
('admin','221.196.0.0')
]
forusername,ipinaccess_log:
user_login(username,ip)Copy