ServiceTwitter.save_data() - Code Metrics - Inspection of "few improvements" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 0f0f78...29b212 )

by Fox

created 2016-05-23 20:56 UTC

ServiceTwitter.save_data() D

↳ Parent: ServiceTwitter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	4
Bugs	0	Features	0

Metric	Value
cc	8
c	4
b	0
f	0
dl	0
loc	38
rs	4

# coding: utf-8
import arrow

# Twitter lib
from twython import Twython

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

# django_th classes
from django_th.services.services import ServicesMgr
from th_twitter.models import Twitter

"""
    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):

    def __init__(self, token=None):
        super(ServiceTwitter, self).__init__(token)
        self.consumer_key = settings.TH_TWITTER['consumer_key']
        self.consumer_secret = settings.TH_TWITTER['consumer_secret']
        self.token = token
        if self.token is not None:
            token_key, token_secret = self.token.split('#TH#')
            self.twitter_api = Twython(self.consumer_key, self.consumer_secret,
                                       token_key, token_secret)

    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 = 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
                statuses = self.twitter_api.get_user_timeline(**search)

            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
        tags = []
        # set the title and content of the data
        title, content = super(ServiceTwitter, self).save_data(
            trigger_id, data, kwargs={})

        if data.get('link') and len(data.get('link')) > 0:
            # get the Twitter data of this trigger
            trigger = Twitter.objects.get(trigger_id=trigger_id)

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

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

            # TODO : need to check the size of the content and tags to add
            if len(tags) > 0:
                content += ' ' + str(','.join(tags))

            try:
                self.twitter_api.update_status(status=content)
                status = True
            except Exception as inst:
                logger.critical("Twitter ERR {}".format(inst))
                status = False
        return status

    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')

        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
        """
        kwargs = {'access_token': '', 'service': 'ServiceTwitter',
                  'return': 'twitter'}
        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


1			# coding: utf-8
2			import arrow
3
4			# Twitter lib
5			from twython import Twython
6
7			# django classes
8			from django.conf import settings
9			from django.utils.log import getLogger
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 th_twitter.models import Twitter
16
17			"""
18			handle process with twitter
19			put the following in settings.py
20
21			TH_TWITTER = {
22			'consumer_key': 'abcdefghijklmnopqrstuvwxyz',
23			'consumer_secret': 'abcdefghijklmnopqrstuvwxyz',
24			}
25
26			"""
27
28			logger = getLogger('django_th.trigger_happy')
29			cache = caches['th_twitter']
30
31
32			class ServiceTwitter(ServicesMgr):
33
34			def __init__(self, token=None):
35			super(ServiceTwitter, self).__init__(token)
36			self.consumer_key = settings.TH_TWITTER['consumer_key']
37			self.consumer_secret = settings.TH_TWITTER['consumer_secret']
38			self.token = token
39			if self.token is not None:
40			token_key, token_secret = self.token.split('#TH#')
41			self.twitter_api = Twython(self.consumer_key, self.consumer_secret,
42			token_key, token_secret)
43
44			def read_data(self, **kwargs):
45			"""
46			get the data from the service
47
48			:param kwargs: contain keyword args : trigger_id at least
49			:type kwargs: dict
50			:rtype: list
51			"""
52			twitter_url = 'https://www.twitter.com/{}/status/{}'
53			now = arrow.utcnow().to(settings.TIME_ZONE)
54			my_tweets = []
55			search = {}
56			since_id = None
57			trigger_id = kwargs['trigger_id']
58			date_triggered = kwargs['date_triggered']
59
60			def _get_tweets(twitter_obj, search):
61			"""
62			get the tweets from twitter and return the filters to use :
63			search and count
64
65			:param twitter_obj: from Twitter model
66			:param search: filter used for twython.search() or
67			twython.get_user_timeline())
68			:type twitter_obj: Object
69			:type search: dict
70			:return: count that limit the quantity of tweet to retrieve,
71			the filter named search, the tweets
72			:rtype: list
73			"""
74
75			"""
76			explanations about statuses :
77			when we want to track the tweet of a screen
78			statuses contain all of them
79			when we want to track all the tweet matching a tag
80			statuses contain statuses + metadata array
81			this is why we need to do
82			statuses = statuses['statuses']
83			to be able to handle the result as for screen_name
84			"""
85
86			# get the tweets for a given tag
87			# https://dev.twitter.com/docs/api/1.1/get/search/tweets
88			statuses = ''
89			count = 100
90			if twitter_obj.tag != '':
91			count = 100
92			search['count'] = count
93			search['q'] = twitter_obj.tag
94			search['result_type'] = 'recent'
95			# do a search
96			statuses = self.twitter_api.search(**search)
97			# just return the content of te statuses array
98			statuses = statuses['statuses']
99
100			# get the tweets from a given user
101			# https://dev.twitter.com/docs/api/1.1/get/statuses/user_timeline
102			elif twitter_obj.screen != '':
103			count = 200
104			search['count'] = count
105			search['screen_name'] = twitter_obj.screen
106			# call the user timeline and get his tweet
107			statuses = self.twitter_api.get_user_timeline(**search)
108
109			return count, search, statuses
110
111			if self.token is not None:
112			kw = {'model_name': 'Twitter', 'trigger_id': trigger_id}
113			twitter_obj = super(ServiceTwitter, self).read_data(**kw)
114
115			# https://dev.twitter.com/rest/public/timelines
116			if twitter_obj.since_id is not None and twitter_obj.since_id > 0:
117			since_id = twitter_obj.since_id
118			search = {'since_id': twitter_obj.since_id}
119
120			# first request to Twitter
121			count, search, statuses = _get_tweets(twitter_obj, search)
122
123			if len(statuses) > 0:
124			newest = None
125			for status in statuses:
126			if newest is None:
127			newest = True
128			# first query ; get the max id
129			search['max_id'] = max_id = status['id']
130
131			since_id = search['since_id'] = statuses[-1]['id'] - 1
132
133			count, search, statuses = _get_tweets(twitter_obj, search)
134
135			newest = None
136			if len(statuses) > 0:
137			my_tweets = []
138			for s in statuses:
139			if newest is None:
140			newest = True
141			max_id = s['id'] - 1
142			screen_name = s['user']['screen_name']
143			# get the text of the tweet + url to this one
144			url = twitter_url.format(screen_name,
145			s['id_str'])
146			title = _('Tweet from @{}'.format(screen_name))
147			# Wed Aug 29 17:12:58 +0000 2012
148			my_date = arrow.get(s['created_at'],
149			'ddd MMM DD HH:mm:ss Z YYYY')
150			published = arrow.get(my_date).to(settings.TIME_ZONE)
151			if date_triggered is not None and \
152			published is not None and \
153			now >= published >= date_triggered:
154			my_tweets.append({'title': title,
155			'content': s['text'],
156			'link': url,
157			'my_date': my_date})
158			cache.set('th_twitter_' + str(trigger_id), my_tweets)
159			Twitter.objects.filter(trigger_id=trigger_id).update(
160			since_id=since_id,
161			max_id=max_id,
162			count=count)
163			return my_tweets
164
165			def save_data(self, trigger_id, **data):
166			"""
167			let's save the data
168
169			:param trigger_id: trigger ID from which to save data
170			:param data: the data to check to be used and save
171			:type trigger_id: int
172			:type data: dict
173			:return: the status of the save statement
174			:rtype: boolean
175			"""
176			status = False
177			tags = []
178			# set the title and content of the data
179			title, content = super(ServiceTwitter, self).save_data(
180			trigger_id, data, kwargs={})
181
182			if data.get('link') and len(data.get('link')) > 0:
183			# get the Twitter data of this trigger
184			trigger = Twitter.objects.get(trigger_id=trigger_id)
185
186			if trigger.tag:
187			# is there several tag ?
188			tags = ["#" + tag.strip() for tag in trigger.tag.split(',')] if ',' in trigger.tag else "#" + trigger.tag
189
190			content = str("{title} {link}").format(title=title, link=data.get('link'))
191
192			# TODO : need to check the size of the content and tags to add
193			if len(tags) > 0:
194			content += ' ' + str(','.join(tags))
195
196			try:
197			self.twitter_api.update_status(status=content)
198			status = True
199			except Exception as inst:
200			logger.critical("Twitter ERR {}".format(inst))
201			status = False
202			return status
203
204			def auth(self, request):
205			"""
206			build the request to access to the Twitter
207			website with all its required parms
208			:param request: makes the url to call Twitter + the callback url
209			:return: go to the Twitter website to ask to the user
210			to allow the access of TriggerHappy
211			"""
212			callback_url = self.callback_url(request, 'twitter')
213
214			twitter = Twython(self.consumer_key, self.consumer_secret)
215
216			req_token = twitter.get_authentication_tokens(
217			callback_url=callback_url)
218			request.session['oauth_token'] = req_token['oauth_token']
219			request.session['oauth_token_secret'] = req_token['oauth_token_secret']
220
221			return req_token['auth_url']
222
223			def callback(self, request, **kwargs):
224			"""
225			Called from the Service when the user accept to activate it
226			"""
227			kwargs = {'access_token': '', 'service': 'ServiceTwitter',
228			'return': 'twitter'}
229			return super(ServiceTwitter, self).callback(request, **kwargs)
230
231			def get_access_token(
232			self, oauth_token, oauth_token_secret, oauth_verifier
233			):
234			"""
235			:param oauth_token: oauth_token retrieve by the API Twython
236			get_authentication_tokens()
237			:param oauth_token_secret: oauth_token_secret retrieve by the
238			API Twython get_authentication_tokens()
239			:param oauth_verifier: oauth_verifier retrieve from Twitter
240			:type oauth_token: string
241			:type oauth_token_secret: string
242			:type oauth_verifier: string
243			:return: access_token
244			:rtype: dict
245			"""
246			twitter = Twython(self.consumer_key,
247			self.consumer_secret,
248			oauth_token,
249			oauth_token_secret)
250			access_token = twitter.get_authorized_tokens(oauth_verifier)
251			return access_token
252

foxmask / django-th

Push — master ( 0f0f78...29b212 )

ServiceTwitter.save_data() D

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like