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

UpdateUser.account_function() B
last analyzed 2018-03-05 01:11 UTC

↳ Parent: UpdateUser

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	20

Importance

Changes

Metric	Value
dl	0
loc	31
ccs	0
cts	14
cp	0
rs	8.5806
c	0
b	0
f	0
cc	4
crap	20

from plugin.core.filters import Filters
from plugin.core.helpers.variable import to_integer
from plugin.managers.core.base import Get, Manager, Update
from plugin.managers.core.exceptions import UserFilteredException
from plugin.models import User, UserRule, PlexAccount

from exception_wrappers.libraries import apsw
import logging
import peewee

log = logging.getLogger(__name__)


class GetUser(Get):

    def __call__(self, user):

        user = self.manager.parse_user(user)

        if not user:
            return None

        return super(GetUser, self).__call__(
            User.key == to_integer(user['key'])
        )

    def or_create(self, user, fetch=False, match=False, filtered_exception=False):

        user = self.manager.parse_user(user)

        if not user:
            return None

        try:
            # Create new user
            obj = self.manager.create(
                key=to_integer(user['key'])
            )

            # Update newly created object
            self.manager.update(
                obj, user,

                fetch=fetch,
                match=match,
                filtered_exception=filtered_exception
            )

            return obj
        except (apsw.ConstraintError, peewee.IntegrityError):
            # Return existing user
            obj = self(user)

            if fetch or match:
                # Update existing `User`
                self.manager.update(
                    obj, user,

                    fetch=fetch,
                    match=match,
                    filtered_exception=filtered_exception
                )

            return obj


class UpdateUser(Update):
    def __call__(self, obj, user, fetch=False, match=False, filtered_exception=False):

        user = self.manager.parse_user(user)

        if not user:
            return None

        filtered, data = self.to_dict(
            obj, user,

            fetch=fetch,
            match=match
        )

        updated = super(UpdateUser, self).__call__(
            obj, data
        )

        if filtered and filtered_exception:
            raise UserFilteredException

        return updated

    def to_dict(self, obj, user, fetch=False, match=False):

        result = {}

        # Fill `result` with available fields
        if user.get('title'):
            result['name'] = user['title']

        if user.get('thumb'):
            result['thumb'] = user['thumb']

        filtered = False

        if match:
            # Try match `User` against rules
            filtered, result = self.match(
                result, user
            )

        return filtered, result

    @classmethod
    def match(cls, result, user):
        # Apply global filters
        if not Filters.is_valid_user(user):
            # User didn't pass filters, update `account` attribute and return
            result['account'] = None

            return True, result

        # Find matching `UserRule`
        rule = (UserRule
            .select()
            .where(
                (UserRule.name == user['title']) |
                (UserRule.name == '*') |
                (UserRule.name == None)
            )
            .order_by(
                UserRule.priority.asc()
            )
            .first()
        )

        log.debug('Activity matched against rule: %r', rule)

        if rule:
            # Process rule
            if rule.account_function is not None:
                result['account'] = cls.account_function(user, rule)
            elif rule.account_id is not None:
                result['account'] = rule.account_id
            else:
                return True, result
        else:
            result['account'] = None

        return False, result

    @staticmethod
    def account_function(user, rule):
        func = rule.account_function

        # Handle account function
        account_id = None

        if func == '@':
            # Map, try automatically finding matching `PlexAccount`
            plex_account = (PlexAccount
                .select()
                .where(
                    (PlexAccount.username == user['title']) |
                    (PlexAccount.title == user['title'])
                )
                .first()
            )

            if plex_account:
                account_id = plex_account.account_id
        else:
            log.warn('Unknown account function: %r', func)
            return None

        # Ensure `account_id` is valid
        if account_id is None:
            log.info('Unable to match user %r against any account', user['title'])
            return None

        log.debug('Matched user %r to account %r', user['title'], account_id)
        return account_id


class UserManager(Manager):
    get = GetUser
    update = UpdateUser

    model = User

    @classmethod
    def parse_user(cls, user):
        if type(user) is not dict:
            # Build user dict from object
            user = {
                'key': user.id,
                'title': user.title,
                'thumb': user.thumb
            }

        # Validate `user`
        if not user.get('key'):
            return None

        return user


1		from plugin.core.filters import Filters
2		from plugin.core.helpers.variable import to_integer
3		from plugin.managers.core.base import Get, Manager, Update
4		from plugin.managers.core.exceptions import UserFilteredException
5		from plugin.models import User, UserRule, PlexAccount
6
7		from exception_wrappers.libraries import apsw
8		import logging
9		import peewee
10
11		log = logging.getLogger(__name__)
12
13
14	View Code Duplication	class GetUser(Get):
		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...
15		def __call__(self, user):
		0 ignored issues – show Bug introduced 2016-11-10 04:18 UTC by Report Bug Copy Issue Report Arguments number differs from overridden '__call__' method Loading history...
16		user = self.manager.parse_user(user)
17
18		if not user:
19		return None
20
21		return super(GetUser, self).__call__(
22		User.key == to_integer(user['key'])
23		)
24
25		def or_create(self, user, fetch=False, match=False, filtered_exception=False):
		0 ignored issues – show Bug introduced 2016-11-10 04:18 UTC by Report Bug Copy Issue Report Arguments number differs from overridden 'or_create' method Loading history...
26		user = self.manager.parse_user(user)
27
28		if not user:
29		return None
30
31		try:
32		# Create new user
33		obj = self.manager.create(
34		key=to_integer(user['key'])
35		)
36
37		# Update newly created object
38		self.manager.update(
39		obj, user,
40
41		fetch=fetch,
42		match=match,
43		filtered_exception=filtered_exception
44		)
45
46		return obj
47		except (apsw.ConstraintError, peewee.IntegrityError):
48		# Return existing user
49		obj = self(user)
50
51		if fetch or match:
52		# Update existing `User`
53		self.manager.update(
54		obj, user,
55
56		fetch=fetch,
57		match=match,
58		filtered_exception=filtered_exception
59		)
60
61		return obj
62
63
64		class UpdateUser(Update):
65		def __call__(self, obj, user, fetch=False, match=False, filtered_exception=False):
		0 ignored issues – show Bug introduced 2016-11-10 04:18 UTC by Report Bug Copy Issue Report Arguments number differs from overridden '__call__' method Loading history...
66		user = self.manager.parse_user(user)
67
68		if not user:
69		return None
70
71		filtered, data = self.to_dict(
72		obj, user,
73
74		fetch=fetch,
75		match=match
76		)
77
78		updated = super(UpdateUser, self).__call__(
79		obj, data
80		)
81
82		if filtered and filtered_exception:
83		raise UserFilteredException
84
85		return updated
86
87		def to_dict(self, obj, user, fetch=False, match=False):
		0 ignored issues – show Unused Code introduced 2015-12-19 01:32 UTC by Report Bug Copy Issue Report The argument `obj` seems to be unused. Loading history... Unused Code introduced 2015-12-19 01:32 UTC by Report Bug Copy Issue Report The argument `fetch` seems to be unused. Loading history...
88		result = {}
89
90		# Fill `result` with available fields
91		if user.get('title'):
92		result['name'] = user['title']
93
94		if user.get('thumb'):
95		result['thumb'] = user['thumb']
96
97		filtered = False
98
99		if match:
100		# Try match `User` against rules
101		filtered, result = self.match(
102		result, user
103		)
104
105		return filtered, result
106
107		@classmethod
108		def match(cls, result, user):
109		# Apply global filters
110		if not Filters.is_valid_user(user):
111		# User didn't pass filters, update `account` attribute and return
112		result['account'] = None
113
114		return True, result
115
116		# Find matching `UserRule`
117		rule = (UserRule
118		.select()
119		.where(
120		(UserRule.name == user['title']) \|
121		(UserRule.name == '*') \|
122		(UserRule.name == None)
123		)
124		.order_by(
125		UserRule.priority.asc()
126		)
127		.first()
128		)
129
130		log.debug('Activity matched against rule: %r', rule)
131
132		if rule:
133		# Process rule
134		if rule.account_function is not None:
135		result['account'] = cls.account_function(user, rule)
136		elif rule.account_id is not None:
137		result['account'] = rule.account_id
138		else:
139		return True, result
140		else:
141		result['account'] = None
142
143		return False, result
144
145		@staticmethod
146		def account_function(user, rule):
147		func = rule.account_function
148
149		# Handle account function
150		account_id = None
151
152		if func == '@':
153		# Map, try automatically finding matching `PlexAccount`
154		plex_account = (PlexAccount
155		.select()
156		.where(
157		(PlexAccount.username == user['title']) \|
158		(PlexAccount.title == user['title'])
159		)
160		.first()
161		)
162
163		if plex_account:
164		account_id = plex_account.account_id
165		else:
166		log.warn('Unknown account function: %r', func)
167		return None
168
169		# Ensure `account_id` is valid
170		if account_id is None:
171		log.info('Unable to match user %r against any account', user['title'])
172		return None
173
174		log.debug('Matched user %r to account %r', user['title'], account_id)
175		return account_id
176
177
178		class UserManager(Manager):
179		get = GetUser
180		update = UpdateUser
181
182		model = User
183
184		@classmethod
185		def parse_user(cls, user):
186		if type(user) is not dict:
187		# Build user dict from object
188		user = {
189		'key': user.id,
190		'title': user.title,
191		'thumb': user.thumb
192		}
193
194		# Validate `user`
195		if not user.get('key'):
196		return None
197
198		return user
199

fuzeman / Plex-Trakt-Scrobbler

UpdateUser.account_function() B last analyzed 2018-03-05 01:11 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

UpdateUser.account_function() B
last analyzed 2018-03-05 01:11 UTC