ServiceWallabag._new_token() - Code Metrics - Inspection of "simplified th_wallabag" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 001a1c...67edad )

by Fox

created 2017-09-25 18:38 UTC

ServiceWallabag._new_token() B

↳ Parent: ServiceWallabag

Complexity

Conditions

Size

Total Lines

Duplication

Lines	21
Ratio	61.76 %

Importance

Changes	4
Bugs	1	Features	0

Metric	Value
cc	3
dl	21
loc	34
rs	8.8571
c	4
b	1
f	0

# coding: utf-8
# add here the call of any native lib of python like datetime etc.
import arrow
import requests
from requests import HTTPError
# add the python API here if needed
from wallabag_api.wallabag import Wallabag as Wall
# django classes
from django.core.urlresolvers import reverse
from logging import getLogger
from django.core.cache import caches

# django_th classes
from django_th.services.services import ServicesMgr
from django_th.html_entities import HtmlEntities
from django_th.models import UserService, ServicesActivated, update_result
from th_wallabag.models import Wallabag

"""
    handle process with wallabag
    put the following in settings.py

    TH_SERVICES = (
        ...
        'th_wallabag.my_wallabag.ServiceWallabag',
        ...
    )
"""

logger = getLogger('django_th.trigger_happy')

cache = caches['th_wallabag']


class ServiceWallabag(ServicesMgr):
    """
        Service Wallabag
    """
    def __init__(self, token=None, **kwargs):
        super(ServiceWallabag, self).__init__(token, **kwargs)
        self.token = token
        self.user = kwargs.get('user')

    def _get_wall_data(self):
        us = UserService.objects.get(user=self.user, name='ServiceWallabag')

        params = dict({'access_token': self.token,
                       'archive': 0,
                       'star': 0,
                       'delete': 0,
                       'sort': 'created',
                       'order': 'desc',
                       'page': 1,
                       'perPage': 30,
                       'tags': []})

        responses = requests.get(us.host + '/api/entries.json',
                                 params=params)
        if responses.status_code == 401:
            params['access_token'] = Wall.get_token(host=us.host, **params)
            responses = requests.get(us.host + '/api/entries.json',
                                     params=params)
        elif responses.status_code != 200:
            raise HTTPError(responses.status_code, responses.json())

        return responses

    def read_data(self, **kwargs):
        """
            get the data from the service
            as the pocket service does not have any date
            in its API linked to the note,
            add the triggered date to the dict data
            thus the service will be triggered when data will be found

            :param kwargs: contain keyword args : trigger_id at least
            :type kwargs: dict

            :rtype: list
        """
        self.date_triggered = arrow.get(kwargs.get('date_triggered'))
        self.trigger_id = kwargs.get('trigger_id')
        self.user = kwargs.get('user', '')

        responses = self._get_wall_data()

        data = []
        try:
            json_data = responses.json()

            for d in json_data['_embedded']['items']:
                created_at = arrow.get(d.get('created_at'))
                date_triggered = arrow.get(self.date_triggered)

                if created_at > date_triggered:
                    data.append({'title': d.get('title'),
                                 'content': d.get('content')})
                    # digester
                    self.send_digest_event(self.trigger_id,
                                           d.get('title'),
                                           link='')
            if len(data) > 0:
                cache.set('th_wallabag_' + str(self.trigger_id), data)
        except Exception as e:
                logger.critical(e)
                update_result(self.trigger_id, msg=e, status=False)
        return data

    def wall(self):
        """
            refresh the token from the API
            then call a Wallabag instance
            then store the token
        :return: wall instance
        """
        us = UserService.objects.get(user=self.user, name='ServiceWallabag')
        params = {
            'client_id': us.client_id,
            'client_secret': us.client_secret,
            'username': us.username,
            'password': us.password,
        }
        try:
            token = Wall.get_token(host=us.host, **params)
        except Exception as e:
            update_result(self.trigger_id, msg=e, status=False)
            logger.critical('{} {}'.format(self.user, e))
            return False

        wall = Wall(host=us.host, client_secret=us.client_secret,
                    client_id=us.client_id, token=token)

        UserService.objects.filter(user=self.user,
                                   name='ServiceWallabag').update(token=token)

        return wall

    def _create_entry(self, title, data, tags):
        """
            create an entry
            :param title: string
            :param data: dict
            :param tags: list
            :return: boolean
        """
        status = False
        if data.get('link') and len(data.get('link')) > 0:
            wall = self.wall()
            if wall is not False:
                try:
                    wall.post_entries(url=data.get('link').encode(),
                                      title=title,
                                      tags=(tags.lower()))
                    logger.debug('wallabag {} created'.format(data.get('link')))
                    status = True
                except Exception as e:
                    logger.critical('issue with something else that a token'
                                    ' link ? : {}'.format(data.get('link')))
                    logger.critical(e)
                    update_result(self.trigger_id, msg=e, status=False)
                    status = False
        else:
            status = True  # we ignore empty link
        return status

    def save_data(self, trigger_id, **data):
        """
            let's save the data

            :param trigger_id: trigger ID from which to save data
            :param data: the data to check to be used and save
            :type trigger_id: int
            :type data:  dict
            :return: the status of the save statement
            :rtype: boolean
        """
        self.trigger_id = trigger_id

        trigger = Wallabag.objects.get(trigger_id=trigger_id)

        title = self.set_title(data)
        if title is not None:
            # convert htmlentities
            title = HtmlEntities(title).html_entity_decode

            return self._create_entry(title, data, trigger.tag)
        else:
            # we ignore data without title so return True to let
            # the process continue without
            # raising exception
            return True

    def auth(self, request):
        """
            let's auth the user to the Service
            :param request: request object
            :return: callback url
            :rtype: string that contains the url to redirect after auth

        """
        service = UserService.objects.get(
            user=request.user, name='ServiceWallabag')
        callback_url = 'http://%s%s' % (
            request.get_host(), reverse('wallabag_callback'))
        params = {'username': service.username,
                  'password': service.password,
                  'client_id': service.client_id,
                  'client_secret': service.client_secret}
        access_token = Wall.get_token(host=service.host, **params)
        request.session['oauth_token'] = access_token
        return callback_url

    def callback(self, request, **kwargs):
        """
            Called from the Service when the user accept to activate it
            :param request: request object
            :return: callback url
            :rtype: string , path to the template
        """

        try:
            UserService.objects.filter(
                user=request.user,
                name=ServicesActivated.objects.get(name='ServiceWallabag')
            )
        except KeyError:
            return '/'

        return 'wallabag/callback.html'


1			# coding: utf-8
2			# add here the call of any native lib of python like datetime etc.
3			import arrow
4			import requests
5			from requests import HTTPError
6			# add the python API here if needed
7			from wallabag_api.wallabag import Wallabag as Wall
8			# django classes
9			from django.core.urlresolvers import reverse
10			from logging import getLogger
11			from django.core.cache import caches
12
13			# django_th classes
14			from django_th.services.services import ServicesMgr
15			from django_th.html_entities import HtmlEntities
16			from django_th.models import UserService, ServicesActivated, update_result
17			from th_wallabag.models import Wallabag
18
19			"""
20			handle process with wallabag
21			put the following in settings.py
22
23			TH_SERVICES = (
24			...
25			'th_wallabag.my_wallabag.ServiceWallabag',
26			...
27			)
28			"""
29
30			logger = getLogger('django_th.trigger_happy')
31
32			cache = caches['th_wallabag']
33
34
35			class ServiceWallabag(ServicesMgr):
36			"""
37			Service Wallabag
38			"""
39			def __init__(self, token=None, **kwargs):
40			super(ServiceWallabag, self).__init__(token, **kwargs)
41			self.token = token
42			self.user = kwargs.get('user')
43
44			def _get_wall_data(self):
45			us = UserService.objects.get(user=self.user, name='ServiceWallabag')
46
47			params = dict({'access_token': self.token,
48			'archive': 0,
49			'star': 0,
50			'delete': 0,
51			'sort': 'created',
52			'order': 'desc',
53			'page': 1,
54			'perPage': 30,
55			'tags': []})
56
57			responses = requests.get(us.host + '/api/entries.json',
58			params=params)
59			if responses.status_code == 401:
60			params['access_token'] = Wall.get_token(host=us.host, **params)
61			responses = requests.get(us.host + '/api/entries.json',
62			params=params)
63			elif responses.status_code != 200:
64			raise HTTPError(responses.status_code, responses.json())
65
66			return responses
67
68			def read_data(self, **kwargs):
69			"""
70			get the data from the service
71			as the pocket service does not have any date
72			in its API linked to the note,
73			add the triggered date to the dict data
74			thus the service will be triggered when data will be found
75
76			:param kwargs: contain keyword args : trigger_id at least
77			:type kwargs: dict
78
79			:rtype: list
80			"""
81			self.date_triggered = arrow.get(kwargs.get('date_triggered'))
82			self.trigger_id = kwargs.get('trigger_id')
83			self.user = kwargs.get('user', '')
84
85			responses = self._get_wall_data()
86
87			data = []
88			try:
89			json_data = responses.json()
90
91			for d in json_data['_embedded']['items']:
92			created_at = arrow.get(d.get('created_at'))
93			date_triggered = arrow.get(self.date_triggered)
94
95			if created_at > date_triggered:
96			data.append({'title': d.get('title'),
97			'content': d.get('content')})
98			# digester
99			self.send_digest_event(self.trigger_id,
100			d.get('title'),
101			link='')
102			if len(data) > 0:
103			cache.set('th_wallabag_' + str(self.trigger_id), data)
104			except Exception as e:
105			logger.critical(e)
106			update_result(self.trigger_id, msg=e, status=False)
107			return data
108
109			def wall(self):
110			"""
111			refresh the token from the API
112			then call a Wallabag instance
113			then store the token
114			:return: wall instance
115			"""
116			us = UserService.objects.get(user=self.user, name='ServiceWallabag')
117			params = {
118			'client_id': us.client_id,
119			'client_secret': us.client_secret,
120			'username': us.username,
121			'password': us.password,
122			}
123			try:
124			token = Wall.get_token(host=us.host, **params)
125			except Exception as e:
126			update_result(self.trigger_id, msg=e, status=False)
127			logger.critical('{} {}'.format(self.user, e))
128			return False
129
130			wall = Wall(host=us.host, client_secret=us.client_secret,
131			client_id=us.client_id, token=token)
132
133			UserService.objects.filter(user=self.user,
134			name='ServiceWallabag').update(token=token)
135
136			return wall
137
138			def _create_entry(self, title, data, tags):
139			"""
140			create an entry
141			:param title: string
142			:param data: dict
143			:param tags: list
144			:return: boolean
145			"""
146			status = False
147			if data.get('link') and len(data.get('link')) > 0:
148			wall = self.wall()
149			if wall is not False:
150			try:
151			wall.post_entries(url=data.get('link').encode(),
152			title=title,
153			tags=(tags.lower()))
154			logger.debug('wallabag {} created'.format(data.get('link')))
155			status = True
156			except Exception as e:
157			logger.critical('issue with something else that a token'
158			' link ? : {}'.format(data.get('link')))
159			logger.critical(e)
160			update_result(self.trigger_id, msg=e, status=False)
161			status = False
162			else:
163			status = True # we ignore empty link
164			return status
165
166			def save_data(self, trigger_id, **data):
167			"""
168			let's save the data
169
170			:param trigger_id: trigger ID from which to save data
171			:param data: the data to check to be used and save
172			:type trigger_id: int
173			:type data: dict
174			:return: the status of the save statement
175			:rtype: boolean
176			"""
177			self.trigger_id = trigger_id
178
179			trigger = Wallabag.objects.get(trigger_id=trigger_id)
180
181			title = self.set_title(data)
182			if title is not None:
183			# convert htmlentities
184			title = HtmlEntities(title).html_entity_decode
185
186			return self._create_entry(title, data, trigger.tag)
187			else:
188			# we ignore data without title so return True to let
189			# the process continue without
190			# raising exception
191			return True
192
193			def auth(self, request):
194			"""
195			let's auth the user to the Service
196			:param request: request object
197			:return: callback url
198			:rtype: string that contains the url to redirect after auth
199
200			"""
201			service = UserService.objects.get(
202			user=request.user, name='ServiceWallabag')
203			callback_url = 'http://%s%s' % (
204			request.get_host(), reverse('wallabag_callback'))
205			params = {'username': service.username,
206			'password': service.password,
207			'client_id': service.client_id,
208			'client_secret': service.client_secret}
209			access_token = Wall.get_token(host=service.host, **params)
210			request.session['oauth_token'] = access_token
211			return callback_url
212
213			def callback(self, request, **kwargs):
214			"""
215			Called from the Service when the user accept to activate it
216			:param request: request object
217			:return: callback url
218			:rtype: string , path to the template
219			"""
220
221			try:
222			UserService.objects.filter(
223			user=request.user,
224			name=ServicesActivated.objects.get(name='ServiceWallabag')
225			)
226			except KeyError:
227			return '/'
228
229			return 'wallabag/callback.html'
230

foxmask / django-th

Push — master ( 001a1c...67edad )

ServiceWallabag._new_token() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like