ServiceTwitter._get_tweets() - Code Metrics - Inspection of "Get Twitter data from favorites too" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d48528...1348df )

by Fox

created 2017-09-19 21:27 UTC

ServiceTwitter._get_tweets() B

↳ Parent: ServiceTwitter.read_data()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
dl	0
loc	62
rs	8.3192
c	0
b	0
f	0

How to fix Long Method

# 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_status_url = 'https://www.twitter.com/{}/status/{}'
        twitter_fav_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:
                    if twitter_obj.fav:
                        count = 20
                        search['count'] = 20
                        # get the favorites https://dev.twitter.com/rest/
                        # reference/get/favorites/list
                        statuses = self.twitter_api.get_favorites(**search)
                    else:
                        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
                        if twitter_obj.fav:
                            url = twitter_fav_url.format(screen_name,
                                                         s['id_str'])
                            title = _('Tweet Fav from @{}'.format(screen_name))
                        else:
                            url = twitter_status_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})
                            # digester
                            self.send_digest_event(trigger_id, title, url)
                    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_status_url = 'https://www.twitter.com/{}/status/{}'
71			twitter_fav_url = 'https://www.twitter.com/{}/status/{}'
72			now = arrow.utcnow().to(settings.TIME_ZONE)
73			my_tweets = []
74			search = {}
75			since_id = None
76			trigger_id = kwargs['trigger_id']
77			date_triggered = arrow.get(kwargs['date_triggered'])
78
79			def _get_tweets(twitter_obj, search):
80			"""
81			get the tweets from twitter and return the filters to use :
82			search and count
83
84			:param twitter_obj: from Twitter model
85			:param search: filter used for twython.search() or
86			twython.get_user_timeline())
87			:type twitter_obj: Object
88			:type search: dict
89			:return: count that limit the quantity of tweet to retrieve,
90			the filter named search, the tweets
91			:rtype: list
92			"""
93
94			"""
95			explanations about statuses :
96			when we want to track the tweet of a screen
97			statuses contain all of them
98			when we want to track all the tweet matching a tag
99			statuses contain statuses + metadata array
100			this is why we need to do
101			statuses = statuses['statuses']
102			to be able to handle the result as for screen_name
103			"""
104
105			# get the tweets for a given tag
106			# https://dev.twitter.com/docs/api/1.1/get/search/tweets
107			statuses = ''
108			count = 100
109			if twitter_obj.tag:
110			count = 100
111			search['count'] = count
112			search['q'] = twitter_obj.tag
113			search['result_type'] = 'recent'
114			# do a search
115			statuses = self.twitter_api.search(**search)
116			# just return the content of te statuses array
117			statuses = statuses['statuses']
118
119			# get the tweets from a given user
120			# https://dev.twitter.com/docs/api/1.1/get/statuses/user_timeline
121			elif twitter_obj.screen:
122			count = 200
123			search['count'] = count
124			search['screen_name'] = twitter_obj.screen
125
126			# call the user timeline and get his tweet
127			try:
128			if twitter_obj.fav:
129			count = 20
130			search['count'] = 20
131			# get the favorites https://dev.twitter.com/rest/
132			# reference/get/favorites/list
133			statuses = self.twitter_api.get_favorites(**search)
134			else:
135			statuses = self.twitter_api.get_user_timeline(**search)
136			except TwythonAuthError as e:
137			logger.error(e.msg, e.error_code)
138			update_result(trigger_id, msg=e.msg, status=False)
139
140			return count, search, statuses
141
142			if self.token is not None:
143			kw = {'model_name': 'Twitter', 'trigger_id': trigger_id}
144			twitter_obj = super(ServiceTwitter, self).read_data(**kw)
145
146			# https://dev.twitter.com/rest/public/timelines
147			if twitter_obj.since_id is not None and twitter_obj.since_id > 0:
148			since_id = twitter_obj.since_id
149			search = {'since_id': twitter_obj.since_id}
150
151			# first request to Twitter
152			count, search, statuses = _get_tweets(twitter_obj, search)
153
154			if len(statuses) > 0:
155			newest = None
156			for status in statuses:
157			if newest is None:
158			newest = True
159			# first query ; get the max id
160			search['max_id'] = max_id = status['id']
161
162			since_id = search['since_id'] = statuses[-1]['id'] - 1
163
164			count, search, statuses = _get_tweets(twitter_obj, search)
165
166			newest = None
167			if len(statuses) > 0:
168			my_tweets = []
169			for s in statuses:
170			if newest is None:
171			newest = True
172			max_id = s['id'] - 1
173			screen_name = s['user']['screen_name']
174			# get the text of the tweet + url to this one
175			if twitter_obj.fav:
176			url = twitter_fav_url.format(screen_name,
177			s['id_str'])
178			title = _('Tweet Fav from @{}'.format(screen_name))
179			else:
180			url = twitter_status_url.format(screen_name,
181			s['id_str'])
182			title = _('Tweet from @{}'.format(screen_name))
183			# Wed Aug 29 17:12:58 +0000 2012
184			my_date = arrow.get(s['created_at'],
185			'ddd MMM DD HH:mm:ss Z YYYY')
186			published = arrow.get(my_date).to(settings.TIME_ZONE)
187			if date_triggered is not None and \
188			published is not None and \
189			now >= published >= date_triggered:
190			my_tweets.append({'title': title,
191			'content': s['text'],
192			'link': url,
193			'my_date': my_date})
194			# digester
195			self.send_digest_event(trigger_id, title, url)
196			cache.set('th_twitter_' + str(trigger_id), my_tweets)
197			Twitter.objects.filter(trigger_id=trigger_id).update(
198			since_id=since_id,
199			max_id=max_id,
200			count=count)
201			return my_tweets
202
203			def save_data(self, trigger_id, **data):
204			"""
205			let's save the data
206
207			:param trigger_id: trigger ID from which to save data
208			:param data: the data to check to be used and save
209			:type trigger_id: int
210			:type data: dict
211			:return: the status of the save statement
212			:rtype: boolean
213			"""
214			status = False
215			# set the title and content of the data
216			title, content = super(ServiceTwitter, self).save_data(
217			trigger_id, **data)
218
219			if data.get('link') and len(data.get('link')) > 0:
220
221			if self.title_or_content(title):
222
223			content = str("{title} {link}").format(
224			title=title, link=data.get('link'))
225
226			content += self.get_tags(trigger_id)
227			else:
228			content = self.set_twitter_content(content)
229
230			try:
231			self.twitter_api.update_status(status=content)
232			status = True
233			except Exception as inst:
234			logger.critical("Twitter ERR {}".format(inst))
235			update_result(trigger_id, msg=inst, status=False)
236			status = False
237			return status
238
239			def get_tags(self, trigger_id):
240			"""
241			get the tags if any
242			:param trigger_id: the id of the related trigger
243			:return: tags string
244			"""
245
246			# get the Twitter data of this trigger
247			trigger = Twitter.objects.get(trigger_id=trigger_id)
248
249			tags = ''
250
251			if len(trigger.tag) > 0:
252			# is there several tag ?
253			tags = ["#" + tag.strip() for tag in trigger.tag.split(',')
254			] if ',' in trigger.tag else "#" + trigger.tag
255
256			tags = str(','.join(tags)) if isinstance(tags, list) else tags
257			tags = ' ' + tags
258
259			return tags
260
261			def auth(self, request):
262			"""
263			build the request to access to the Twitter
264			website with all its required parms
265			:param request: makes the url to call Twitter + the callback url
266			:return: go to the Twitter website to ask to the user
267			to allow the access of TriggerHappy
268			"""
269			callback_url = self.callback_url(request)
270
271			twitter = Twython(self.consumer_key, self.consumer_secret)
272
273			req_token = twitter.get_authentication_tokens(
274			callback_url=callback_url)
275			request.session['oauth_token'] = req_token['oauth_token']
276			request.session['oauth_token_secret'] = req_token['oauth_token_secret']
277
278			return req_token['auth_url']
279
280			def callback(self, request, **kwargs):
281			"""
282			Called from the Service when the user accept to activate it
283			"""
284			return super(ServiceTwitter, self).callback(request, **kwargs)
285
286			def get_access_token(
287			self, oauth_token, oauth_token_secret, oauth_verifier
288			):
289			"""
290			:param oauth_token: oauth_token retrieve by the API Twython
291			get_authentication_tokens()
292			:param oauth_token_secret: oauth_token_secret retrieve by the
293			API Twython get_authentication_tokens()
294			:param oauth_verifier: oauth_verifier retrieve from Twitter
295			:type oauth_token: string
296			:type oauth_token_secret: string
297			:type oauth_verifier: string
298			:return: access_token
299			:rtype: dict
300			"""
301			twitter = Twython(self.consumer_key,
302			self.consumer_secret,
303			oauth_token,
304			oauth_token_secret)
305			access_token = twitter.get_authorized_tokens(oauth_verifier)
306			return access_token
307
308			def title_or_content(self, title):
309			"""
310			If the title always contains 'New status from'
311			drop the title and get 'the content' instead
312			:param title:
313			:return:
314			"""
315			if "New status by" in title:
316			return False
317			return True
318
319			def set_twitter_content(self, content):
320			"""
321			cleaning content by removing any existing html tag
322			:param content:
323			:return:
324			"""
325			content = html.strip_tags(content)
326
327			if len(content) > 140:
328			return content[:140]
329
330			return content
331

foxmask / django-th

Push — master ( d48528...1348df )

ServiceTwitter._get_tweets() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like