File size: 3,727 Bytes
c0ef6d4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 |
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' # 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
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
# Global weather service instance
weather_service = WeatherService()
|