refactor(parse): add base html and json parsers
This commit is contained in:
@@ -1,14 +1,15 @@
|
||||
import datetime
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from typing import Literal, NamedTuple
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Literal
|
||||
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
from bs4 import Tag
|
||||
|
||||
from gallery.sketch.parse.core import ParseError, Parser
|
||||
from gallery.sketch.parse.core import ParsedContext, ParseError, Parser
|
||||
from gallery.sketch.parse.date import parse_date
|
||||
from gallery.sketch.parse.html import TableParser, TableValueParser
|
||||
from gallery.sketch.parse.html import HTMLContextParser, TableParser, TableValueParser
|
||||
from gallery.sketch.parse.json import JSONParser
|
||||
from gallery.sketch.weather.model import (
|
||||
Cloudness,
|
||||
Location,
|
||||
@@ -198,42 +199,41 @@ class GismeteoTableParser(TableParser[WeatherValue]):
|
||||
return WeatherValue.model_validate(data)
|
||||
|
||||
|
||||
class ForecastData(NamedTuple):
|
||||
@dataclass
|
||||
class ForecastData:
|
||||
date: datetime.date
|
||||
period: Literal["day", "days"]
|
||||
data: str
|
||||
values: int | None = None
|
||||
|
||||
|
||||
class GismeteoForecastParser(Parser[ForecastData, WeatherResponse]):
|
||||
class GismeteoForecastParser(HTMLContextParser[ForecastData, WeatherResponse]):
|
||||
location_parser = LocationNameParser()
|
||||
table_parser = GismeteoTableParser()
|
||||
|
||||
def parse(self, data: ForecastData) -> WeatherResponse:
|
||||
location = self.location_parser.parse(data.data)
|
||||
soup = BeautifulSoup(data.data, features="html.parser")
|
||||
values = self.table_parser.parse(soup)
|
||||
if data.values and len(values) > data.values:
|
||||
n = int(len((values)) / data.values)
|
||||
def extract(self, data: ParsedContext[ForecastData, Tag]) -> WeatherResponse:
|
||||
location = self.location_parser.parse(data.context.data)
|
||||
values = self.table_parser.parse(data.data)
|
||||
if data.context.values and len(values) > data.context.values:
|
||||
n = int(len((values)) / data.context.values)
|
||||
values = [
|
||||
merge_weather_values(values[i].date, values[i : i + n], "average") for i in range(0, len(values), n)
|
||||
]
|
||||
if data.period == "days":
|
||||
if data.context.period == "days":
|
||||
fix_dates_sequence(values)
|
||||
return WeatherResponse(
|
||||
location=location,
|
||||
date=data.date,
|
||||
period=data.period,
|
||||
date=data.context.date,
|
||||
period=data.context.period,
|
||||
values=values,
|
||||
)
|
||||
|
||||
|
||||
class GismeteoLocationParser(Parser[str, list[Location]]):
|
||||
class GismeteoLocationParser(JSONParser[list[Location]]):
|
||||
|
||||
def parse(self, data: str) -> list[Location]:
|
||||
content = json.loads(data)
|
||||
def extract(self, data: Any) -> list[Location]:
|
||||
result = []
|
||||
for item in content["data"]:
|
||||
for item in data["data"]:
|
||||
result.append(
|
||||
Location(
|
||||
id=f"{item['slug']}-{item['id']}",
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
import datetime
|
||||
import json
|
||||
import re
|
||||
from typing import NamedTuple
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from gallery.sketch.parse.core import Parser
|
||||
from gallery.sketch.parse.core import ParsedContext, Parser
|
||||
from gallery.sketch.parse.json import JSONContextParser
|
||||
from gallery.sketch.schedule.model import Channel, Schedule, ScheduleValue
|
||||
|
||||
from . import PROVIDER
|
||||
@@ -15,20 +16,20 @@ class DateParser(Parser[str, datetime.date]):
|
||||
return datetime.date.strptime(matched.group(1), data)
|
||||
|
||||
|
||||
class ScheduleData(NamedTuple):
|
||||
@dataclass
|
||||
class ScheduleData:
|
||||
date: datetime.date
|
||||
data: str
|
||||
|
||||
|
||||
class MatchTVScheduleParser(Parser[ScheduleData, Schedule]):
|
||||
class MatchTVScheduleParser(JSONContextParser[ScheduleData, Schedule]):
|
||||
date_parser = DateParser()
|
||||
|
||||
def parse(self, data: ScheduleData) -> Schedule:
|
||||
content = json.loads(data.data)
|
||||
channel_data = content["result"]["channels"][0]
|
||||
def extract(self, data: ParsedContext[ScheduleData, Any]) -> Schedule:
|
||||
channel_data = data.data["result"]["channels"][0]
|
||||
channel = Channel(id=channel_data["alias"], name=channel_data["name"], provider=PROVIDER)
|
||||
values = []
|
||||
start = datetime.datetime.combine(data.date, datetime.time.min)
|
||||
start = datetime.datetime.combine(data.context.date, datetime.time.min)
|
||||
end = start + datetime.timedelta(days=1, hours=6)
|
||||
prev_value: ScheduleValue | None = None
|
||||
for item in channel_data["schedule"]:
|
||||
@@ -47,22 +48,22 @@ class MatchTVScheduleParser(Parser[ScheduleData, Schedule]):
|
||||
prev_value = value
|
||||
return Schedule(
|
||||
channel=channel,
|
||||
date=data.date,
|
||||
date=data.context.date,
|
||||
values=values,
|
||||
)
|
||||
|
||||
|
||||
class ChannelData(NamedTuple):
|
||||
@dataclass
|
||||
class ChannelData:
|
||||
query: str
|
||||
data: str
|
||||
|
||||
|
||||
class MatchTVChannelParser(Parser[ChannelData, list[Channel]]):
|
||||
def parse(self, data: ChannelData) -> list[Channel]:
|
||||
context = json.loads(data.data)
|
||||
class MatchTVChannelParser(JSONContextParser[ChannelData, list[Channel]]):
|
||||
def extract(self, data: ParsedContext[ChannelData, Any]) -> list[Channel]:
|
||||
result = []
|
||||
query = data.query.lower()
|
||||
for item in context["result"]:
|
||||
query = data.context.query.lower()
|
||||
for item in data.data["result"]:
|
||||
if query in item["name"].lower() or query in item["alias"]:
|
||||
channel_id = item["alias"]
|
||||
name = item["name"].split("|")[0].strip()
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
import datetime
|
||||
import re
|
||||
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
from bs4 import Tag
|
||||
|
||||
from gallery.sketch.parse.core import Parser
|
||||
from gallery.sketch.parse.date import parse_date
|
||||
from gallery.sketch.parse.html import (
|
||||
HTMLParser,
|
||||
TableParser,
|
||||
TableValueParser,
|
||||
TagIterParser,
|
||||
@@ -82,6 +82,7 @@ class PrecipitationParser(TableValueParser[tuple[Precipitation, float]]):
|
||||
PRECIPITATION_MAP = {
|
||||
"Явления погоды отсутствуют": Precipitation.NO,
|
||||
"Cлабый дождь": Precipitation.SMALL_RAIN,
|
||||
"Умеренный дождь": Precipitation.RAIN,
|
||||
}
|
||||
|
||||
def extract(self, data: Tag) -> tuple[Precipitation, float]:
|
||||
@@ -92,7 +93,7 @@ class PrecipitationParser(TableValueParser[tuple[Precipitation, float]]):
|
||||
if precipitation_value_matched:
|
||||
# one hour value
|
||||
precipitation_value = float(precipitation_value_matched.group(1)) * 6
|
||||
if "(" in precipitation_str:
|
||||
if " (" in precipitation_str:
|
||||
precipitation_str = precipitation_str.split(" (")[0]
|
||||
return self.PRECIPITATION_MAP[precipitation_str], precipitation_value
|
||||
|
||||
@@ -163,16 +164,15 @@ class RP5TableParser(TableParser[WeatherValue]):
|
||||
return WeatherValue.model_validate(data)
|
||||
|
||||
|
||||
class RP5ForecastParser(Parser[str, list[WeatherResponse]]):
|
||||
class RP5ForecastParser(HTMLParser[list[WeatherResponse]]):
|
||||
location_parser = LocationNameParser()
|
||||
dates_parser = DatesParser()
|
||||
table_parser = RP5TableParser()
|
||||
|
||||
def parse(self, data: str) -> list[WeatherResponse]:
|
||||
soup = BeautifulSoup(data, features="html.parser")
|
||||
location = self.location_parser.parse(soup)
|
||||
dates = self.dates_parser.parse(soup)
|
||||
values = self.table_parser.parse(soup)
|
||||
def extract(self, data: Tag) -> list[WeatherResponse]:
|
||||
location = self.location_parser.parse(data)
|
||||
dates = self.dates_parser.parse(data)
|
||||
values = self.table_parser.parse(data)
|
||||
responses = []
|
||||
for date, count in dates:
|
||||
item_values, values = values[:count], values[count:]
|
||||
@@ -190,13 +190,12 @@ class RP5ForecastParser(Parser[str, list[WeatherResponse]]):
|
||||
return responses
|
||||
|
||||
|
||||
class RP5LocationParser(Parser[str, list[Location]]):
|
||||
class RP5LocationParser(HTMLParser[list[Location]]):
|
||||
selector = ".searchResults"
|
||||
|
||||
def parse(self, data: str) -> list[Location]:
|
||||
soup = BeautifulSoup(data, features="html.parser")
|
||||
table = soup.select_one(".searchResults")
|
||||
def extract(self, data: Tag) -> list[Location]:
|
||||
result = []
|
||||
for row in table.select("tr")[1:]:
|
||||
for row in data.select("tr")[1:]:
|
||||
cells = row.select("td")
|
||||
name = cells[0].text.strip()
|
||||
location_id = cells[0].select_one("a").attrs["href"].lstrip("/")
|
||||
|
||||
@@ -1,32 +1,34 @@
|
||||
import datetime
|
||||
import json
|
||||
from typing import NamedTuple
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from bs4 import BeautifulSoup
|
||||
from bs4 import Tag
|
||||
|
||||
from gallery.sketch.parse.core import Parser
|
||||
from gallery.sketch.parse.core import ParsedContext
|
||||
from gallery.sketch.parse.json import JSONParser
|
||||
from gallery.sketch.schedule.model import Channel, Schedule, ScheduleValue
|
||||
|
||||
from ...sketch.parse.html import HTMLContextParser
|
||||
from . import PROVIDER
|
||||
|
||||
|
||||
class ScheduleData(NamedTuple):
|
||||
@dataclass
|
||||
class ScheduleData:
|
||||
date: datetime.date
|
||||
channel_id: str
|
||||
data: str
|
||||
|
||||
|
||||
class YandexTVScheduleParser(Parser[ScheduleData, Schedule]):
|
||||
def parse(self, data: ScheduleData) -> Schedule:
|
||||
soup = BeautifulSoup(data.data, features="html.parser")
|
||||
if soup.select_one(".CheckboxCaptcha") is not None:
|
||||
class YandexTVScheduleParser(HTMLContextParser[ScheduleData, Schedule]):
|
||||
def extract(self, data: ParsedContext[ScheduleData, Tag]) -> Schedule:
|
||||
if data.data.select_one(".CheckboxCaptcha") is not None:
|
||||
raise RuntimeError("Captcha")
|
||||
values = []
|
||||
channel_name = soup.select_one(".channel-header__text").text.strip()
|
||||
current_day = datetime.datetime.combine(data.date, datetime.datetime.min.time())
|
||||
channel_name = data.data.select_one(".channel-header__text").text.strip()
|
||||
current_day = datetime.datetime.combine(data.context.date, datetime.datetime.min.time())
|
||||
end = current_day + datetime.timedelta(days=1, hours=6)
|
||||
prev_value: ScheduleValue | None = None
|
||||
for item in soup.select(".channel-schedule .channel-schedule__event"):
|
||||
for item in data.data.select(".channel-schedule .channel-schedule__event"):
|
||||
title = item.select_one(".channel-schedule__title").text.strip()
|
||||
time_str = item.select_one(".channel-schedule__time").text.strip()
|
||||
hours, minutes = map(int, time_str.split(":"))
|
||||
@@ -41,15 +43,15 @@ class YandexTVScheduleParser(Parser[ScheduleData, Schedule]):
|
||||
prev_value.end = item_date
|
||||
prev_value = value
|
||||
return Schedule(
|
||||
channel=Channel(id=data.channel_id, name=channel_name, provider=PROVIDER),
|
||||
date=data.date,
|
||||
channel=Channel(id=data.context.channel_id, name=channel_name, provider=PROVIDER),
|
||||
date=data.context.date,
|
||||
values=values,
|
||||
)
|
||||
|
||||
|
||||
class YandexTVChannelParser(Parser[str, list[Channel]]):
|
||||
def parse(self, data: str) -> list[Channel]:
|
||||
_, values = json.loads(data)
|
||||
class YandexTVChannelParser(JSONParser[list[Channel]]):
|
||||
def extract(self, data: Any) -> list[Channel]:
|
||||
_, values = data
|
||||
result = []
|
||||
for _, name, content in values:
|
||||
if content["label"] == "Каналы":
|
||||
|
||||
@@ -1,12 +1,17 @@
|
||||
import datetime
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
from bs4 import Tag
|
||||
|
||||
from gallery.sketch.parse.core import Parser
|
||||
from gallery.sketch.parse.date import parse_date
|
||||
from gallery.sketch.parse.html import TableParser, TableValueParser, TagParser
|
||||
from gallery.sketch.parse.html import (
|
||||
HTMLParser,
|
||||
TableParser,
|
||||
TableValueParser,
|
||||
TagParser,
|
||||
)
|
||||
from gallery.sketch.parse.json import JSONParser
|
||||
from gallery.sketch.weather.model import (
|
||||
Cloudness,
|
||||
Location,
|
||||
@@ -143,15 +148,14 @@ class YandexTableParser(TableParser[WeatherValue]):
|
||||
return WeatherValue.model_validate(data)
|
||||
|
||||
|
||||
class YandexForecastParser(Parser[str, list[WeatherResponse]]):
|
||||
class YandexForecastParser(HTMLParser[list[WeatherResponse]]):
|
||||
table_parser = YandexTableParser()
|
||||
date_parser = DateParser()
|
||||
location_parser = LocationNameParser()
|
||||
|
||||
def parse(self, data: str) -> list[WeatherResponse]:
|
||||
soup = BeautifulSoup(data, features="html.parser")
|
||||
location = self.location_parser.parse(soup)
|
||||
articles = soup.select("article[class^=AppForecastDay]")
|
||||
def extract(self, data: Tag) -> list[WeatherResponse]:
|
||||
location = self.location_parser.parse(data)
|
||||
articles = data.select("article[class^=AppForecastDay]")
|
||||
return [
|
||||
WeatherResponse(
|
||||
location=location,
|
||||
@@ -163,10 +167,9 @@ class YandexForecastParser(Parser[str, list[WeatherResponse]]):
|
||||
]
|
||||
|
||||
|
||||
class YandexLocationParser(Parser[str, list[Location]]):
|
||||
class YandexLocationParser(JSONParser[list[Location]]):
|
||||
|
||||
def parse(self, data: str) -> list[Location]:
|
||||
content = json.loads(data)
|
||||
def extract(self, data: Any) -> list[Location]:
|
||||
return [
|
||||
Location(
|
||||
id=str(item["geoId"]),
|
||||
@@ -179,5 +182,5 @@ class YandexLocationParser(Parser[str, list[Location]]):
|
||||
district=item["description"],
|
||||
subdistrict="",
|
||||
)
|
||||
for item in content
|
||||
for item in data
|
||||
]
|
||||
|
||||
@@ -1,13 +1,49 @@
|
||||
from typing import Generic, TypeVar
|
||||
|
||||
D = TypeVar("D")
|
||||
R = TypeVar("R")
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Generic, NamedTuple, Protocol, TypeVar
|
||||
|
||||
|
||||
class ParseError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class Parser(Generic[D, R]):
|
||||
D = TypeVar("D")
|
||||
R = TypeVar("R")
|
||||
|
||||
|
||||
class Parser[D, R](Protocol):
|
||||
def parse(self, data: D) -> R: ...
|
||||
|
||||
|
||||
P = TypeVar("P")
|
||||
|
||||
|
||||
class PrepareParser(ABC, Parser[D, R], Generic[D, P, R]):
|
||||
|
||||
@abstractmethod
|
||||
def prepare(self, data: D) -> P: ...
|
||||
|
||||
@abstractmethod
|
||||
def extract(self, data: P) -> R: ...
|
||||
|
||||
def parse(self, data: D) -> R:
|
||||
raise NotImplementedError
|
||||
return self.extract(self.prepare(data))
|
||||
|
||||
|
||||
class Context(Protocol):
|
||||
data: str
|
||||
|
||||
|
||||
C = TypeVar("C", bound=Context)
|
||||
|
||||
|
||||
class ParsedContext(NamedTuple, Generic[C, D]):
|
||||
context: C
|
||||
data: D
|
||||
|
||||
|
||||
class ContextParser(PrepareParser[C, ParsedContext[C, P], R], Generic[C, P, R]):
|
||||
@abstractmethod
|
||||
def prepare_context(self, data: str) -> P: ...
|
||||
|
||||
def prepare(self, data: C) -> ParsedContext[C, P]:
|
||||
return ParsedContext(context=data, data=self.prepare_context(data.data))
|
||||
|
||||
@@ -1,22 +1,47 @@
|
||||
from typing import Generic, Iterable
|
||||
|
||||
from bs4 import Tag
|
||||
from bs4 import BeautifulSoup, Tag
|
||||
|
||||
from .core import ParseError, Parser, R
|
||||
from .core import C, ContextParser, ParseError, Parser, PrepareParser, R
|
||||
from .util import set_dict_nested_value
|
||||
|
||||
|
||||
class TagParser(Parser[Tag, R], Generic[R]):
|
||||
selector: str
|
||||
|
||||
def extract(self, data: Tag) -> R:
|
||||
raise NotImplementedError
|
||||
|
||||
def parse(self, data: Tag) -> R:
|
||||
tag = data.select_one(self.selector) if self.selector else data
|
||||
def _resolve_tag(source: Tag, selector: str | None = None) -> Tag:
|
||||
if selector:
|
||||
tag = source.select_one(selector)
|
||||
if tag is None:
|
||||
raise ParseError("Tag not found", self.selector)
|
||||
return self.extract(tag)
|
||||
raise ParseError("Tag not found", selector)
|
||||
return tag
|
||||
return source
|
||||
|
||||
|
||||
class _HTMLParser(Parser[str, BeautifulSoup]):
|
||||
def parse(self, data: str) -> BeautifulSoup:
|
||||
return BeautifulSoup(data, features="html.parser")
|
||||
|
||||
|
||||
HTML_PARSER = _HTMLParser()
|
||||
|
||||
|
||||
class HTMLParser(PrepareParser[str, Tag, R], Generic[R]):
|
||||
selector: str | None = None
|
||||
|
||||
def prepare(self, data: str) -> Tag:
|
||||
return _resolve_tag(HTML_PARSER.parse(data), self.selector)
|
||||
|
||||
|
||||
class HTMLContextParser(ContextParser[C, Tag, R], Generic[C, R]):
|
||||
selector: str | None = None
|
||||
|
||||
def prepare_context(self, data: str) -> Tag:
|
||||
return _resolve_tag(HTML_PARSER.parse(data), self.selector)
|
||||
|
||||
|
||||
class TagParser(PrepareParser[Tag, Tag, R], Generic[R]):
|
||||
selector: str | None = None
|
||||
|
||||
def prepare(self, data: Tag) -> Tag:
|
||||
return _resolve_tag(data, self.selector)
|
||||
|
||||
|
||||
class TagIterParser(Parser[Tag, Iterable[R]], Generic[R]):
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
import json
|
||||
from typing import Any, Generic
|
||||
|
||||
from .core import C, ContextParser, Parser, PrepareParser, R
|
||||
|
||||
|
||||
class _JSONParser(Parser[str, Any]):
|
||||
def parse(self, data: str) -> Any:
|
||||
return json.loads(data)
|
||||
|
||||
|
||||
JSON_PARSER = _JSONParser()
|
||||
|
||||
|
||||
class JSONParser(PrepareParser[str, Any, R], Generic[R]):
|
||||
def prepare(self, data: str) -> Any:
|
||||
return JSON_PARSER.parse(data)
|
||||
|
||||
|
||||
class JSONContextParser(ContextParser[C, Any, R], Generic[C, R]):
|
||||
def prepare_context(self, data: str) -> Any:
|
||||
return JSON_PARSER.parse(data)
|
||||
Reference in New Issue
Block a user