import requests import os from typing import Optional, Dict, Any from utils.config import config from functools import lru_cache import time class WeatherService: """Service for fetching weather information with caching""" def __init__(self): self.api_key = config.openweather_api_key self.base_url = "http://api.openweathermap.org/data/2.5" def _get_ttl_hash(self, seconds=300): """Helper function to invalidate cache periodically""" return round(time.time() / seconds) @lru_cache(maxsize=128) def get_current_weather_cached(self, city: str, ttl_hash=None): """Cached version of weather API calls""" return self.get_current_weather(city) def get_current_weather(self, city: str) -> Optional[Dict[str, Any]]: """Get current weather for a city""" if not self.api_key: print("OpenWeather API key not configured") return None try: params = { 'q': city, 'appid': self.api_key, 'units': 'metric' # Celsius } response = requests.get( f"{self.base_url}/weather", params=params, timeout=10 ) if response.status_code == 200: data = response.json() return { 'city': data['name'], 'country': data['sys']['country'], 'temperature': data['main']['temp'], 'feels_like': data['main']['feels_like'], 'humidity': data['main']['humidity'], 'description': data['weather'][0]['description'], 'icon': data['weather'][0]['icon'] } else: print(f"Weather API error: {response.status_code} - {response.text}") return None except Exception as e: print(f"Error fetching weather data: {e}") return None def get_forecast(self, city: str, days: int = 5) -> Optional[Dict[str, Any]]: """Get weather forecast for a city""" if not self.api_key: print("OpenWeather API key not configured") return None try: params = { 'q': city, 'appid': self.api_key, 'units': 'metric', 'cnt': days } response = requests.get( f"{self.base_url}/forecast", params=params, timeout=10 ) if response.status_code == 200: data = response.json() forecasts = [] for item in data['list']: forecasts.append({ 'datetime': item['dt_txt'], 'temperature': item['main']['temp'], 'description': item['weather'][0]['description'], 'icon': item['weather'][0]['icon'] }) return { 'city': data['city']['name'], 'country': data['city']['country'], 'forecasts': forecasts } else: print(f"Forecast API error: {response.status_code} - {response.text}") return None except Exception as e: print(f"Error fetching forecast data: {e}") return None def get_weather_summary(self, city="New York") -> str: """Get formatted weather summary with caching""" try: weather = self.get_current_weather_cached( city, ttl_hash=self._get_ttl_hash(300) ) if weather: return f"{weather.get('temperature', 'N/A')}°C, {weather.get('description', 'Clear skies')}" else: return "Clear skies" except: return "Clear skies" # Global weather service instance weather_service = WeatherService()