249 lines
10 KiB
Python
249 lines
10 KiB
Python
"""Погодный сервис: геокодинг 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}")
|