Now lists all items from feed and caches links

main
Micke Nordin 3 years ago
parent 9d166d8b42
commit cc6c244769
Signed by: micke
GPG Key ID: 0DA0A7A5708FE257

@ -1,49 +1,97 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import hashlib
import io
import json
import os
import pickle
import ChannelProvider import ChannelProvider
import feedparser import feedparser
import requests import requests
import wx
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from Items import Item from Items import Item
import json
class SVT(ChannelProvider.ChannelProvider): class SVT(ChannelProvider.ChannelProvider):
m_cache: dict = dict()
m_cachefile = '/tmp/svt_cache'
def __init__(self) -> None: def __init__(self) -> None:
super().__init__('SVT', 'http://www.svtplay.se/rss.xml') super().__init__('SVT', 'http://www.svtplay.se/rss.xml')
if os.path.exists(self.m_cachefile):
print("Found cache file: {}".format(self.m_cachefile))
with open(self.m_cachefile, 'rb') as cachehandle:
self.m_cache = pickle.load(cachehandle)
self.m_items: list[Item] = self.parse_feed()
def refresh_items(self):
self.m_items: list[Item] = self.parse_feed()
def parse_feed(self) -> list[Item]: def parse_feed(self) -> list[Item]:
feed = feedparser.parse(self.get_feed()) feed = feedparser.parse(self.get_feed())
entries = feed['entries'] entries = feed['entries']
items: list[Item] = list() items: list[Item] = list()
for entry in entries: for entry in entries:
thumbnail = '' key = hashlib.sha256(entry['link'].encode('utf-8')).hexdigest()
svt_id = ''
for link in entry['links']: if key in self.m_cache.keys():
if str(link['type']).startswith('image/'): print("Cache hit: {}".format(key))
thumbnail = str(link['href']) thumbnail_link = self.m_cache[key]['thumbnail_link']
page = requests.get(str(entry['link'])) resolved_link = self.m_cache[key]['resolved_link']
soup = BeautifulSoup(page.text, 'html.parser') description = self.m_cache[key]['description']
published_parsed = self.m_cache[key]['published_parsed']
for element in soup.find_all('a'): title = self.m_cache[key]['title']
href = element.get('href') else:
datart= element.get('data-rt') svt_id = ''
if datart == 'top-area-play-button': for link in entry['links']:
svt_id = href.split('=')[1].split('&')[0] if str(link['type']).startswith('image/'):
thumbnail_link = str(link['href'])
api = json.loads(requests.get('https://api.svt.se/video/{}'.format(svt_id)).text) break
#print(api['videoReferences']) page = requests.get(str(entry['link']))
resolved_link = '' soup = BeautifulSoup(page.text, 'html.parser')
for reference in api['videoReferences']:
if reference['format'] == "dashhbbtv": for element in soup.find_all('a'):
resolved_link = reference['url'] href = element.get('href')
print(resolved_link) datart = element.get('data-rt')
item = Item(str(entry['description']), resolved_link, if datart == 'top-area-play-button':
self.m_provider_name, entry['published_parsed'], svt_id = href.split('=')[1].split('&')[0]
thumbnail, str(entry['title']))
api = json.loads(
requests.get(
'https://api.svt.se/video/{}'.format(svt_id)).text)
resolved_link = ''
for reference in api['videoReferences']:
if reference['format'] == "dashhbbtv":
resolved_link = reference['url']
print(resolved_link)
description = str(entry['description'])
published_parsed = entry['published_parsed']
title = str(entry['title'])
self.m_cache[key] = {'thumbnail_link': thumbnail_link}
self.m_cache[key]['resolved_link'] = resolved_link
self.m_cache[key]['description'] = description
self.m_cache[key]['published_parsed'] = published_parsed
self.m_cache[key]['title'] = title
res = requests.get(thumbnail_link)
content_bytes = io.BytesIO(res.content)
image = wx.Image(content_bytes,
type=wx.BITMAP_TYPE_ANY,
index=-1)
thumbnail = wx.Bitmap(image)
item = Item(description, resolved_link, self.m_provider_name,
published_parsed, thumbnail, title)
items.append(item) items.append(item)
# write to cache file
with open(self.m_cachefile, 'wb') as cachehandle:
pickle.dump(self.m_cache, cachehandle)
return items return items

@ -9,7 +9,7 @@ class ChannelProvider:
def __init__(self, provider_name: str, feed: str) -> None: def __init__(self, provider_name: str, feed: str) -> None:
self.m_provider_name = provider_name self.m_provider_name = provider_name
self.m_feed = feed self.m_feed = feed
self.m_items: Union[list[Item], None] = self.parse_feed() self.m_items: Union[list[Item], None] = None
def get_feed(self) -> str: def get_feed(self) -> str:
return self.m_feed return self.m_feed

@ -1,5 +1,5 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import wx
from datetime import datetime from datetime import datetime
@ -9,7 +9,7 @@ class Item(dict):
link: str, link: str,
provider_id: str, provider_id: str,
published: datetime, published: datetime,
thumbnail: str, thumbnail: wx.Bitmap,
title: str, title: str,
watchtime: int = 0): watchtime: int = 0):
self.__dict__['description'] = description self.__dict__['description'] = description

@ -1,8 +1,8 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import hashlib
from typing import NewType from typing import NewType
import wx import wx
import wx.lib.scrolledpanel as scrolled
import wx.media import wx.media
from ChannelProvider import SVT, ChannelProvider from ChannelProvider import SVT, ChannelProvider
@ -11,53 +11,84 @@ ChannelProvider = NewType('ChannelProvider', ChannelProvider)
class Cast(wx.Frame): class Cast(wx.Frame):
m_channels: list[ChannelProvider] = list()
def __init__(self, *args, **kw): def __init__(self, *args, **kw):
"""__init__. """__init__.
:param args: :param args:
:param kw: :param kw:
""" """
super().__init__(*args, **kw) super().__init__(*args, **kw)
self.m_channels.append(SVT.SVT()) self.m_index = 0
self.m_sizer: wx.Sizer = wx.BoxSizer(wx.VERTICAL)
self.m_panel: wx.lib.scrolledpanel.ScrolledPanel = scrolled.ScrolledPanel(
self, -1, style=wx.VSCROLL)
self.m_control = None
self.m_panel.SetupScrolling()
self.m_panel.SetSizer(self.m_sizer)
self.m_channels: list[ChannelProvider] = [SVT.SVT()]
self.show_list(None)
def show_list(self, _):
self.m_sizer.Clear(delete_windows=True)
self.m_sizer = wx.BoxSizer(wx.VERTICAL)
for channel in self.m_channels:
for item in channel.get_items():
title = wx.StaticText(self.m_panel, -1, item['title'])
description = wx.StaticText(self.m_panel, -1,
item['description'])
bitmap = item['thumbnail']
btn = wx.BitmapButton(self.m_panel, id=self.m_index, bitmap=bitmap)
btn.Bind(wx.EVT_BUTTON,
lambda event, link=item['link']: self.show_player(
event, link))
self.m_sizer.Add(title)
self.m_sizer.Add(btn)
self.m_sizer.Add(description)
self.m_index = self.m_index + 1
self.m_panel.SetSizer(self.m_sizer)
self.m_panel.Layout()
def show_player(self, _, uri):
self.m_sizer.Clear(delete_windows=True)
self.m_sizer = wx.GridBagSizer()
self.m_control = wx.media.MediaCtrl( self.m_control = wx.media.MediaCtrl(
self, self.m_panel,
size=(480, 480), size=(480, 480),
style=wx.SIMPLE_BORDER, style=wx.SIMPLE_BORDER,
szBackend=wx.media.MEDIABACKEND_GSTREAMER) szBackend=wx.media.MEDIABACKEND_GSTREAMER)
play_button = wx.Button(self, -1, "Play") play_button = wx.Button(self.m_panel, -1, "Play")
self.Bind(wx.EVT_BUTTON, self.play, play_button) play_button.Bind(wx.EVT_BUTTON, self.play)
pause_button = wx.Button(self, -1, "Pause") pause_button = wx.Button(self.m_panel, -1, "Pause")
self.Bind(wx.EVT_BUTTON, self.pause, pause_button) pause_button.Bind(wx.EVT_BUTTON, self.pause)
stop_button = wx.Button(self, -1, "Stop") back_button = wx.Button(self.m_panel, -1, "Back")
self.Bind(wx.EVT_BUTTON, self.quit, stop_button) back_button.Bind(wx.EVT_BUTTON, self.show_list)
sizer = wx.GridBagSizer() self.m_sizer.Add(self.m_control, (0, 0))
sizer.Add(self.m_control, (0, 0)) self.m_sizer.SetItemSpan(0, (0, 6))
sizer.SetItemSpan(0, (0, 6)) self.m_sizer.Add(play_button, (1, 1))
sizer.Add(play_button, (1, 1)) self.m_sizer.Add(pause_button, (1, 2))
sizer.Add(pause_button, (1, 2)) self.m_sizer.Add(back_button, (1, 3))
sizer.Add(stop_button, (1, 3))
self.SetSizer(sizer) self.Bind(wx.media.EVT_MEDIA_LOADED, self.play)
self.Bind(wx.media.EVT_MEDIA_FINISHED, self.show_list)
self.m_control.Bind(wx.media.EVT_MEDIA_LOADED, self.play) self.load_uri(uri)
self.m_control.Bind(wx.media.EVT_MEDIA_FINISHED, self.quit) self.m_panel.SetSizer(self.m_sizer)
self.load_video(self.m_channels[0].get_items()[0]['link']) self.m_panel.Layout()
self.Show(True) self.m_panel.ScrollChildIntoView(self.m_control)
sizer.Layout()
def load_uri(self, uri):
def load_video(self, uri):
self.m_control.LoadURI(uri) self.m_control.LoadURI(uri)
def play(self, event): def play(self, _):
self.m_control.Play() self.m_control.Play()
def pause(self, event): def pause(self, _):
self.m_control.Pause() self.m_control.Pause()
def quit(self, event): def quit(self, _):
self.Destroy() self.Destroy()

Loading…
Cancel
Save