main.py

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)