init

gismeteo/api.py

@@ -0,0 +1,51 @@
+import datetime
+from typing import Any, Dict, List, NamedTuple
+
+import aiohttp
+from bs4 import BeautifulSoup
+
+from .location import LOCATION_BUNDLE
+from .parser import ROW_PARSERS, OneDayParser
+
+
+class WeatherValue(NamedTuple):
+    date: datetime.datetime
+    cloudness: str
+    temperature: int
+    wind_speed: int
+    wind_gust: int
+    wind_direction: str
+    precipitation: float
+    pressure: int
+    humidity: int
+
+
+class GismeteoApi:
+    BASE_URL = "https://www.gismeteo.ru"
+
+    async def _request(self, endpoint: str) -> str:
+        url = f"{self.BASE_URL}/{endpoint}"
+        async with aiohttp.ClientSession(raise_for_status=True) as session:
+            async with session.request("GET", url) as response:
+                return await response.text()
+
+    def _parse_oneday(self, data: str) -> List[WeatherValue]:
+        result: List[Dict[str, Any]] = []
+        soup = BeautifulSoup(data, features="html.parser")
+        widget = OneDayParser().parse_widget(soup)
+        for parser in ROW_PARSERS:
+            for index, value in enumerate(parser.parse_row(widget)):
+                while len(result) < index + 1:
+                    result.append({})
+                result[index][parser.KEY] = value
+        return [WeatherValue(**item) for item in result]
+
+    async def today(self, location_id: str) -> List[WeatherValue]:
+        location = LOCATION_BUNDLE.parse(location_id)
+        data = await self._request(f"weather-{location}/today")
+        return self._parse_oneday(data)
+
+    async def tomorrow(self, location_id: str) -> List[WeatherValue]:
+        location = LOCATION_BUNDLE.parse(location_id)
+        data = await self._request(f"weather-{location}/tomorrow")
+        return self._parse_oneday(data)

gismeteo/app.py

@@ -0,0 +1,29 @@
+from os import environ
+
+import uvicorn
+from fastapi import FastAPI
+
+from gismeteo import custom_doc
+from gismeteo.api import GismeteoApi
+
+
+app = FastAPI(docs_url=None, redoc_url=None)
+custom_doc.apply(app)
+
+
+@app.get("/")
+async def root():
+    return {}
+
+
+@app.get("/weather/{location_id}")
+async def get_weather(location_id: str):
+    api = GismeteoApi()
+    result = await api.today(location_id)
+    return [item._asdict() for item in result]
+
+
+def run():
+    uvicorn.run(
+        "gismeteo.app:app", host="0.0.0.0", port=8000, reload="DEBUG" in environ
+    )

gismeteo/core.py

@@ -0,0 +1,31 @@
+from typing import Generic, Iterable, Optional, TypeVar
+
+from bs4 import Tag
+
+T = TypeVar("T")
+
+
+class WidgetParser:
+    def parse_widget(self, tag: Tag) -> Tag:
+        raise NotImplementedError
+
+
+class BaseWidgetParser(WidgetParser):
+    SELECT: str
+
+    def __init__(self, select: Optional[str] = None):
+        super().__init__()
+        self._select = select or self.SELECT
+
+    def parse_widget(self, tag: Tag) -> Tag:
+        widget = tag.select_one(self._select)
+        if widget is None:
+            raise ValueError(self._select)
+        return widget
+
+
+class RowParser(Generic[T]):
+    KEY: str
+
+    def parse_row(self, tag: Tag) -> Iterable[T]:
+        raise NotImplementedError

gismeteo/custom_doc.py

@@ -0,0 +1,36 @@
+from anyio import Path
+from fastapi import FastAPI
+from fastapi.openapi.docs import (
+    get_redoc_html,
+    get_swagger_ui_html,
+    get_swagger_ui_oauth2_redirect_html,
+)
+from fastapi.staticfiles import StaticFiles
+
+
+def apply(app: FastAPI):
+    app.mount(
+        "/docs/static", StaticFiles(directory=Path(__file__).parent.parent / "static/docs")
+    )
+
+    @app.get("/docs", include_in_schema=False)
+    async def custom_swagger_ui_html():
+        return get_swagger_ui_html(
+            openapi_url=app.openapi_url,
+            title=app.title + " - Swagger UI",
+            oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
+            swagger_js_url="docs/static/swagger-ui-bundle.js",
+            swagger_css_url="docs/static/swagger-ui.css",
+        )
+
+    @app.get(app.swagger_ui_oauth2_redirect_url, include_in_schema=False)
+    async def swagger_ui_redirect():
+        return get_swagger_ui_oauth2_redirect_html()
+
+    @app.get("/redoc", include_in_schema=False)
+    async def redoc_html():
+        return get_redoc_html(
+            openapi_url=app.openapi_url,
+            title=app.title + " - ReDoc",
+            redoc_js_url="docs/static/redoc.standalone.js",
+        )

gismeteo/location.py

@@ -0,0 +1,34 @@
+from typing import NamedTuple
+
+
+class LocationValue(NamedTuple):
+    location_id: int
+    name: str
+
+    def __str__(self) -> str:
+        return f"{self.name}-{self. location_id}"
+
+    @classmethod
+    def parse(cls, source: str) -> "LocationValue":
+        location, name = source.split("-")
+        return cls(int(location), name)
+
+
+class LocationBundle:
+    def __init__(self, *values: LocationValue) -> None:
+        self._values = values
+        self._values_by_name = {value.name: value for value in self._values}
+        self._values_by_id = {value.location_id: value for value in self._values}
+
+    def parse(self, value: str) -> LocationValue:
+        if str.isdigit(value):
+            return self._values_by_id[int(value)]
+        if value in self._values_by_name:
+            return self._values_by_name[value]
+        return LocationValue.parse(value)
+
+
+LOCATION_BUNDLE = LocationBundle(
+    LocationValue(4432, "orel"),
+    LocationValue(184640, "zmiyevka"),
+)

gismeteo/parser.py

@@ -0,0 +1,118 @@
+import datetime
+from typing import Dict, Iterable, List
+
+import dateparser
+from bs4 import Tag
+
+from .core import BaseWidgetParser, RowParser
+
+
+class OneDayParser(BaseWidgetParser):
+    SELECT = ".widget.widget-oneday .widget-items"
+
+
+class DateParser(RowParser[datetime.datetime]):
+    KEY = "date"
+
+    def parse_row(self, tag: Tag) -> Iterable[datetime.datetime]:
+        date_str = (
+            tag.select_one(".widget-row.widget-row-datetime-date > .row-item")
+            .find(text=True, recursive=False)
+            .text
+        )
+        date = dateparser.parse(date_str, languages=["ru"])
+        for item in tag.select(".widget-row.widget-row-datetime-time > .row-item"):
+            time_str = item.text
+            time = dateparser.parse(time_str, languages=["ru"])
+            time = time.replace(year=date.year, month=date.month, day=date.day)
+            yield time
+
+
+class CloudnessParser(RowParser[str]):
+    KEY = "cloudness"
+
+    def parse_row(self, tag: Tag) -> Iterable[str]:
+        for item in tag.select(".widget-row[data-row=icon-tooltip] > .row-item"):
+            yield item.attrs["data-tooltip"]
+
+
+class TemperatureParser(RowParser[int]):
+    KEY = "temperature"
+
+    def parse_row(self, tag: Tag) -> Iterable[int]:
+        for item in tag.select(
+            ".widget-row-chart[data-row=temperature-air] > .chart > .values > .value > temperature-value"
+        ):
+            yield int(item.attrs["value"])
+
+
+class WindSpeedParser(RowParser[int]):
+    KEY = "wind_speed"
+
+    def parse_row(self, tag: Tag) -> Iterable[int]:
+        for item in tag.select(
+            ".widget-row[data-row=wind-speed] > .row-item > speed-value"
+        ):
+            yield int(item.attrs["value"])
+
+
+class WindGustParser(RowParser[int]):
+    KEY = "wind_gust"
+
+    def parse_row(self, tag: Tag) -> Iterable[int]:
+        for item in tag.select(".widget-row[data-row=wind-gust] > .row-item"):
+            value = item.select_one("speed-value")
+            yield int(value.attrs["value"]) if value else 0
+
+
+class WindDirectionParser(RowParser[str]):
+    KEY = "wind_direction"
+
+    def parse_row(self, tag: Tag) -> Iterable[str]:
+        for item in tag.select(
+            ".widget-row[data-row=wind-direction] > .row-item > .direction"
+        ):
+            yield item.text
+
+
+class WindPrecipitationParser(RowParser[float]):
+    KEY = "precipitation"
+
+    def parse_row(self, tag: Tag) -> Iterable[float]:
+        for item in tag.select(
+            ".widget-row[data-row=precipitation-bars] > .row-item > .item-unit"
+        ):
+            yield float(item.text.replace(",", "."))
+
+
+class PressureParser(RowParser[int]):
+    KEY = "pressure"
+
+    def parse_row(self, tag: Tag) -> Iterable[int]:
+        for item in tag.select(
+            ".widget-row-chart[data-row=pressure] > .chart > .values > .value > pressure-value"
+        ):
+            yield int(item.attrs["value"])
+
+
+class HumidityParser(RowParser[int]):
+    KEY = "humidity"
+
+    def parse_row(self, tag: Tag) -> Iterable[int]:
+        for item in tag.select(".widget-row[data-row=humidity] > .row-item"):
+            yield int(item.text)
+
+
+ROW_PARSERS: List[RowParser] = [
+    DateParser(),
+    CloudnessParser(),
+    TemperatureParser(),
+    WindSpeedParser(),
+    WindGustParser(),
+    WindDirectionParser(),
+    WindPrecipitationParser(),
+    PressureParser(),
+    HumidityParser(),
+]
+
+ROW_PARSERS_MAP: Dict[str, RowParser] = {parser.KEY: parser for parser in ROW_PARSERS}