245 lines
9.0 KiB
Python
245 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
||
"""Получение прогноза через API Яндекс.Погоды для тарифа «Умный дом».
|
||
|
||
Ключ берётся из переменной окружения YANDEX_WEATHER_KEY или из локального
|
||
файла .env рядом со скриптом. Ключ никогда не выводится в консоль.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.parse import urlencode, urlparse, urlunparse
|
||
from urllib.request import Request, urlopen
|
||
|
||
|
||
API_URL = "https://api.weather.yandex.ru/v2/forecast"
|
||
PROJECT_DIR = Path(__file__).resolve().parent
|
||
LOCATIONS_PATH = PROJECT_DIR / "locations.json"
|
||
|
||
CONDITIONS = {
|
||
"clear": "ясно",
|
||
"partly-cloudy": "малооблачно",
|
||
"cloudy": "облачно с прояснениями",
|
||
"overcast": "пасмурно",
|
||
"light-rain": "небольшой дождь",
|
||
"rain": "дождь",
|
||
"heavy-rain": "сильный дождь",
|
||
"showers": "ливень",
|
||
"wet-snow": "дождь со снегом",
|
||
"light-snow": "небольшой снег",
|
||
"snow": "снег",
|
||
"snow-showers": "снегопад",
|
||
"hail": "град",
|
||
"thunderstorm": "гроза",
|
||
"thunderstorm-with-rain": "гроза с дождём",
|
||
"thunderstorm-with-hail": "гроза с градом",
|
||
}
|
||
|
||
|
||
def load_locations() -> dict[str, dict[str, Any]]:
|
||
if not LOCATIONS_PATH.exists():
|
||
return {}
|
||
return json.loads(LOCATIONS_PATH.read_text(encoding="utf-8"))
|
||
|
||
|
||
def normalize_location(value: str) -> str:
|
||
return " ".join(value.casefold().replace(",", " ").split())
|
||
|
||
|
||
def find_location(query: str, locations: dict[str, dict[str, Any]]) -> tuple[str, dict[str, Any]]:
|
||
normalized_query = normalize_location(query)
|
||
for key, location in locations.items():
|
||
candidates = [key, location.get("name", ""), *location.get("aliases", [])]
|
||
if any(normalized_query == normalize_location(str(candidate)) for candidate in candidates):
|
||
return key, location
|
||
available = ", ".join(locations) or "нет сохранённых локаций"
|
||
raise ValueError(f"Локация не найдена: {query}. Доступны: {available}")
|
||
|
||
def load_dotenv(path: Path) -> None:
|
||
"""Минимально загружает KEY=VALUE из .env без сторонней зависимости."""
|
||
if not path.exists():
|
||
return
|
||
|
||
for raw_line in path.read_text(encoding="utf-8").splitlines():
|
||
line = raw_line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, value = line.split("=", 1)
|
||
key = key.strip()
|
||
value = value.strip().strip('"').strip("'")
|
||
os.environ.setdefault(key, value)
|
||
|
||
|
||
def get_map_url(info_url: str | None, lat: float | None = None, lon: float | None = None) -> str:
|
||
"""Добавляет страницу карты осадков к URL населённого пункта Яндекса."""
|
||
if info_url:
|
||
parsed = urlparse(info_url)
|
||
segments = [segment for segment in parsed.path.split("/") if segment]
|
||
if len(segments) >= 2 and segments[-1] not in {"pogoda", "weather"}:
|
||
path = parsed.path.rstrip("/") + "/maps/nowcast"
|
||
return urlunparse((parsed.scheme, parsed.netloc, path, "", "", ""))
|
||
|
||
if lat is not None and lon is not None:
|
||
return "https://yandex.ru/pogoda/maps/nowcast?" + urlencode({"lat": lat, "lon": lon})
|
||
return "https://yandex.ru/pogoda/maps/nowcast"
|
||
|
||
|
||
def fetch_forecast(
|
||
access_key: str,
|
||
lat: float,
|
||
lon: float,
|
||
days: int = 2,
|
||
) -> dict[str, Any]:
|
||
params = {
|
||
"lat": lat,
|
||
"lon": lon,
|
||
"lang": "ru_RU",
|
||
"limit": days,
|
||
"hours": "true",
|
||
}
|
||
url = f"{API_URL}?{urlencode(params)}"
|
||
request = Request(url, headers={"X-Yandex-Weather-Key": access_key})
|
||
try:
|
||
with urlopen(request, timeout=20) as response:
|
||
return json.load(response)
|
||
except HTTPError as exc:
|
||
body = exc.read().decode("utf-8", errors="replace")
|
||
raise RuntimeError(
|
||
f"Яндекс.Погода вернула HTTP {exc.code}: {body[:300]}"
|
||
) from exc
|
||
except URLError as exc:
|
||
raise RuntimeError(f"Не удалось подключиться к API Яндекс.Погоды: {exc.reason}") from exc
|
||
|
||
|
||
def describe_condition(value: str | None) -> str:
|
||
return CONDITIONS.get(value or "", value or "неизвестно")
|
||
|
||
|
||
def format_forecast(data: dict[str, Any], location_label: str | None = None) -> str:
|
||
info = data.get("info") or {}
|
||
fact = data.get("fact") or {}
|
||
forecasts = data.get("forecasts") or []
|
||
location_url = info.get("url")
|
||
location_segments = [
|
||
segment for segment in urlparse(location_url or "").path.split("/") if segment
|
||
]
|
||
if location_label:
|
||
location = location_label
|
||
elif location_segments and location_segments[-1] not in {"pogoda", "weather"}:
|
||
location = location_segments[-1]
|
||
elif info.get("name"):
|
||
location = str(info["name"])
|
||
elif info.get("lat") is not None and info.get("lon") is not None:
|
||
location = f"{info['lat']:.4f}, {info['lon']:.4f}"
|
||
else:
|
||
location = "локация"
|
||
|
||
lines = [f"Яндекс.Погода: {location}"]
|
||
lines.append(
|
||
"Сейчас: "
|
||
f"{fact.get('temp', '—')}°C, "
|
||
f"{describe_condition(fact.get('condition'))}; "
|
||
f"ощущается как {fact.get('feels_like', '—')}°C"
|
||
)
|
||
lines.append(
|
||
"Ветер: "
|
||
f"{fact.get('wind_speed', '—')} м/с {fact.get('wind_dir', '')}; "
|
||
f"влажность {fact.get('humidity', '—')}%"
|
||
)
|
||
|
||
lines.append("")
|
||
lines.append("Ближайшие дни:")
|
||
for forecast in forecasts:
|
||
parts = forecast.get("parts") or {}
|
||
day = parts.get("day") or parts.get("day_short") or {}
|
||
lines.append(
|
||
f"{forecast.get('date', '—')}: "
|
||
f"{day.get('temp_min', '—')}…{day.get('temp_max', '—')}°C, "
|
||
f"{describe_condition(day.get('condition'))}"
|
||
)
|
||
|
||
lines.append("")
|
||
lines.append(f"Карта осадков: {get_map_url(location_url, info.get('lat'), info.get('lon'))}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument(
|
||
"--location", help="Сохранённая локация или её название/адрес"
|
||
)
|
||
parser.add_argument("--lat", type=float, help="Широта, если локация не сохранена")
|
||
parser.add_argument("--lon", type=float, help="Долгота, если локация не сохранена")
|
||
parser.add_argument(
|
||
"--list-locations", action="store_true",
|
||
help="Показать сохранённые локации и завершить работу",
|
||
)
|
||
parser.add_argument(
|
||
"--days", type=int, default=2, choices=range(1, 3),
|
||
help="Количество дней прогноза (1–2, по умолчанию: 2)",
|
||
)
|
||
parser.add_argument(
|
||
"--json", action="store_true", dest="as_json",
|
||
help="Вывести исходный JSON вместо краткого текста",
|
||
)
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> int:
|
||
load_dotenv(PROJECT_DIR / ".env")
|
||
args = parse_args()
|
||
locations = load_locations()
|
||
|
||
if args.list_locations:
|
||
for key, location in locations.items():
|
||
print(f"{key}: {location['name']} ({location['lat']}, {location['lon']})")
|
||
return 0
|
||
|
||
location_label = None
|
||
if args.location:
|
||
try:
|
||
_, selected = find_location(args.location, locations)
|
||
except ValueError as exc:
|
||
print(str(exc), file=sys.stderr)
|
||
return 2
|
||
lat = float(selected["lat"])
|
||
lon = float(selected["lon"])
|
||
location_label = str(selected["name"])
|
||
elif args.lat is not None and args.lon is not None:
|
||
lat = args.lat
|
||
lon = args.lon
|
||
else:
|
||
print("Укажите --location, либо одновременно --lat и --lon.", file=sys.stderr)
|
||
return 2
|
||
|
||
access_key = os.environ.get("YANDEX_WEATHER_KEY")
|
||
if not access_key:
|
||
print(
|
||
"Не найден YANDEX_WEATHER_KEY. Создайте .env по примеру .env.example "
|
||
"или задайте переменную окружения.",
|
||
file=sys.stderr,
|
||
)
|
||
return 2
|
||
|
||
try:
|
||
data = fetch_forecast(access_key, lat, lon, args.days)
|
||
except (OSError, RuntimeError, ValueError) as exc:
|
||
print(f"Ошибка получения прогноза: {exc}", file=sys.stderr)
|
||
return 1
|
||
|
||
if args.as_json:
|
||
print(json.dumps(data, ensure_ascii=False, indent=2))
|
||
else:
|
||
print(format_forecast(data, location_label=location_label))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|