87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
import datetime
|
|
import logging
|
|
|
|
import aiohttp
|
|
from aiocache import cached
|
|
from bs4 import BeautifulSoup
|
|
|
|
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 ...sketch.headers import build_headers
|
|
from .parser import parse_page
|
|
|
|
logger = logging.getLogger("rp5")
|
|
|
|
|
|
class RP5WeatherApi(WeatherApi):
|
|
PROVIDER = "rp5"
|
|
|
|
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})
|
|
response = await self.source.request(endpoint, method="POST", data=data)
|
|
soup = BeautifulSoup(response, features="html.parser")
|
|
table = soup.select_one(".searchResults")
|
|
result = []
|
|
for row in table.select("tr")[1:]:
|
|
cells = row.select("td")
|
|
name = cells[0].text.strip()
|
|
location_id = cells[0].select_one("a").attrs["href"].lstrip("/")
|
|
country = cells[1].text.strip()
|
|
district = cells[2].text.strip()
|
|
subdistrict = cells[3].text.strip()
|
|
result.append(
|
|
Location(
|
|
id=location_id,
|
|
name=name,
|
|
provider=self.provider,
|
|
lat=0,
|
|
lon=0,
|
|
country=country,
|
|
country_code="ru", # TODO: contry_code from country
|
|
district=district,
|
|
subdistrict=subdistrict,
|
|
)
|
|
)
|
|
return result
|
|
|
|
@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)
|
|
soup = BeautifulSoup(content, features="html.parser")
|
|
return parse_page(soup)
|
|
|
|
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,
|
|
)
|