ServiceWallabag.save_data() - Code Metrics - Inspection of "ignore article without title" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 31b96c...839eeb )

by Fox

created 2017-06-15 21:02 UTC

ServiceWallabag.save_data() B

↳ Parent: ServiceWallabag

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	9
Bugs	1	Features	0

Metric	Value
cc	2
dl	0
loc	24
rs	8.9713
c	9
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(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', '')

        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)
                update_result(self.trigger_id, msg=e, status=False)
        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)
                    update_result(self.trigger_id, msg=e, status=False)
                    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)
            update_result(self.trigger_id, msg=e, status=False)
            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)
        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(token=self.token, 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'] = self._refresh_token()
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			if len(data) > 0:
99			cache.set('th_wallabag_' + str(self.trigger_id), data)
100			except Exception as e:
101			logger.critical(e)
102			update_result(self.trigger_id, msg=e, status=False)
103			return data
104
105			def new_wall(self, token):
106			"""
107			produce a new wall instance
108			:param token: to get
109			:return: Wall instance
110			"""
111			if token:
112			try:
113			us = UserService.objects.get(
114			token=self.token, name='ServiceWallabag')
115			except UserService.DoesNotExist:
116			us = UserService.objects.get(
117			user=self.user, name='ServiceWallabag')
118			finally:
119			return Wall(host=us.host, client_secret=us.client_secret,
120			client_id=us.client_id, token=us.token)
121
122			def _create_entry(self, title, data, tags):
123			"""
124			create an entry
125			:param title: string
126			:param data: dict
127			:param tags: list
128			:return: boolean
129			"""
130			if data.get('link') and len(data.get('link')) > 0:
131			wall = self.new_wall(self.token)
132			try:
133			wall.post_entries(url=data.get('link').encode(),
134			title=title,
135			tags=(tags.lower()))
136			logger.debug('wallabag {} created'.format(data.get('link')))
137			status = True
138			except Exception as e:
139			if e.errno == 401:
140			status = self._new_token(data.get('userservice_id'),
141			data.get('link').encode(),
142			title,
143			tags.lower())
144			else:
145			logger.critical('issue with something else that a token'
146			' link ? : {}'.format(data.get('link')))
147			logger.critical(e.errno, e.strerror)
148			update_result(self.trigger_id, msg=e, status=False)
149			status = False
150			else:
151			status = True # we ignore empty link
152			return status
153
154			def _refresh_token(self):
155			"""
156			refresh the expired token
157			get the token of the service Wallabag
158			for the user that uses Wallabag
159			:return: boolean
160			"""
161			us = UserService.objects.get(user=self.user, name='ServiceWallabag')
162			params = {'username': us.username,
163			'password': us.password,
164			'client_id': us.client_id,
165			'client_secret': us.client_secret}
166			return Wall.get_token(host=us.host, **params)
167
168			def _new_token(self, userservice_id, link, title, tags):
169			"""
170			create a new token
171			:param userservice_id: id of the UserService
172			:param link: string
173			:param title: string
174			:param tags: list
175			:return: boolean
176			"""
177			new_token = self._refresh_token()
178			logger.info('new token : {}'.format(new_token))
179			UserService.objects.filter(id=userservice_id).update(token=new_token)
180
181			new_wall = self.new_wall(new_token)
182			try:
183			status = new_wall.post_entries(url=link, title=title, tags=tags)
184			except Exception as e:
185			logger.critical('could not create a post link ? : {}'.format(link))
186			logger.critical(e)
187			update_result(self.trigger_id, msg=e, status=False)
188			status = False
189			return status
190
191			def save_data(self, trigger_id, **data):
192			"""
193			let's save the data
194
195			:param trigger_id: trigger ID from which to save data
196			:param data: the data to check to be used and save
197			:type trigger_id: int
198			:type data: dict
199			:return: the status of the save statement
200			:rtype: boolean
201			"""
202			trigger = Wallabag.objects.get(trigger_id=trigger_id)
203
204			title = self.set_title(data)
205			if title is not None:
206			# convert htmlentities
207			title = HtmlEntities(title).html_entity_decode
208
209			return self._create_entry(title, data, trigger.tag)
210			else:
211			# we ignore data without title so return True to let
212			# the process continue without
213			# raising exception
214			return True
215
216			def auth(self, request):
217			"""
218			let's auth the user to the Service
219			:param request: request object
220			:return: callback url
221			:rtype: string that contains the url to redirect after auth
222
223			"""
224			service = UserService.objects.get(
225			user=request.user, name='ServiceWallabag')
226			callback_url = 'http://%s%s' % (
227			request.get_host(), reverse('wallabag_callback'))
228			params = {'username': service.username,
229			'password': service.password,
230			'client_id': service.client_id,
231			'client_secret': service.client_secret}
232			access_token = Wall.get_token(host=service.host, **params)
233			request.session['oauth_token'] = access_token
234			return callback_url
235
236			def callback(self, request, **kwargs):
237			"""
238			Called from the Service when the user accept to activate it
239			:param request: request object
240			:return: callback url
241			:rtype: string , path to the template
242			"""
243
244			try:
245			UserService.objects.filter(
246			user=request.user,
247			name=ServicesActivated.objects.get(name='ServiceWallabag')
248			)
249			except KeyError:
250			return '/'
251
252			return 'wallabag/callback.html'
253

foxmask / django-th

Push — master ( 31b96c...839eeb )

ServiceWallabag.save_data() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like