|
import ipaddress |
|
import maxminddb |
|
from fastapi import FastAPI, Request |
|
import json |
|
import datetime |
|
import logging |
|
import sys |
|
import os |
|
from logging.handlers import RotatingFileHandler |
|
|
|
LOG_FILE = os.path.join('/code', 'ip_query.log') |
|
|
|
try: |
|
formatter = logging.Formatter('%(message)s') |
|
log_handler = RotatingFileHandler( |
|
LOG_FILE, |
|
maxBytes=10*1024*1024, |
|
backupCount=5, |
|
encoding='utf-8' |
|
) |
|
log_handler.setFormatter(formatter) |
|
logger = logging.getLogger('ip_query') |
|
logger.setLevel(logging.INFO) |
|
logger.addHandler(log_handler) |
|
startup_log = { |
|
"时间": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), |
|
"事件": "系统启动", |
|
"状态": "成功" |
|
} |
|
logger.info(json.dumps(startup_log, ensure_ascii=False)) |
|
except Exception as e: |
|
print(f"日志初始化失败: {e}") |
|
sys.exit(1) |
|
|
|
city_reader = maxminddb.open_database('GeoLite2-City.mmdb') |
|
asn_reader = maxminddb.open_database('GeoLite2-ASN.mmdb') |
|
cn_reader = maxminddb.open_database('GeoCN.mmdb') |
|
lang = ["zh-CN", "en"] |
|
asn_map = { |
|
9812: "东方有线", |
|
9389: "中国长城", |
|
17962: "天威视讯", |
|
17429: "歌华有线", |
|
7497: "科技网", |
|
24139: "华数", |
|
9801: "中关村", |
|
4538: "教育网", |
|
24151: "CNNIC", |
|
38019: "中国移动", 139080: "中国移动", 9808: "中国移动", 24400: "中国移动", 134810: "中国移动", 24547: "中国移动", |
|
56040: "中国移动", 56041: "中国移动", 56042: "中国移动", 56044: "中国移动", 132525: "中国移动", 56046: "中国移动", |
|
56047: "中国移动", 56048: "中国移动", 59257: "中国移动", 24444: "中国移动", |
|
24445: "中国移动", 137872: "中国移动", 9231: "中国移动", 58453: "中国移动", |
|
4134: "中国电信", 4812: "中国电信", 23724: "中国电信", 136188: "中国电信", 137693: "中国电信", 17638: "中国电信", |
|
140553: "中国电信", 4847: "中国电信", 140061: "中国电信", 136195: "中国电信", 17799: "中国电信", 139018: "中国电信", |
|
134764: "中国电信", 4837: "中国联通", 4808: "中国联通", 134542: "中国联通", 134543: "中国联通", |
|
59019: "金山云", |
|
135377: "优刻云", |
|
45062: "网易云", |
|
37963: "阿里云", 45102: "阿里云国际", |
|
45090: "腾讯云", 132203: "腾讯云国际", |
|
55967: "百度云", 38365: "百度云", |
|
58519: "华为云", 55990: "华为云", 136907: "华为云", |
|
4609: "澳門電訊", |
|
13335: "Cloudflare", |
|
55960: "亚马逊云", 14618: "亚马逊云", 16509: "亚马逊云", |
|
15169: "谷歌云", 396982: "谷歌云", 36492: "谷歌云", |
|
} |
|
|
|
def get_as_info(number): |
|
r = asn_map.get(number) |
|
if r: |
|
return r |
|
|
|
def get_des(d): |
|
for i in lang: |
|
if i in d['names']: |
|
return d['names'][i] |
|
return d['names']['en'] |
|
|
|
def get_country(d): |
|
r = get_des(d) |
|
if r in ["香港", "澳门", "台湾"]: |
|
return "中国" + r |
|
return r |
|
|
|
def province_match(s): |
|
arr = ['内蒙古', '黑龙江', '河北', '山西', '吉林', '辽宁', '江苏', '浙江', '安徽', '福建', '江西', '山东', '河南', '湖北', '湖南', '广东', '海南', '四川', '贵州', '云南', '陕西', '甘肃', '青海', '广西', '西藏', '宁夏', '新疆', '北京', '天津', '上海', '重庆'] |
|
for i in arr: |
|
if i in s: |
|
return i |
|
return '' |
|
|
|
def de_duplicate(regions): |
|
regions = filter(bool, regions) |
|
ret = [] |
|
[ret.append(i) for i in regions if i not in ret] |
|
return ret |
|
|
|
def get_addr(ip, mask): |
|
network = ipaddress.ip_network(f"{ip}/{mask}", strict=False) |
|
first_ip = network.network_address |
|
return f"{first_ip}/{mask}" |
|
|
|
def get_maxmind(ip: str): |
|
ret = {"ip": ip} |
|
asn_info = asn_reader.get(ip) |
|
if asn_info: |
|
as_ = {"number": asn_info["autonomous_system_number"], "name": asn_info["autonomous_system_organization"]} |
|
info = get_as_info(as_["number"]) |
|
if info: |
|
as_["info"] = info |
|
ret["as"] = as_ |
|
|
|
city_info, prefix = city_reader.get_with_prefix_len(ip) |
|
ret["addr"] = get_addr(ip, prefix) |
|
if not city_info: |
|
return ret |
|
|
|
if "location" in city_info: |
|
location = city_info["location"] |
|
ret["location"] = { |
|
"latitude": location.get("latitude"), |
|
"longitude": location.get("longitude") |
|
} |
|
|
|
if "country" in city_info: |
|
country_code = city_info["country"]["iso_code"] |
|
country_name = get_country(city_info["country"]) |
|
ret["country"] = {"code": country_code, "name": country_name} |
|
|
|
if "registered_country" in city_info: |
|
registered_country_code = city_info["registered_country"]["iso_code"] |
|
ret["registered_country"] = {"code": registered_country_code, "name": get_country(city_info["registered_country"])} |
|
|
|
regions = [get_des(i) for i in city_info.get('subdivisions', [])] |
|
|
|
if "city" in city_info: |
|
c = get_des(city_info["city"]) |
|
if (not regions or c not in regions[-1]) and c not in country_name: |
|
regions.append(c) |
|
|
|
regions = de_duplicate(regions) |
|
if regions: |
|
ret["regions"] = regions |
|
|
|
return ret |
|
|
|
def get_cn(ip: str, info={}): |
|
ret, prefix = cn_reader.get_with_prefix_len(ip) |
|
if not ret: |
|
return |
|
info["addr"] = get_addr(ip, prefix) |
|
regions = de_duplicate([ret["province"], ret["city"], ret["districts"]]) |
|
if regions: |
|
info["regions"] = regions |
|
info["regions_short"] = de_duplicate([province_match(ret["province"]), ret["city"].replace('市', ''), ret["districts"]]) |
|
if "as" not in info: |
|
info["as"] = {} |
|
info["as"]["info"] = ret['isp'] |
|
if ret['net']: |
|
info["type"] = ret['net'] |
|
return ret |
|
|
|
def get_ip_info(ip): |
|
info = get_maxmind(ip) |
|
if "country" in info and info["country"]["code"] == "CN" and ("registered_country" not in info or info["registered_country"]["code"] == "CN"): |
|
get_cn(ip, info) |
|
return info |
|
|
|
def query(): |
|
while True: |
|
try: |
|
ip = input('IP： \t').strip() |
|
info = get_ip_info(ip) |
|
print(f"网段：\t{info['addr']}") |
|
if "location" in info: |
|
print(f"经纬度：\t{info['location']['latitude']}, {info['location']['longitude']}") |
|
if "as" in info: |
|
print(f"ISP：\t", end=' ') |
|
if "info" in info["as"]: |
|
print(info["as"]["info"], end=' ') |
|
else: |
|
print(info["as"]["name"], end=' ') |
|
if "type" in info: |
|
print(f"({info['type']})", end=' ') |
|
print(f"ASN{info['as']['number']}", end=' ') |
|
print(info['as']["name"]) |
|
if "registered_country" in info and ("country" not in info or info["country"]["code"] != info["registered_country"]["code"]): |
|
print(f"注册地：\t{info['registered_country']['name']}") |
|
if "country" in info: |
|
print(f"使用地：\t{info['country']['name']}") |
|
if "regions" in info: |
|
print(f"位置： \t{' '.join(info['regions'])}") |
|
except Exception as e: |
|
print(e) |
|
raise e |
|
finally: |
|
print("\n") |
|
|
|
app = FastAPI() |
|
|
|
@app.get("/") |
|
async def api(request: Request, ip: str = None): |
|
client_ip = request.headers.get("x-forwarded-for") or request.headers.get("x-real-ip") or request.client.host |
|
query_ip = ip.strip() if ip else client_ip |
|
result = get_ip_info(query_ip) |
|
log_data = { |
|
"时间": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), |
|
"访问IP": client_ip, |
|
"查询IP": query_ip, |
|
"请求头": dict(request.headers), |
|
"查询结果": result |
|
} |
|
logger.info(json.dumps(log_data, ensure_ascii=False)) |
|
return result |
|
|
|
@app.get("/{ip}") |
|
async def path_api(request: Request, ip: str): |
|
client_ip = request.headers.get("x-forwarded-for") or request.headers.get("x-real-ip") or request.client.host |
|
result = get_ip_info(ip) |
|
log_data = { |
|
"时间": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), |
|
"访问IP": client_ip, |
|
"查询IP": ip, |
|
"请求头": dict(request.headers), |
|
"查询结果": result |
|
} |
|
logger.info(json.dumps(log_data, ensure_ascii=False)) |
|
return result |
|
|
|
if __name__ == '__main__': |
|
query() |
|
import uvicorn |
|
uvicorn.run(app, host="0.0.0.0", port=8080, server_header=False, proxy_headers=True) |
|
|
