ServiceTwitter.save_data() - Code Metrics - Inspection of "fix #146 do a tweet with one tag" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 2107f4...08e703 )

by Fox

created 2016-09-02 13:01 UTC

ServiceTwitter.save_data() B

↳ Parent: ServiceTwitter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	4
Bugs	1	Features	0

Metric	Value
cc	4
c	4
b	1
f	0
dl	0
loc	30
rs	8.5806

# 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, **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#')
            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
        # 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:

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

            content += self.get_tags(trigger_id)

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


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

foxmask / django-th

Push — master ( 2107f4...08e703 )

ServiceTwitter.save_data() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like