zxwreader/IPAPI

1import ipaddress2import maxminddb3from fastapi import FastAPI, Request4import json5import datetime6import logging7import sys8import os9from logging.handlers import RotatingFileHandler10 11LOG_FILE = os.path.join('/code', 'ip_query.log')12 13try:14    formatter = logging.Formatter('%(message)s')15    log_handler = RotatingFileHandler(16        LOG_FILE,17        maxBytes=10*1024*1024,18        backupCount=5,19        encoding='utf-8'20    )21    log_handler.setFormatter(formatter)22    logger = logging.getLogger('ip_query')23    logger.setLevel(logging.INFO)24    logger.addHandler(log_handler)25    startup_log = {26        "时间": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),27        "事件": "系统启动",28        "状态": "成功"29    }30    logger.info(json.dumps(startup_log, ensure_ascii=False))31except Exception as e:32    print(f"日志初始化失败: {e}")33    sys.exit(1)34 35city_reader = maxminddb.open_database('GeoLite2-City.mmdb')36asn_reader = maxminddb.open_database('GeoLite2-ASN.mmdb')37cn_reader = maxminddb.open_database('GeoCN.mmdb')38lang = ["zh-CN", "en"]39asn_map = {40    9812: "东方有线",41    9389: "中国长城",42    17962: "天威视讯",43    17429: "歌华有线",44    7497: "科技网",45    24139: "华数",46    9801: "中关村",47    4538: "教育网",48    24151: "CNNIC",49    38019: "中国移动", 139080: "中国移动", 9808: "中国移动", 24400: "中国移动", 134810: "中国移动", 24547: "中国移动",50    56040: "中国移动", 56041: "中国移动", 56042: "中国移动", 56044: "中国移动", 132525: "中国移动", 56046: "中国移动",51    56047: "中国移动", 56048: "中国移动", 59257: "中国移动", 24444: "中国移动",52    24445: "中国移动", 137872: "中国移动", 9231: "中国移动", 58453: "中国移动",53    4134: "中国电信", 4812: "中国电信", 23724: "中国电信", 136188: "中国电信", 137693: "中国电信", 17638: "中国电信",54    140553: "中国电信", 4847: "中国电信", 140061: "中国电信", 136195: "中国电信", 17799: "中国电信", 139018: "中国电信",55    134764: "中国电信", 4837: "中国联通", 4808: "中国联通", 134542: "中国联通", 134543: "中国联通",56    59019: "金山云",57    135377: "优刻云",58    45062: "网易云",59    37963: "阿里云", 45102: "阿里云国际",60    45090: "腾讯云", 132203: "腾讯云国际",61    55967: "百度云", 38365: "百度云",62    58519: "华为云", 55990: "华为云", 136907: "华为云",63    4609: "澳門電訊",64    13335: "Cloudflare",65    55960: "亚马逊云", 14618: "亚马逊云", 16509: "亚马逊云",66    15169: "谷歌云", 396982: "谷歌云", 36492: "谷歌云",67}68 69def get_as_info(number):70    r = asn_map.get(number)71    if r:72        return r73 74def get_des(d):75    for i in lang:76        if i in d['names']:77            return d['names'][i]78    return d['names']['en']79 80def get_country(d):81    r = get_des(d)82    if r in ["香港", "澳门", "台湾"]:83        return "中国" + r84    return r85 86def province_match(s):87    arr = ['内蒙古', '黑龙江', '河北', '山西', '吉林', '辽宁', '江苏', '浙江', '安徽', '福建', '江西', '山东', '河南', '湖北', '湖南', '广东', '海南', '四川', '贵州', '云南', '陕西', '甘肃', '青海', '广西', '西藏', '宁夏', '新疆', '北京', '天津', '上海', '重庆']88    for i in arr:89        if i in s:90            return i91    return ''92 93def de_duplicate(regions):94    regions = filter(bool, regions)95    ret = []96    [ret.append(i) for i in regions if i not in ret]97    return ret98 99def get_addr(ip, mask):100    network = ipaddress.ip_network(f"{ip}/{mask}", strict=False)101    first_ip = network.network_address102    return f"{first_ip}/{mask}"103 104def get_maxmind(ip: str):105    ret = {"ip": ip}106    asn_info = asn_reader.get(ip)107    if asn_info:108        as_ = {"number": asn_info["autonomous_system_number"], "name": asn_info["autonomous_system_organization"]}109        info = get_as_info(as_["number"])110        if info:111            as_["info"] = info112        ret["as"] = as_113 114    city_info, prefix = city_reader.get_with_prefix_len(ip)115    ret["addr"] = get_addr(ip, prefix)116    if not city_info:117        return ret118    119    if "location" in city_info:120        location = city_info["location"]121        ret["location"] = {122            "latitude": location.get("latitude"),123            "longitude": location.get("longitude")124        }125    126    if "country" in city_info:127        country_code = city_info["country"]["iso_code"]128        country_name = get_country(city_info["country"])129        ret["country"] = {"code": country_code, "name": country_name}130    131    if "registered_country" in city_info:132        registered_country_code = city_info["registered_country"]["iso_code"]133        ret["registered_country"] = {"code": registered_country_code, "name": get_country(city_info["registered_country"])}134        135    regions = [get_des(i) for i in city_info.get('subdivisions', [])]136 137    if "city" in city_info:138        c = get_des(city_info["city"])139        if (not regions or c not in regions[-1]) and c not in country_name:140            regions.append(c)141            142    regions = de_duplicate(regions)143    if regions:144        ret["regions"] = regions145    146    return ret147 148def get_cn(ip: str, info={}):149    ret, prefix = cn_reader.get_with_prefix_len(ip)150    if not ret:151        return152    info["addr"] = get_addr(ip, prefix)153    regions = de_duplicate([ret["province"], ret["city"], ret["districts"]])154    if regions:155        info["regions"] = regions156        info["regions_short"] = de_duplicate([province_match(ret["province"]), ret["city"].replace('市', ''), ret["districts"]])157    if "as" not in info:158        info["as"] = {}159    info["as"]["info"] = ret['isp']160    if ret['net']:161        info["type"] = ret['net']162    return ret163 164def get_ip_info(ip):165    info = get_maxmind(ip)166    if "country" in info and info["country"]["code"] == "CN" and ("registered_country" not in info or info["registered_country"]["code"] == "CN"):167        get_cn(ip, info)168    return info169 170def query():171    while True:172        try:173            ip = input('IP：   \t').strip()174            info = get_ip_info(ip)175            print(f"网段：\t{info['addr']}")176            if "location" in info:177                print(f"经纬度：\t{info['location']['latitude']}, {info['location']['longitude']}")178            if "as" in info:179                print(f"ISP：\t", end=' ')180                if "info" in info["as"]:181                    print(info["as"]["info"], end=' ')182                else:183                    print(info["as"]["name"], end=' ')184                if "type" in info:185                    print(f"({info['type']})", end=' ')186                print(f"ASN{info['as']['number']}", end=' ')187                print(info['as']["name"])188            if "registered_country" in info and ("country" not in info or info["country"]["code"] != info["registered_country"]["code"]):189                print(f"注册地：\t{info['registered_country']['name']}")190            if "country" in info:191                print(f"使用地：\t{info['country']['name']}")192            if "regions" in info:193                print(f"位置：    \t{' '.join(info['regions'])}")194        except Exception as e:195            print(e)196            raise e197        finally:198            print("\n")199            200app = FastAPI()201 202@app.get("/")203async def api(request: Request, ip: str = None):204    client_ip = request.headers.get("x-forwarded-for") or request.headers.get("x-real-ip") or request.client.host205    query_ip = ip.strip() if ip else client_ip206    result = get_ip_info(query_ip)207    log_data = {208        "时间": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),209        "访问IP": client_ip,210        "查询IP": query_ip,211        "请求头": dict(request.headers),212        "查询结果": result213    }214    logger.info(json.dumps(log_data, ensure_ascii=False))215    return result216 217@app.get("/{ip}")218async def path_api(request: Request, ip: str):219    client_ip = request.headers.get("x-forwarded-for") or request.headers.get("x-real-ip") or request.client.host220    result = get_ip_info(ip)221    log_data = {222        "时间": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),223        "访问IP": client_ip,224        "查询IP": ip,225        "请求头": dict(request.headers),226        "查询结果": result227    }228    logger.info(json.dumps(log_data, ensure_ascii=False))229    return result230 231if __name__ == '__main__':232    query()233    import uvicorn234    uvicorn.run(app, host="0.0.0.0", port=8080, server_header=False, proxy_headers=True)235