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

AccountMigration.get_trakt_username() A
last analyzed 2018-03-05 01:11 UTC

↳ Parent: AccountMigration

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	8.2077

Importance

Changes

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

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

from exception_wrappers.libraries 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) as 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) as 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) as 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	from exception_wrappers.libraries 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):
		0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Arguments number differs from overridden 'run' method Loading history...
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,
		0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The Class `Account` does not seem to have a member named `id`. 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...
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)
		0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The Class `Account` does not seem to have a member named `id`. 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...
41		except Account.DoesNotExist:
		0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The Class `Account` does not seem to have a member named `DoesNotExist`. 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...
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)
		0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The Class `Account` does not seem to have a member named `id`. 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...
53		except Account.DoesNotExist:
		0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The Class `Account` does not seem to have a member named `DoesNotExist`. 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...
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:
		0 ignored issues – show Coding Style Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report General except handlers without types should be used sparingly. Typically, you would use general except handlers when you intend to specifically handle all types of errors, f.e. when logging. Otherwise, such general error handlers can mask errors in your application that you want to know of. Loading history...
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:
		0 ignored issues – show Coding Style Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report General except handlers without types should be used sparingly. Typically, you would use general except handlers when you intend to specifically handle all types of errors, f.e. when logging. Otherwise, such general error handlers can mask errors in your application that you want to know of. Loading history...
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:
		0 ignored issues – show Coding Style Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report General except handlers without types should be used sparingly. Typically, you would use general except handlers when you intend to specifically handle all types of errors, f.e. when logging. Otherwise, such general error handlers can mask errors in your application that you want to know of. Loading history...
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:
		0 ignored issues – show Coding Style Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report General except handlers without types should be used sparingly. Typically, you would use general except handlers when you intend to specifically handle all types of errors, f.e. when logging. Otherwise, such general error handlers can mask errors in your application that you want to know of. Loading history...
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) as 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) as 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) as 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

AccountMigration.get_trakt_username() A last analyzed 2018-03-05 01:11 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

AccountMigration.get_trakt_username() A
last analyzed 2018-03-05 01:11 UTC