cast/src/Utils/__init__.py

#!/usr/bin/env python3
import io
import json
import sqlite3
from datetime import datetime
from os import environ, makedirs, path
from typing import Callable, Union
from urllib.parse import urlparse
from Items import Item

import requests
import wx

SIZE = wx.Size(100, 68)
MYPATH = path.dirname(path.abspath(__file__))
SCREEN_WIDTH = int(720 / 2)
BASEPATH = path.join(str(environ.get("HOME")), '.config/cast')
DB_FILE_NAME = 'cast.db'
SUB_TABLE = 'subscriptions'
VIDEO_TABLE = 'videos'


def add_subscription(channel_id: str,
                     name: str,
                     basepath: str = BASEPATH,
                     filename: str = DB_FILE_NAME) -> None:
    fullpath = path.join(basepath, filename)
    if not path.isdir(basepath):
        makedirs(basepath)
    con = sqlite3.connect(fullpath)
    cur = con.cursor()
    create_query: str = '''CREATE TABLE IF NOT EXISTS {}
    (channel_id TEXT PRIMARY KEY, channel_name TEXT)'''.format(SUB_TABLE)
    cur.execute(create_query)
    con.commit()
    upsert_query: str = '''INSERT INTO {} (channel_id, channel_name)
    VALUES('{}',"{}") ON CONFLICT(channel_id) DO NOTHING'''.format(
        SUB_TABLE, channel_id, name)
    cur.execute(upsert_query)
    con.commit()


def add_video(video_id: str,
              channel_id: str,
              provider_id: str,
              description: str,
              link: str,
              published: datetime,
              bitmap: wx.Bitmap,
              title: str,
              watchtime: str,
              basepath: str = BASEPATH,
              filename: str = DB_FILE_NAME) -> None:
    thumbnail = bitmap.GetData()
    fullpath = path.join(basepath, filename)
    if not path.isdir(basepath):
        makedirs(basepath)
    con = sqlite3.connect(fullpath)
    cur = con.cursor()
    create_query: str = '''CREATE TABLE IF NOT EXISTS {}
    (video_id TEXT PRIMARY KEY, channel_id TEXT, provider_id TEXT,
    title TEXT, link text, description TEXT, thumbnail BLOB, published DATETIME)'''.format(
        VIDEO_TABLE)
    cur.execute(create_query)
    con.commit()

    upsert_query: str = '''INSERT INTO {} (video_id, channel_id, provider_id, title, link, description, thumbnail, published, watchtime)
    VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(video_id) DO NOTHING'''.format(
        VIDEO_TABLE)

    cur.execute(upsert_query, video_id, channel_id, provider_id, title, link,
                description, thumbnail, published, watchtime)

    con.commit()


def get_default_logo(providerid: str = 'default') -> wx.Bitmap:
    if providerid == 'SVT':
        return wx.Bitmap('{}/assets/SVT.png'.format(MYPATH))
    if providerid == 'YouTube':
        return wx.Bitmap('{}/assets/YouTube.png'.format(MYPATH))
    else:
        return wx.Bitmap('{}/assets/Default.png'.format(MYPATH))


def get_subscriptions(basepath: str = BASEPATH,
                      filename: str = DB_FILE_NAME) -> list[tuple[str, str]]:
    subscriptions = list()
    fullpath = path.join(basepath, filename)
    con = sqlite3.connect(fullpath)
    cur = con.cursor()
    select_query = '''SELECT * FROM {}'''.format(SUB_TABLE)
    cur.execute(select_query)
    for result in cur.fetchall():
        subscriptions.append(result)
    return subscriptions

def get_videos(channel_id: str,
              basepath: str = BASEPATH,
              filename: str = DB_FILE_NAME) -> list[Item]:
    videos = list()
    fullpath = path.join(basepath, filename)
    con = sqlite3.connect(fullpath)
    cur = con.cursor()
    select_query = '''SELECT * FROM {} WHERE channel_id = ?'''.format(VIDEO_TABLE)
    cur.execute(select_query,channel_id)
    for result in cur.fetchall():
        pass
      #  videos.append(Item()) # Make an item from db
    return videos

def import_from_newpipe(filename) -> None:

    if path.isfile(filename):
        with open(filename, 'r') as subs:
            sub_data = json.loads(subs.read())

            for channel in sub_data['subscriptions']:
                if channel['service_id'] == 0:
                    channel_id = urlparse(channel['url']).path.split('/').pop()
                    add_subscription(channel_id, channel['name'])


def make_sized_button(parent_pnl: wx.Panel, bitmap_or_str: Union[wx.Bitmap,
                                                                 str],
                      text: str, callback: Callable) -> wx.BoxSizer:
    btn_sizer = wx.BoxSizer(wx.HORIZONTAL)
    if type(bitmap_or_str) == type(str):
        if bitmap_or_str.startswith('http'):  # type: ignore
            bitmap = make_bitmap_from_url(bitmap_or_str)  # type: ignore
        else:
            bitmap = wx.Bitmap(bitmap_or_str, wx.BITMAP_TYPE_ANY)
    else:
        bitmap = bitmap_or_str
    btn_style = wx.BORDER_NONE | wx.BU_AUTODRAW | wx.BU_EXACTFIT | wx.BU_NOTEXT
    btn_logo = wx.BitmapButton(parent_pnl, wx.ID_ANY, bitmap, style=btn_style)
    btn_logo.SetMinSize(SIZE)
    btn_logo.SetToolTip(text)
    btn_sizer.Add(btn_logo, 0, wx.BOTTOM | wx.EXPAND | wx.LEFT | wx.TOP, 1)

    btn_text = wx.Button(parent_pnl,
                         wx.ID_ANY,
                         text,
                         style=wx.BORDER_NONE | wx.BU_AUTODRAW)
    btn_text.SetMinSize(
        wx.Size(SCREEN_WIDTH - SIZE.GetWidth(), SIZE.GetHeight()))
    btn_text.SetToolTip(text)
    btn_sizer.Add(btn_text, 0, wx.BOTTOM | wx.RIGHT | wx.TOP | wx.EXPAND, 1)
    parent_pnl.Bind(wx.EVT_BUTTON, callback, btn_logo)
    parent_pnl.Bind(wx.EVT_BUTTON, callback, btn_text)

    return btn_sizer


def make_bitmap_from_url(logo_url: str, size: wx.Size = SIZE) -> wx.Bitmap:
    res = requests.get(logo_url)
    content = res.content
    content_bytes = io.BytesIO(content)
    image = wx.Image(content_bytes, type=wx.BITMAP_TYPE_ANY, index=-1)
    scale_factor = image.GetWidth() / SCREEN_WIDTH
    size.SetWidth(SCREEN_WIDTH)
    height = image.GetHeight()
    size.SetHeight(height / scale_factor)
    image.Rescale(size.GetWidth(), size.GetHeight())
    return wx.Bitmap(image)


def make_bitmap_from_file(path, size: wx.Size = SIZE) -> wx.Bitmap:
    image = wx.Image(path, type=wx.BITMAP_TYPE_ANY, index=-1)
    # scale_factor = image.GetWidth() / SCREEN_WIDTH
    # size.SetWidth(SCREEN_WIDTH)
    # height = image.GetHeight()
    # size.SetHeight(height/scale_factor)
    # image.Rescale(size.GetWidth(), size.GetHeight() )
    return wx.Bitmap(image)


def resolve_svt_channel(svt_id: str) -> dict:

    channels = {
        "ch-barnkanalen": {
            "name":
            "Barnkanalen",
            "thumbnail":
            make_bitmap_from_file('{}/assets/Barnkanalen.png'.format(MYPATH))
        },
        "ch-svt1": {
            "name": "SVT 1",
            "thumbnail":
            make_bitmap_from_file('{}/assets/SVT1.png'.format(MYPATH))
        },
        "ch-svt2": {
            "name": "SVT 2",
            "thumbnail":
            make_bitmap_from_file('{}/assets/SVT2.png'.format(MYPATH))
        },
        "ch-svt24": {
            "name":
            "SVT 24",
            "thumbnail":
            make_bitmap_from_file('{}/assets/SVT24.png'.format(MYPATH))
        },
        "ch-kunskapskanalen": {
            "name":
            "Kunskapskanalen",
            "thumbnail":
            make_bitmap_from_file(
                '{}/assets/Kunskapskanalen.png'.format(MYPATH))
        },
        "feed": {
            "name": "Senaste program",
            "thumbnail":
            make_bitmap_from_file('{}/assets/SVT.png'.format(MYPATH))
        },
    }

    return channels[svt_id]