Mode.trakt() - Code Metrics - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Mode.trakt() A
last analyzed 2018-03-05 01:11 UTC

↳ Parent: Mode

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	6.7968

Importance

Changes

Metric	Value
dl	0
loc	6
ccs	1
cts	4
cp	0.25
rs	9.4285
c	0
b	0
f	0
cc	3
crap	6.7968

from plugin.core.filters import Filters
from plugin.sync.core.enums import SyncData, SyncMedia, SyncMode
from plugin.sync.core.guid import GuidMatch

from plex import Plex
import elapsed
import itertools
import logging

log = logging.getLogger(__name__)

TRAKT_DATA_MAP = {
    SyncMedia.Movies: [
        SyncData.Collection,
        SyncData.Playback,
        SyncData.Ratings,
        SyncData.Watched,
        # SyncData.Watchlist
    ],
    SyncMedia.Shows: [
        SyncData.Ratings
    ],
    SyncMedia.Seasons: [
        SyncData.Ratings
    ],
    SyncMedia.Episodes: [
        SyncData.Collection,
        SyncData.Playback,
        SyncData.Ratings,
        SyncData.Watched,
        # SyncData.Watchlist
    ]
}

DATA_PREFERENCE_MAP = {
    SyncData.Collection:    'sync.collection.mode',
    SyncData.Playback:      'sync.playback.mode',
    SyncData.Ratings:       'sync.ratings.mode',
    SyncData.Watched:       'sync.watched.mode',

    # Lists
    SyncData.Liked:         'sync.lists.liked.mode',
    SyncData.Personal:      'sync.lists.personal.mode',
    SyncData.Watchlist:     'sync.lists.watchlist.mode',
}


class Mode(object):
    data = None
    mode = None

    children = []

    def __init__(self, task):
        self.__task = task

        self.children = [c(task) for c in self.children]

        # Retrieve enabled data
        self.enabled_data = self.get_enabled_data()

        # Determine if mode should be enabled
        self.enabled = len(self.enabled_data) > 0

        if not self.enabled:
            log.debug('Mode %r disabled on: %r', self.mode, self)

    @property
    def current(self):
        return self.__task

    @property
    def configuration(self):
        return self.__task.configuration

    @property
    def handlers(self):
        return self.__task.handlers

    @property
    def modes(self):
        return self.__task.modes

    @property
    def plex(self):
        if not self.current or not self.current.state:
            return None

        return self.current.state.plex

    @property
    def trakt(self):
        if not self.current or not self.current.state:
            return None

        return self.current.state.trakt

    def construct(self):
        pass

    def start(self):
        pass

    def run(self):
        raise NotImplementedError

    def finish(self):
        pass

    def stop(self):
        pass

    def checkpoint(self):
        if self.current is None:
            return

        self.current.checkpoint()

    def execute_children(self, name, force=None):
        # Run method on children
        for c in self.children:
            if not force and not c.enabled:
                log.debug('Ignoring %s() call on child: %r', name, c)
                continue

            # Find method `name` in child
            log.info('Executing %s() on child: %r', name, c)

            func = getattr(c, name, None)

            if not func:
                log.warn('Unknown method: %r', name)
                continue

            # Run method on child
            func()

    def execute_episode_action(self, mode, data, ids, match, p_show, p_episode, t_item, **kwargs):
        # Process episode
        if match.media == GuidMatch.Media.Episode:
            # Process episode
            self.execute_handlers(
                mode, SyncMedia.Episodes, data,
                key=ids['episode'],

                p_item=p_episode,
                t_item=t_item,
                **kwargs
            )

            return True

        # Process movie
        if match.media == GuidMatch.Media.Movie:
            # Build movie item from plex episode
            p_movie = p_episode.copy()

            p_movie['title'] = p_show.get('title')
            p_movie['year'] = p_show.get('year')

            # Process movie
            self.execute_handlers(
                mode, SyncMedia.Movies, data,
                key=ids['episode'],

                p_item=p_episode,
                t_item=t_item,
                **kwargs
            )
            return True

        raise ValueError('Unknown media type: %r' % (match.media,))

    @elapsed.clock
    def execute_handlers(self, mode, media, data, *args, **kwargs):
        if type(media) is not list:
            media = [media]

        if type(data) is not list:
            data = [data]

        for m, d in itertools.product(media, data):
            if d not in self.handlers:
                log.debug('Unable to find handler for data: %r', d)
                continue

            try:
                self.handlers[d].run(m, mode, *args, **kwargs)
            except Exception as ex:

                log.warn('Exception raised in handlers[%r].run(%r, ...): %s', d, m, ex, exc_info=True)

    def get_enabled_data(self):
        config = self.configuration

        # Determine accepted modes
        modes = [SyncMode.Full]

        if self.mode == SyncMode.Full:
            modes.extend([
                SyncMode.FastPull,
                SyncMode.Pull,
                SyncMode.Push
            ])
        elif self.mode == SyncMode.FastPull:
            modes.extend([
                self.mode,
                SyncMode.Pull
            ])
        else:
            modes.append(self.mode)

        # Retrieve enabled data
        result = []

        if config['sync.watched.mode'] in modes:
            result.append(SyncData.Watched)

        if config['sync.ratings.mode'] in modes:
            result.append(SyncData.Ratings)

        if config['sync.playback.mode'] in modes:
            result.append(SyncData.Playback)

        if config['sync.collection.mode'] in modes:
            result.append(SyncData.Collection)

        # Lists
        if config['sync.lists.watchlist.mode'] in modes:
            result.append(SyncData.Watchlist)

        if config['sync.lists.liked.mode'] in modes:
            result.append(SyncData.Liked)

        if config['sync.lists.personal.mode'] in modes:
            result.append(SyncData.Personal)

        # Filter `result` to data provided by this mode
        if self.data is None:
            log.warn('No "data" property defined on %r', self)
            return result

        if self.data == SyncData.All:
            return result

        return [
            data for data in result
            if data in self.data
        ]

    def get_data(self, media):
        for data in TRAKT_DATA_MAP[media]:
            if not self.is_data_enabled(data):
                continue

            yield data

    @elapsed.clock
    def is_data_enabled(self, data):
        return data in self.enabled_data

    def run_episode_action(self, mode, data, ids, match, p_show, p_episode, t_item, **kwargs):
        if match.media == GuidMatch.Media.Movie:
            # Process movie
            self.execute_episode_action(
                mode, data, ids, match,
                p_show, p_episode, t_item,
                **kwargs
            )
        elif match.media == GuidMatch.Media.Episode:
            # Ensure `match` contains episodes
            if not match.episodes:
                log.info('No episodes returned for: %s/%s', match.guid.service, match.guid.id)
                return

            # Process each episode
            for season_num, episode_num in match.episodes:
                t_season = self._get_show_season(t_item, season_num)

                if t_season is None:
                    # Unable to find matching season in `t_show`
                    continue

                t_episode = self._get_season_episode(t_season, episode_num)

                if t_episode is None:
                    # Unable to find matching episode in `t_season`
                    continue

                self.execute_episode_action(
                    mode, data, ids, match,
                    p_show, p_episode, t_episode,
                    **kwargs
                )

    def sections(self, section_type=None):
        # Retrieve "section" for current task
        section_key = self.current.kwargs.get('section', None)

        # Fetch sections from server
        p_sections = Plex['library'].sections()

        if p_sections is None:
            return None

        # Filter sections, map to dictionary
        result = {}

        for section in p_sections.filter(section_type, section_key):
            # Apply section name filter
            if not Filters.is_valid_section_name(section.title):
                continue

            try:
                key = int(section.key)
            except Exception as ex:

                log.warn('Unable to cast section key %r to integer: %s', section.key, ex, exc_info=True)
                continue

            result[key] = section.uuid

        return [(key, ) for key in result.keys()], result

    @staticmethod
    def _get_show_season(t_show, season_num):
        if type(t_show) is dict:
            return t_show.get('seasons', {}).get(season_num)

        return t_show.seasons.get(season_num)

    @staticmethod
    def _get_season_episode(t_season, episode_num):
        if type(t_season) is dict:
            return t_season.get('episodes', {}).get(episode_num)

        return t_season.episodes.get(episode_num)


1	1	from plugin.core.filters import Filters
2	1	from plugin.sync.core.enums import SyncData, SyncMedia, SyncMode
3	1	from plugin.sync.core.guid import GuidMatch
4
5	1	from plex import Plex
6	1	import elapsed
7	1	import itertools
8	1	import logging
9
10	1	log = logging.getLogger(__name__)
11
12	1	TRAKT_DATA_MAP = {
13		SyncMedia.Movies: [
14		SyncData.Collection,
15		SyncData.Playback,
16		SyncData.Ratings,
17		SyncData.Watched,
18		# SyncData.Watchlist
19		],
20		SyncMedia.Shows: [
21		SyncData.Ratings
22		],
23		SyncMedia.Seasons: [
24		SyncData.Ratings
25		],
26		SyncMedia.Episodes: [
27		SyncData.Collection,
28		SyncData.Playback,
29		SyncData.Ratings,
30		SyncData.Watched,
31		# SyncData.Watchlist
32		]
33		}
34
35	1	DATA_PREFERENCE_MAP = {
36		SyncData.Collection: 'sync.collection.mode',
37		SyncData.Playback: 'sync.playback.mode',
38		SyncData.Ratings: 'sync.ratings.mode',
39		SyncData.Watched: 'sync.watched.mode',
40
41		# Lists
42		SyncData.Liked: 'sync.lists.liked.mode',
43		SyncData.Personal: 'sync.lists.personal.mode',
44		SyncData.Watchlist: 'sync.lists.watchlist.mode',
45		}
46
47
48	1	class Mode(object):
49	1	data = None
50	1	mode = None
51
52	1	children = []
53
54	1	def __init__(self, task):
55		self.__task = task
56
57		self.children = [c(task) for c in self.children]
58
59		# Retrieve enabled data
60		self.enabled_data = self.get_enabled_data()
61
62		# Determine if mode should be enabled
63		self.enabled = len(self.enabled_data) > 0
64
65		if not self.enabled:
66		log.debug('Mode %r disabled on: %r', self.mode, self)
67
68	1	@property
69		def current(self):
70		return self.__task
71
72	1	@property
73		def configuration(self):
74		return self.__task.configuration
75
76	1	@property
77		def handlers(self):
78		return self.__task.handlers
79
80	1	@property
81		def modes(self):
82		return self.__task.modes
83
84	1	@property
85		def plex(self):
86		if not self.current or not self.current.state:
87		return None
88
89		return self.current.state.plex
90
91	1	@property
92		def trakt(self):
93		if not self.current or not self.current.state:
94		return None
95
96		return self.current.state.trakt
97
98	1	def construct(self):
99		pass
100
101	1	def start(self):
102		pass
103
104	1	def run(self):
105		raise NotImplementedError
106
107	1	def finish(self):
108		pass
109
110	1	def stop(self):
111		pass
112
113	1	def checkpoint(self):
114		if self.current is None:
115		return
116
117		self.current.checkpoint()
118
119	1	def execute_children(self, name, force=None):
120		# Run method on children
121		for c in self.children:
122		if not force and not c.enabled:
123		log.debug('Ignoring %s() call on child: %r', name, c)
124		continue
125
126		# Find method `name` in child
127		log.info('Executing %s() on child: %r', name, c)
128
129		func = getattr(c, name, None)
130
131		if not func:
132		log.warn('Unknown method: %r', name)
133		continue
134
135		# Run method on child
136		func()
137
138	1	def execute_episode_action(self, mode, data, ids, match, p_show, p_episode, t_item, **kwargs):
139		# Process episode
140		if match.media == GuidMatch.Media.Episode:
141		# Process episode
142		self.execute_handlers(
143		mode, SyncMedia.Episodes, data,
144		key=ids['episode'],
145
146		p_item=p_episode,
147		t_item=t_item,
148		**kwargs
149		)
150
151		return True
152
153		# Process movie
154		if match.media == GuidMatch.Media.Movie:
155		# Build movie item from plex episode
156		p_movie = p_episode.copy()
157
158		p_movie['title'] = p_show.get('title')
159		p_movie['year'] = p_show.get('year')
160
161		# Process movie
162		self.execute_handlers(
163		mode, SyncMedia.Movies, data,
164		key=ids['episode'],
165
166		p_item=p_episode,
167		t_item=t_item,
168		**kwargs
169		)
170		return True
171
172		raise ValueError('Unknown media type: %r' % (match.media,))
173
174	1	@elapsed.clock
175		def execute_handlers(self, mode, media, data, args, *kwargs):
176		if type(media) is not list:
177		media = [media]
178
179		if type(data) is not list:
180		data = [data]
181
182		for m, d in itertools.product(media, data):
183		if d not in self.handlers:
184		log.debug('Unable to find handler for data: %r', d)
185		continue
186
187		try:
188		self.handlers[d].run(m, mode, args, *kwargs)
189		except Exception as ex:
		0 ignored issues – show Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
190		log.warn('Exception raised in handlers[%r].run(%r, ...): %s', d, m, ex, exc_info=True)
191
192	1	def get_enabled_data(self):
193		config = self.configuration
194
195		# Determine accepted modes
196		modes = [SyncMode.Full]
197
198		if self.mode == SyncMode.Full:
199		modes.extend([
200		SyncMode.FastPull,
201		SyncMode.Pull,
202		SyncMode.Push
203		])
204		elif self.mode == SyncMode.FastPull:
205		modes.extend([
206		self.mode,
207		SyncMode.Pull
208		])
209		else:
210		modes.append(self.mode)
211
212		# Retrieve enabled data
213		result = []
214
215		if config['sync.watched.mode'] in modes:
216		result.append(SyncData.Watched)
217
218		if config['sync.ratings.mode'] in modes:
219		result.append(SyncData.Ratings)
220
221		if config['sync.playback.mode'] in modes:
222		result.append(SyncData.Playback)
223
224		if config['sync.collection.mode'] in modes:
225		result.append(SyncData.Collection)
226
227		# Lists
228		if config['sync.lists.watchlist.mode'] in modes:
229		result.append(SyncData.Watchlist)
230
231		if config['sync.lists.liked.mode'] in modes:
232		result.append(SyncData.Liked)
233
234		if config['sync.lists.personal.mode'] in modes:
235		result.append(SyncData.Personal)
236
237		# Filter `result` to data provided by this mode
238		if self.data is None:
239		log.warn('No "data" property defined on %r', self)
240		return result
241
242		if self.data == SyncData.All:
243		return result
244
245		return [
246		data for data in result
247		if data in self.data
248		]
249
250	1	def get_data(self, media):
251		for data in TRAKT_DATA_MAP[media]:
252		if not self.is_data_enabled(data):
253		continue
254
255		yield data
256
257	1	@elapsed.clock
258		def is_data_enabled(self, data):
259		return data in self.enabled_data
260
261	1	def run_episode_action(self, mode, data, ids, match, p_show, p_episode, t_item, **kwargs):
262		if match.media == GuidMatch.Media.Movie:
263		# Process movie
264		self.execute_episode_action(
265		mode, data, ids, match,
266		p_show, p_episode, t_item,
267		**kwargs
268		)
269		elif match.media == GuidMatch.Media.Episode:
270		# Ensure `match` contains episodes
271		if not match.episodes:
272		log.info('No episodes returned for: %s/%s', match.guid.service, match.guid.id)
273		return
274
275		# Process each episode
276		for season_num, episode_num in match.episodes:
277		t_season = self._get_show_season(t_item, season_num)
278
279		if t_season is None:
280		# Unable to find matching season in `t_show`
281		continue
282
283		t_episode = self._get_season_episode(t_season, episode_num)
284
285		if t_episode is None:
286		# Unable to find matching episode in `t_season`
287		continue
288
289		self.execute_episode_action(
290		mode, data, ids, match,
291		p_show, p_episode, t_episode,
292		**kwargs
293		)
294
295	1	def sections(self, section_type=None):
296		# Retrieve "section" for current task
297		section_key = self.current.kwargs.get('section', None)
298
299		# Fetch sections from server
300		p_sections = Plex['library'].sections()
301
302		if p_sections is None:
303		return None
304
305		# Filter sections, map to dictionary
306		result = {}
307
308		for section in p_sections.filter(section_type, section_key):
309		# Apply section name filter
310		if not Filters.is_valid_section_name(section.title):
311		continue
312
313		try:
314		key = int(section.key)
315		except Exception as ex:
		0 ignored issues – show Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
316		log.warn('Unable to cast section key %r to integer: %s', section.key, ex, exc_info=True)
317		continue
318
319		result[key] = section.uuid
320
321		return [(key, ) for key in result.keys()], result
322
323	1	@staticmethod
324		def _get_show_season(t_show, season_num):
325		if type(t_show) is dict:
326		return t_show.get('seasons', {}).get(season_num)
327
328		return t_show.seasons.get(season_num)
329
330	1	@staticmethod
331		def _get_season_episode(t_season, episode_num):
332		if type(t_season) is dict:
333		return t_season.get('episodes', {}).get(episode_num)
334
335		return t_season.episodes.get(episode_num)
336

fuzeman / Plex-Trakt-Scrobbler

Mode.trakt() A last analyzed 2018-03-05 01:11 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

Mode.trakt() A
last analyzed 2018-03-05 01:11 UTC