| 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) |
|
|
