WebSocket.to_events() - Code Metrics - Inspection of "Merge branch 'hotfix/1.0.1.2'" - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( b25270...aeb165 )

by Dean

created 2016-03-14 20:45 UTC

WebSocket.to_events() B

↳ Parent: WebSocket

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	20

Metric	Value
cc	4
dl	0
loc	22
ccs	0
cts	11
cp	0
crap	20
rs	8.9197

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_websocket import WSessionManager
from plugin.scrobbler.core.constants import IGNORED_EVENTS
from plugin.scrobbler.core.engine import SessionEngine
from plugin.scrobbler.methods.core.base import Base

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

log = logging.getLogger(__name__)


class WebSocket(Base):
    name = 'websocket'

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

        self.engine = SessionEngine()

    def on_playing(self, info):
        # Create or retrieve existing session
        session = WSessionManager.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')
            WSessionManager.update(session, info, fetch=True)
            return

        if session.duration is None or session.view_offset is None:
            # Update session
            WSessionManager.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

        # Check for changed media
        media_changed = session.rating_key != to_integer(info.get('ratingKey'))

        # 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
        WSessionManager.update(session, info, fetch=media_changed)

    @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 []

        # 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': to_integer(info.get('viewOffset'))
            })
        ]

    @classmethod
    def test(cls):
        if Plex['status'].sessions() is None:
            log.info("Error while retrieving sessions, assuming WebSocket method isn't available")
            return False

        detail = Plex.detail()

        if detail is None:
            log.info('Error while retrieving server info for testing')
            return False

        if not detail.multiuser:
            log.info("Server info indicates multi-user support isn't available, WebSocket method not available")
            return False

        return True


Push — master ( b25270...aeb165 )

WebSocket.to_events() B

Complexity

Size

Duplication

Code Coverage

1. Missing Dependencies

2. Missing init.py files

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_websocket import WSessionManager
4			from plugin.scrobbler.core.constants import IGNORED_EVENTS
5			from plugin.scrobbler.core.engine import SessionEngine
6			from plugin.scrobbler.methods.core.base import Base
7
8			from datetime import datetime, timedelta
9			from plex import Plex
10			from plex_activity import Activity
11			import logging
12
13			log = logging.getLogger(__name__)
14
15
16			class WebSocket(Base):
17			name = 'websocket'
18
19			def __init__(self):
20			Activity.on('websocket.playing', self.on_playing)
21
22			self.engine = SessionEngine()
23
24			def on_playing(self, info):
25			# Create or retrieve existing session
26			session = WSessionManager.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			WSessionManager.update(session, info, fetch=True)
32			return
33
34			if session.duration is None or session.view_offset is None:
35			# Update session
36			WSessionManager.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			# Check for changed media
49			media_changed = session.rating_key != to_integer(info.get('ratingKey'))
50
51			# Parse `events`
52			actions = self.engine.process(session, events)
53
54			for action, payload in actions:
55			# Build request for the event
56			request = self.build_request(
57			session,
58			rating_key=payload.get('rating_key'),
59			view_offset=payload.get('view_offset')
60			)
61
62			if not request:
63			continue
64
65			# Queue request to be sent
66			ActionManager.queue('/'.join(['scrobble', action]), request, session)
67
68			# Update session
69			WSessionManager.update(session, info, fetch=media_changed)
70
71			@classmethod
72			def to_events(cls, session, info):
73			# Validate `state`
74			state = info.get('state')
75
76			if not state:
77			log.warn('Event has an invalid state %r', state)
78			return []
79
80			if state in IGNORED_EVENTS:
81			log.debug('Ignored "%s" event: %r', state, info)
82			return []
83
84			# Check for session `view_offset` jump
85			if cls.session_jumped(session, info.get('viewOffset')):
86			return []
87
88			# Build event
89			return [
90			(state, {
91			'rating_key': to_integer(info.get('ratingKey')),
92			'view_offset': to_integer(info.get('viewOffset'))
93			})
94			]
95
96			@classmethod
97			def test(cls):
98			if Plex['status'].sessions() is None:
99			log.info("Error while retrieving sessions, assuming WebSocket method isn't available")
100			return False
101
102			detail = Plex.detail()
			0 ignored issues – show Bug introduced 2015-12-19 01:32 UTC by Report Bug Copy Issue Report The Class `Plex` does not seem to have a member named `detail`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
103			if detail is None:
104			log.info('Error while retrieving server info for testing')
105			return False
106
107			if not detail.multiuser:
108			log.info("Server info indicates multi-user support isn't available, WebSocket method not available")
109			return False
110
111			return True
112

fuzeman / Plex-Trakt-Scrobbler

Push — master ( b25270...aeb165 )

WebSocket.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