63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
import datetime
|
|
|
|
import aiohttp
|
|
from aiocache import cached
|
|
|
|
from gallery.sketch.headers import build_headers
|
|
from gallery.sketch.source import ApiSource
|
|
from gallery.sketch.weather.api import WeatherApi
|
|
from gallery.sketch.weather.model import (
|
|
Location,
|
|
WeatherResponse,
|
|
)
|
|
from gallery.sketch.weather.util import merge_weather_values
|
|
from gallery.util import TimeUnit
|
|
|
|
from . import PROVIDER
|
|
from .parser import RP5ForecastParser, RP5LocationParser
|
|
|
|
|
|
class RP5WeatherApi(WeatherApi):
|
|
PROVIDER = PROVIDER
|
|
|
|
forecast_parser = RP5ForecastParser()
|
|
location_parser = RP5LocationParser()
|
|
|
|
def _build_source(self) -> ApiSource:
|
|
return ApiSource("https://rp5.ru", headers=build_headers("rp5.ru"))
|
|
|
|
async def find_locations(self, query: str) -> list[Location]:
|
|
endpoint = "docs/search/ru"
|
|
data = aiohttp.FormData({"searchStr": query})
|
|
content = await self.source.request(endpoint, method="POST", data=data)
|
|
return self.location_parser.parse(content)
|
|
|
|
@cached(
|
|
key_builder=lambda fun, self, location_id: f"api.weather.{self.provider}.source.{location_id}.forecast",
|
|
alias="redis",
|
|
ttl=TimeUnit.HOUR,
|
|
)
|
|
async def _get_location_forecast(self, location_id: str) -> list[WeatherResponse]:
|
|
endpoint = f"{location_id}"
|
|
content = await self.source.request(endpoint)
|
|
return self.forecast_parser.parse(content)
|
|
|
|
async def get_day(self, location_id: str, date: datetime.date) -> WeatherResponse:
|
|
forecast = await self._get_location_forecast(location_id)
|
|
for response in forecast:
|
|
if response.date == date:
|
|
return response
|
|
if date == datetime.date.today():
|
|
return WeatherResponse(location=location_id, date=date, period="day", values=[])
|
|
raise ValueError(f"Unsupported date: {date}")
|
|
|
|
async def get_days(self, location_id: str, days: int) -> WeatherResponse:
|
|
forecast = await self._get_location_forecast(location_id)
|
|
values = [merge_weather_values(item.date, item.values) for item in forecast]
|
|
return WeatherResponse(
|
|
location=forecast[0].location,
|
|
date=datetime.date.today(),
|
|
period="days",
|
|
values=values,
|
|
)
|