mediathek-viewer/src/gtk_mediathek_player/mediathek_request.py

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