Logging.to_events() - Code Metrics - Inspection of "Merge branch 'feature/error-reporter-improvements'..." - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 20bb73...3c901c )

by Dean

created 2016-03-25 06:13 UTC

Logging.to_events() B

↳ Parent: Logging

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	30

Metric	Value
cc	5
dl	0
loc	29
ccs	0
cts	9
cp	0
crap	30
rs	8.0894

from plugin.core.helpers.variable import to_integer
from plugin.managers.action import ActionManager
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3
from plugin.managers.session.s_logging import LSessionManager
from plugin.scrobbler.core import SessionEngine
from plugin.scrobbler.core.constants import IGNORED_EVENTS
from plugin.scrobbler.methods.core.base import Base

from datetime import datetime, timedelta
from plex_activity import Activity
import logging


log = logging.getLogger(__name__)


class Logging(Base):
    name = 'logging'

    def __init__(self):
        Activity.on('logging.playing', self.on_playing)

        self.engine = SessionEngine()

    def on_playing(self, info):
        # Create or retrieve existing session
        session = LSessionManager.get.or_create(info, fetch=True)


        # Validate session
        if session.updated_at is None or (datetime.utcnow() - session.updated_at) > timedelta(minutes=5):
            log.info('Updating session, last update was over 5 minutes ago')
            LSessionManager.update(session, info, fetch=True)
            return

        if session.duration is None or session.view_offset is None:
            # Update session
            LSessionManager.update(session, info, fetch=lambda s, i: (
                s.rating_key != to_integer(i.get('ratingKey')) or
                s.duration is None
            ))
            return

        # Parse `info` to events
        events = self.to_events(session, info)

        if not events:
            return

        # Parse `events`
        actions = self.engine.process(session, events)

        for action, payload in actions:
            # Build request for the event
            request = self.build_request(
                session,
                rating_key=payload.get('rating_key'),
                view_offset=payload.get('view_offset')
            )

            if not request:
                continue

            # Queue request to be sent
            ActionManager.queue('/'.join(['scrobble', action]), request, session)

        # Update session
        LSessionManager.update(session, info)

    @classmethod
    def to_events(cls, session, info):
        # Validate `state`
        state = info.get('state')

        if not state:
            log.warn('Event has an invalid state %r', state)
            return []

        if state in IGNORED_EVENTS:
            log.debug('Ignored "%s" event: %r', state, info)
            return []

        # Validate `view_offset`
        view_offset = to_integer(info.get('viewOffset'))

        if view_offset is None:
            log.info('Event has an invalid view offset %r', view_offset)
            return []

        # Check for session `view_offset` jump
        if cls.session_jumped(session, info.get('viewOffset')):
            return []

        # Build event
        return [
            (state, {
                'rating_key': to_integer(info.get('ratingKey')),
                'view_offset': view_offset
            })
        ]

    @classmethod
    def test(cls):
        return True


1			from plugin.core.helpers.variable import to_integer
2			from plugin.managers.action import ActionManager
			0 ignored issues – show Bug introduced 2016-02-09 02:13 UTC by Report Bug Copy Issue Report The name `action` does not seem to exist in module `plugin.managers`. Loading history... Configuration introduced 2016-02-09 02:13 UTC by Report Bug Copy Issue Report Unable to import 'plugin.managers.action' (invalid syntax (<string>, line 72)) This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
3			from plugin.managers.session.s_logging import LSessionManager
4			from plugin.scrobbler.core import SessionEngine
5			from plugin.scrobbler.core.constants import IGNORED_EVENTS
6			from plugin.scrobbler.methods.core.base import Base
7
8			from datetime import datetime, timedelta
9			from plex_activity import Activity
10			import logging
11
12
13			log = logging.getLogger(__name__)
14
15
16			class Logging(Base):
17			name = 'logging'
18
19			def __init__(self):
20			Activity.on('logging.playing', self.on_playing)
21
22			self.engine = SessionEngine()
23
24			def on_playing(self, info):
25			# Create or retrieve existing session
26			session = LSessionManager.get.or_create(info, fetch=True)
			0 ignored issues – show Bug introduced 2015-12-19 01:32 UTC by Report Bug Copy Issue Report It seems like a value for argument `info` is missing in the unbound method call. Loading history...
27
28			# Validate session
29			if session.updated_at is None or (datetime.utcnow() - session.updated_at) > timedelta(minutes=5):
30			log.info('Updating session, last update was over 5 minutes ago')
31			LSessionManager.update(session, info, fetch=True)
32			return
33
34			if session.duration is None or session.view_offset is None:
35			# Update session
36			LSessionManager.update(session, info, fetch=lambda s, i: (
37			s.rating_key != to_integer(i.get('ratingKey')) or
38			s.duration is None
39			))
40			return
41
42			# Parse `info` to events
43			events = self.to_events(session, info)
44
45			if not events:
46			return
47
48			# Parse `events`
49			actions = self.engine.process(session, events)
50
51			for action, payload in actions:
52			# Build request for the event
53			request = self.build_request(
54			session,
55			rating_key=payload.get('rating_key'),
56			view_offset=payload.get('view_offset')
57			)
58
59			if not request:
60			continue
61
62			# Queue request to be sent
63			ActionManager.queue('/'.join(['scrobble', action]), request, session)
64
65			# Update session
66			LSessionManager.update(session, info)
67
68			@classmethod
69			def to_events(cls, session, info):
70			# Validate `state`
71			state = info.get('state')
72
73			if not state:
74			log.warn('Event has an invalid state %r', state)
75			return []
76
77			if state in IGNORED_EVENTS:
78			log.debug('Ignored "%s" event: %r', state, info)
79			return []
80
81			# Validate `view_offset`
82			view_offset = to_integer(info.get('viewOffset'))
83
84			if view_offset is None:
85			log.info('Event has an invalid view offset %r', view_offset)
86			return []
87
88			# Check for session `view_offset` jump
89			if cls.session_jumped(session, info.get('viewOffset')):
90			return []
91
92			# Build event
93			return [
94			(state, {
95			'rating_key': to_integer(info.get('ratingKey')),
96			'view_offset': view_offset
97			})
98			]
99
100			@classmethod
101			def test(cls):
102			return True
103

fuzeman / Plex-Trakt-Scrobbler

Push — develop ( 20bb73...3c901c )

Logging.to_events() B

Complexity

Size

Duplication

Code Coverage

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files