PlexKodiConnect/resources/lib/library_sync/full_sync.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import absolute_import, division, unicode_literals
from logging import getLogger
import xbmc

from .get_metadata import GetMetadataTask, reset_collections
from .process_metadata import InitNewSection, UpdateLastSync, ProcessMetadata
from . import common, sections
from .. import utils, backgroundthread, variables as v, state
from .. import plex_functions as PF, itemtypes
from ..plex_db import PlexDB

if (v.PLATFORM != 'Microsoft UWP' and
        utils.settings('enablePlaylistSync') == 'true'):
    # Xbox cannot use watchdog, a dependency for PKC playlist features
    from .. import playlists
    PLAYLIST_SYNC_ENABLED = True
else:
    PLAYLIST_SYNC_ENABLED = False

LOG = getLogger('PLEX.sync.full_sync')


class FullSync(common.libsync_mixin):
    def __init__(self, repair, callback, show_dialog):
        """
        repair=True: force sync EVERY item
        """
        self._canceled = False
        self.repair = repair
        self.callback = callback
        self.show_dialog = show_dialog
        self.queue = None
        self.process_thread = None
        self.current_sync = None
        self.plexdb = None
        self.plex_type = None
        self.section_type = None
        self.processing_thread = None
        self.install_sync_done = utils.settings('SyncInstallRunDone') == 'true'
        self.threader = backgroundthread.ThreaderManager(
            worker=backgroundthread.NonstoppingBackgroundWorker)
        super(FullSync, self).__init__()

    def process_item(self, xml_item):
        """
        Processes a single library item
        """
        plex_id = int(xml_item.get('ratingKey'))
        if not self.repair and self.plexdb.checksum(plex_id, self.plex_type) == \
                int('%s%s' % (plex_id,
                              xml_item.get('updatedAt',
                                           xml_item.get('addedAt', 1541572987)))):
            # Already got EXACTLY this item in our DB. BUT need to collect all
            # DB updates within the same thread
            self.queue.put(UpdateLastSync(plex_id))
            return
        task = GetMetadataTask()
        task.setup(self.queue, plex_id, self.plex_type, self.get_children)
        self.threader.addTask(task)

    def process_delete(self):
        """
        Removes all the items that have NOT been updated (last_sync timestamp
        is different)
        """
        with self.context(self.current_sync) as c:
            for plex_id in self.plexdb.plex_id_by_last_sync(self.plex_type,
                                                            self.current_sync):
                if self.isCanceled():
                    return
                c.remove(plex_id, plex_type=self.plex_type)

    @utils.log_time
    def process_playstate(self, iterator):
        """
        Updates the playstate (resume point, number of views, userrating, last
        played date, etc.) for all elements in the (xml-)iterator
        """
        with self.context(self.current_sync) as c:
            for xml_item in iterator:
                if self.isCanceled():
                    return False
                c.update_userdata(xml_item, self.plex_type)

    @utils.log_time
    def process_kind(self):
        """
        """
        successful = True
        LOG.debug('Start processing %ss', self.section_type)
        sects = (x for x in sections.SECTIONS
                 if x['plex_type'] == self.section_type)
        for section in sects:
            LOG.debug('Processing library section %s', section)
            if self.isCanceled():
                return False
            if not self.install_sync_done:
                state.PATH_VERIFIED = False
            try:
                # Sync new, updated and deleted items
                iterator = PF.SectionItems(section['section_id'],
                                           plex_type=self.plex_type)
                # Tell the processing thread about this new section
                queue_info = InitNewSection(self.context,
                                            iterator.total,
                                            iterator.get('librarySectionTitle'),
                                            section['section_id'],
                                            self.plex_type)
                self.queue.put(queue_info)
                with PlexDB() as self.plexdb:
                    for xml_item in iterator:
                        if self.isCanceled():
                            return False
                        self.process_item(xml_item)
            except RuntimeError:
                LOG.error('Could not entirely process section %s', section)
                successful = False
                continue
            LOG.debug('Waiting for download threads to finish')
            while self.threader.threader.working():
                xbmc.sleep(100)
            LOG.debug('Waiting for processing thread to finish section')
            self.queue.join()
            reset_collections()
            try:
                # Sync playstate of every item
                iterator = PF.SectionItems(section['section_id'],
                                           plex_type=self.plex_type)
                # Tell the processing thread that we're syncing playstate
                queue_info = InitNewSection(self.context,
                                            iterator.total,
                                            iterator.get('librarySectionTitle'),
                                            section['section_id'],
                                            self.plex_type)
                self.queue.put(queue_info)
                # Ensure that the DB connection is closed to commit the
                # changes above - avoids "Item not yet synced" error
                self.queue.join()
                self.process_playstate(iterator)
            except RuntimeError:
                LOG.error('Could not process playstate for section %s', section)
                successful = False
                continue
            LOG.debug('Done processing playstate for section')

        LOG.debug('Finished processing %ss', self.plex_type)
        return successful

    def full_library_sync(self):
        """
        """
        kinds = [
            (v.PLEX_TYPE_MOVIE, v.PLEX_TYPE_MOVIE, itemtypes.Movie, False),
            (v.PLEX_TYPE_SHOW, v.PLEX_TYPE_SHOW, itemtypes.Show, False),
            (v.PLEX_TYPE_SEASON, v.PLEX_TYPE_SHOW, itemtypes.Season, False),
            (v.PLEX_TYPE_EPISODE, v.PLEX_TYPE_SHOW, itemtypes.Episode, False)
        ]
        if state.ENABLE_MUSIC:
            kinds.extend([
                (v.PLEX_TYPE_ARTIST, v.PLEX_TYPE_ARTIST, itemtypes.Artist, False),
                (v.PLEX_TYPE_ALBUM, v.PLEX_TYPE_ARTIST, itemtypes.Album, True),
            ])
        with PlexDB() as self.plexdb:
            for kind in kinds:
                # Setup our variables
                self.plex_type = kind[0]
                self.section_type = kind[1]
                self.context = kind[2]
                self.get_children = kind[3]
                # Now do the heavy lifting
                if self.isCanceled() or not self.process_kind():
                    return False
                # Delete movies that are not on Plex anymore
                self.process_delete()
        return True

    @utils.log_time
    def run(self):
        if self.isCanceled():
            return
        successful = False
        self.current_sync = utils.unix_timestamp()
        # Delete playlist and video node files from Kodi
        utils.delete_playlists()
        utils.delete_nodes()
        # Get latest Plex libraries and build playlist and video node files
        if not sections.sync_from_pms():
            return
        try:
            # Fire up our single processing thread
            self.queue = backgroundthread.Queue.Queue(maxsize=600)
            self.processing_thread = ProcessMetadata(self.queue,
                                                     self.current_sync,
                                                     self.show_dialog)
            self.processing_thread.start()

            # Actual syncing - do only new items first
            LOG.info('Running full_library_sync with repair=%s',
                     self.repair)
            if not self.full_library_sync():
                return
            # Tell the processing thread to exit with one last element None
            self.queue.put(None)
            if self.isCanceled():
                return
            if PLAYLIST_SYNC_ENABLED and not playlists.full_sync():
                return
            successful = True
        except:
            utils.ERROR(txt='full_sync.py crashed', notify=True)
        finally:
            # This will block until the processing thread really exits
            LOG.debug('Waiting for processing thread to exit')
            self.processing_thread.join()
            common.update_kodi_library(video=True, music=True)
            self.threader.shutdown()
            if self.callback:
                self.callback(successful)
            LOG.info('Done full_sync')


def start(show_dialog, repair=False, callback=None):
    """
    """
    # FullSync(repair, callback, show_dialog).start()
    FullSync(repair, callback, show_dialog).run()