Spaces:
Sleeping
Sleeping
| """ | |
| IP and location logging service with Redis gating. | |
| """ | |
| import os | |
| import logging | |
| import requests | |
| from typing import Optional, Dict, Any | |
| from .redis_service import get_redis_service | |
| from .mongodb_service import get_mongodb_service | |
| logger = logging.getLogger(__name__) | |
| class IPLocationService: | |
| """ | |
| Service for logging IP and location data. | |
| Uses Redis gate to limit IPinfo API calls (once per device per 5 minutes). | |
| """ | |
| def __init__(self): | |
| self.redis_service = get_redis_service() | |
| self.mongodb_service = get_mongodb_service() | |
| self.ipinfo_token = os.getenv("IPINFO_TOKEN") | |
| if not self.ipinfo_token: | |
| logger.warning("IPINFO_TOKEN not set. IP location logging disabled.") | |
| def _fetch_ip_location(self, ip_address: str) -> Optional[Dict[str, Any]]: | |
| """ | |
| Fetch IP location from IPinfo API. | |
| Args: | |
| ip_address: IP address to lookup | |
| Returns: | |
| Location data or None | |
| """ | |
| if not self.ipinfo_token: | |
| return None | |
| try: | |
| url = f"https://ipinfo.io/{ip_address}" | |
| headers = {"Authorization": f"Bearer {self.ipinfo_token}"} | |
| response = requests.get(url, headers=headers, timeout=5) | |
| if response.status_code == 200: | |
| data = response.json() | |
| # Extract relevant fields | |
| location_data = { | |
| "ip": data.get("ip"), | |
| "city": data.get("city"), | |
| "region": data.get("region"), | |
| "country": data.get("country"), | |
| "loc": data.get("loc"), # latitude,longitude | |
| "org": data.get("org"), # ISP/organization | |
| "timezone": data.get("timezone") | |
| } | |
| return location_data | |
| else: | |
| logger.warning(f"IPinfo API returned status {response.status_code}") | |
| return None | |
| except Exception as e: | |
| logger.error(f"Failed to fetch IP location: {str(e)}") | |
| return None | |
| def log_session_metadata( | |
| self, | |
| device_id: str, | |
| ip_address: str, | |
| user_id: Optional[str] = None, | |
| user_agent: Optional[str] = None | |
| ) -> bool: | |
| """ | |
| Log session metadata with IP and location (gated by Redis). | |
| Only logs if Redis gate allows (once per device per 5 minutes). | |
| Args: | |
| device_id: Device identifier | |
| ip_address: Client IP address | |
| user_id: Optional user identifier | |
| user_agent: Optional user agent string | |
| Returns: | |
| True if logged, False if skipped or failed | |
| """ | |
| # Check Redis gate (5 minute TTL) | |
| should_log = self.redis_service.should_log_ip(device_id, ttl_seconds=300) | |
| if not should_log: | |
| logger.debug(f"IP logging skipped for device {device_id} (within TTL window)") | |
| return False | |
| # Fetch IP location | |
| location_data = self._fetch_ip_location(ip_address) | |
| # Prepare metadata | |
| metadata = { | |
| "ip": ip_address, | |
| "user_agent": user_agent | |
| } | |
| # Add location data if available | |
| if location_data: | |
| metadata.update(location_data) | |
| # Log SESSION_METADATA event to MongoDB | |
| success = self.mongodb_service.log_event( | |
| event_type="SESSION_METADATA", | |
| device_id=device_id, | |
| user_id=user_id, | |
| metadata=metadata | |
| ) | |
| if success: | |
| logger.info(f"Session metadata logged for device {device_id}") | |
| else: | |
| logger.warning(f"Failed to log session metadata for device {device_id}") | |
| return success | |
| # Singleton instance | |
| _ip_location_service = None | |
| def get_ip_location_service() -> IPLocationService: | |
| """Get IP location service singleton.""" | |
| global _ip_location_service | |
| if _ip_location_service is None: | |
| _ip_location_service = IPLocationService() | |
| return _ip_location_service | |