library.py

# -*- coding: utf-8 -*-
from __future__ import division, absolute_import, print_function, unicode_literals

##################################################################################################

import threading
from datetime import datetime, timedelta

import queue

import xbmc
import xbmcgui

from .objects import Movies, TVShows, MusicVideos, Music
from .objects.kodi import Movies as KodiDb
from .database import Database, jellyfin_db, get_sync, save_sync
from .full_sync import FullSync
from .views import Views
from .downloader import GetItemWorker
from .helper import translate, api, stop, settings, window, dialog, event, LazyLogger
from .helper.utils import split_list, set_screensaver, get_screensaver
from .helper.exceptions import LibraryException
from .jellyfin import Jellyfin

##################################################################################################

LOG = LazyLogger(__name__)
LIMIT = int(settings("limitIndex") or 15)
DTHREADS = int(settings("limitThreads") or 3)
TARGET_DB_VERSION = 1

##################################################################################################


class Library(threading.Thread):

    started = False
    stop_thread = False
    suspend = False
    pending_refresh = False
    screensaver = None
    progress_updates = None
    total_updates = 0

    def __init__(self, monitor):

        self.direct_path = settings("useDirectPaths") == "1"
        self.progress_display = int(settings("syncProgress") or 50)
        self.monitor = monitor
        self.player = monitor.monitor.player
        self.server = Jellyfin().get_client()
        self.updated_queue = queue.Queue()
        self.userdata_queue = queue.Queue()
        self.removed_queue = queue.Queue()
        self.updated_output = self.__new_queues__()
        self.userdata_output = self.__new_queues__()
        self.removed_output = self.__new_queues__()
        self.notify_output = queue.Queue()

        self.jellyfin_threads = []
        self.download_threads = []
        self.notify_threads = []
        self.writer_threads = {"updated": [], "userdata": [], "removed": []}
        self.database_lock = threading.Lock()
        self.music_database_lock = threading.Lock()

        threading.Thread.__init__(self)

    def __new_queues__(self):
        return {
            "Movie": queue.Queue(),
            "BoxSet": queue.Queue(),
            "MusicVideo": queue.Queue(),
            "Series": queue.Queue(),
            "Season": queue.Queue(),
            "Episode": queue.Queue(),
            "MusicAlbum": queue.Queue(),
            "MusicArtist": queue.Queue(),
            "AlbumArtist": queue.Queue(),
            "Audio": queue.Queue(),
        }

    def run(self):

        LOG.info("--->[ library ]")

        if not self.startup():
            self.stop_client()

        window("jellyfin_startup.bool", True)

        while not self.stop_thread:

            try:
                self.service()
            except LibraryException:
                break
            except Exception as error:
                LOG.exception(error)

                break

            if self.monitor.waitForAbort(2):
                break

        LOG.info("---<[ library ]")

    def test_databases(self):
        """Open the databases to test if the file exists."""
        with Database("video"), Database("music"):
            pass

    def check_version(self):
        """
        Checks database version and triggers any required data migrations
        """
        with Database("jellyfin") as jellyfindb:
            db = jellyfin_db.JellyfinDatabase(jellyfindb.cursor)
            db_version = db.get_version()

            if not db_version:
                # Make sure we always have a version in the database
                db.add_version((TARGET_DB_VERSION))

        # Video Database Migrations
        with Database("video") as videodb:
            vid_db = KodiDb(videodb.cursor)
            if vid_db.migrations():
                LOG.info("changes detected, reloading skin")
                xbmc.executebuiltin("UpdateLibrary(video)")
                xbmc.executebuiltin("ReloadSkin()")

    @stop
    def service(self):
        """If error is encountered, it will rerun this function.
        Start new "daemon threads" to process library updates.
        (actual daemon thread is not supported in Kodi)
        """
        self.download_threads = [
            thread for thread in self.download_threads if not thread.is_done
        ]
        self.writer_threads["updated"] = [
            thread for thread in self.writer_threads["updated"] if not thread.is_done
        ]
        self.writer_threads["userdata"] = [
            thread for thread in self.writer_threads["userdata"] if not thread.is_done
        ]
        self.writer_threads["removed"] = [
            thread for thread in self.writer_threads["removed"] if not thread.is_done
        ]

        if (
            not self.player.isPlayingVideo()
            or settings("syncDuringPlay.bool")
            or xbmc.getCondVisibility("VideoPlayer.Content(livetv)")
        ):

            self.worker_downloads()
            self.worker_sort()

            self.worker_updates()
            self.worker_userdata()
            self.worker_remove()
            self.worker_notify()

        if self.pending_refresh:
            window("jellyfin_sync.bool", True)

            if self.total_updates > self.progress_display:
                queue_size = self.worker_queue_size()

                if self.progress_updates is None:

                    self.progress_updates = xbmcgui.DialogProgressBG()
                    self.progress_updates.create(
                        translate("addon_name"), translate(33178)
                    )
                    self.progress_updates.update(
                        int(
                            (
                                float(self.total_updates - queue_size)
                                / float(self.total_updates)
                            )
                            * 100
                        ),
                        message="%s: %s" % (translate(33178), queue_size),
                    )
                elif queue_size:
                    self.progress_updates.update(
                        int(
                            (
                                float(self.total_updates - queue_size)
                                / float(self.total_updates)
                            )
                            * 100
                        ),
                        message="%s: %s" % (translate(33178), queue_size),
                    )
                else:
                    self.progress_updates.update(
                        int(
                            (
                                float(self.total_updates - queue_size)
                                / float(self.total_updates)
                            )
                            * 100
                        ),
                        message=translate(33178),
                    )

            if not settings("dbSyncScreensaver.bool") and self.screensaver is None:

                xbmc.executebuiltin("InhibitIdleShutdown(true)")
                self.screensaver = get_screensaver()
                set_screensaver(value="")

        if (
            self.pending_refresh
            and not self.download_threads
            and not self.writer_threads["updated"]
            and not self.writer_threads["userdata"]
            and not self.writer_threads["removed"]
        ):
            self.pending_refresh = False
            self.save_last_sync()
            self.total_updates = 0
            window("jellyfin_sync", clear=True)

            if self.progress_updates:

                self.progress_updates.close()
                self.progress_updates = None

            if not settings("dbSyncScreensaver.bool") and self.screensaver is not None:

                xbmc.executebuiltin("InhibitIdleShutdown(false)")
                set_screensaver(value=self.screensaver)
                self.screensaver = None

            if xbmc.getCondVisibility(
                "Container.Content(musicvideos)"
            ):  # Prevent cursor from moving
                xbmc.executebuiltin("Container.Refresh")
            else:  # Update widgets
                xbmc.executebuiltin("UpdateLibrary(video)")

                if xbmc.getCondVisibility("Window.IsMedia"):
                    xbmc.executebuiltin("Container.Refresh")

    def stop_client(self):
        self.stop_thread = True

    def enable_pending_refresh(self):
        """When there's an active thread. Let the main thread know."""
        self.pending_refresh = True
        window("jellyfin_sync.bool", True)

    def worker_queue_size(self):
        """Get how many items are queued up for worker threads."""
        total = 0

        for queues in self.updated_output:
            total += self.updated_output[queues].qsize()

        for queues in self.userdata_output:
            total += self.userdata_output[queues].qsize()

        for queues in self.removed_output:
            total += self.removed_output[queues].qsize()

        return total

    def worker_downloads(self):
        """Get items from jellyfin and place them in the appropriate queues."""
        for work_queue in (
            (self.updated_queue, self.updated_output),
            (self.userdata_queue, self.userdata_output),
        ):
            if work_queue[0].qsize() and len(self.download_threads) < DTHREADS:

                new_thread = GetItemWorker(self.server, work_queue[0], work_queue[1])
                new_thread.start()
                LOG.info("-->[ q:download/%s ]", id(new_thread))
                self.download_threads.append(new_thread)

    def worker_sort(self):
        """Get items based on the local jellyfin database and place item in appropriate queues."""
        if self.removed_queue.qsize() and len(self.jellyfin_threads) < 2:

            new_thread = SortWorker(self.removed_queue, self.removed_output)
            new_thread.start()
            LOG.info("-->[ q:sort/%s ]", id(new_thread))

    def worker_updates(self):
        """Update items in the Kodi database."""
        for queues in self.updated_output:
            queue = self.updated_output[queues]

            if queue.qsize() and not len(self.writer_threads["updated"]):

                if queues in ("Audio", "MusicArtist", "AlbumArtist", "MusicAlbum"):
                    new_thread = UpdateWorker(
                        queue,
                        self.notify_output,
                        self.music_database_lock,
                        "music",
                        self.server,
                        self.direct_path,
                    )
                else:
                    new_thread = UpdateWorker(
                        queue,
                        self.notify_output,
                        self.database_lock,
                        "video",
                        self.server,
                        self.direct_path,
                    )

                new_thread.start()
                LOG.info("-->[ q:updated/%s/%s ]", queues, id(new_thread))
                self.writer_threads["updated"].append(new_thread)
                self.enable_pending_refresh()

    def worker_userdata(self):
        """Update userdata in the Kodi database."""
        for queues in self.userdata_output:
            queue = self.userdata_output[queues]

            if queue.qsize() and not len(self.writer_threads["userdata"]):

                if queues in ("Audio", "MusicArtist", "AlbumArtist", "MusicAlbum"):
                    new_thread = UserDataWorker(
                        queue,
                        self.music_database_lock,
                        "music",
                        self.server,
                        self.direct_path,
                    )
                else:
                    new_thread = UserDataWorker(
                        queue,
                        self.database_lock,
                        "video",
                        self.server,
                        self.direct_path,
                    )

                new_thread.start()
                LOG.info("-->[ q:userdata/%s/%s ]", queues, id(new_thread))
                self.writer_threads["userdata"].append(new_thread)
                self.enable_pending_refresh()

    def worker_remove(self):
        """Remove items from the Kodi database."""
        for queues in self.removed_output:
            queue = self.removed_output[queues]

            if queue.qsize() and not len(self.writer_threads["removed"]):

                if queues in ("Audio", "MusicArtist", "AlbumArtist", "MusicAlbum"):
                    new_thread = RemovedWorker(
                        queue,
                        self.music_database_lock,
                        "music",
                        self.server,
                        self.direct_path,
                    )
                else:
                    new_thread = RemovedWorker(
                        queue,
                        self.database_lock,
                        "video",
                        self.server,
                        self.direct_path,
                    )

                new_thread.start()
                LOG.info("-->[ q:removed/%s/%s ]", queues, id(new_thread))
                self.writer_threads["removed"].append(new_thread)
                self.enable_pending_refresh()

    def worker_notify(self):
        """Notify the user of new additions."""
        if self.notify_output.qsize() and not len(self.notify_threads):

            new_thread = NotifyWorker(self.notify_output, self.player)
            new_thread.start()
            LOG.info("-->[ q:notify/%s ]", id(new_thread))
            self.notify_threads.append(new_thread)

    def startup(self):
        """Run at startup.
        Check databases.
        Check for the server plugin.
        """
        self.test_databases()
        self.check_version()

        Views().get_views()
        Views().get_nodes()

        try:
            if get_sync()["Libraries"]:

                try:
                    with FullSync(self, self.server) as sync:
                        sync.libraries()

                    Views().get_nodes()
                except Exception as error:
                    LOG.exception(error)

            elif not settings("SyncInstallRunDone.bool"):

                with FullSync(self, self.server) as sync:
                    sync.libraries()

                Views().get_nodes()

                return True

            if settings("SyncInstallRunDone.bool") and settings("kodiCompanion.bool"):
                # None == Unknown
                if self.server.jellyfin.check_companion_enabled() is not False:

                    if not self.fast_sync():
                        dialog("ok", "{jellyfin}", translate(33128))

                        raise Exception("Failed to retrieve latest updates")

                    LOG.info("--<[ retrieve changes ]")

                # is False
                else:
                    dialog("ok", "{jellyfin}", translate(33099))
                    settings("kodiCompanion.bool", False)
                    return True

            return True
        except LibraryException as error:
            LOG.error(error.status)

            if error.status in "SyncLibraryLater":

                dialog("ok", "{jellyfin}", translate(33129))
                settings("SyncInstallRunDone.bool", True)
                sync = get_sync()
                sync["Libraries"] = []
                save_sync(sync)

                return True

        except Exception as error:
            LOG.exception(error)

        return False

    def fast_sync(self):
        """Movie and userdata not provided by server yet."""
        last_sync = settings("LastIncrementalSync")
        include = []
        filters = ["tvshows", "boxsets", "musicvideos", "music", "movies"]
        sync = get_sync()
        whitelist = [x.replace("Mixed:", "") for x in sync["Whitelist"]]
        LOG.info("--[ retrieve changes ] %s", last_sync)

        # Get the item type of each synced library and build list of types to request
        for item_id in whitelist:
            library = self.server.jellyfin.get_item(item_id)
            library_type = library.get("CollectionType")
            if library_type in filters:
                include.append(library_type)

        # Include boxsets if movies are synced
        if "movies" in include:
            include.append("boxsets")

        # Filter down to the list of library types we want to exclude
        query_filter = list(set(filters) - set(include))

        try:
            # Get list of updates from server for synced library types and populate work queues
            result = self.server.jellyfin.get_sync_queue(
                last_sync, ",".join([x for x in query_filter])
            )

            if result is None:
                return True

            updated = []
            userdata = []
            removed = []

            updated.extend(result["ItemsAdded"])
            updated.extend(result["ItemsUpdated"])
            userdata.extend(result["UserDataChanged"])
            removed.extend(result["ItemsRemoved"])

            total = len(updated) + len(userdata)

            if total > int(settings("syncIndicator") or 99):

                """Inverse yes no, in case the dialog is forced closed by Kodi."""
                if dialog(
                    "yesno",
                    "{jellyfin}",
                    translate(33172).replace("{number}", str(total)),
                    nolabel=translate(107),
                    yeslabel=translate(106),
                ):
                    LOG.warning("Large updates skipped.")

                    return True

            self.updated(updated)
            self.userdata(userdata)
            self.removed(removed)

        except Exception as error:
            LOG.exception(error)

            return False

        return True

    def save_last_sync(self):

        try:
            time_now = datetime.strptime(
                self.server.config.data["server-time"].split(", ", 1)[1],
                "%d %b %Y %H:%M:%S GMT",
            ) - timedelta(minutes=2)
        except Exception as error:

            LOG.exception(error)
            time_now = datetime.utcnow() - timedelta(minutes=2)

        last_sync = time_now.strftime("%Y-%m-%dT%H:%M:%Sz")
        settings("LastIncrementalSync", value=last_sync)
        LOG.info("--[ sync/%s ]", last_sync)

    def select_libraries(self, mode=None):
        """Select from libraries synced. Either update or repair libraries.
        Send event back to service.py
        """
        modes = {
            "SyncLibrarySelection": "SyncLibrary",
            "RepairLibrarySelection": "RepairLibrary",
            "AddLibrarySelection": "SyncLibrary",
            "RemoveLibrarySelection": "RemoveLibrary",
        }
        sync = get_sync()
        whitelist = [x.replace("Mixed:", "") for x in sync["Whitelist"]]
        libraries = []

        with Database("jellyfin") as jellyfindb:
            db = jellyfin_db.JellyfinDatabase(jellyfindb.cursor)

            if mode in (
                "SyncLibrarySelection",
                "RepairLibrarySelection",
                "RemoveLibrarySelection",
            ):
                for library in sync["Whitelist"]:

                    name = db.get_view_name(library.replace("Mixed:", ""))
                    libraries.append({"Id": library, "Name": name})
            else:
                available = [x for x in sync["SortedViews"] if x not in whitelist]

                for library in available:
                    view = db.get_view(library)

                    if view.media_type in (
                        "movies",
                        "tvshows",
                        "musicvideos",
                        "mixed",
                        "music",
                    ):
                        libraries.append({"Id": view.view_id, "Name": view.view_name})

        choices = [x["Name"] for x in libraries]
        choices.insert(0, translate(33121))

        titles = {
            "RepairLibrarySelection": 33199,
            "SyncLibrarySelection": 33198,
            "RemoveLibrarySelection": 33200,
            "AddLibrarySelection": 33120,
        }
        title = titles.get(mode, "Failed to get title {}".format(mode))

        selection = dialog("multi", translate(title), choices)

        if selection is None:
            return

        if 0 in selection:
            selection = list(range(1, len(libraries) + 1))

        selected_libraries = []

        for x in selection:

            library = libraries[x - 1]
            selected_libraries.append(library["Id"])

        event(
            modes[mode],
            {
                "Id": ",".join([libraries[x - 1]["Id"] for x in selection]),
                "Update": mode == "SyncLibrarySelection",
            },
        )

    def add_library(self, library_id, update=False):

        try:
            with FullSync(self, server=self.server) as sync:
                sync.libraries(library_id, update)
        except Exception as error:
            LOG.exception(error)

            return False

        Views().get_nodes()

        return True

    def remove_library(self, library_id):

        try:
            with FullSync(self, self.server) as sync:
                sync.remove_library(library_id)

            Views().remove_library(library_id)
        except Exception as error:
            LOG.exception(error)

            return False

        Views().get_views()
        Views().get_nodes()

        return True

    def userdata(self, data):
        """Add item_id to userdata queue."""
        if not data:
            return

        items = [x["ItemId"] for x in data]

        for item in split_list(items, LIMIT):
            self.userdata_queue.put(item)

        self.total_updates += len(items)
        LOG.info("---[ userdata:%s ]", len(items))

    def updated(self, data):
        """Add item_id to updated queue."""
        if not data:
            return

        for item in split_list(data, LIMIT):
            self.updated_queue.put(item)

        self.total_updates += len(data)
        LOG.info("---[ updated:%s ]", len(data))

    def removed(self, data):
        """Add item_id to removed queue."""
        if not data:
            return

        for item in data:

            if item in list(self.removed_queue.queue):
                continue

            self.removed_queue.put(item)

        self.total_updates += len(data)
        LOG.info("---[ removed:%s ]", len(data))


class UpdateWorker(threading.Thread):

    is_done = False

    def __init__(
        self, queue, notify, lock, database, server=None, direct_path=None, *args
    ):
        self.queue = queue
        self.notify_output = notify
        self.notify = settings("newContent.bool")
        self.lock = lock
        self.database = Database(database)
        self.args = args
        self.server = server
        self.direct_path = direct_path
        threading.Thread.__init__(self)

    def run(self):
        with self.lock, self.database as kodidb, Database("jellyfin") as jellyfindb:
            default_args = (self.server, jellyfindb, kodidb, self.direct_path)
            if kodidb.db_file == "video":
                movies = Movies(*default_args)
                tvshows = TVShows(*default_args)
                musicvideos = MusicVideos(*default_args)
            elif kodidb.db_file == "music":
                music = Music(*default_args)
            else:
                # this should not happen
                LOG.error(
                    '"{}" is not a valid Kodi library type.'.format(kodidb.db_file)
                )
                return

            while True:

                try:
                    item = self.queue.get(timeout=1)
                except queue.Empty:
                    break

                try:
                    LOG.debug("{} - {}".format(item["Type"], item["Name"]))
                    if item["Type"] == "Movie":
                        movies.movie(item)
                    elif item["Type"] == "BoxSet":
                        movies.boxset(item)
                    elif item["Type"] == "Series":
                        tvshows.tvshow(item)
                    elif item["Type"] == "Season":
                        tvshows.season(item)
                    elif item["Type"] == "Episode":
                        tvshows.episode(item)
                    elif item["Type"] == "MusicVideo":
                        musicvideos.musicvideo(item)
                    elif item["Type"] == "MusicAlbum":
                        music.album(item)
                    elif item["Type"] == "MusicArtist":
                        music.artist(item)
                    elif item["Type"] == "AlbumArtist":
                        music.albumartist(item)
                    elif item["Type"] == "Audio":
                        music.song(item)

                    if self.notify:
                        self.notify_output.put(
                            (item["Type"], api.API(item).get_naming())
                        )
                except LibraryException as error:
                    if error.status == "StopCalled":
                        break
                except Exception as error:
                    LOG.exception(error)

                self.queue.task_done()

                if window("jellyfin_should_stop.bool"):
                    break

        LOG.info("--<[ q:updated/%s ]", id(self))
        self.is_done = True


class UserDataWorker(threading.Thread):

    is_done = False

    def __init__(self, queue, lock, database, server, direct_path):

        self.queue = queue
        self.lock = lock
        self.database = Database(database)
        self.server = server
        self.direct_path = direct_path

        threading.Thread.__init__(self)

    def run(self):

        with self.lock, self.database as kodidb, Database("jellyfin") as jellyfindb:
            default_args = (self.server, jellyfindb, kodidb, self.direct_path)
            if kodidb.db_file == "video":
                movies = Movies(*default_args)
                tvshows = TVShows(*default_args)
            elif kodidb.db_file == "music":
                music = Music(*default_args)
            else:
                # this should not happen
                LOG.error(
                    '"{}" is not a valid Kodi library type.'.format(kodidb.db_file)
                )
                return

            while True:

                try:
                    item = self.queue.get(timeout=1)
                except queue.Empty:
                    break

                try:
                    if item["Type"] == "Movie":
                        movies.userdata(item)
                    elif item["Type"] in ["Series", "Season", "Episode"]:
                        tvshows.userdata(item)
                    elif item["Type"] == "MusicAlbum":
                        music.album(item)
                    elif item["Type"] == "MusicArtist":
                        music.artist(item)
                    elif item["Type"] == "AlbumArtist":
                        music.albumartist(item)
                    elif item["Type"] == "Audio":
                        music.userdata(item)
                except LibraryException as error:
                    if error.status == "StopCalled":
                        break
                except Exception as error:
                    LOG.exception(error)

                self.queue.task_done()

                if window("jellyfin_should_stop.bool"):
                    break

        LOG.info("--<[ q:userdata/%s ]", id(self))
        self.is_done = True


class SortWorker(threading.Thread):

    is_done = False

    def __init__(self, queue, output, *args):

        self.queue = queue
        self.output = output
        self.args = args
        threading.Thread.__init__(self)

    def run(self):

        with Database("jellyfin") as jellyfindb:
            database = jellyfin_db.JellyfinDatabase(jellyfindb.cursor)

            while True:

                try:
                    item_id = self.queue.get(timeout=1)
                except queue.Empty:
                    break

                try:
                    media = database.get_media_by_id(item_id)
                    if media:
                        self.output[media].put({"Id": item_id, "Type": media})
                    else:
                        items = database.get_media_by_parent_id(item_id)

                        if not items:
                            LOG.info(
                                "Could not find media %s in the jellyfin database.",
                                item_id,
                            )
                        else:
                            for item in items:
                                self.output[item[1]].put(
                                    {"Id": item[0], "Type": item[1]}
                                )
                except Exception as error:
                    LOG.exception(error)

                self.queue.task_done()

                if window("jellyfin_should_stop.bool"):
                    break

        LOG.info("--<[ q:sort/%s ]", id(self))
        self.is_done = True


class RemovedWorker(threading.Thread):

    is_done = False

    def __init__(self, queue, lock, database, server, direct_path):

        self.queue = queue
        self.lock = lock
        self.database = Database(database)
        self.server = server
        self.direct_path = direct_path
        threading.Thread.__init__(self)

    def run(self):

        with self.lock, self.database as kodidb, Database("jellyfin") as jellyfindb:
            default_args = (self.server, jellyfindb, kodidb, self.direct_path)
            if kodidb.db_file == "video":
                movies = Movies(*default_args)
                tvshows = TVShows(*default_args)
                musicvideos = MusicVideos(*default_args)
            elif kodidb.db_file == "music":
                music = Music(*default_args)
            else:
                # this should not happen
                LOG.error(
                    '"{}" is not a valid Kodi library type.'.format(kodidb.db_file)
                )
                return

            while True:

                try:
                    item = self.queue.get(timeout=1)
                except queue.Empty:
                    break

                if item["Type"] == "Movie":
                    obj = movies.remove
                elif item["Type"] in ["Series", "Season", "Episode"]:
                    obj = tvshows.remove
                elif item["Type"] in [
                    "MusicAlbum",
                    "MusicArtist",
                    "AlbumArtist",
                    "Audio",
                ]:
                    obj = music.remove
                elif item["Type"] == "MusicVideo":
                    obj = musicvideos.remove

                try:
                    obj(item["Id"])
                except LibraryException as error:
                    if error.status == "StopCalled":
                        break
                except Exception as error:
                    LOG.exception(error)
                finally:
                    self.queue.task_done()

                if window("jellyfin_should_stop.bool"):
                    break

        LOG.info("--<[ q:removed/%s ]", id(self))
        self.is_done = True


class NotifyWorker(threading.Thread):

    is_done = False

    def __init__(self, queue, player):

        self.queue = queue
        self.video_time = int(settings("newvideotime")) * 1000
        self.music_time = int(settings("newmusictime")) * 1000
        self.player = player
        threading.Thread.__init__(self)

    def run(self):

        while True:

            try:
                item = self.queue.get(timeout=3)
            except queue.Empty:
                break

            time = self.music_time if item[0] == "Audio" else self.video_time

            if time and (
                not self.player.isPlayingVideo()
                or xbmc.getCondVisibility("VideoPlayer.Content(livetv)")
            ):
                dialog(
                    "notification",
                    heading="%s %s" % (translate(33049), item[0]),
                    message=item[1],
                    icon="{jellyfin}",
                    time=time,
                    sound=False,
                )

            self.queue.task_done()

            if window("jellyfin_should_stop.bool"):
                break

        LOG.info("--<[ q:notify/%s ]", id(self))
        self.is_done = True