Jellyfin/jellyfin-kodi
1
0
Fork 0
mirror of https://github.com/jellyfin/jellyfin-kodi.git synced 2024-11-13 21:56:11 +00:00
Code Issues Projects Releases Packages Wiki Activity
jellyfin-kodi/jellyfin_kodi/library.py

998 lines
32 KiB
Python
Raw Blame History

# -*- 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
Reference in a new issue View git blame Copy permalink