PlexAccount.basic_authorization() - Code Metrics - Inspection of "Merge branch 'develop' into beta" - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — beta ( ff30bd...8befa3 )

by Dean

created 2016-07-01 00:29 UTC

PlexAccount.basic_authorization() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	15.664

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	16
ccs	1
cts	10
cp	0.1
rs	9.2
cc	4
crap	15.664

from plugin.core.exceptions import AccountAuthenticationError
from plugin.models.core import db
from plugin.models.account import Account

from datetime import datetime, timedelta
from playhouse.apsw_ext import *
from plex import Plex
from urllib import urlencode
from urlparse import urlparse, parse_qsl
from xml.etree import ElementTree
import logging
import requests

REFRESH_INTERVAL = timedelta(days=1)

log = logging.getLogger(__name__)


class PlexAccount(Model):
    class Meta:
        database = db
        db_table = 'plex.account'

    account = ForeignKeyField(Account, 'plex_accounts', unique=True)

    key = IntegerField(null=True, unique=True)
    username = CharField(null=True, unique=True)

    title = CharField(null=True)
    thumb = TextField(null=True)

    refreshed_at = DateTimeField(null=True)

    def __init__(self, *args, **kwargs):
        super(PlexAccount, self).__init__(*args, **kwargs)

        self._basic_credential = None

    @property
    def account_id(self):
        return self._data.get('account')

    @property
    def basic(self):
        if self._basic_credential:
            return self._basic_credential

        return self.basic_credentials.first()

    @basic.setter
    def basic(self, value):
        self._basic_credential = value

    def authorization(self):
        # Basic
        basic = self.basic

        if basic:
            return self.basic_authorization(basic)

        # No account authorization available
        raise AccountAuthenticationError("Plex account hasn't been authenticated")

    def basic_authorization(self, basic_credential=None):
        if basic_credential is None:
            basic_credential = self.basic

        # Ensure token exists
        if basic_credential.token_server is None:
            raise AccountAuthenticationError("Plex account is missing the server token")

        # Handle anonymous authentication
        if basic_credential.token_server == 'anonymous':
            log.debug('Using anonymous authorization for %r', self)
            return Plex.configuration.authentication(token=None)

        # Configure client
        log.debug('Using basic authorization for %r', self)
        return Plex.configuration.authentication(basic_credential.token_server)

    def refresh(self, force=False, save=True):
        # Retrieve credentials
        basic = self.basic

        if not basic:
            return False

        # Check if refresh is required
        if self.refresh_required(basic):
            force = True

        # Only refresh account every `REFRESH_INTERVAL`
        if not force and self.refreshed_at:
            since_refresh = datetime.utcnow() - self.refreshed_at

            if since_refresh < REFRESH_INTERVAL:
                return False

        # Refresh account details
        if not self.refresh_details(basic):
            return False

        if not basic.refresh():
            return False

        # Store changes in database
        self.refreshed_at = datetime.utcnow()

        if save:
            self.save()

        return True

    def refresh_details(self, basic):
        if basic.token_plex == 'anonymous':
            return self.refresh_anonymous()

        # Fetch account details
        response = requests.get('https://plex.tv/users/account', headers={
            'X-Plex-Token': basic.token_plex
        })

        if not (200 <= response.status_code < 300):
            # Invalid response
            return False

        user = ElementTree.fromstring(response.content)

        # Update details
        self.username = user.attrib.get('username') or None

        self.title = user.attrib.get('title')
        self.thumb = user.attrib.get('thumb')

        # Update `key`
        if self.id == 1:
            # Use administrator `key`
            self.key = 1
        else:
            # Retrieve user id from plex.tv details
            try:
                user_id = int(user.attrib.get('id'))
            except Exception, ex:
                log.warn('Unable to cast user id to integer: %s', ex, exc_info=True)
                user_id = None

            # Update `key`
            self.key = user_id

        return True

    def refresh_anonymous(self):
        log.debug('Refreshing anonymous plex account')

        self.username = 'administrator'

        self.title = 'Administrator'
        self.thumb = None

        if self.id == 1:
            self.key = 1
        else:
            self.key = None

        return True

    def refresh_required(self, basic):
        if self.key is None:
            return True

        if self.title is None:
            return True

        if basic.token_server is None:
            return True

        return False

    def thumb_url(self, default=None, rating='pg', size=256):
        if not self.thumb:
            return None

        thumb = urlparse(self.thumb)

        if thumb.netloc.endswith('plex.tv'):
            return self.thumb

        if not thumb.netloc.endswith('gravatar.com'):
            return None

        result = 'https://secure.gravatar.com%s' % thumb.path

        if default is None:
            query = dict(parse_qsl(thumb.query))

            default = query.get('d') or query.get('default')

        return result + '?' + urlencode({
            'd': default,
            'r': rating,
            's': size
        })

    def to_json(self, full=False):
        result = {
            'id': self.id,
            'username': self.username,

            'title': self.title,
            'thumb_url': self.thumb_url()
        }

        if not full:
            return result

        # Merge authorization details
        result['authorization'] = {
            'basic': {'state': 'empty'}
        }

        # - Basic credentials
        basic = self.basic

        if basic is not None:
            result['authorization']['basic'] = basic.to_json(self)

        return result

    def __repr__(self):
        return '<PlexAccount username: %r>' % (
            self.username,
        )


1	1	from plugin.core.exceptions import AccountAuthenticationError
2	1	from plugin.models.core import db
3	1	from plugin.models.account import Account
4
5	1	from datetime import datetime, timedelta
6	1	from playhouse.apsw_ext import *
7	1	from plex import Plex
8	1	from urllib import urlencode
9	1	from urlparse import urlparse, parse_qsl
10	1	from xml.etree import ElementTree
11	1	import logging
12	1	import requests
13
14	1	REFRESH_INTERVAL = timedelta(days=1)
15
16	1	log = logging.getLogger(__name__)
17
18
19	1	class PlexAccount(Model):
20	1	class Meta:
21	1	database = db
22	1	db_table = 'plex.account'
23
24	1	account = ForeignKeyField(Account, 'plex_accounts', unique=True)
25
26	1	key = IntegerField(null=True, unique=True)
27	1	username = CharField(null=True, unique=True)
28
29	1	title = CharField(null=True)
30	1	thumb = TextField(null=True)
31
32	1	refreshed_at = DateTimeField(null=True)
33
34	1	def __init__(self, args, *kwargs):
35		super(PlexAccount, self).__init__(args, *kwargs)
36
37		self._basic_credential = None
38
39	1	@property
40		def account_id(self):
41		return self._data.get('account')
42
43	1	@property
44		def basic(self):
45		if self._basic_credential:
46		return self._basic_credential
47
48		return self.basic_credentials.first()
49
50	1	@basic.setter
51		def basic(self, value):
52		self._basic_credential = value
53
54	1	def authorization(self):
55		# Basic
56		basic = self.basic
57
58		if basic:
59		return self.basic_authorization(basic)
60
61		# No account authorization available
62		raise AccountAuthenticationError("Plex account hasn't been authenticated")
63
64	1	def basic_authorization(self, basic_credential=None):
65		if basic_credential is None:
66		basic_credential = self.basic
67
68		# Ensure token exists
69		if basic_credential.token_server is None:
70		raise AccountAuthenticationError("Plex account is missing the server token")
71
72		# Handle anonymous authentication
73		if basic_credential.token_server == 'anonymous':
74		log.debug('Using anonymous authorization for %r', self)
75		return Plex.configuration.authentication(token=None)
76
77		# Configure client
78		log.debug('Using basic authorization for %r', self)
79		return Plex.configuration.authentication(basic_credential.token_server)
80
81	1	def refresh(self, force=False, save=True):
82		# Retrieve credentials
83		basic = self.basic
84
85		if not basic:
86		return False
87
88		# Check if refresh is required
89		if self.refresh_required(basic):
90		force = True
91
92		# Only refresh account every `REFRESH_INTERVAL`
93		if not force and self.refreshed_at:
94		since_refresh = datetime.utcnow() - self.refreshed_at
95
96		if since_refresh < REFRESH_INTERVAL:
97		return False
98
99		# Refresh account details
100		if not self.refresh_details(basic):
101		return False
102
103		if not basic.refresh():
104		return False
105
106		# Store changes in database
107		self.refreshed_at = datetime.utcnow()
108
109		if save:
110		self.save()
111
112		return True
113
114	1	def refresh_details(self, basic):
115		if basic.token_plex == 'anonymous':
116		return self.refresh_anonymous()
117
118		# Fetch account details
119		response = requests.get('https://plex.tv/users/account', headers={
120		'X-Plex-Token': basic.token_plex
121		})
122
123		if not (200 <= response.status_code < 300):
124		# Invalid response
125		return False
126
127		user = ElementTree.fromstring(response.content)
128
129		# Update details
130		self.username = user.attrib.get('username') or None
131
132		self.title = user.attrib.get('title')
133		self.thumb = user.attrib.get('thumb')
134
135		# Update `key`
136		if self.id == 1:
137		# Use administrator `key`
138		self.key = 1
139		else:
140		# Retrieve user id from plex.tv details
141		try:
142		user_id = int(user.attrib.get('id'))
143		except Exception, ex:
144		log.warn('Unable to cast user id to integer: %s', ex, exc_info=True)
145		user_id = None
146
147		# Update `key`
148		self.key = user_id
149
150		return True
151
152	1	def refresh_anonymous(self):
153		log.debug('Refreshing anonymous plex account')
154
155		self.username = 'administrator'
156
157		self.title = 'Administrator'
158		self.thumb = None
159
160		if self.id == 1:
161		self.key = 1
162		else:
163		self.key = None
164
165		return True
166
167	1	def refresh_required(self, basic):
168		if self.key is None:
169		return True
170
171		if self.title is None:
172		return True
173
174		if basic.token_server is None:
175		return True
176
177		return False
178
179	1	def thumb_url(self, default=None, rating='pg', size=256):
180		if not self.thumb:
181		return None
182
183		thumb = urlparse(self.thumb)
184
185		if thumb.netloc.endswith('plex.tv'):
186		return self.thumb
187
188		if not thumb.netloc.endswith('gravatar.com'):
189		return None
190
191		result = 'https://secure.gravatar.com%s' % thumb.path
192
193		if default is None:
194		query = dict(parse_qsl(thumb.query))
195
196		default = query.get('d') or query.get('default')
197
198		return result + '?' + urlencode({
199		'd': default,
200		'r': rating,
201		's': size
202		})
203
204	1	def to_json(self, full=False):
205		result = {
206		'id': self.id,
207		'username': self.username,
208
209		'title': self.title,
210		'thumb_url': self.thumb_url()
211		}
212
213		if not full:
214		return result
215
216		# Merge authorization details
217		result['authorization'] = {
218		'basic': {'state': 'empty'}
219		}
220
221		# - Basic credentials
222		basic = self.basic
223
224		if basic is not None:
225		result['authorization']['basic'] = basic.to_json(self)
226
227		return result
228
229	1	def __repr__(self):
230		return '<PlexAccount username: %r>' % (
231		self.username,
232		)
233

fuzeman / Plex-Trakt-Scrobbler

Push — beta ( ff30bd...8befa3 )

PlexAccount.basic_authorization() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like