jellyfin-kodi/jellyfin_kodi/library.py
Chuddah 2bf2206a1e Move the responsibility of the maximum number of threads to the UI.
Previously the user was able to set any number to limitThreads UI
component but internally that was reduced to be a maximum of 50 which is
deceiving to the user. Set this is set as a boundary in the UI.

This also fixes a bug that the user could set the number of threads to
zero which follows from GIGO; but let the UI assist by placing a min
number of threads to 1.
2020-02-17 11:59:48 +00:00

840 lines
28 KiB
Python

# -*- coding: utf-8 -*-
from __future__ import division, absolute_import, print_function, unicode_literals
##################################################################################################
import logging
import threading
from datetime import datetime, timedelta
from six.moves import queue as Queue
from kodi_six import xbmc, xbmcgui
from objects import Movies, TVShows, MusicVideos, Music
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, LibraryException
from helper.utils import split_list, set_screensaver, get_screensaver
from jellyfin import Jellyfin
##################################################################################################
LOG = logging.getLogger("JELLYFIN." + __name__)
LIMIT = int(settings('limitIndex') or 15)
DTHREADS = int(settings('limitThreads') or 3)
MEDIA = {
'Movie': Movies,
'BoxSet': Movies,
'MusicVideo': MusicVideos,
'Series': TVShows,
'Season': TVShows,
'Episode': TVShows,
'MusicAlbum': Music,
'MusicArtist': Music,
'AlbumArtist': Music,
'Audio': Music
}
##################################################################################################
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.media = {'Movies': Movies, 'TVShows': TVShows, 'MusicVideos': MusicVideos, 'Music': Music}
self.MEDIA = MEDIA
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 as error:
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
@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)
'''
for threads in (self.download_threads, self.writer_threads['updated'],
self.writer_threads['userdata'], self.writer_threads['removed']):
for thread in threads:
if thread.is_done:
threads.remove(thread)
if not self.player.isPlayingVideo() or settings('syncDuringPlay.bool') or xbmc.getCondVisibility('VideoPlayer.Content(livetv)'):
while self.worker_downloads():
pass
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.
'''
added_threads = False
for queue in ((self.updated_queue, self.updated_output), (self.userdata_queue, self.userdata_output)):
if queue[0].qsize() and len(self.download_threads) < DTHREADS:
new_thread = GetItemWorker(self.server, queue[0], queue[1])
new_thread.start()
LOG.info("-->[ q:download/%s ]", id(new_thread))
self.download_threads.append(new_thread)
added_threads = True
return added_threads
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 = UpdatedWorker(queue, self.notify_output, self.music_database_lock, "music", self.server, self.direct_path)
else:
new_thread = UpdatedWorker(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()
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'):
if settings('kodiCompanion.bool'):
if self.server.jellyfin.check_companion_installed():
if not self.fast_sync():
dialog("ok", heading="{jellyfin}", line1=translate(33128))
raise Exception("Failed to retrieve latest updates")
LOG.info("--<[ retrieve changes ]")
else:
raise LibraryException('CompanionMissing')
return True
except LibraryException as error:
LOG.error(error.status)
if error.status in 'SyncLibraryLater':
dialog("ok", heading="{jellyfin}", line1=translate(33129))
settings('SyncInstallRunDone.bool', True)
sync = get_sync()
sync['Libraries'] = []
save_sync(sync)
return True
elif error.status == 'CompanionMissing':
dialog("ok", heading="{jellyfin}", line1=translate(33099))
settings('kodiCompanion.bool', False)
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')
filters = ["tvshows", "boxsets", "musicvideos", "music", "movies"]
sync = get_sync()
LOG.info("--[ retrieve changes ] %s", last_sync)
try:
updated = []
userdata = []
removed = []
for media in filters:
result = self.server.jellyfin.get_sync_queue(last_sync, ",".join([x for x in filters if x != media]))
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", heading="{jellyfin}", line1=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)
"""
result = self.server.jellyfin.get_sync_queue(last_sync)
self.userdata(result['UserDataChanged'])
self.removed(result['ItemsRemoved'])
filters.extend(["tvshows", "boxsets", "musicvideos", "music"])
# Get only movies.
result = self.server.jellyfin.get_sync_queue(last_sync, ",".join(filters))
self.updated(result['ItemsAdded'])
self.updated(result['ItemsUpdated'])
self.userdata(result['UserDataChanged'])
self.removed(result['ItemsRemoved'])
"""
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:
name, media = db.get_view(library)
if media in ('movies', 'tvshows', 'musicvideos', 'mixed', 'music'):
libraries.append({'Id': library, 'Name': 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 UpdatedWorker(threading.Thread):
is_done = False
def __init__(self, queue, notify, lock, database, *args):
self.queue = queue
self.notify_output = notify
self.notify = settings('newContent.bool')
self.lock = lock
self.database = Database(database)
self.args = args
threading.Thread.__init__(self)
def run(self):
with self.lock, self.database as kodidb, Database('jellyfin') as jellyfindb:
while True:
try:
item = self.queue.get(timeout=1)
except Queue.Empty:
break
if item['Type'] == 'Movie':
obj = Movies(self.args[0], jellyfindb, kodidb, self.args[1]).movie
elif item['Type'] == 'Boxset':
obj = Movies(self.args[0], jellyfindb, kodidb, self.args[1]).boxset
elif item['Type'] == 'Series':
obj = TVShows(self.args[0], jellyfindb, kodidb, self.args[1]).tvshow
elif item['Type'] == 'Season':
obj = TVShows(self.args[0], jellyfindb, kodidb, self.args[1]).season
elif item['Type'] == 'Episode':
obj = TVShows(self.args[0], jellyfindb, kodidb, self.args[1]).episode
elif item['Type'] == 'MusicVideo':
obj = MusicVideos(self.args[0], jellyfindb, kodidb, self.args[1]).musicvideo
elif item['Type'] == 'MusicAlbum':
obj = Music(self.args[0], jellyfindb, kodidb, self.args[1]).album
elif item['Type'] == 'MusicArtist':
obj = Music(self.args[0], jellyfindb, kodidb, self.args[1]).artist
elif item['Type'] == 'AlbumArtist':
obj = Music(self.args[0], jellyfindb, kodidb, self.args[1]).albumartist
elif item['Type'] == 'Audio':
obj = Music(self.args[0], jellyfindb, kodidb, self.args[1]).song
try:
if obj(item) and 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, *args):
self.queue = queue
self.lock = lock
self.database = Database(database)
self.args = args
threading.Thread.__init__(self)
def run(self):
with self.lock, self.database as kodidb, Database('jellyfin') as jellyfindb:
while True:
try:
item = self.queue.get(timeout=1)
except Queue.Empty:
break
if item['Type'] == 'Movie':
obj = Movies(self.args[0], jellyfindb, kodidb, self.args[1]).userdata(item)
elif item['Type'] in ['Series', 'Season', 'Episode']:
obj = TVShows(self.args[0], jellyfindb, kodidb, self.args[1]).userdata(item)
elif item['Type'] == 'MusicAlbum':
obj = Music(self.args[0], jellyfindb, kodidb, self.args[1]).album
elif item['Type'] == 'MusicArtist':
obj = Music(self.args[0], jellyfindb, kodidb, self.args[1]).artist
elif item['Type'] == 'AlbumArtist':
obj = Music(self.args[0], jellyfindb, kodidb, self.args[1]).albumartist
elif item['Type'] == 'Audio':
obj = Music(self.args[0], jellyfindb, kodidb, self.args[1]).song
try:
obj(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)
self.output[media].put({'Id': item_id, 'Type': media})
except Exception as error:
LOG.exception(error)
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]})
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, *args):
self.queue = queue
self.lock = lock
self.database = Database(database)
self.args = args
threading.Thread.__init__(self)
def run(self):
with self.lock, self.database as kodidb, Database('jellyfin') as jellyfindb:
while True:
try:
item = self.queue.get(timeout=1)
except Queue.Empty:
break
if item['Type'] == 'Movie':
obj = Movies(self.args[0], jellyfindb, kodidb, self.args[1]).remove
elif item['Type'] in ['Series', 'Season', 'Episode']:
obj = TVShows(self.args[0], jellyfindb, kodidb, self.args[1]).remove
elif item['Type'] in ['MusicAlbum', 'MusicArtist', 'AlbumArtist', 'Audio']:
obj = Music(self.args[0], jellyfindb, kodidb, self.args[1]).remove
elif item['Type'] == 'MusicVideo':
obj = MusicVideos(self.args[0], jellyfindb, kodidb, self.args[1]).remove
try:
obj(item['Id'])
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: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