PlexAccount.refresh_details() - Code Metrics - Inspection of "Updated the `Mapper` module to support oem-client..." - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — beta ( a8392a...918fe2 )

by Dean

created 2016-07-19 04:24 UTC

PlexAccount.refresh_details() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	26.5972

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	39
ccs	1
cts	21
cp	0.0476
rs	8.0894
cc	5
crap	26.5972

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(force=True):
            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()

        log.info('Refreshing plex account: %r', self)

        # 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):
            log.warn('Unable to retrieve account details from plex.tv (status_code: %s)', response.status_code)
            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.info('Refreshing plex account: %r (anonymous)', self)

        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 not basic.token_server:
            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(force=True):
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		log.info('Refreshing plex account: %r', self)
119
120		# Fetch account details
121		response = requests.get('https://plex.tv/users/account', headers={
122		'X-Plex-Token': basic.token_plex
123		})
124
125		if not (200 <= response.status_code < 300):
126		log.warn('Unable to retrieve account details from plex.tv (status_code: %s)', response.status_code)
127		return False
128
129		user = ElementTree.fromstring(response.content)
130
131		# Update details
132		self.username = user.attrib.get('username') or None
133
134		self.title = user.attrib.get('title')
135		self.thumb = user.attrib.get('thumb')
136
137		# Update `key`
138		if self.id == 1:
139		# Use administrator `key`
140		self.key = 1
141		else:
142		# Retrieve user id from plex.tv details
143		try:
144		user_id = int(user.attrib.get('id'))
145		except Exception, ex:
146		log.warn('Unable to cast user id to integer: %s', ex, exc_info=True)
147		user_id = None
148
149		# Update `key`
150		self.key = user_id
151
152		return True
153
154	1	def refresh_anonymous(self):
155		log.info('Refreshing plex account: %r (anonymous)', self)
156
157		self.username = 'administrator'
158
159		self.title = 'Administrator'
160		self.thumb = None
161
162		if self.id == 1:
163		self.key = 1
164		else:
165		self.key = None
166
167		return True
168
169	1	def refresh_required(self, basic):
170		if self.key is None:
171		return True
172
173		if self.title is None:
174		return True
175
176		if not basic.token_server:
177		return True
178
179		return False
180
181	1	def thumb_url(self, default=None, rating='pg', size=256):
182		if not self.thumb:
183		return None
184
185		thumb = urlparse(self.thumb)
186
187		if thumb.netloc.endswith('plex.tv'):
188		return self.thumb
189
190		if not thumb.netloc.endswith('gravatar.com'):
191		return None
192
193		result = 'https://secure.gravatar.com%s' % thumb.path
194
195		if default is None:
196		query = dict(parse_qsl(thumb.query))
197
198		default = query.get('d') or query.get('default')
199
200		return result + '?' + urlencode({
201		'd': default,
202		'r': rating,
203		's': size
204		})
205
206	1	def to_json(self, full=False):
207		result = {
208		'id': self.id,
209		'username': self.username,
210
211		'title': self.title,
212		'thumb_url': self.thumb_url()
213		}
214
215		if not full:
216		return result
217
218		# Merge authorization details
219		result['authorization'] = {
220		'basic': {'state': 'empty'}
221		}
222
223		# - Basic credentials
224		basic = self.basic
225
226		if basic is not None:
227		result['authorization']['basic'] = basic.to_json(self)
228
229		return result
230
231	1	def __repr__(self):
232		return '<PlexAccount username: %r>' % (
233		self.username,
234		)
235

fuzeman / Plex-Trakt-Scrobbler

Push — beta ( a8392a...918fe2 )

PlexAccount.refresh_details() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like