-
Notifications
You must be signed in to change notification settings - Fork 1
/
utils.py
71 lines (56 loc) · 2 KB
/
utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
import logging
import random
import httpx
from fastapi import Request
from models import GeoIpApi, GeoIpApiRtypeEnum
logger = logging.getLogger("inspector.utils")
GEOIP_API = {
"country.is": GeoIpApi(
url="https://api.country.is/{ip}",
rtype=GeoIpApiRtypeEnum.json,
lookup="country",
),
"ip2c.org": GeoIpApi(
url="https://ip2c.org/{ip}",
rtype=GeoIpApiRtypeEnum.text,
delimiter=";",
lookup="1", # index
),
"geoplugin.net": GeoIpApi(
url="http://www.geoplugin.net/json.gp?ip={ip}",
rtype=GeoIpApiRtypeEnum.json,
lookup="geoplugin_countryCode",
),
}
def request_to_ip(request: Request) -> str:
logger.debug("Getting IP address from request...")
# get the ip address of the client. First check the proxy headers, then the client host
client_ip = request.client.host if request.client else ""
proxy_ip = request.headers.get("x-forwarded-for", "")
ip = proxy_ip or client_ip
logger.debug(f"Returning {ip=}")
return ip
async def ip_to_country(ip: str) -> str:
logger.debug(f"Getting country code for {ip}...")
if not ip:
return ""
geoip_api = GEOIP_API.get(random.choice(list(GEOIP_API.keys())))
if not geoip_api:
return ""
# build the url
url = geoip_api.url.format(ip=ip)
async with httpx.AsyncClient() as client:
try:
response = await client.get(url)
response.raise_for_status()
except httpx.HTTPError as e:
logger.info(f"Error getting country code for {ip}: {e}")
return ""
country = "" # alpha-2 country code, i.e. "AU"
# parse the response and lookup the country code
if geoip_api.rtype == GeoIpApiRtypeEnum.json:
country = response.json().get(geoip_api.lookup) or ""
elif geoip_api.rtype == GeoIpApiRtypeEnum.text:
country = response.text.split(geoip_api.delimiter)[int(geoip_api.lookup)] or ""
logger.debug(f"Returning {country=}")
return country