ServiceTwitter.set_twitter_content() - Code Metrics - Inspection of "little tweak to handle atom feed and a Mastodon be..." - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 81f6dd...bebd4c )

by Fox

created 2017-04-22 16:22 UTC

ServiceTwitter.set_twitter_content() A

↳ Parent: ServiceTwitter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

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

# coding: utf-8
import arrow

# Twitter lib
from twython import Twython, TwythonAuthError, TwythonRateLimitError

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

# django_th classes
from django_th.services.services import ServicesMgr
from django_th.models import update_result, UserService
from th_twitter.models import Twitter

from logging import getLogger

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

    TH_TWITTER = {
        'consumer_key': 'abcdefghijklmnopqrstuvwxyz',
        'consumer_secret': 'abcdefghijklmnopqrstuvwxyz',
    }

"""

logger = getLogger('django_th.trigger_happy')
cache = caches['th_twitter']


class ServiceTwitter(ServicesMgr):
    """
        Service Twitter
    """
    def __init__(self, token=None, **kwargs):
        """

        :param token:
        :param kwargs:
        """
        super(ServiceTwitter, self).__init__(token, **kwargs)
        self.consumer_key = settings.TH_TWITTER['consumer_key']
        self.consumer_secret = settings.TH_TWITTER['consumer_secret']
        self.token = token
        self.oauth = 'oauth1'
        self.service = 'ServiceTwitter'
        if self.token is not None:
            token_key, token_secret = self.token.split('#TH#')
            try:
                self.twitter_api = Twython(self.consumer_key,
                                           self.consumer_secret,
                                           token_key, token_secret)
            except (TwythonAuthError, TwythonRateLimitError) as e:
                us = UserService.objects.get(token=token)
                logger.error(e.msg, e.error_code)
                update_result(us.trigger_id, msg=e.msg, status=False)

    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
        """
        twitter_url = 'https://www.twitter.com/{}/status/{}'
        now = arrow.utcnow().to(settings.TIME_ZONE)
        my_tweets = []
        search = {}
        since_id = None
        trigger_id = kwargs['trigger_id']
        date_triggered = arrow.get(kwargs['date_triggered'])

        def _get_tweets(twitter_obj, search):
            """
                get the tweets from twitter and return the filters to use :
                search and count

                :param twitter_obj: from Twitter model
                :param search: filter used for twython.search() or
                twython.get_user_timeline())
                :type twitter_obj: Object
                :type search: dict
                :return: count that limit the quantity of tweet to retrieve,
                the filter named search, the tweets
                :rtype: list
            """

            """
                explanations about statuses :
                when we want to track the tweet of a screen
                statuses contain all of them
                when we want to track all the tweet matching a tag
                statuses contain statuses + metadata array
                this is why we need to do
                statuses = statuses['statuses']
                to be able to handle the result as for screen_name
            """

            # get the tweets for a given tag
            # https://dev.twitter.com/docs/api/1.1/get/search/tweets
            statuses = ''
            count = 100
            if twitter_obj.tag != '':
                count = 100
                search['count'] = count
                search['q'] = twitter_obj.tag
                search['result_type'] = 'recent'
                # do a search
                statuses = self.twitter_api.search(**search)
                # just return the content of te statuses array
                statuses = statuses['statuses']

            # get the tweets from a given user
            # https://dev.twitter.com/docs/api/1.1/get/statuses/user_timeline
            elif twitter_obj.screen != '':
                count = 200
                search['count'] = count
                search['screen_name'] = twitter_obj.screen
                # call the user timeline and get his tweet
                try:
                    statuses = self.twitter_api.get_user_timeline(**search)
                except TwythonAuthError as e:
                    logger.error(e.msg, e.error_code)
                    update_result(trigger_id, msg=e.msg, status=False)

            return count, search, statuses

        if self.token is not None:
            kw = {'model_name': 'Twitter', 'trigger_id': trigger_id}
            twitter_obj = super(ServiceTwitter, self).read_data(**kw)

            # https://dev.twitter.com/rest/public/timelines
            if twitter_obj.since_id is not None and twitter_obj.since_id > 0:
                since_id = twitter_obj.since_id
                search = {'since_id': twitter_obj.since_id}

            # first request to Twitter
            count, search, statuses = _get_tweets(twitter_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_tweets(twitter_obj, search)

                newest = None
                if len(statuses) > 0:
                    my_tweets = []
                    for s in statuses:
                        if newest is None:
                            newest = True
                            max_id = s['id'] - 1
                        screen_name = s['user']['screen_name']
                        # get the text of the tweet + url to this one
                        url = twitter_url.format(screen_name,
                                                 s['id_str'])
                        title = _('Tweet from @{}'.format(screen_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_tweets.append({'title': title,
                                              'content': s['text'],
                                              'link': url,
                                              'my_date': my_date})
                    cache.set('th_twitter_' + str(trigger_id), my_tweets)
                    Twitter.objects.filter(trigger_id=trigger_id).update(
                        since_id=since_id,
                        max_id=max_id,
                        count=count)
        return my_tweets

    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
        """
        status = False
        # set the title and content of the data
        title, content = super(ServiceTwitter, self).save_data(
            trigger_id, **data)

        if data.get('link') and len(data.get('link')) > 0:

            if self.title_or_content(title):

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

                content += self.get_tags(trigger_id)
            else:
                content = self.set_twitter_content(content)

            try:
                self.twitter_api.update_status(status=content)
                status = True
            except Exception as inst:
                logger.critical("Twitter 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 Twitter data of this trigger
        trigger = Twitter.objects.get(trigger_id=trigger_id)

        tags = ''

        if len(trigger.tag) > 0:
            # 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 auth(self, request):
        """
        build the request to access to the Twitter
        website with all its required parms
        :param request: makes the url to call Twitter + the callback url
        :return: go to the Twitter website to ask to the user
        to allow the access of TriggerHappy
        """
        callback_url = self.callback_url(request)

        twitter = Twython(self.consumer_key, self.consumer_secret)

        req_token = twitter.get_authentication_tokens(
            callback_url=callback_url)
        request.session['oauth_token'] = req_token['oauth_token']
        request.session['oauth_token_secret'] = req_token['oauth_token_secret']

        return req_token['auth_url']

    def callback(self, request, **kwargs):
        """
            Called from the Service when the user accept to activate it
        """
        return super(ServiceTwitter, self).callback(request, **kwargs)

    def get_access_token(
        self, oauth_token, oauth_token_secret, oauth_verifier
    ):
        """
        :param oauth_token: oauth_token retrieve by the API Twython
        get_authentication_tokens()
        :param oauth_token_secret: oauth_token_secret retrieve by the
        API Twython get_authentication_tokens()
        :param oauth_verifier: oauth_verifier retrieve from Twitter
        :type oauth_token: string
        :type oauth_token_secret: string
        :type oauth_verifier: string
        :return: access_token
        :rtype: dict
        """
        twitter = Twython(self.consumer_key,
                          self.consumer_secret,
                          oauth_token,
                          oauth_token_secret)
        access_token = twitter.get_authorized_tokens(oauth_verifier)
        return access_token

    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 set_twitter_content(self, content):
        """
        cleaning content by removing any existing html tag 
        :param content: 
        :return: 
        """
        content = html.strip_tags(content)

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

        return content


1			# coding: utf-8
2			import arrow
3
4			# Twitter lib
5			from twython import Twython, TwythonAuthError, TwythonRateLimitError
6
7			# django classes
8			from django.conf import settings
9			from django.utils import html
10			from django.utils.translation import ugettext as _
11			from django.core.cache import caches
12
13			# django_th classes
14			from django_th.services.services import ServicesMgr
15			from django_th.models import update_result, UserService
16			from th_twitter.models import Twitter
17
18			from logging import getLogger
19
20			"""
21			handle process with twitter
22			put the following in settings.py
23
24			TH_TWITTER = {
25			'consumer_key': 'abcdefghijklmnopqrstuvwxyz',
26			'consumer_secret': 'abcdefghijklmnopqrstuvwxyz',
27			}
28
29			"""
30
31			logger = getLogger('django_th.trigger_happy')
32			cache = caches['th_twitter']
33
34
35			class ServiceTwitter(ServicesMgr):
36			"""
37			Service Twitter
38			"""
39			def __init__(self, token=None, **kwargs):
40			"""
41
42			:param token:
43			:param kwargs:
44			"""
45			super(ServiceTwitter, self).__init__(token, **kwargs)
46			self.consumer_key = settings.TH_TWITTER['consumer_key']
47			self.consumer_secret = settings.TH_TWITTER['consumer_secret']
48			self.token = token
49			self.oauth = 'oauth1'
50			self.service = 'ServiceTwitter'
51			if self.token is not None:
52			token_key, token_secret = self.token.split('#TH#')
53			try:
54			self.twitter_api = Twython(self.consumer_key,
55			self.consumer_secret,
56			token_key, token_secret)
57			except (TwythonAuthError, TwythonRateLimitError) as e:
58			us = UserService.objects.get(token=token)
59			logger.error(e.msg, e.error_code)
60			update_result(us.trigger_id, msg=e.msg, status=False)
61
62			def read_data(self, **kwargs):
63			"""
64			get the data from the service
65
66			:param kwargs: contain keyword args : trigger_id at least
67			:type kwargs: dict
68			:rtype: list
69			"""
70			twitter_url = 'https://www.twitter.com/{}/status/{}'
71			now = arrow.utcnow().to(settings.TIME_ZONE)
72			my_tweets = []
73			search = {}
74			since_id = None
75			trigger_id = kwargs['trigger_id']
76			date_triggered = arrow.get(kwargs['date_triggered'])
77
78			def _get_tweets(twitter_obj, search):
79			"""
80			get the tweets from twitter and return the filters to use :
81			search and count
82
83			:param twitter_obj: from Twitter model
84			:param search: filter used for twython.search() or
85			twython.get_user_timeline())
86			:type twitter_obj: Object
87			:type search: dict
88			:return: count that limit the quantity of tweet to retrieve,
89			the filter named search, the tweets
90			:rtype: list
91			"""
92
93			"""
94			explanations about statuses :
95			when we want to track the tweet of a screen
96			statuses contain all of them
97			when we want to track all the tweet matching a tag
98			statuses contain statuses + metadata array
99			this is why we need to do
100			statuses = statuses['statuses']
101			to be able to handle the result as for screen_name
102			"""
103
104			# get the tweets for a given tag
105			# https://dev.twitter.com/docs/api/1.1/get/search/tweets
106			statuses = ''
107			count = 100
108			if twitter_obj.tag != '':
109			count = 100
110			search['count'] = count
111			search['q'] = twitter_obj.tag
112			search['result_type'] = 'recent'
113			# do a search
114			statuses = self.twitter_api.search(**search)
115			# just return the content of te statuses array
116			statuses = statuses['statuses']
117
118			# get the tweets from a given user
119			# https://dev.twitter.com/docs/api/1.1/get/statuses/user_timeline
120			elif twitter_obj.screen != '':
121			count = 200
122			search['count'] = count
123			search['screen_name'] = twitter_obj.screen
124			# call the user timeline and get his tweet
125			try:
126			statuses = self.twitter_api.get_user_timeline(**search)
127			except TwythonAuthError as e:
128			logger.error(e.msg, e.error_code)
129			update_result(trigger_id, msg=e.msg, status=False)
130
131			return count, search, statuses
132
133			if self.token is not None:
134			kw = {'model_name': 'Twitter', 'trigger_id': trigger_id}
135			twitter_obj = super(ServiceTwitter, self).read_data(**kw)
136
137			# https://dev.twitter.com/rest/public/timelines
138			if twitter_obj.since_id is not None and twitter_obj.since_id > 0:
139			since_id = twitter_obj.since_id
140			search = {'since_id': twitter_obj.since_id}
141
142			# first request to Twitter
143			count, search, statuses = _get_tweets(twitter_obj, search)
144
145			if len(statuses) > 0:
146			newest = None
147			for status in statuses:
148			if newest is None:
149			newest = True
150			# first query ; get the max id
151			search['max_id'] = max_id = status['id']
152
153			since_id = search['since_id'] = statuses[-1]['id'] - 1
154
155			count, search, statuses = _get_tweets(twitter_obj, search)
156
157			newest = None
158			if len(statuses) > 0:
159			my_tweets = []
160			for s in statuses:
161			if newest is None:
162			newest = True
163			max_id = s['id'] - 1
164			screen_name = s['user']['screen_name']
165			# get the text of the tweet + url to this one
166			url = twitter_url.format(screen_name,
167			s['id_str'])
168			title = _('Tweet from @{}'.format(screen_name))
169			# Wed Aug 29 17:12:58 +0000 2012
170			my_date = arrow.get(s['created_at'],
171			'ddd MMM DD HH:mm:ss Z YYYY')
172			published = arrow.get(my_date).to(settings.TIME_ZONE)
173			if date_triggered is not None and \
174			published is not None and \
175			now >= published >= date_triggered:
176			my_tweets.append({'title': title,
177			'content': s['text'],
178			'link': url,
179			'my_date': my_date})
180			cache.set('th_twitter_' + str(trigger_id), my_tweets)
181			Twitter.objects.filter(trigger_id=trigger_id).update(
182			since_id=since_id,
183			max_id=max_id,
184			count=count)
185			return my_tweets
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			status = False
199			# set the title and content of the data
200			title, content = super(ServiceTwitter, self).save_data(
201			trigger_id, **data)
202
203			if data.get('link') and len(data.get('link')) > 0:
204
205			if self.title_or_content(title):
206
207			content = str("{title} {link}").format(
208			title=title, link=data.get('link'))
209
210			content += self.get_tags(trigger_id)
211			else:
212			content = self.set_twitter_content(content)
213
214			try:
215			self.twitter_api.update_status(status=content)
216			status = True
217			except Exception as inst:
218			logger.critical("Twitter ERR {}".format(inst))
219			update_result(trigger_id, msg=inst, status=False)
220			status = False
221			return status
222
223			def get_tags(self, trigger_id):
224			"""
225			get the tags if any
226			:param trigger_id: the id of the related trigger
227			:return: tags string
228			"""
229
230			# get the Twitter data of this trigger
231			trigger = Twitter.objects.get(trigger_id=trigger_id)
232
233			tags = ''
234
235			if len(trigger.tag) > 0:
236			# is there several tag ?
237			tags = ["#" + tag.strip() for tag in trigger.tag.split(',')
238			] if ',' in trigger.tag else "#" + trigger.tag
239
240			tags = str(','.join(tags)) if isinstance(tags, list) else tags
241			tags = ' ' + tags
242
243			return tags
244
245			def auth(self, request):
246			"""
247			build the request to access to the Twitter
248			website with all its required parms
249			:param request: makes the url to call Twitter + the callback url
250			:return: go to the Twitter website to ask to the user
251			to allow the access of TriggerHappy
252			"""
253			callback_url = self.callback_url(request)
254
255			twitter = Twython(self.consumer_key, self.consumer_secret)
256
257			req_token = twitter.get_authentication_tokens(
258			callback_url=callback_url)
259			request.session['oauth_token'] = req_token['oauth_token']
260			request.session['oauth_token_secret'] = req_token['oauth_token_secret']
261
262			return req_token['auth_url']
263
264			def callback(self, request, **kwargs):
265			"""
266			Called from the Service when the user accept to activate it
267			"""
268			return super(ServiceTwitter, self).callback(request, **kwargs)
269
270			def get_access_token(
271			self, oauth_token, oauth_token_secret, oauth_verifier
272			):
273			"""
274			:param oauth_token: oauth_token retrieve by the API Twython
275			get_authentication_tokens()
276			:param oauth_token_secret: oauth_token_secret retrieve by the
277			API Twython get_authentication_tokens()
278			:param oauth_verifier: oauth_verifier retrieve from Twitter
279			:type oauth_token: string
280			:type oauth_token_secret: string
281			:type oauth_verifier: string
282			:return: access_token
283			:rtype: dict
284			"""
285			twitter = Twython(self.consumer_key,
286			self.consumer_secret,
287			oauth_token,
288			oauth_token_secret)
289			access_token = twitter.get_authorized_tokens(oauth_verifier)
290			return access_token
291
292			def title_or_content(self, title):
293			"""
294			If the title always contains 'New status from'
295			drop the title and get 'the content' instead
296			:param title:
297			:return:
298			"""
299			if "New status by" in title:
300			return False
301			return True
302
303			def set_twitter_content(self, content):
304			"""
305			cleaning content by removing any existing html tag
306			:param content:
307			:return:
308			"""
309			content = html.strip_tags(content)
310
311			if len(content) > 140:
312			return content[:140]
313
314			return content
315

foxmask / django-th

Push — master ( 81f6dd...bebd4c )

ServiceTwitter.set_twitter_content() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like