72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
import datetime
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
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
|
|
|
|
|
|
class DateParser(Parser[str, datetime.date]):
|
|
def parse(self, data: str) -> datetime.date:
|
|
matched = re.compile('{\\"date\\":(\\d+)}').search(data)
|
|
return datetime.date.strptime(matched.group(1), data)
|
|
|
|
|
|
@dataclass
|
|
class ScheduleData:
|
|
date: datetime.date
|
|
data: str
|
|
|
|
|
|
class MatchTVScheduleParser(JSONContextParser[ScheduleData, Schedule]):
|
|
date_parser = DateParser()
|
|
|
|
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.context.date, datetime.time.min)
|
|
end = start + datetime.timedelta(days=1, hours=6)
|
|
prev_value: ScheduleValue | None = None
|
|
for item in channel_data["schedule"]:
|
|
title = item["title"]
|
|
time_str = item["time"]
|
|
hours, minutes = map(int, time_str.split(":"))
|
|
item_date = start.replace(hour=hours, minute=minutes)
|
|
if prev_value is not None and item_date.hour < prev_value.start.hour:
|
|
start += datetime.timedelta(days=1)
|
|
item_date += datetime.timedelta(days=1)
|
|
live = "Прямая трансляция" in title
|
|
value = ScheduleValue(start=item_date, end=end, label=title, live=live)
|
|
values.append(value)
|
|
if prev_value is not None:
|
|
prev_value.end = item_date
|
|
prev_value = value
|
|
return Schedule(
|
|
channel=channel,
|
|
date=data.context.date,
|
|
values=values,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class ChannelData:
|
|
query: str
|
|
data: str
|
|
|
|
|
|
class MatchTVChannelParser(JSONContextParser[ChannelData, list[Channel]]):
|
|
def extract(self, data: ParsedContext[ChannelData, Any]) -> list[Channel]:
|
|
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()
|
|
result.append(Channel(id=channel_id, name=name, provider=PROVIDER))
|
|
return result
|