add offline ip location resolution (#78)

Author: wu-clan

backend/app/common/exception/exception_handler.py

@@ -108,6 +108,7 @@ def all_exception_handler(request: Request, exc: Exception):
 
         else:
             import traceback
+
             log.exception(traceback.format_exc())
             return JSONResponse(
                 status_code=500,

backend/app/common/log.py

@@ -15,7 +15,6 @@
 
 
 class Logger:
-
     def __init__(self):
         self.log_path = path_conf.LogPath
 

backend/app/core/conf.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 from functools import lru_cache
+from typing import Literal
 
 from pydantic import BaseSettings, root_validator
 
@@ -55,7 +56,7 @@ def validator_api_url(cls, values):
     STATIC_FILES: bool = False
 
     # Location Parse
-    LOCATION_PARSE: bool = True  # 将会导致登录延时，建议关闭，有条件自行使用第三方离线数据库
+    LOCATION_PARSE: Literal['online', 'offline', 'false'] = 'offline'
 
     # MySQL
     DB_ECHO: bool = False

backend/app/core/path_conf.py

@@ -17,3 +17,6 @@
 
 # RBAC model.conf 文件路径
 RBAC_MODEL_CONF = os.path.join(BasePath, 'app', 'core', settings.CASBIN_RBAC_MODEL_NAME)
+
+# 离线 IP 数据库路径
+IP2REGION_XDB = os.path.join(BasePath, 'app', 'static', 'ip2region.xdb')

backend/app/services/login_log_service.py

@@ -30,7 +30,12 @@ async def create(
             ip = await request_parse.get_request_ip(request)
             user_agent = request.headers.get('User-Agent')
             _, os_info, browser = str(parse(user_agent)).replace(' ', '').split('/')
-            location = await request_parse.get_location(ip, user_agent) if settings.LOCATION_PARSE else '未知'
+            if settings.LOCATION_PARSE == 'online':
+                location = await request_parse.get_location_online(ip, user_agent)
+            elif settings.LOCATION_PARSE == 'offline':
+                location = request_parse.get_location_offline(ip)
+            else:
+                location = '未知'
             obj_in = CreateLoginLog(
                 user_uuid=user.user_uuid,
                 username=user.username,

backend/app/static/ip2region.xdb

@@ -1,12 +1,15 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 import httpx
+from XdbSearchIP.xdbSearcher import XdbSearcher
 from httpx import HTTPError
 from fastapi import Request
 
+from backend.app.core.path_conf import IP2REGION_XDB
+
 
 async def get_request_ip(request: Request) -> str:
-    """获取请求的ip地址"""
+    """获取请求的 ip 地址"""
     real = request.headers.get('X-Real-IP')
     if real:
         ip = real
@@ -19,8 +22,14 @@ async def get_request_ip(request: Request) -> str:
     return ip
 
 
-async def get_location(ipaddr: str, user_agent: str) -> str:
-    """获取ip地址归属地(临时)"""
+async def get_location_online(ipaddr: str, user_agent: str) -> str:
+    """
+    在线获取 ip 地址属地，无法保证可用性，准确率较高
+
+    :param ipaddr:
+    :param user_agent:
+    :return:
+    """
     async with httpx.AsyncClient(timeout=3) as client:
         ip_api_url = f'http://ip-api.com/json/{ipaddr}?lang=zh-CN'
         whois_url = f'http://whois.pconline.com.cn/ipJson.jsp?ip={ipaddr}&json=true'
@@ -35,3 +44,21 @@ async def get_location(ipaddr: str, user_agent: str) -> str:
             except (HTTPError, KeyError):
                 city = None
     return city or '未知' if city != '' else '未知'
+
+
+def get_location_offline(ipaddr: str) -> str:
+    """
+    离线获取 ip 地址属地，无法保证准确率，100%可用
+
+    :param ipaddr:
+    :return:
+    """
+    cb = XdbSearcher.loadContentFromFile(dbfile=IP2REGION_XDB)
+    searcher = XdbSearcher(contentBuff=cb)
+    data = searcher.search(ipaddr)
+    searcher.close()
+    location_info = data.split('|')
+    country = location_info[0]
+    province = location_info[2]
+    city = location_info[3]
+    return city if city != '0' else province if province != '0' else country if country != '0' else '未知'

backend/app/utils/request_parse.py

@@ -35,3 +35,4 @@ tzlocal==2.1
 user_agents==2.2.0
 uvicorn[standard]==0.13.4
 wait-for-it==2.2.1
+XdbSearchIP==1.0.2