| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465 |
- from __future__ import annotations
- from dataclasses import dataclass
- from typing import Optional
- from app.config import settings
- @dataclass
- class GeoInfo:
- country: str
- city: str
- latitude: Optional[float]
- longitude: Optional[float]
- org: Optional[str] = None
- _UNKNOWN = GeoInfo("Unknown", "Unknown", None, None, None)
- class GeoResolver:
- def __init__(self, db_path: str, asn_db_path: str) -> None:
- self._db_path = db_path
- self._asn_db_path = asn_db_path
- self._reader = None
- self._asn_reader = None
- def _get_reader(self):
- if self._reader is None:
- import geoip2.database
- self._reader = geoip2.database.Reader(self._db_path)
- return self._reader
- def _get_asn_reader(self):
- if self._asn_reader is None:
- import geoip2.database
- self._asn_reader = geoip2.database.Reader(self._asn_db_path)
- return self._asn_reader
- def resolve(self, ip: str) -> GeoInfo:
- # Private / loopback addresses have no GeoIP entry
- if ip in ("127.0.0.1", "::1", "localhost") or ip.startswith("192.168.") or ip.startswith("10.") or ip.startswith("172."):
- return GeoInfo("Local", "Loopback", None, None, None)
- try:
- reader = self._get_reader()
- response = reader.city(ip)
- country = response.country.name or "Unknown"
- city = response.city.name or "Unknown"
- lat = response.location.latitude
- lon = response.location.longitude
- except Exception:
- country, city, lat, lon = "Unknown", "Unknown", None, None
- org = None
- try:
- asn_reader = self._get_asn_reader()
- asn_response = asn_reader.asn(ip)
- org = asn_response.autonomous_system_organization or None
- except Exception:
- pass
- return GeoInfo(country, city, lat, lon, org)
- geo_resolver = GeoResolver(settings.geoip_db_path, settings.geoip_asn_db_path)
|
