MCP Weather & Accruals Server

#!/usr/bin/env python3 """ Модуль для работы с OpenWeatherMap API. Содержит функции для получения текущей погоды и прогноза. """ import logging from typing import Any, Dict, Optional import httpx from modules.config import Config logger = logging.getLogger("WeatherAPI") # Вспомогательная функция для выполнения запросов к OpenWeatherMap API. async def make_weather_request( endpoint: str, params: Dict[str, Any] ) -> Optional[Dict[str, Any]]: """ Вспомогательная функция для выполнения запросов к OpenWeatherMap API. Обрабатывает HTTP-ошибки и другие исключения. Args: endpoint: Конечная точка API (например, "weather" или "forecast") params: Параметры запроса (q, units и т.д.) Returns: JSON ответ от API (словарь) или None в случае ошибки. Если произошла ошибка, словарь содержит "cod" и "message". """ # Добавляем API ключ к параметрам запроса params["appid"] = Config.OPENWEATHER_API_KEY params["lang"] = "ru" # Русский язык для описаний params["units"] = params.get("units", "metric") # Формируем полный URL url = f"{Config.OPENWEATHER_BASE_URL}/{endpoint}" try: async with httpx.AsyncClient() as client: logger.debug(f"Запрос к {url} с параметрами: {params}") response = await client.get(url, params=params, timeout=Config.API_TIMEOUT) response.raise_for_status() data = response.json() logger.debug(f"Получен ответ от {endpoint}") return data except httpx.HTTPStatusError as e: error_msg = f"HTTP ошибка {e.response.status_code}: {e.response.text}" logger.error(error_msg) return {"cod": e.response.status_code, "message": e.response.text} except httpx.RequestError as e: error_msg = f"Ошибка сети: {str(e)}" logger.error(error_msg) return {"cod": 500, "message": f"Ошибка сети: {str(e)}"} except Exception as e: error_msg = f"Непредвиденная ошибка: {str(e)}" logger.error(error_msg) return {"cod": 500, "message": f"Внутренняя ошибка: {str(e)}"} # Получить текущую погоду для указанного города. async def get_current_weather(city: str, units: str = "metric") -> str: """ Получить текущую погоду для указанного города. Args: city: Название города на русском или английском (например, "Москва" или "Moscow") units: Система измерения - "metric" (Цельсий) или "imperial" (Фаренгейт) Returns: Строка с описанием текущей погоды или сообщение об ошибке. """ logger.info(f"Запрос текущей погоды для города: '{city}'") params = { "q": city, "units": units } data = await make_weather_request("weather", params) # Проверяем, вернул ли запрос ошибку if not data or ("cod" in data and data["cod"] not in [200, "200"]): error_message = data.get("message", "Неизвестная ошибка API") if data else "Ошибка соединения" logger.warning(f"Не удалось получить погоду для '{city}': {error_message}") return f"❌ Не удалось получить погоду для города '{city}'. Причина: {error_message}." try: main = data.get("main", {}) weather_list = data.get("weather", []) weather = weather_list[0] if weather_list else {} wind = data.get("wind", {}) sys_info = data.get("sys", {}) temp_unit = "°C" if units == "metric" else "°F" wind_unit = "м/с" if units == "metric" else "миль/ч" city_name = data.get('name', city.capitalize()) country_code = sys_info.get('country', 'Н/Д') result = f""" 🌍 Погода в городе {city_name}, {country_code} 🌡️ Температура: {main.get('temp', 0.0):.1f}{temp_unit} 🤔 Ощущается как: {main.get('feels_like', 0.0):.1f}{temp_unit} 📊 Мин/Макс: {main.get('temp_min', 0.0):.1f}{temp_unit} / {main.get('temp_max', 0.0):.1f}{temp_unit} ☁️ Условия: {weather.get('description', 'Неизвестно').capitalize()} 💧 Влажность: {main.get('humidity', 0)}% 🎚️ Давление: {main.get('pressure', 0)} гПа 💨 Ветер: {wind.get('speed', 0.0):.1f} {wind_unit}, направление {wind.get('deg', 'н/д')}° """.strip() logger.info(f"Успешно получена погода для '{city_name}'") return result except Exception as e: logger.error(f"Ошибка при обработке данных текущей погоды: {str(e)}") return f"❌ Ошибка при обработке данных о погоде для города '{city}'." # Форматирует прогноз на один день на основе данных за несколько временных интервалов. def format_day_forecast(day_data: list, temp_unit: str) -> str: """ Форматирует прогноз на один день на основе данных за несколько временных интервалов. Args: day_data: Список данных о погоде за день (каждые 3 часа) temp_unit: Символ единицы температуры (°C или °F) Returns: Отформатированная строка с прогнозом на день. """ if not day_data: return "Нет данных для этого дня." date_str = day_data[0].get("dt_txt", "Неизвестная дата").split(" ")[0] try: temps = [item.get("main", {}).get("temp", 0.0) for item in day_data] avg_temp = sum(temps) / len(temps) if temps else 0.0 min_temp = min(temps) if temps else 0.0 max_temp = max(temps) if temps else 0.0 descriptions = [item.get("weather", [{}])[0].get("description", "Неизвестно") for item in day_data] most_common_desc = max(set(descriptions), key=descriptions.count) if descriptions else "Неизвестно" humidities = [item.get("main", {}).get("humidity", 0) for item in day_data] avg_humidity = sum(humidities) / len(humidities) if humidities else 0 wind_speeds = [item.get("wind", {}).get("speed", 0.0) for item in day_data] avg_wind = sum(wind_speeds) / len(wind_speeds) if wind_speeds else 0.0 return f"""📆 Дата: {date_str} 🌡️ Средняя температура: {avg_temp:.1f}{temp_unit} 📊 Мин/Макс: {min_temp:.1f}{temp_unit} / {max_temp:.1f}{temp_unit} ☁️ Условия: {most_common_desc.capitalize()} 💧 Влажность: {avg_humidity:.0f}% 💨 Ветер: {avg_wind:.1f} м/с""" except Exception as e: logger.error(f"Ошибка при форматировании прогноза: {str(e)}") return f"❌ Ошибка при форматировании прогноза за {date_str}." # Получить прогноз погоды на несколько дней. async def get_forecast(city: str, days: int = 3, units: str = "metric") -> str: """ Получить прогноз погоды на несколько дней. Args: city: Название города на русском или английском days: Количество дней для прогноза (автоматически ограничивается от 1 до 5) units: Система измерения - "metric" (Цельсий) или "imperial" (Фаренгейт) Returns: Строка с прогнозом погоды или сообщение об ошибке. """ logger.info(f"Запрос прогноза для города: '{city}' на {days} дней") # Ограничиваем количество дней days = min(max(days, Config.MIN_FORECAST_DAYS), Config.MAX_FORECAST_DAYS) params = { "q": city, "units": units, "cnt": days * 8 # API возвращает данные каждые 3 часа, 8 записей = 1 день } data = await make_weather_request("forecast", params) # Проверяем, вернул ли запрос ошибку if not data or ("cod" in data and str(data["cod"]) not in ["200", 200]): error_message = data.get("message", "Неизвестная ошибка API") if data else "Ошибка соединения" logger.warning(f"Не удалось получить прогноз для '{city}': {error_message}") return f"❌ Не удалось получить прогноз для города '{city}'. Причина: {error_message}." try: city_info = data.get("city", {}) forecast_list = data.get("list", []) if not forecast_list: return f"❌ Прогноз для города '{city}' доступен, но данных о прогнозе нет." temp_unit = "°C" if units == "metric" else "°F" # Группируем данные по дням forecasts_by_day = {} for item in forecast_list: date_time = item.get("dt_txt", "").split(" ") if len(date_time) > 0: date = date_time[0] if date not in forecasts_by_day: forecasts_by_day[date] = [] forecasts_by_day[date].append(item) formatted_forecasts = [] # Сортируем дни, чтобы они шли по порядку sorted_dates = sorted(forecasts_by_day.keys()) for date in sorted_dates: if len(formatted_forecasts) >= days: break formatted_forecasts.append(format_day_forecast(forecasts_by_day[date], temp_unit)) city_name = city_info.get('name', city.capitalize()) country_code = city_info.get('country', 'Н/Д') result = f"📅 Прогноз погоды для {city_name}, {country_code} на {len(formatted_forecasts)} {'день' if len(formatted_forecasts) == 1 else 'дня' if 1 < len(formatted_forecasts) < 5 else 'дней'}:\n\n" result += "\n\n".join(formatted_forecasts) logger.info(f"Успешно получен прогноз для '{city_name}' на {len(formatted_forecasts)} дней") return result except Exception as e: logger.error(f"Ошибка при обработке данных прогноза: {str(e)}") return f"❌ Ошибка при обработке данных прогноза для города '{city}'."