ServiceMastodon.callback_url() - Code Metrics - Inspection of "Merge pull request #204 from foxmask/mastodon" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( e7ede0...eadaef )

by Fox

created 2017-10-04 21:22 UTC

ServiceMastodon.callback_url() A

↳ Parent: ServiceMastodon

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	12
rs	9.4285

# coding: utf-8
import arrow
from logging import getLogger

from mastodon import Mastodon as MastodonAPI

# django classes
from django.conf import settings
from django.core.cache import caches
from django.shortcuts import reverse
from django.utils import html
from django.utils.translation import ugettext as _

# django_th classes
from django_th.services.services import ServicesMgr
from django_th.models import update_result, UserService
from th_mastodon.models import Mastodon


logger = getLogger('django_th.trigger_happy')

cache = caches['django_th']


class ServiceMastodon(ServicesMgr):
    """
        Service Mastodon
    """
    def __init__(self, token=None, **kwargs):
        super(ServiceMastodon, self).__init__(token, **kwargs)

        self.token = token
        self.service = 'ServiceMastodon'
        self.user = kwargs.get('user')

    def read_data(self, **kwargs):
        """
            get the data from the service

            :param kwargs: contain keyword args : trigger_id at least
            :type kwargs: dict
            :rtype: list
        """
        now = arrow.utcnow().to(settings.TIME_ZONE)
        my_toots = []
        search = {}
        since_id = None
        trigger_id = kwargs['trigger_id']
        date_triggered = arrow.get(kwargs['date_triggered'])

        def _get_toots(toot_api, toot_obj, search):
            """
                get the toots from mastodon and return the filters to use :
                search and count

                :param toot_obj: from Mastodon model
                :param search: filter used for MastodonAPI.search()
                :type toot_obj: Object
                :type search: dict
                :return: count that limit the quantity of tweet to retrieve,
                the filter named search, the tweets
                :rtype: list
            """

            # get the toots for a given tag
            statuses = ''
            count = 100
            if toot_obj.tag:
                count = 100
                search['count'] = count
                search['q'] = toot_obj.tag
                search['result_type'] = 'recent'
                # do a search
                statuses = toot_api.search(**search)
                # just return the content of te statuses array
                statuses = statuses['statuses']

            # get the tweets from a given user
            elif toot_obj.tooter:
                count = 200
                search['count'] = count
                search['username'] = toot_obj.tooter

                # call the user timeline and get his toot
                if toot_obj.fav:
                    count = 20
                    search['count'] = 20
                    statuses = toot_api.favourites(max_id=max_id,
                                                   since_id=since_id,
                                                   limit=count)
                else:
                    statuses = toot_api.account_statuses(
                        id=toot_obj.tooter,
                        max_id=max_id,
                        since_id=since_id,
                        limit=count)

            return count, search, statuses

        if self.token is not None:
            kw = {'app_label': 'th_mastodon', 'model_name': 'Mastodon',
                  'trigger_id': trigger_id}
            toot_obj = super(ServiceMastodon, self).read_data(**kw)

            us = UserService.objects.get(user=self.user,
                                         token=self.token,
                                         name='ServiceMastodon')
            try:
                toot_api = MastodonAPI(
                    client_id=us.client_id,
                    client_secret=us.client_secret,
                    access_token=self.token,
                    api_base_url=us.host
                )
            except ValueError as e:
                logger.error(e)
                update_result(trigger_id, msg=e, status=False)

            if toot_obj.since_id is not None and toot_obj.since_id > 0:
                since_id = toot_obj.since_id
                search = {'since_id': toot_obj.since_id}

            # first request to Mastodon
            count, search, statuses = _get_toots(toot_api, toot_obj, search)

            if len(statuses) > 0:
                newest = None
                for status in statuses:
                    if newest is None:
                        newest = True
                        # first query ; get the max id
                        search['max_id'] = max_id = status['id']

                since_id = search['since_id'] = statuses[-1]['id'] - 1

                count, search, statuses = _get_toots(toot_api, toot_obj,
                                                     search)

                newest = None
                if len(statuses) > 0:
                    my_toots = []
                    for s in statuses:
                        if newest is None:
                            newest = True
                            max_id = s['id'] - 1
                        toot_name = s['account']['username']
                        # get the text of the tweet + url to this one
                        if toot_obj.fav:
                            url = '{0}/api/v1/statuses/{1}'.format(
                                self.api_base_url, s['id'])
                            title = _('Toot Fav from @{}'.format(toot_name))
                        else:
                            url = '{0}/api/v1/accounts/{1}/statuses'.format(
                                self.api_base_url, s['id'])
                            title = _('Toot from @{}'.format(toot_name))
                        # Wed Aug 29 17:12:58 +0000 2012
                        my_date = arrow.get(s['created_at'],
                                            'ddd MMM DD HH:mm:ss Z YYYY')
                        published = arrow.get(my_date).to(settings.TIME_ZONE)
                        if date_triggered is not None and \
                           published is not None and \
                           now >= published >= date_triggered:
                            my_toots.append({'title': title,
                                             'content': s['content'],
                                             'link': url,
                                             'my_date': my_date})
                            # digester
                            self.send_digest_event(trigger_id, title, url)
                    cache.set('th_mastodon_' + str(trigger_id), my_toots)
                    Mastodon.objects.filter(trigger_id=trigger_id).update(
                        since_id=since_id,
                        max_id=max_id,
                        count=count)
        return my_toots

    def save_data(self, trigger_id, **data):
        """
            get the data from the service

            :param trigger_id: id of the trigger
            :params data, dict
            :rtype: dict
        """
        title, content = super(ServiceMastodon, self).save_data(
            trigger_id, **data)

        if self.title_or_content(title):

            content = str("{title} {link}").format(
                title=title, link=data.get('link'))

        content += self.get_tags(trigger_id)

        content = self.set_mastodon_content(content)

        us = UserService.objects.get(user=self.user,
                                     token=self.token,
                                     name='ServiceMastodon')

        try:
            toot_api = MastodonAPI(
                    client_id=us.client_id,
                    client_secret=us.client_secret,
                    access_token=self.token,
                    api_base_url=us.host
            )
        except ValueError as e:
            logger.error(e)
            update_result(trigger_id, msg=e, status=False)

        try:
            toot_api.toot(content)
            status = True
        except Exception as inst:
            logger.critical("Mastodon ERR {}".format(inst))
            update_result(trigger_id, msg=inst, status=False)
            status = False

        return status

    def get_tags(self, trigger_id):
        """
        get the tags if any
        :param trigger_id: the id of the related trigger
        :return: tags string
        """

        # get the Mastodon data of this trigger
        trigger = Mastodon.objects.get(trigger_id=trigger_id)

        tags = ''

        if trigger.tag is not None:
            # is there several tag ?
            tags = ["#" + tag.strip() for tag in trigger.tag.split(',')
                    ] if ',' in trigger.tag else "#" + trigger.tag

            tags = str(','.join(tags)) if isinstance(tags, list) else tags
            tags = ' ' + tags

        return tags

    def set_mastodon_content(self, content):
        """
        cleaning content by removing any existing html tag
        :param content:
        :return:
        """
        content = html.strip_tags(content)

        if len(content) > 560:
            return content[:560]

        return content

    def title_or_content(self, title):
        """
        If the title always contains 'New status from'
        drop the title and get 'the content' instead
        :param title:
        :return:
        """
        if "New status by" in title:
            return False
        return True

    def auth(self, request):
        """
            get the auth of the services
            :param request: contains the current session
            :type request: dict
            :rtype: dict
        """
        # create app
        redirect_uris = '%s://%s%s' % (request.scheme, request.get_host(),
                                       reverse('mastodon_callback'))
        us = UserService.objects.get(user=request.user,
                                     name='ServiceMastodon')
        client_id, client_secret = MastodonAPI.create_app(
            client_name="TriggerHappy", api_base_url=us.host,
            redirect_uris=redirect_uris)

        us.client_id = client_id
        us.client_secret = client_secret
        us.save()

        us = UserService.objects.get(user=request.user,
                                     name='ServiceMastodon')
        # get the token by logging in
        mastodon = MastodonAPI(
            client_id=client_id,
            client_secret=client_secret,
            api_base_url=us.host
        )
        token = mastodon.log_in(username=us.username, password=us.password)
        us.token = token
        us.save()
        return self.callback_url(request)

    def callback_url(self, request):
        us = UserService.objects.get(user=request.user,
                                     name='ServiceMastodon')
        mastodon = MastodonAPI(
            client_id=us.client_id,
            client_secret=us.client_secret,
            access_token=us.token,
            api_base_url=us.host
        )
        redirect_uris = '%s://%s%s' % (request.scheme, request.get_host(),
                                       reverse('mastodon_callback'))
        return mastodon.auth_request_url(redirect_uris=redirect_uris)

    def callback(self, request, **kwargs):
        """
            Called from the Service when the user accept to activate it
            the url to go back after the external service call
            :param request: contains the current session
            :param kwargs: keyword args
            :type request: dict
            :type kwargs: dict
            :rtype: string
        """
        return 'mastodon/callback.html'


1			# coding: utf-8
2			import arrow
3			from logging import getLogger
4
5			from mastodon import Mastodon as MastodonAPI
6
7			# django classes
8			from django.conf import settings
9			from django.core.cache import caches
10			from django.shortcuts import reverse
11			from django.utils import html
12			from django.utils.translation import ugettext as _
13
14			# django_th classes
15			from django_th.services.services import ServicesMgr
16			from django_th.models import update_result, UserService
17			from th_mastodon.models import Mastodon
18
19
20			logger = getLogger('django_th.trigger_happy')
21
22			cache = caches['django_th']
23
24
25			class ServiceMastodon(ServicesMgr):
26			"""
27			Service Mastodon
28			"""
29			def __init__(self, token=None, **kwargs):
30			super(ServiceMastodon, self).__init__(token, **kwargs)
31
32			self.token = token
33			self.service = 'ServiceMastodon'
34			self.user = kwargs.get('user')
35
36			def read_data(self, **kwargs):
37			"""
38			get the data from the service
39
40			:param kwargs: contain keyword args : trigger_id at least
41			:type kwargs: dict
42			:rtype: list
43			"""
44			now = arrow.utcnow().to(settings.TIME_ZONE)
45			my_toots = []
46			search = {}
47			since_id = None
48			trigger_id = kwargs['trigger_id']
49			date_triggered = arrow.get(kwargs['date_triggered'])
50
51			def _get_toots(toot_api, toot_obj, search):
52			"""
53			get the toots from mastodon and return the filters to use :
54			search and count
55
56			:param toot_obj: from Mastodon model
57			:param search: filter used for MastodonAPI.search()
58			:type toot_obj: Object
59			:type search: dict
60			:return: count that limit the quantity of tweet to retrieve,
61			the filter named search, the tweets
62			:rtype: list
63			"""
64
65			# get the toots for a given tag
66			statuses = ''
67			count = 100
68			if toot_obj.tag:
69			count = 100
70			search['count'] = count
71			search['q'] = toot_obj.tag
72			search['result_type'] = 'recent'
73			# do a search
74			statuses = toot_api.search(**search)
75			# just return the content of te statuses array
76			statuses = statuses['statuses']
77
78			# get the tweets from a given user
79			elif toot_obj.tooter:
80			count = 200
81			search['count'] = count
82			search['username'] = toot_obj.tooter
83
84			# call the user timeline and get his toot
85			if toot_obj.fav:
86			count = 20
87			search['count'] = 20
88			statuses = toot_api.favourites(max_id=max_id,
89			since_id=since_id,
90			limit=count)
91			else:
92			statuses = toot_api.account_statuses(
93			id=toot_obj.tooter,
94			max_id=max_id,
95			since_id=since_id,
96			limit=count)
97
98			return count, search, statuses
99
100			if self.token is not None:
101			kw = {'app_label': 'th_mastodon', 'model_name': 'Mastodon',
102			'trigger_id': trigger_id}
103			toot_obj = super(ServiceMastodon, self).read_data(**kw)
104
105			us = UserService.objects.get(user=self.user,
106			token=self.token,
107			name='ServiceMastodon')
108			try:
109			toot_api = MastodonAPI(
110			client_id=us.client_id,
111			client_secret=us.client_secret,
112			access_token=self.token,
113			api_base_url=us.host
114			)
115			except ValueError as e:
116			logger.error(e)
117			update_result(trigger_id, msg=e, status=False)
118
119			if toot_obj.since_id is not None and toot_obj.since_id > 0:
120			since_id = toot_obj.since_id
121			search = {'since_id': toot_obj.since_id}
122
123			# first request to Mastodon
124			count, search, statuses = _get_toots(toot_api, toot_obj, search)
125
126			if len(statuses) > 0:
127			newest = None
128			for status in statuses:
129			if newest is None:
130			newest = True
131			# first query ; get the max id
132			search['max_id'] = max_id = status['id']
133
134			since_id = search['since_id'] = statuses[-1]['id'] - 1
135
136			count, search, statuses = _get_toots(toot_api, toot_obj,
137			search)
138
139			newest = None
140			if len(statuses) > 0:
141			my_toots = []
142			for s in statuses:
143			if newest is None:
144			newest = True
145			max_id = s['id'] - 1
146			toot_name = s['account']['username']
147			# get the text of the tweet + url to this one
148			if toot_obj.fav:
149			url = '{0}/api/v1/statuses/{1}'.format(
150			self.api_base_url, s['id'])
151			title = _('Toot Fav from @{}'.format(toot_name))
152			else:
153			url = '{0}/api/v1/accounts/{1}/statuses'.format(
154			self.api_base_url, s['id'])
155			title = _('Toot from @{}'.format(toot_name))
156			# Wed Aug 29 17:12:58 +0000 2012
157			my_date = arrow.get(s['created_at'],
158			'ddd MMM DD HH:mm:ss Z YYYY')
159			published = arrow.get(my_date).to(settings.TIME_ZONE)
160			if date_triggered is not None and \
161			published is not None and \
162			now >= published >= date_triggered:
163			my_toots.append({'title': title,
164			'content': s['content'],
165			'link': url,
166			'my_date': my_date})
167			# digester
168			self.send_digest_event(trigger_id, title, url)
169			cache.set('th_mastodon_' + str(trigger_id), my_toots)
170			Mastodon.objects.filter(trigger_id=trigger_id).update(
171			since_id=since_id,
172			max_id=max_id,
173			count=count)
174			return my_toots
175
176			def save_data(self, trigger_id, **data):
177			"""
178			get the data from the service
179
180			:param trigger_id: id of the trigger
181			:params data, dict
182			:rtype: dict
183			"""
184			title, content = super(ServiceMastodon, self).save_data(
185			trigger_id, **data)
186
187			if self.title_or_content(title):
188
189			content = str("{title} {link}").format(
190			title=title, link=data.get('link'))
191
192			content += self.get_tags(trigger_id)
193
194			content = self.set_mastodon_content(content)
195
196			us = UserService.objects.get(user=self.user,
197			token=self.token,
198			name='ServiceMastodon')
199
200			try:
201			toot_api = MastodonAPI(
202			client_id=us.client_id,
203			client_secret=us.client_secret,
204			access_token=self.token,
205			api_base_url=us.host
206			)
207			except ValueError as e:
208			logger.error(e)
209			update_result(trigger_id, msg=e, status=False)
210
211			try:
212			toot_api.toot(content)
213			status = True
214			except Exception as inst:
215			logger.critical("Mastodon ERR {}".format(inst))
216			update_result(trigger_id, msg=inst, status=False)
217			status = False
218
219			return status
220
221			def get_tags(self, trigger_id):
222			"""
223			get the tags if any
224			:param trigger_id: the id of the related trigger
225			:return: tags string
226			"""
227
228			# get the Mastodon data of this trigger
229			trigger = Mastodon.objects.get(trigger_id=trigger_id)
230
231			tags = ''
232
233			if trigger.tag is not None:
234			# is there several tag ?
235			tags = ["#" + tag.strip() for tag in trigger.tag.split(',')
236			] if ',' in trigger.tag else "#" + trigger.tag
237
238			tags = str(','.join(tags)) if isinstance(tags, list) else tags
239			tags = ' ' + tags
240
241			return tags
242
243			def set_mastodon_content(self, content):
244			"""
245			cleaning content by removing any existing html tag
246			:param content:
247			:return:
248			"""
249			content = html.strip_tags(content)
250
251			if len(content) > 560:
252			return content[:560]
253
254			return content
255
256			def title_or_content(self, title):
257			"""
258			If the title always contains 'New status from'
259			drop the title and get 'the content' instead
260			:param title:
261			:return:
262			"""
263			if "New status by" in title:
264			return False
265			return True
266
267			def auth(self, request):
268			"""
269			get the auth of the services
270			:param request: contains the current session
271			:type request: dict
272			:rtype: dict
273			"""
274			# create app
275			redirect_uris = '%s://%s%s' % (request.scheme, request.get_host(),
276			reverse('mastodon_callback'))
277			us = UserService.objects.get(user=request.user,
278			name='ServiceMastodon')
279			client_id, client_secret = MastodonAPI.create_app(
280			client_name="TriggerHappy", api_base_url=us.host,
281			redirect_uris=redirect_uris)
282
283			us.client_id = client_id
284			us.client_secret = client_secret
285			us.save()
286
287			us = UserService.objects.get(user=request.user,
288			name='ServiceMastodon')
289			# get the token by logging in
290			mastodon = MastodonAPI(
291			client_id=client_id,
292			client_secret=client_secret,
293			api_base_url=us.host
294			)
295			token = mastodon.log_in(username=us.username, password=us.password)
296			us.token = token
297			us.save()
298			return self.callback_url(request)
299
300			def callback_url(self, request):
301			us = UserService.objects.get(user=request.user,
302			name='ServiceMastodon')
303			mastodon = MastodonAPI(
304			client_id=us.client_id,
305			client_secret=us.client_secret,
306			access_token=us.token,
307			api_base_url=us.host
308			)
309			redirect_uris = '%s://%s%s' % (request.scheme, request.get_host(),
310			reverse('mastodon_callback'))
311			return mastodon.auth_request_url(redirect_uris=redirect_uris)
312
313			def callback(self, request, **kwargs):
314			"""
315			Called from the Service when the user accept to activate it
316			the url to go back after the external service call
317			:param request: contains the current session
318			:param kwargs: keyword args
319			:type request: dict
320			:type kwargs: dict
321			:rtype: string
322			"""
323			return 'mastodon/callback.html'
324

foxmask / django-th

Push — master ( e7ede0...eadaef )

ServiceMastodon.callback_url() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like