ServiceWallabag._new_token() - Code Metrics - Inspection of "wallabag now can provide data too" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 87b3e4...a9f319 )

by Fox

created 2016-06-28 19:40 UTC

ServiceWallabag._new_token() A

↳ Parent: ServiceWallabag

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	2
c	2
b	0
f	0
dl	0
loc	21
rs	9.3142

# 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 django.utils.log 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
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):

    def __init__(self, token=None, **kwargs):
        super(ServiceWallabag, self).__init__(token, **kwargs)
        self.token = token
        self.user = kwargs.get('user')

    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 = kwargs.get('date_triggered')
        self.trigger_id = kwargs.get('trigger_id')
        self.user = kwargs.get('user') if kwargs.get('user') else ''

        us = UserService.objects.get(token=self.token, 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'] = self._refresh_token()
            responses = requests.get(us.host + '/api/entries.json',
                                     params=params)
        elif responses.status_code != 200:
            raise HTTPError(responses.status_code, responses.json())

        json_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')})
            if len(data) > 0:
                cache.set('th_wallabag_' + str(self.trigger_id), data)
        except Exception as e:
            if json_data.get('errors'):
                for error in json_data['errors']:
                    error_json = json_data['errors'][error]['content']
                    logger.critical("Wallabag: {error}".format(
                        error=error_json))
                    logger.critical(e)
        return data

    def new_wall(self, token):
        """
            produce a new wall instance
            :param token: to get
            :return: Wall instance
        """
        if token:
            try:
                us = UserService.objects.get(
                    token=self.token, name='ServiceWallabag')
            except UserService.DoesNotExist:
                us = UserService.objects.get(
                    user=self.user, name='ServiceWallabag')
            finally:
                return Wall(host=us.host, client_secret=us.client_secret,
                            client_id=us.client_id, token=us.token)

    def _create_entry(self, title, data, tags):
        """
            create an entry
            :param title: string
            :param data: dict
            :param tags: list
            :return: boolean
        """
        if data.get('link') and len(data.get('link')) > 0:
            wall = self.new_wall(self.token)
            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:
                if e.errno == 401:
                    status = self._new_token(data.get('userservice_id'),
                                             data.get('link').encode(),
                                             title,
                                             tags.lower())
                else:
                    logger.critical('issue with something else that a token'
                                    ' link ? : {}'.format(data.get('link')))
                    logger.critical(e.errno, e.strerror)
                    status = False
        else:
            status = True  # we ignore empty link
        return status

    def _refresh_token(self):
        """
            refresh the expired token
            get the token of the service Wallabag
            for the user that uses Wallabag
            :return: boolean
        """
        us = UserService.objects.get(user=self.user, name='ServiceWallabag')
        params = {'username': us.username,
                  'password': us.password,
                  'client_id': us.client_id,
                  'client_secret': us.client_secret}
        return Wall.get_token(host=us.host, **params)

    def _new_token(self, userservice_id, link, title, tags):
        """
            create a new token
            :param userservice_id: id of the UserService
            :param link: string
            :param title: string
            :param tags: list
            :return: boolean
        """
        new_token = self._refresh_token()
        logger.info('new token : {}'.format(new_token))
        UserService.objects.filter(id=userservice_id).update(token=new_token)

        new_wall = self.new_wall(new_token)
        try:
            status = new_wall.post_entries(url=link, title=title, tags=tags)
        except Exception as e:
            logger.critical('could not create a post link ? : {}'.format(link))
            logger.critical(e)
            status = False
        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
        """
        trigger = Wallabag.objects.get(trigger_id=trigger_id)

        title = self.set_title(data)
        # convert htmlentities
        title = HtmlEntities(title).html_entity_decode

        return self._create_entry(title, data, trigger.tag)

    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 django.utils.log 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
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			def __init__(self, token=None, **kwargs):
38			super(ServiceWallabag, self).__init__(token, **kwargs)
39			self.token = token
40			self.user = kwargs.get('user')
41
42			def read_data(self, **kwargs):
43			"""
44			get the data from the service
45			as the pocket service does not have any date
46			in its API linked to the note,
47			add the triggered date to the dict data
48			thus the service will be triggered when data will be found
49
50			:param kwargs: contain keyword args : trigger_id at least
51			:type kwargs: dict
52
53			:rtype: list
54			"""
55			self.date_triggered = kwargs.get('date_triggered')
56			self.trigger_id = kwargs.get('trigger_id')
57			self.user = kwargs.get('user') if kwargs.get('user') else ''
58
59			us = UserService.objects.get(token=self.token, name='ServiceWallabag')
60
61			params = dict({'access_token': self.token,
62			'archive': 0,
63			'star': 0,
64			'delete': 0,
65			'sort': 'created',
66			'order': 'desc',
67			'page': 1,
68			'perPage': 30,
69			'tags': []})
70
71			responses = requests.get(us.host + '/api/entries.json',
72			params=params)
73			if responses.status_code == 401:
74			params['access_token'] = self._refresh_token()
75			responses = requests.get(us.host + '/api/entries.json',
76			params=params)
77			elif responses.status_code != 200:
78			raise HTTPError(responses.status_code, responses.json())
79
80			json_data = {}
81			data = []
82			try:
83			json_data = responses.json()
84
85			for d in json_data['_embedded']['items']:
86			created_at = arrow.get(d.get('created_at'))
87			date_triggered = arrow.get(self.date_triggered)
88
89			if created_at > date_triggered:
90			data.append({'title': d.get('title'),
91			'content': d.get('content')})
92			if len(data) > 0:
93			cache.set('th_wallabag_' + str(self.trigger_id), data)
94			except Exception as e:
95			if json_data.get('errors'):
96			for error in json_data['errors']:
97			error_json = json_data['errors'][error]['content']
98			logger.critical("Wallabag: {error}".format(
99			error=error_json))
100			logger.critical(e)
101			return data
102
103			def new_wall(self, token):
104			"""
105			produce a new wall instance
106			:param token: to get
107			:return: Wall instance
108			"""
109			if token:
110			try:
111			us = UserService.objects.get(
112			token=self.token, name='ServiceWallabag')
113			except UserService.DoesNotExist:
114			us = UserService.objects.get(
115			user=self.user, name='ServiceWallabag')
116			finally:
117			return Wall(host=us.host, client_secret=us.client_secret,
118			client_id=us.client_id, token=us.token)
119
120			def _create_entry(self, title, data, tags):
121			"""
122			create an entry
123			:param title: string
124			:param data: dict
125			:param tags: list
126			:return: boolean
127			"""
128			if data.get('link') and len(data.get('link')) > 0:
129			wall = self.new_wall(self.token)
130			try:
131			wall.post_entries(url=data.get('link').encode(),
132			title=title,
133			tags=(tags.lower()))
134			logger.debug('wallabag {} created'.format(data.get('link')))
135			status = True
136			except Exception as e:
137			if e.errno == 401:
138			status = self._new_token(data.get('userservice_id'),
139			data.get('link').encode(),
140			title,
141			tags.lower())
142			else:
143			logger.critical('issue with something else that a token'
144			' link ? : {}'.format(data.get('link')))
145			logger.critical(e.errno, e.strerror)
146			status = False
147			else:
148			status = True # we ignore empty link
149			return status
150
151			def _refresh_token(self):
152			"""
153			refresh the expired token
154			get the token of the service Wallabag
155			for the user that uses Wallabag
156			:return: boolean
157			"""
158			us = UserService.objects.get(user=self.user, name='ServiceWallabag')
159			params = {'username': us.username,
160			'password': us.password,
161			'client_id': us.client_id,
162			'client_secret': us.client_secret}
163			return Wall.get_token(host=us.host, **params)
164
165			def _new_token(self, userservice_id, link, title, tags):
166			"""
167			create a new token
168			:param userservice_id: id of the UserService
169			:param link: string
170			:param title: string
171			:param tags: list
172			:return: boolean
173			"""
174			new_token = self._refresh_token()
175			logger.info('new token : {}'.format(new_token))
176			UserService.objects.filter(id=userservice_id).update(token=new_token)
177
178			new_wall = self.new_wall(new_token)
179			try:
180			status = new_wall.post_entries(url=link, title=title, tags=tags)
181			except Exception as e:
182			logger.critical('could not create a post link ? : {}'.format(link))
183			logger.critical(e)
184			status = False
185			return status
186
187			def save_data(self, trigger_id, **data):
188			"""
189			let's save the data
190
191			:param trigger_id: trigger ID from which to save data
192			:param data: the data to check to be used and save
193			:type trigger_id: int
194			:type data: dict
195			:return: the status of the save statement
196			:rtype: boolean
197			"""
198			trigger = Wallabag.objects.get(trigger_id=trigger_id)
199
200			title = self.set_title(data)
201			# convert htmlentities
202			title = HtmlEntities(title).html_entity_decode
203
204			return self._create_entry(title, data, trigger.tag)
205
206			def auth(self, request):
207			"""
208			let's auth the user to the Service
209			:param request: request object
210			:return: callback url
211			:rtype: string that contains the url to redirect after auth
212
213			"""
214			service = UserService.objects.get(
215			user=request.user, name='ServiceWallabag')
216			callback_url = 'http://%s%s' % (
217			request.get_host(), reverse('wallabag_callback'))
218			params = {'username': service.username,
219			'password': service.password,
220			'client_id': service.client_id,
221			'client_secret': service.client_secret}
222			access_token = Wall.get_token(host=service.host, **params)
223			request.session['oauth_token'] = access_token
224			return callback_url
225
226			def callback(self, request, **kwargs):
227			"""
228			Called from the Service when the user accept to activate it
229			:param request: request object
230			:return: callback url
231			:rtype: string , path to the template
232			"""
233
234			try:
235			UserService.objects.filter(
236			user=request.user,
237			name=ServicesActivated.objects.get(name='ServiceWallabag')
238			)
239			except KeyError:
240			return '/'
241
242			return 'wallabag/callback.html'
243

foxmask / django-th

Push — master ( 87b3e4...a9f319 )

ServiceWallabag._new_token() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like