weather-fetch-lab/weather_client.py

import datetime
from math import inf
import os
from dataclasses import dataclass
from urllib.parse import urlencode

from cnet import HttpError, as_dict, as_float, as_int, as_list, as_str, fetch_json
from db_models import DailyForecast
from geo_locator import Geo


messages = {
    401: "[401] Wrong WEATHER_API_KEY",
    404: "[404] City not found",
    429: "[429] Limit exceeded",
}


class WeatherAPIError(Exception): ...


@dataclass
class _Item:
    date: datetime.date
    hour: int
    temp_min: float
    temp_max: float
    humidity: float
    wind_speed: float
    description: str


@dataclass
class _DayMetrics:
    t_mins: float = inf
    t_maxs: float = -inf
    h_sum: float = 0
    n: int = 0
    wind_max: float = -inf
    noon_dst: int | None = None
    description: str = ""


def _timezone_offset(city: dict[str, object]) -> datetime.timedelta:
    shift = as_float(city.get("timezone"), default=0.0) or 0.0
    return datetime.timedelta(seconds=int(shift))


def _parse_item(raw: object, offset: datetime.timedelta) -> _Item | None:
    entry = as_dict(raw)
    main = as_dict(entry.get("main"))
    wind = as_dict(entry.get("wind"))
    weather_list = as_list(entry.get("weather"))
    weather = as_dict(weather_list[0]) if weather_list else {}

    dt = as_float(entry.get("dt"), default=0.0)
    if dt is None or dt <= 0:
        return None
    time = datetime.datetime.fromtimestamp(dt, tz=datetime.timezone.utc) + offset

    return _Item(
        date=time.date(),
        hour=time.hour,
        temp_min=as_float(main.get("temp_min"), default=0.0) or 0.0,
        temp_max=as_float(main.get("temp_max"), default=0.0) or 0.0,
        humidity=as_float(main.get("humidity"), default=0.0) or 0.0,
        wind_speed=as_float(wind.get("speed"), default=0.0) or 0.0,
        description=as_str(weather.get("description"), default=""),
    )


def _calc(intervals: list[_Item], city: str) -> list[DailyForecast]:
    days_metrics: dict[datetime.date, _DayMetrics] = {}
    for i in intervals:
        m = days_metrics.setdefault(i.date, _DayMetrics())
        m.t_mins = min(m.t_mins, i.temp_min)
        m.t_maxs = max(m.t_maxs, i.temp_max)

        m.h_sum += i.humidity
        m.n += 1
        m.wind_max = max(m.wind_max, i.wind_speed)
        dst = abs(i.hour - 12)
        if m.noon_dst is None or dst < m.noon_dst:
            m.noon_dst = dst
            m.description = i.description

    result: list[DailyForecast] = []
    for day, m in sorted(days_metrics.items()):
        result.append(
            DailyForecast(
                city=city,
                forecast_date=day,
                temp_min=m.t_mins,
                temp_max=m.t_maxs,
                humidity=round(m.h_sum / m.n) if m.n else 0,
                wind_speed=m.wind_max,
                description=m.description,
            )
        )
    return result[:4]


def get_forecast(geo: Geo) -> list[DailyForecast]:
    api_key = os.getenv("WEATHER_API_KEY", "").strip()
    if not api_key:
        raise WeatherAPIError("no WEATHER_API_KEY")
    base_url = os.getenv(
        "WEATHER_API_BASE_URL", "https://api.openweathermap.org/data/2.5/forecast"
    )

    params: dict[str, str] = {"appid": api_key, "units": "metric", "lang": "ru"}
    if geo.lat is not None and geo.lon is not None:
        params["lat"] = f"{geo.lat}"
        params["lon"] = f"{geo.lon}"
    else:
        params["q"] = geo.city

    url = f"{base_url}?{urlencode(params)}"

    try:
        raw = fetch_json(url)
    except HttpError as e:
        if e.status_code == HttpError.NO_CODE:
            raise WeatherAPIError(f"weather request failed: {e.body}") from e
        raise WeatherAPIError(
            messages.get(e.status_code, f"[{e.status_code}] weather error")
        ) from e

    resp = as_dict(raw)

    cod = as_int(resp.get("cod"), 200)
    if cod != 200:
        raise WeatherAPIError(messages.get(cod, f"[{cod}] weather error"))

    resp_city = as_dict(resp.get("city"))
    city = as_str(resp_city.get("name"), default=geo.city)
    offset = _timezone_offset(resp_city)
    items = [
        i
        for i in (_parse_item(entry, offset) for entry in as_list(resp.get("list")))
        if i is not None
    ]
    if not items:
        raise WeatherAPIError("weather is empty")

    return _calc(items, city)