"""Погодный сервис: геокодинг Nominatim (OpenStreetMap) + прогноз met.no. Модуль намеренно ничего не знает про AMS и pydantic-ai — это чистый слой получения данных. Его можно запустить и проверить отдельно: python weather.py Казань Москва Поддерживаются русские (кириллица) и латинские названия городов. API-ключи не нужны: оба сервиса бесплатные. """ from __future__ import annotations import logging import threading import time from dataclasses import dataclass from functools import lru_cache from typing import Any import requests logger = logging.getLogger(__name__) # --- Внешние сервисы ------------------------------------------------------ # Nominatim (OSM) корректно ищет по кириллице, в отличие от GeoNames-индекса. GEOCODE_URL = "https://nominatim.openstreetmap.org/search" FORECAST_URL = "https://api.met.no/weatherapi/locationforecast/2.0/compact" # Оба сервиса требуют осмысленный User-Agent с контактом. USER_AGENT = "ams-weather-agent/1.0 (+https://git.totmin.ru/en2zmax/ams-weather)" # (connect, read) в секундах. Держим короткими, чтобы не подвешивать диалог. TIMEOUT = (4, 7) RETRIES = 2 # общее число попыток на один HTTP-запрос # Политика использования Nominatim: не чаще одного запроса в секунду. NOMINATIM_MIN_INTERVAL = 1.0 # Переиспользуем TCP-соединения между вызовами — заметно быстрее серии запросов. _session = requests.Session() _session.headers["User-Agent"] = USER_AGENT # Ограничитель частоты запросов к Nominatim (общий на процесс). _rate_lock = threading.Lock() _last_request_at = 0.0 class WeatherError(Exception): """Ожидаемая ошибка получения погоды. Разделяем два уровня информации: * message — короткий безопасный текст для пользователя; * detail — подробности для лога (пользователю не показываем). """ def __init__(self, message: str, detail: str | None = None) -> None: super().__init__(message) self.message = message self.detail = detail or message @dataclass(frozen=True, slots=True) class Location: """Найденный населённый пункт.""" name: str region: str | None country: str | None latitude: float longitude: float @property def title(self) -> str: """Человекочитаемое имя места: «Казань, Татарстан, Россия».""" return ", ".join(part for part in (self.name, self.region, self.country) if part) @dataclass(frozen=True, slots=True) class Weather: """Текущие погодные условия.""" temperature_c: float humidity_pct: int wind_ms: float wind_dir_deg: int precipitation_mm: float description: str # met.no отдаёт symbol_code вида «partlycloudy_day»; суффикс дня/ночи отбрасываем. _SYMBOLS: dict[str, str] = { "clearsky": "ясно", "fair": "малооблачно", "partlycloudy": "переменная облачность", "cloudy": "облачно", "fog": "туман", "lightrain": "небольшой дождь", "rain": "дождь", "heavyrain": "сильный дождь", "lightrainshowers": "небольшой ливень", "rainshowers": "ливень", "heavyrainshowers": "сильный ливень", "lightsleet": "небольшой дождь со снегом", "sleet": "дождь со снегом", "heavysleet": "сильный дождь со снегом", "lightsnow": "небольшой снег", "snow": "снег", "heavysnow": "сильный снег", "lightsnowshowers": "небольшой снегопад", "snowshowers": "снегопад", "heavysnowshowers": "сильный снегопад", "rainandthunder": "дождь с грозой", "sleetandthunder": "дождь со снегом и грозой", "snowandthunder": "снег с грозой", "thunderstorm": "гроза", } _UNKNOWN_DESCRIPTION = "нет данных" def _throttle_nominatim() -> None: """Выдержать паузу между запросами к Nominatim (правила OSM).""" global _last_request_at with _rate_lock: wait = NOMINATIM_MIN_INTERVAL - (time.monotonic() - _last_request_at) if wait > 0: time.sleep(wait) _last_request_at = time.monotonic() def _get(url: str, params: dict[str, Any]) -> Any: """GET с короткими ретраями и единообразной обработкой ошибок. Повторяем только временные сбои: сетевые ошибки, 5xx и 429. Постоянные 4xx (например, 400) сразу превращаем в WeatherError — повторять их бессмысленно. """ last_error: Exception | None = None for attempt in range(1, RETRIES + 1): try: response = _session.get(url, params=params, timeout=TIMEOUT) except requests.RequestException as exc: last_error = exc else: status = response.status_code if 400 <= status < 500 and status != 429: raise WeatherError("внешний сервис вернул ошибку", f"{url}: HTTP {status}") if status < 500 and status != 429: try: return response.json() except ValueError as exc: # ответ не является корректным JSON last_error = exc else: last_error = requests.HTTPError(f"HTTP {status}") logger.warning("Запрос %s не удался (попытка %d/%d): %s", url, attempt, RETRIES, last_error) if attempt < RETRIES: time.sleep(0.4 * attempt) # короткий нарастающий backoff raise WeatherError("внешний сервис недоступен", f"{url}: {last_error}") @lru_cache(maxsize=256) def geocode(city: str) -> Location: """Найти город по названию (кириллица или латиница). Результат кэшируется: повторный запрос того же города отвечает мгновенно. """ city = city.strip() if not city: raise WeatherError("не указан город") _throttle_nominatim() payload = _get( GEOCODE_URL, {"q": city, "format": "jsonv2", "limit": 1, "addressdetails": 1, "accept-language": "ru"}, ) results = payload if isinstance(payload, list) else [] if not results: raise WeatherError(f"город «{city}» не найден") top = results[0] address = top.get("address") or {} name = ( address.get("city") or address.get("town") or address.get("village") or address.get("municipality") or city ) try: latitude = round(float(top["lat"]), 4) # met.no принимает до 4 знаков longitude = round(float(top["lon"]), 4) except (KeyError, TypeError, ValueError) as exc: raise WeatherError("геокодер вернул неожиданный формат", repr(exc)) from exc return Location( name=name, region=address.get("state") or address.get("region"), country=address.get("country"), latitude=latitude, longitude=longitude, ) def fetch_weather(location: Location) -> Weather: """Текущая погода по координатам (met.no).""" payload = _get(FORECAST_URL, {"lat": location.latitude, "lon": location.longitude}) try: series = payload["properties"]["timeseries"][0] details = series["data"]["instant"]["details"] next_hour = series["data"].get("next_1_hours") or {} symbol = (next_hour.get("summary") or {}).get("symbol_code", "") precipitation = (next_hour.get("details") or {}).get("precipitation_amount", 0.0) return Weather( temperature_c=round(float(details["air_temperature"]), 1), humidity_pct=round(float(details["relative_humidity"])), wind_ms=round(float(details["wind_speed"]), 1), wind_dir_deg=round(float(details["wind_from_direction"])), precipitation_mm=round(float(precipitation), 1), description=_SYMBOLS.get(symbol.rsplit("_", 1)[0], _UNKNOWN_DESCRIPTION), ) except (KeyError, IndexError, TypeError, ValueError) as exc: raise WeatherError("сервис прогноза вернул неожиданный формат", repr(exc)) from exc def format_summary(location: Location, weather: Weather) -> str: """Детерминированная сводка. Используется как фолбэк, если языковая модель недоступна, — пользователь в любом случае получает ответ по существу. """ return ( f"{location.title}: {weather.description}, {weather.temperature_c:g} °C, " f"влажность {weather.humidity_pct} %, ветер {weather.wind_ms:g} м/с." ) if __name__ == "__main__": # ручной smoke-тест: python weather.py Казань import sys logging.basicConfig(level=logging.INFO) for query in sys.argv[1:] or ["Казань"]: try: found = geocode(query) print(format_summary(found, fetch_weather(found))) except WeatherError as exc: print(f"{query}: ошибка — {exc.message}")