cast/Channel/YouTube/__init__.py

import hashlib
import io
import os
import pickle
import threading
from typing import Union

import feedparser
import requests
import wx
from bs4 import BeautifulSoup
from youtube_dl import YoutubeDL as yt

from Channel import Channel
from Items import Item
from Utils import get_default_log_url, make_bitmap_from_url


class YouTube(Channel):
    m_cache: dict = dict()
    m_cachefile = '/tmp/yt_cache'

    def __init__(self, channel_id) -> None:
        self.m_channel_id = channel_id
        rss_url = 'https://www.youtube.com/feeds/videos.xml?channel_id={}'.format(
            channel_id)
        logo_url = get_default_log_url('YouTube')
        super().__init__(channel_id, rss_url, logo_url)
        self.set_avatar()
        self.m_items: Union[list[Item], None] = None

        if os.path.exists(self.m_cachefile):
            with open(self.m_cachefile, 'rb') as cachehandle:
                self.m_cache = pickle.load(cachehandle)
        self.m_thr = threading.Thread(target=self.parse_feed,
                                      args=(),
                                      kwargs={})
        self.m_thr.start()

    def set_avatar(self) -> None:
        info = get_info(self.m_channel_id)
        bmap = self.get_logo_as_bitmap()
        title = info['title']
        dc = wx.MemoryDC(bmap)
        cblack = wx.Colour(0, 0, 0)
        cwhite = wx.Colour(255, 255, 255)
        dc.SetTextForeground(cwhite)
        dc.SetTextBackground(cblack)
        dc.SetFont(wx.Font().Bold())
        dc.SetBackgroundMode(wx.BRUSHSTYLE_SOLID)
        w, h = dc.GetSize()
        tw, th = dc.GetTextExtent(title)
        dc.DrawText(title, (w - tw) / 2, (h - th) / 2)  #display text in center
        del dc
        self.m_logo = bmap

    def wait(self) -> bool:
        return self.m_thr.is_alive()

    def parse_feed(self) -> None:
        feed = feedparser.parse(self.get_feed())

        ydl_opts = {
            'format':
            'worstvideo[ext=mp4]+worstaudio[ext=m4a]/worstvideo+worstaudio',
            'container': 'webm_dash',
        }
        entries = feed['entries']
        self.m_items: list[Item] = list()

        for entry in entries:
            key = hashlib.sha256(entry['link'].encode('utf-8')).hexdigest()

            if key in self.m_cache.keys():
                thumbnail_link = self.m_cache[key]['thumbnail_link']
                resolved_link = self.m_cache[key]['resolved_link']
                description = self.m_cache[key]['description']
                published_parsed = self.m_cache[key]['published_parsed']
                title = self.m_cache[key]['title']
            else:

                title = str(entry['title'])
                thumbnail_link = str(entry['media_thumbnail'][0]['url'])
                description = str(entry['description'])
                link = ''
                with yt(ydl_opts) as ydl:
                    video = ydl.extract_info(entry['link'], download=False)

                    for form in video['formats']:
                        if form['height']:
                            if form['height'] < 480 and form[
                                    'acodec'] != 'none':
                                link = form['url']

                resolved_link = link

                published_parsed = entry['published_parsed']

            if not resolved_link:
                continue
            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
            thumbnail = make_bitmap_from_url(thumbnail_link)
            item = Item(description, resolved_link, self.m_provider_name,
                        published_parsed, thumbnail, title)
            self.m_items.append(item)
        # write to cache file
        with open(self.m_cachefile, 'wb') as cachehandle:
            pickle.dump(self.m_cache, cachehandle)


def get_info(channel_id: str) -> str:
    link = 'https://www.youtube.com/channel/{}'.format(channel_id)
    ydl_opts = {'extract_flat': True, 'skip_download': True}
    with yt(ydl_opts) as ydl:
        info = ydl.extract_info(link, download=False)
    return info