AccountMigration.get_trakt_username() - Code Metrics - Inspection of "Updated trakt.py (v2.6.1)" - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 456c9f...9a88f4 )

by Dean

created 2016-05-19 03:23 UTC

AccountMigration.get_trakt_username() A

↳ Parent: AccountMigration

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	8.2077

Metric	Value
cc	3
dl	0
loc	9
ccs	1
cts	6
cp	0.1666
crap	8.2077
rs	9.6666

from plugin.core.environment import Environment
from plugin.models import (
    Account, ClientRule, UserRule,
    PlexAccount, PlexBasicCredential,
    TraktAccount, TraktBasicCredential, TraktOAuthCredential
)
from plugin.modules.migrations.core.base import Migration

import apsw
import logging
import os
import peewee
import requests

log = logging.getLogger(__name__)


class AccountMigration(Migration):
    def run(self, token_plex=None):
        # Ensure server `Account` exists
        self.create_server_account()

        # Ensure administrator `Account` exists
        self.create_administrator_account(token_plex=token_plex)

        # Refresh extra accounts
        accounts = Account.select().where(
            Account.id > 1,
            Account.deleted == False
        )

        for account in accounts:
            self.refresh_account(account)

        return True

    @classmethod
    def create_server_account(cls):
        try:
            Account.get(Account.id == 0)
        except Account.DoesNotExist:
            Account.create(
                id=0,
                name=''
            )

    @classmethod
    def create_administrator_account(cls, token_plex=None):
        username = cls.get_trakt_username()

        try:
            account = Account.get(Account.id == 1)
        except Account.DoesNotExist:
            account = Account.create(
                id=1,
                name=username
            )

            # Create default rules for account
            cls.create_rules(account)

        # Ensure plex account details exist
        p_created, p_account = cls.create_plex_account(account)

        cls.create_plex_basic_credential(p_account, token_plex=token_plex)

        # Refresh plex account details
        try:
            p_refreshed = p_account.refresh(force=p_created)
        except:
            log.warn('Unable to refresh plex account (not authenticated?)', exc_info=True)
            p_refreshed = False

        # Ensure trakt account details exist
        t_created, t_account = cls.create_trakt_account(account, username)

        cls.create_trakt_basic_credential(t_account)
        cls.create_trakt_oauth_credential(t_account)

        # Refresh trakt account details
        try:
            t_refreshed = t_account.refresh(force=t_created)
        except:
            log.warn('Unable to refresh trakt account (not authenticated?)', exc_info=True)
            t_refreshed = False

        # Refresh account
        account.refresh(force=p_refreshed or t_refreshed)

    @classmethod
    def refresh_account(cls, account):
        if not account or account.deleted:
            return

        log.debug('Refreshing account: %r', account)

        # Refresh plex account details
        p_account = account.plex
        p_refreshed = False

        if p_account:
            try:
                p_refreshed = p_account.refresh()
            except:
                log.info('Unable to refresh plex account (not authenticated?)', exc_info=True)
                p_refreshed = False

        # Refresh trakt account details
        t_account = account.trakt
        t_refreshed = False

        if t_account:
            try:
                t_refreshed = t_account.refresh()
            except:
                log.info('Unable to refresh trakt account (not authenticated?)', exc_info=True)
                t_refreshed = False

        # Refresh account
        account.refresh(force=p_refreshed or t_refreshed)

    @classmethod
    def create_rules(cls, account):
        ClientRule.create(account=account, priority=1)
        UserRule.create(account=account, priority=1)

    #
    # Plex
    #

    @classmethod
    def create_plex_account(cls, account):
        try:
            return True, PlexAccount.create(
                account=account
            )
        except (apsw.ConstraintError, peewee.IntegrityError):
            return False, PlexAccount.get(
                account=account
            )

    @classmethod
    def create_plex_basic_credential(cls, plex_account, token_plex=None):
        if token_plex is None:
            token_plex = cls.get_token()

        if not token_plex:
            log.warn('No plex token available, unable to authenticate plex account')
            return False

        try:
            PlexBasicCredential.create(
                account=plex_account,

                token_plex=token_plex
            )
        except (apsw.ConstraintError, peewee.IntegrityError), ex:
            # Ensure basic credential has a token
            rows_updated = PlexBasicCredential.update(
                token_plex=token_plex,
                token_server=None
            ).where(
                PlexBasicCredential.account == plex_account,
                PlexBasicCredential.token_plex != token_plex
            ).execute()

            # Check if basic credential was updated
            if rows_updated:
                return True

            log.debug('Ignoring basic credential update for %r, already exists (%s)', plex_account, ex)
            return False

        return True

    #
    # Trakt
    #

    @classmethod
    def create_trakt_account(cls, account, username):
        try:
            return True, TraktAccount.create(
                account=account,
                username=username
            )
        except (apsw.ConstraintError, peewee.IntegrityError):
            return False, TraktAccount.get(
                account=account
            )

    @classmethod
    def create_trakt_basic_credential(cls, trakt_account):
        if not Environment.dict['trakt.token']:
            return False

        try:
            TraktBasicCredential.create(
                account=trakt_account,
                password=Environment.get_pref('password'),

                token=Environment.dict['trakt.token']
            )
        except (apsw.ConstraintError, peewee.IntegrityError), ex:
            log.debug('Ignoring basic credential update for %r, already exists (%s)', trakt_account, ex)
            return False

        return True

    @classmethod
    def create_trakt_oauth_credential(cls, trakt_account):
        if not Environment.dict['trakt.pin.code'] or not Environment.dict['trakt.pin.authorization']:
            return False

        try:
            TraktOAuthCredential.create(
                account=trakt_account,
                code=Environment.dict['trakt.pin.code'],

                **Environment.dict['trakt.pin.authorization']
            )
        except (apsw.ConstraintError, peewee.IntegrityError), ex:
            log.debug('Ignoring oauth credential update for %r, already exists (%s)', trakt_account, ex)
            return False

        return True

    @classmethod
    def get_trakt_username(cls):
        if Environment.get_pref('username'):
            return Environment.get_pref('username')

        if Environment.dict['trakt.username']:
            return Environment.dict['trakt.username']

        return None

    @classmethod
    def get_token(cls, request_headers=None):
        # Environment token
        env_token = os.environ.get('PLEXTOKEN')

        if env_token:
            log.info('Plex Token: environment')
            return env_token

        # Check if anonymous access is available
        server = requests.get('http://localhost:32400')

        if server.status_code == 200:
            log.info('Plex Token: anonymous')
            return 'anonymous'

        # No token available
        if request_headers is None:
            log.error('Plex Token: not available')
            return None

        # Try retrieve token from request
        req_token = request_headers.get('X-Plex-Token')

        if req_token:
            log.info('Plex Token: request')
            return req_token

        # No token available in request
        data = {
            'Client': {
                'User-Agent': request_headers.get('User-Agent'),
                'X-Plex-Product': request_headers.get('X-Plex-Product'),
            },
            'Headers': request_headers.keys()
        }

        log.debug('Request details: %r', data)
        log.error('Plex Token: not available', extra={
            'data': data
        })
        return None


1	1	from plugin.core.environment import Environment
2	1	from plugin.models import (
3		Account, ClientRule, UserRule,
4		PlexAccount, PlexBasicCredential,
5		TraktAccount, TraktBasicCredential, TraktOAuthCredential
6		)
7	1	from plugin.modules.migrations.core.base import Migration
8
9	1	import apsw
10	1	import logging
11	1	import os
12	1	import peewee
13	1	import requests
14
15	1	log = logging.getLogger(__name__)
16
17
18	1	class AccountMigration(Migration):
19	1	def run(self, token_plex=None):
20		# Ensure server `Account` exists
21		self.create_server_account()
22
23		# Ensure administrator `Account` exists
24		self.create_administrator_account(token_plex=token_plex)
25
26		# Refresh extra accounts
27		accounts = Account.select().where(
28		Account.id > 1,
29		Account.deleted == False
30		)
31
32		for account in accounts:
33		self.refresh_account(account)
34
35		return True
36
37	1	@classmethod
38		def create_server_account(cls):
39		try:
40		Account.get(Account.id == 0)
41		except Account.DoesNotExist:
42		Account.create(
43		id=0,
44		name=''
45		)
46
47	1	@classmethod
48	1	def create_administrator_account(cls, token_plex=None):
49		username = cls.get_trakt_username()
50
51		try:
52		account = Account.get(Account.id == 1)
53		except Account.DoesNotExist:
54		account = Account.create(
55		id=1,
56		name=username
57		)
58
59		# Create default rules for account
60		cls.create_rules(account)
61
62		# Ensure plex account details exist
63		p_created, p_account = cls.create_plex_account(account)
64
65		cls.create_plex_basic_credential(p_account, token_plex=token_plex)
66
67		# Refresh plex account details
68		try:
69		p_refreshed = p_account.refresh(force=p_created)
70		except:
71		log.warn('Unable to refresh plex account (not authenticated?)', exc_info=True)
72		p_refreshed = False
73
74		# Ensure trakt account details exist
75		t_created, t_account = cls.create_trakt_account(account, username)
76
77		cls.create_trakt_basic_credential(t_account)
78		cls.create_trakt_oauth_credential(t_account)
79
80		# Refresh trakt account details
81		try:
82		t_refreshed = t_account.refresh(force=t_created)
83		except:
84		log.warn('Unable to refresh trakt account (not authenticated?)', exc_info=True)
85		t_refreshed = False
86
87		# Refresh account
88		account.refresh(force=p_refreshed or t_refreshed)
89
90	1	@classmethod
91		def refresh_account(cls, account):
92		if not account or account.deleted:
93		return
94
95		log.debug('Refreshing account: %r', account)
96
97		# Refresh plex account details
98		p_account = account.plex
99		p_refreshed = False
100
101		if p_account:
102		try:
103		p_refreshed = p_account.refresh()
104		except:
105		log.info('Unable to refresh plex account (not authenticated?)', exc_info=True)
106		p_refreshed = False
107
108		# Refresh trakt account details
109		t_account = account.trakt
110		t_refreshed = False
111
112		if t_account:
113		try:
114		t_refreshed = t_account.refresh()
115		except:
116		log.info('Unable to refresh trakt account (not authenticated?)', exc_info=True)
117		t_refreshed = False
118
119		# Refresh account
120		account.refresh(force=p_refreshed or t_refreshed)
121
122	1	@classmethod
123		def create_rules(cls, account):
124		ClientRule.create(account=account, priority=1)
125		UserRule.create(account=account, priority=1)
126
127		#
128		# Plex
129		#
130
131	1	@classmethod
132		def create_plex_account(cls, account):
133		try:
134		return True, PlexAccount.create(
135		account=account
136		)
137		except (apsw.ConstraintError, peewee.IntegrityError):
138		return False, PlexAccount.get(
139		account=account
140		)
141
142	1	@classmethod
143	1	def create_plex_basic_credential(cls, plex_account, token_plex=None):
144		if token_plex is None:
145		token_plex = cls.get_token()
146
147		if not token_plex:
148		log.warn('No plex token available, unable to authenticate plex account')
149		return False
150
151		try:
152		PlexBasicCredential.create(
153		account=plex_account,
154
155		token_plex=token_plex
156		)
157		except (apsw.ConstraintError, peewee.IntegrityError), ex:
158		# Ensure basic credential has a token
159		rows_updated = PlexBasicCredential.update(
160		token_plex=token_plex,
161		token_server=None
162		).where(
163		PlexBasicCredential.account == plex_account,
164		PlexBasicCredential.token_plex != token_plex
165		).execute()
166
167		# Check if basic credential was updated
168		if rows_updated:
169		return True
170
171		log.debug('Ignoring basic credential update for %r, already exists (%s)', plex_account, ex)
172		return False
173
174		return True
175
176		#
177		# Trakt
178		#
179
180	1	@classmethod
181		def create_trakt_account(cls, account, username):
182		try:
183		return True, TraktAccount.create(
184		account=account,
185		username=username
186		)
187		except (apsw.ConstraintError, peewee.IntegrityError):
188		return False, TraktAccount.get(
189		account=account
190		)
191
192	1	@classmethod
193		def create_trakt_basic_credential(cls, trakt_account):
194		if not Environment.dict['trakt.token']:
195		return False
196
197		try:
198		TraktBasicCredential.create(
199		account=trakt_account,
200		password=Environment.get_pref('password'),
201
202		token=Environment.dict['trakt.token']
203		)
204		except (apsw.ConstraintError, peewee.IntegrityError), ex:
205		log.debug('Ignoring basic credential update for %r, already exists (%s)', trakt_account, ex)
206		return False
207
208		return True
209
210	1	@classmethod
211		def create_trakt_oauth_credential(cls, trakt_account):
212		if not Environment.dict['trakt.pin.code'] or not Environment.dict['trakt.pin.authorization']:
213		return False
214
215		try:
216		TraktOAuthCredential.create(
217		account=trakt_account,
218		code=Environment.dict['trakt.pin.code'],
219
220		**Environment.dict['trakt.pin.authorization']
221		)
222		except (apsw.ConstraintError, peewee.IntegrityError), ex:
223		log.debug('Ignoring oauth credential update for %r, already exists (%s)', trakt_account, ex)
224		return False
225
226		return True
227
228	1	@classmethod
229		def get_trakt_username(cls):
230		if Environment.get_pref('username'):
231		return Environment.get_pref('username')
232
233		if Environment.dict['trakt.username']:
234		return Environment.dict['trakt.username']
235
236		return None
237
238	1	@classmethod
239	1	def get_token(cls, request_headers=None):
240		# Environment token
241		env_token = os.environ.get('PLEXTOKEN')
242
243		if env_token:
244		log.info('Plex Token: environment')
245		return env_token
246
247		# Check if anonymous access is available
248		server = requests.get('http://localhost:32400')
249
250		if server.status_code == 200:
251		log.info('Plex Token: anonymous')
252		return 'anonymous'
253
254		# No token available
255		if request_headers is None:
256		log.error('Plex Token: not available')
257		return None
258
259		# Try retrieve token from request
260		req_token = request_headers.get('X-Plex-Token')
261
262		if req_token:
263		log.info('Plex Token: request')
264		return req_token
265
266		# No token available in request
267		data = {
268		'Client': {
269		'User-Agent': request_headers.get('User-Agent'),
270		'X-Plex-Product': request_headers.get('X-Plex-Product'),
271		},
272		'Headers': request_headers.keys()
273		}
274
275		log.debug('Request details: %r', data)
276		log.error('Plex Token: not available', extra={
277		'data': data
278		})
279		return None
280

fuzeman / Plex-Trakt-Scrobbler

Push — develop ( 456c9f...9a88f4 )

AccountMigration.get_trakt_username() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like