#!/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())