WebSocket.to_events() - Code Metrics - Inspection of "Bumped version to 1.1.0.2-beta" - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — beta ( 72a57d...7d0ef0 )

by Dean

created 2016-06-12 12:37 UTC

WebSocket.to_events() B

↳ Parent: WebSocket

Complexity

Conditions

Size

Total Lines

Duplication

Lines	22
Ratio	100 %

Code Coverage

Tests	0
CRAP Score	20

Importance

Changes	1
Bugs	0	Features	0

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

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


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	View Code Duplication	def on_playing(self, info):
		0 ignored issues – show Duplication introduced 2016-05-19 03:26 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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	View Code Duplication	@classmethod
		0 ignored issues – show Duplication introduced 2016-05-19 03:26 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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 — beta ( 72a57d...7d0ef0 )

WebSocket.to_events() B

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files