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 ( a9f319...d94b17 )

by Fox

created 2016-06-28 19:57 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 _get_wall_data(self):
        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())

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

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

foxmask / django-th

Push — master ( a9f319...d94b17 )

ServiceWallabag._new_token() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like