feat(ip-info): Add cached IP info retrieval with fallback tester to avoid rate limiting

This commit is contained in:
Andy
2025-08-15 22:40:07 +00:00
parent e10c760821
commit 50a5a23341
4 changed files with 89 additions and 21 deletions

View File

@@ -24,7 +24,7 @@ from unshackle.core.search_result import SearchResult
from unshackle.core.title_cacher import TitleCacher, get_account_hash, get_region_from_proxy from unshackle.core.title_cacher import TitleCacher, get_account_hash, get_region_from_proxy
from unshackle.core.titles import Title_T, Titles_T from unshackle.core.titles import Title_T, Titles_T
from unshackle.core.tracks import Chapters, Tracks from unshackle.core.tracks import Chapters, Tracks
from unshackle.core.utilities import get_ip_info from unshackle.core.utilities import get_cached_ip_info, get_ip_info
class Service(metaclass=ABCMeta): class Service(metaclass=ABCMeta):
@@ -60,18 +60,24 @@ class Service(metaclass=ABCMeta):
# don't override the explicit proxy set by the user, even if they may be geoblocked # don't override the explicit proxy set by the user, even if they may be geoblocked
with console.status("Checking if current region is Geoblocked...", spinner="dots"): with console.status("Checking if current region is Geoblocked...", spinner="dots"):
if self.GEOFENCE: if self.GEOFENCE:
# no explicit proxy, let's get one to GEOFENCE if needed # Service has geofence - need fresh IP check to determine if proxy needed
try:
current_region = get_ip_info(self.session)["country"].lower() current_region = get_ip_info(self.session)["country"].lower()
if any(x.lower() == current_region for x in self.GEOFENCE): if any(x.lower() == current_region for x in self.GEOFENCE):
self.log.info("Service is not Geoblocked in your region") self.log.info("Service is not Geoblocked in your region")
else: else:
requested_proxy = self.GEOFENCE[0] # first is likely main region requested_proxy = self.GEOFENCE[0] # first is likely main region
self.log.info(f"Service is Geoblocked in your region, getting a Proxy to {requested_proxy}") self.log.info(
f"Service is Geoblocked in your region, getting a Proxy to {requested_proxy}"
)
for proxy_provider in ctx.obj.proxy_providers: for proxy_provider in ctx.obj.proxy_providers:
proxy = proxy_provider.get_proxy(requested_proxy) proxy = proxy_provider.get_proxy(requested_proxy)
if proxy: if proxy:
self.log.info(f"Got Proxy from {proxy_provider.__class__.__name__}") self.log.info(f"Got Proxy from {proxy_provider.__class__.__name__}")
break break
except Exception as e:
self.log.warning(f"Failed to check geofence: {e}")
current_region = None
else: else:
self.log.info("Service has no Geofence") self.log.info("Service has no Geofence")
@@ -86,14 +92,21 @@ class Service(metaclass=ABCMeta):
).decode() ).decode()
} }
) )
# Store region from proxy # Always verify proxy IP - proxies can change exit nodes
try:
proxy_ip_info = get_ip_info(self.session)
self.current_region = proxy_ip_info.get("country", "").lower() if proxy_ip_info else None
except Exception as e:
self.log.warning(f"Failed to verify proxy IP: {e}")
# Fallback to extracting region from proxy config
self.current_region = get_region_from_proxy(proxy) self.current_region = get_region_from_proxy(proxy)
else: else:
# No proxy, try to get current region # No proxy, use cached IP info for title caching (non-critical)
try: try:
ip_info = get_ip_info(self.session) ip_info = get_cached_ip_info(self.session)
self.current_region = ip_info.get("country", "").lower() if ip_info else None self.current_region = ip_info.get("country", "").lower() if ip_info else None
except Exception: except Exception as e:
self.log.debug(f"Failed to get cached IP info: {e}")
self.current_region = None self.current_region = None
# Optional Abstract functions # Optional Abstract functions

View File

@@ -242,6 +242,61 @@ def get_ip_info(session: Optional[requests.Session] = None) -> dict:
return (session or requests.Session()).get("https://ipinfo.io/json").json() return (session or requests.Session()).get("https://ipinfo.io/json").json()
def get_cached_ip_info(session: Optional[requests.Session] = None) -> Optional[dict]:
"""
Get IP location information with 24-hour caching and fallback providers.
This function uses a global cache to avoid repeated API calls when the IP
hasn't changed. Should only be used for local IP checks, not for proxy verification.
Args:
session: Optional requests session (usually without proxy for local IP)
Returns:
Dict with IP info including 'country' key, or None if all providers fail
"""
from unshackle.core.cacher import Cacher
cache = Cacher("global").get("ip_info")
if cache and not cache.expired:
return cache.data
providers = [
"https://ipinfo.io/json",
"https://ipapi.co/json",
]
session = session or requests.Session()
for provider_url in providers:
try:
response = session.get(provider_url, timeout=10)
if response.status_code == 200:
data = response.json()
normalized_data = {}
if "country" in data:
normalized_data = data
elif "country_code" in data:
normalized_data = {
"country": data.get("country_code", "").lower(),
"region": data.get("region", ""),
"city": data.get("city", ""),
"ip": data.get("ip", ""),
}
if normalized_data and "country" in normalized_data:
cache.set(normalized_data, expiration=86400)
return normalized_data
except Exception:
continue
return None
def time_elapsed_since(start: float) -> str: def time_elapsed_since(start: float) -> str:
""" """
Get time elapsed since a timestamp as a string. Get time elapsed since a timestamp as a string.

2
uv.lock generated
View File

@@ -1505,7 +1505,7 @@ wheels = [
[[package]] [[package]]
name = "unshackle" name = "unshackle"
version = "1.4.1" version = "1.4.2"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "appdirs" }, { name = "appdirs" },