You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
257 lines
9.4 KiB
257 lines
9.4 KiB
import logging
|
|
import random
|
|
import requests
|
|
import time
|
|
from requests.adapters import HTTPAdapter
|
|
from urllib3.util.retry import Retry
|
|
|
|
|
|
user_agents = [
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Firefox/89.0',
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Edge/91.0.864.54',
|
|
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15',
|
|
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36',
|
|
'Mozilla/5.0 (iPhone; CPU iPhone OS 14_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 Mobile/15E148 Safari/604.1',
|
|
'Mozilla/5.0 (iPad; CPU OS 14_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 Mobile/15E148 Safari/604.1',
|
|
'Mozilla/5.0 (Android 11; Mobile; rv:89.0) Gecko/89.0 Firefox/89.0',
|
|
'Mozilla/5.0 (Linux; Android 10; SM-G975F) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.105 Mobile Safari/537.36',
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.104 Safari/537.36'
|
|
]
|
|
|
|
|
|
def get_requests_session():
|
|
"""Create a requests session with retry logic."""
|
|
# Disable urllib3 warnings
|
|
import urllib3
|
|
urllib3.disable_warnings()
|
|
|
|
# Disable requests warnings
|
|
import requests.packages.urllib3
|
|
requests.packages.urllib3.disable_warnings()
|
|
|
|
session = requests.Session()
|
|
|
|
# Configure retry strategy without logging
|
|
class SilentRetry(Retry):
|
|
def increment(self, *args, **kwargs):
|
|
# Override to prevent logging
|
|
return super().increment(*args, **kwargs)
|
|
|
|
retry_strategy = SilentRetry(
|
|
total=3, # Maximum number of retries
|
|
backoff_factor=0.5, # Backoff factor for retries
|
|
status_forcelist=[429, 500, 502, 503, 504], # HTTP status codes to retry on
|
|
allowed_methods=["GET"] # Only retry on GET requests
|
|
)
|
|
|
|
# Mount the adapter to the session
|
|
adapter = HTTPAdapter(max_retries=retry_strategy)
|
|
session.mount("http://", adapter)
|
|
session.mount("https://", adapter)
|
|
|
|
return session
|
|
|
|
|
|
def get_country_city_from_nominatim(lat, lon, headers) -> tuple:
|
|
"""Get country and city from OpenStreetMap Nominatim API."""
|
|
try:
|
|
session = get_requests_session()
|
|
resp = session.get(
|
|
f"https://nominatim.openstreetmap.org/reverse?lat={lat}&lon={lon}&format=json&addressdetails=1",
|
|
headers=headers,
|
|
timeout=10
|
|
)
|
|
if resp.status_code == 200:
|
|
address = resp.json().get('address', {})
|
|
country = address.get('country_code', '')
|
|
# Some responses use 'city', others might use 'town' or 'village'
|
|
city = address.get('city', address.get('town', address.get('village', '')))
|
|
return country, city
|
|
else:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
except Exception:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
|
|
def get_country_city_from_bigdatacloud(lat, lon, headers) -> tuple:
|
|
"""Get country and city from BigDataCloud API."""
|
|
try:
|
|
session = get_requests_session()
|
|
resp = session.get(
|
|
f"https://api.bigdatacloud.net/data/reverse-geocode-client?latitude={lat}&longitude={lon}&localityLanguage=en",
|
|
headers=headers,
|
|
timeout=10
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
country = data.get('countryCode', '')
|
|
city = data.get('city', '')
|
|
return country, city
|
|
else:
|
|
return '', ''
|
|
except Exception:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
|
|
def get_country_city_from_geocode_maps(lat, lon, headers) -> tuple:
|
|
"""Get country and city from geocode.maps.co API."""
|
|
try:
|
|
session = get_requests_session()
|
|
resp = session.get(
|
|
f"https://geocode.maps.co/reverse?lat={lat}&lon={lon}",
|
|
headers=headers,
|
|
timeout=10
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
address = data.get('address', {})
|
|
country = address.get('country_code', '')
|
|
city = address.get('city', address.get('town', address.get('village', '')))
|
|
return country, city
|
|
else:
|
|
return '', ''
|
|
except Exception:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
|
|
def get_country_city_from_ipapi_coordinates(lat, lon, headers) -> tuple:
|
|
"""Get country and city from ip-api.com using coordinates."""
|
|
try:
|
|
# This is a fallback that uses IP geolocation based on server IP
|
|
# Not as accurate for the specific coordinates but better than nothing
|
|
session = get_requests_session()
|
|
resp = session.get(
|
|
"http://ip-api.com/json/?fields=status,countryCode,city",
|
|
headers=headers,
|
|
timeout=10
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
if data.get('status') == 'success':
|
|
country = data.get('countryCode', '')
|
|
city = data.get('city', '')
|
|
# Removed info logging
|
|
return country, city
|
|
return '', ''
|
|
except Exception:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
|
|
def get_country_city_from_point(lat, lon) -> tuple:
|
|
"""Get country and city from coordinates using multiple fallback services."""
|
|
headers = {'User-Agent': random.choice(user_agents)}
|
|
|
|
# List of geocoding functions to try
|
|
geocoding_functions = [
|
|
get_country_city_from_nominatim,
|
|
get_country_city_from_bigdatacloud,
|
|
get_country_city_from_geocode_maps,
|
|
get_country_city_from_ipapi_coordinates, # Last resort fallback
|
|
]
|
|
|
|
# Try each geocoding service until we get a result
|
|
for func in geocoding_functions:
|
|
country, city = func(lat, lon, headers)
|
|
if country and city:
|
|
return country, city
|
|
|
|
# If all services fail, silently return empty strings without logging
|
|
return '', ''
|
|
|
|
|
|
def get_country_city_from_ip(ip, headers) -> tuple:
|
|
"""Get country and city from apl.lplocation.net API."""
|
|
try:
|
|
session = get_requests_session()
|
|
resp = session.get(
|
|
f"https://apl.lplocation.net/?ip={ip}",
|
|
headers=headers,
|
|
timeout=5
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
country = data.get('country_code', '')
|
|
city = data.get('city', '')
|
|
return country, city
|
|
else:
|
|
return '', ''
|
|
except Exception:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
|
|
def get_country_city_from_ip_api(ip, headers) -> tuple:
|
|
"""Get country and city from ip-api.com API."""
|
|
try:
|
|
session = get_requests_session()
|
|
resp = session.get(
|
|
f"http://ip-api.com/json/{ip}?fields=status,countryCode,city",
|
|
headers=headers,
|
|
timeout=5
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
if data.get('status') == 'success':
|
|
country = data.get('countryCode', '')
|
|
city = data.get('city', '')
|
|
return country, city
|
|
else:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
else:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
except Exception:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
|
|
|
|
def get_country_city_from_ip2location(ip: str, headers) -> tuple:
|
|
"""Get country and city from ip2location.io API."""
|
|
try:
|
|
session = get_requests_session()
|
|
resp = session.get(
|
|
f"https://api.ip2location.io/?key=A9DE25AC3ADF5255693F8BFAFB23A902&ip={ip}&format=json",
|
|
headers=headers,
|
|
timeout=5
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
country = data.get('country_code', '')
|
|
city = data.get('city_name', '')
|
|
return country, city
|
|
else:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
except Exception:
|
|
# Silently fail without logging
|
|
return '', ''
|
|
|
|
def get_country_city(ip: str) -> tuple:
|
|
"""Get country and city from IP using multiple fallback services."""
|
|
headers = {'User-Agent': random.choice(user_agents)}
|
|
|
|
# List of IP-based geocoding functions to try in order of preference
|
|
functions = [
|
|
get_country_city_from_ip_api, # Most reliable based on common usage
|
|
get_country_city_from_ip2location, # Has API key, likely more reliable
|
|
get_country_city_from_ip, # Third option
|
|
]
|
|
|
|
# Try each service until we get a result
|
|
for i, func in enumerate(functions):
|
|
try:
|
|
country, city = func(ip, headers)
|
|
if country and city:
|
|
# No logging for fallbacks
|
|
return country, city
|
|
# Small delay between API calls to avoid rate limiting
|
|
if i < len(functions) - 1:
|
|
time.sleep(0.2)
|
|
except Exception:
|
|
# Silently continue to the next service without logging
|
|
continue
|
|
|
|
# If all services fail, silently return empty strings without logging
|
|
return '', ''
|