|
import requests |
|
import os |
|
from typing import Optional, Dict, Any |
|
from utils.config import config |
|
|
|
class WeatherService: |
|
"""Service for fetching weather information""" |
|
|
|
def __init__(self): |
|
self.api_key = config.openweather_api_key |
|
self.base_url = "http://api.openweathermap.org/data/2.5" |
|
|
|
def get_current_weather(self, city: str) -> Optional[Dict[str, Any]]: |
|
""" |
|
Get current weather for a city |
|
|
|
Args: |
|
city: Name of the city |
|
|
|
Returns: |
|
Dictionary with weather information or None if failed |
|
""" |
|
if not self.api_key: |
|
print("OpenWeather API key not configured") |
|
return None |
|
|
|
try: |
|
params = { |
|
'q': city, |
|
'appid': self.api_key, |
|
'units': 'metric' |
|
} |
|
|
|
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 |
|
|
|
Args: |
|
city: Name of the city |
|
days: Number of days to forecast (default: 5) |
|
|
|
Returns: |
|
Dictionary with forecast information or None if failed |
|
""" |
|
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 |
|
|
|
|
|
weather_service = WeatherService() |
|
|