196 lines
5.3 KiB
Python
196 lines
5.3 KiB
Python
import requests
|
|
import json
|
|
|
|
import datetime as dt
|
|
|
|
|
|
class MediathekViewWebAnswer(object):
|
|
def __init__(self, answer_json: str):
|
|
|
|
def get_field(field: str):
|
|
if field in answer_json:
|
|
return answer_json[field]
|
|
return None
|
|
|
|
self._raw = answer_json
|
|
|
|
self._channel = get_field('channel')
|
|
self._topic = get_field('topic')
|
|
self._title = get_field('title')
|
|
self._description = get_field('description')
|
|
self._timestamp = get_field('timestamp')
|
|
self._duration = get_field('duration')
|
|
self._size = get_field('size')
|
|
self._url_website = get_field('url_website')
|
|
self._url_subtitle = get_field('url_subtitle')
|
|
self._filmlisteTimestamp = get_field('filmlisteTimestamp')
|
|
self._id = get_field('id')
|
|
self._url_video = get_field('url_video')
|
|
self._url_video_low = get_field('url_video_low')
|
|
self._url_video_hd = get_field('url_video_hd')
|
|
|
|
def get_best_vid_url(self) -> str:
|
|
if self._url_video_hd is not None:
|
|
return self._url_video_hd
|
|
|
|
if self._url_video is not None:
|
|
return self._url_video
|
|
|
|
if self._url_video_low is not None:
|
|
return self._url_video_low
|
|
|
|
return None
|
|
|
|
def get_lowest_vid_url(self) -> str:
|
|
if self._url_video_low is not None:
|
|
return self._url_video_low
|
|
|
|
if self._url_video is not None:
|
|
return self._url_video
|
|
|
|
if self._url_video_hd is not None:
|
|
return self._url_video_hd
|
|
|
|
return None
|
|
|
|
def get_id(self):
|
|
return self._id
|
|
|
|
def get_site(self):
|
|
return self._url_website
|
|
|
|
def get_title(self):
|
|
return self._title
|
|
|
|
def get_channel(self):
|
|
return self._channel
|
|
|
|
def get_topic(self):
|
|
return self._topic
|
|
|
|
def get_description(self):
|
|
return self._description
|
|
|
|
def get_timestamp(self) -> dt.datetime:
|
|
if self._timestamp is None:
|
|
return None
|
|
return dt.datetime.utcfromtimestamp(int(self._timestamp))
|
|
|
|
def get_duration(self):
|
|
return self._duration
|
|
|
|
|
|
class MediathekViewWebRequest(object):
|
|
@staticmethod
|
|
def from_serialization(
|
|
d: dict,
|
|
request_url="https://mediathekviewweb.de/api/query"
|
|
) -> 'MediathekViewWebRequest':
|
|
|
|
json_query = d['queries'][0]
|
|
sort_by = d['sortBy']
|
|
sort_order = d['sortOrder']
|
|
future = d['future']
|
|
offset = d['offset']
|
|
size = d['size']
|
|
|
|
query = json_query['query']
|
|
fields = json_query['fields']
|
|
|
|
search_channel = "channel" in fields
|
|
search_topic = "topic" in fields
|
|
search_title = "title" in fields
|
|
|
|
return MediathekViewWebRequest(
|
|
query=query,
|
|
search_title=search_title,
|
|
search_channel=search_channel,
|
|
search_topic=search_topic,
|
|
sort_by=sort_by,
|
|
sort_order=sort_order,
|
|
future=future,
|
|
size=size,
|
|
offset=offset,
|
|
request_url=request_url
|
|
)
|
|
|
|
@staticmethod
|
|
def from_json(
|
|
js: str,
|
|
request_url="https://mediathekviewweb.de/api/query"
|
|
) -> 'MediathekViewWebRequest':
|
|
return MediathekViewWebRequest.from_serialization(
|
|
d=json.loads(js),
|
|
request_url=request_url
|
|
)
|
|
|
|
def __init__(self,
|
|
query: str,
|
|
search_title: bool = True,
|
|
search_channel: bool = False,
|
|
search_topic: bool = False,
|
|
sort_by: str = 'timestamp',
|
|
sort_order: str = 'desc',
|
|
future: bool = False,
|
|
size: int = 10,
|
|
offset: int = 0,
|
|
request_url="https://mediathekviewweb.de/api/query"):
|
|
|
|
assert search_title or search_channel or search_topic
|
|
assert sort_order == 'desc' or sort_order == 'asc'
|
|
|
|
self._query = query
|
|
self._fields = []
|
|
if search_title:
|
|
self._fields.append("title")
|
|
if search_channel:
|
|
self._fields.append("channel")
|
|
if search_topic:
|
|
self._fields.append("topic")
|
|
|
|
self._sort_by = sort_by
|
|
self._sort_order = sort_order
|
|
self._future = future
|
|
self._size = size
|
|
self._offset = offset
|
|
self._request_url = request_url
|
|
|
|
def serialize(self):
|
|
query = {
|
|
'queries': [
|
|
{
|
|
'fields': self._fields,
|
|
'query': self._query
|
|
}
|
|
],
|
|
|
|
'sortBy': self._sort_by,
|
|
'sortOrder': self._sort_order,
|
|
'future': self._future,
|
|
'offset': self._offset,
|
|
'size': self._size
|
|
}
|
|
return query
|
|
|
|
def to_json(self):
|
|
return json.dumps(self.serialize())
|
|
|
|
def perform_request(self):
|
|
headers = {'Content-type': 'text/plain'}
|
|
req = requests.post(self._request_url,
|
|
data=self.to_json(), headers=headers)
|
|
|
|
if not req.ok:
|
|
return None
|
|
|
|
answer = json.loads(req.text)
|
|
|
|
if 'result' in answer:
|
|
result = answer['result']
|
|
|
|
if 'results' in result:
|
|
results = result['results']
|
|
return [MediathekViewWebAnswer(i) for i in results]
|
|
|
|
return None
|