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 '', ''