__ServicesMgr.__init__() - Code Metrics - Inspection of "Data Digest" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 856bc1...ee6984 )

by Fox

created 2017-09-16 23:26 UTC

ServicesMgr.init__() A

↳ Parent: __ServicesMgr

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# coding: utf-8
# Using OAuth1Session
from requests_oauthlib import OAuth1Session, OAuth2Session

# django stuff
from django.core.cache import caches
from django.core.urlresolvers import reverse

try:
    from django.apps import apps
    get_model = apps.get_model
except ImportError:
    from django.db.models.loading import get_model

# django_th stuff
from django_th import signals
from django_th.models import UserService, ServicesActivated, TriggerService
from django_th.publishing_limit import PublishingLimit
from django_th.html_entities import HtmlEntities


class ServicesMgr(object):
    """
        Main Service Class
    """
    name = ''
    title = ''
    body = ''
    data = {}

    class __ServicesMgr:

        def __init__(self, arg):
            self.val = arg

        def __str__(self):
            return repr(self) + self.val

    instance = None

    def __init__(self, arg, **kwargs):
        base = 'https://www.urltotheserviceapi.com'
        self.AUTH_URL = '{}/api/rest/v1/oauth/authorize/'.format(base)
        self.REQ_TOKEN = '{}/api/rest/v1/oauth/request_token/'.format(base)
        self.ACC_TOKEN = '{}/api/rest/v1/oauth/access_token/'.format(base)
        self.oauth = 'oauth1'
        self.token = ''
        self.service = ''
        self.scope = ''
        if not ServicesMgr.instance:
            ServicesMgr.instance = ServicesMgr.__ServicesMgr(arg)
        else:
            ServicesMgr.instance.val = arg

    def __getattr__(self, name):
        return getattr(self.instance, name)

    def __str__(self):
        return self.name

    @staticmethod
    def _get_content(data, which_content):
        """
            get the content that could be hidden
            in the middle of "content" or "summary detail"
            from the data of the provider
        """
        content = ''
        if data.get(which_content):
            if type(data.get(which_content)) is list or\
               type(data.get(which_content)) is tuple or\
               type(data.get(which_content)) is dict:
                if 'value' in data.get(which_content)[0]:
                    content = data.get(which_content)[0].value
            else:
                if type(data.get(which_content)) is str:
                    content = data.get(which_content)
                else:
                    # if not str or list or tuple
                    # or dict it could be feedparser.FeedParserDict
                    # so get the item value
                    content = data.get(which_content)['value']
        return content

    @staticmethod
    def set_title(data):
        """
            handle the title from the data
            :param data: contains the data from the provider
            :type data: dict
            :rtype: string
        """
        title = (data.get('title') if data.get('title') else data.get('link'))

        return title

    def set_content(self, data):
        """
            handle the content from the data
            :param data: contains the data from the provider
            :type data: dict
            :rtype: string
        """
        content = self._get_content(data, 'content')

        if content == '':
            content = self._get_content(data, 'summary_detail')

        if content == '':
            if data.get('description'):
                content = data.get('description')

        return content

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

            :param kwargs: contain keyword args : trigger_id and model name
            :type kwargs: dict
            :rtype: model
        """
        model = get_model('django_th', kwargs['model_name'])

        return model.objects.get(trigger_id=kwargs['trigger_id'])

    def process_data(self, **kwargs):
        """
             get the data from the cache
            :param kwargs: contain keyword args : trigger_id at least
            :type kwargs: dict
        """
        cache = caches[kwargs.get('cache_stack')]
        cache_data = cache.get(kwargs.get('cache_stack') + '_' +
                               kwargs.get('trigger_id'))
        return PublishingLimit.get_data(kwargs.get('cache_stack'),
                                        cache_data, kwargs.get('trigger_id'))

    def save_data(self, trigger_id, **data):
        """
            used to save data to the service
            but first of all
            make some work about the data to find
            and the data to convert
            :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
        """
        title = self.set_title(data)
        title = HtmlEntities(title).html_entity_decode
        content = self.set_content(data)
        content = HtmlEntities(content).html_entity_decode
        if data.get('output_format'):
            # pandoc to convert tools
            import pypandoc
            content = pypandoc.convert(content,
                                       data.get('output_format'),
                                       format='html')
        return title, content

    def auth(self, request):
        """
            get the auth of the services
            :param request: contains the current session
            :type request: dict
            :rtype: dict
        """
        request_token = self.get_request_token(request)

        return request_token

    def callback_url(self, request):
        """
            the url to go back after the external service call
            :param request: contains the current session
            :type request: dict
            :rtype: string
        """
        service = self.service.split('Service')[1].lower()
        return_to = '{service}_callback'.format(service=service)
        return '%s://%s%s' % (request.scheme, request.get_host(),
                              reverse(return_to))

    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
        """
        if self.oauth == 'oauth1':
            token = self.callback_oauth1(request, **kwargs)
        else:
            token = self.callback_oauth2(request)

        service_name = ServicesActivated.objects.get(name=self.service)

        UserService.objects.filter(user=request.user,
                                   name=service_name
                                   ).update(token=token)
        back = self.service.split('Service')[1].lower()
        back_to = '{back_to}/callback.html'.format(back_to=back)
        return back_to

    def callback_oauth1(self, request, **kwargs):
        """
            Process for oAuth 1
            :param request: contains the current session
            :param kwargs: keyword args
            :type request: dict
            :type kwargs: dict
            :rtype: string
        """
        if kwargs.get('access_token') == '' \
           or kwargs.get('access_token') is None:
            access_token = self.get_access_token(
                request.session['oauth_token'],
                request.session['oauth_token_secret'],
                request.GET.get('oauth_verifier', '')
            )
            print(request.session)
        else:
            access_token = kwargs.get('access_token')

        if type(access_token) == str:
            token = access_token
        else:
            token = '#TH#'.join((access_token.get('oauth_token'),
                                 access_token.get('oauth_token_secret')))
        return token

    def callback_oauth2(self, request):
        """
            Process for oAuth 2
            :param request: contains the current session
            :return:
        """
        callback_url = self.callback_url(request)

        oauth = OAuth2Session(client_id=self.consumer_key,
                              redirect_uri=callback_url,
                              scope=self.scope)
        request_token = oauth.fetch_token(self.REQ_TOKEN,
                                          code=request.GET.get('code', ''),
                                          authorization_response=callback_url,
                                          client_secret=self.consumer_secret,
                                          scope=self.scope,
                                          verify=False)
        return request_token.get('access_token')

    def get_request_token(self, request):
        """
           request the token to the external service
        """
        if self.oauth == 'oauth1':
            oauth = OAuth1Session(self.consumer_key,
                                  client_secret=self.consumer_secret)
            request_token = oauth.fetch_request_token(self.REQ_TOKEN)
            # Save the request token information for later
            request.session['oauth_token'] = request_token['oauth_token']
            request.session['oauth_token_secret'] = request_token[
                'oauth_token_secret']
            print(request.session)
            return request_token
        else:
            callback_url = self.callback_url(request)
            oauth = OAuth2Session(client_id=self.consumer_key,
                                  redirect_uri=callback_url,
                                  scope=self.scope)
            authorization_url, state = oauth.authorization_url(self.AUTH_URL)
            return authorization_url

    def get_access_token(self, oauth_token, oauth_token_secret,
                         oauth_verifier):
        """
           get the access token
            the url to go back after the external service call
            :param oauth_token: oauth token
            :param oauth_token_secret: oauth secret token
            :param oauth_verifier: oauth verifier
            :type oauth_token: string
            :type oauth_token_secret: string
            :type oauth_verifier: string
            :rtype: dict
        """
        # Using OAuth1Session
        oauth = OAuth1Session(self.consumer_key,
                              client_secret=self.consumer_secret,
                              resource_owner_key=oauth_token,
                              resource_owner_secret=oauth_token_secret,
                              verifier=oauth_verifier)
        oauth_tokens = oauth.fetch_access_token(self.ACC_TOKEN)

        return oauth_tokens

    def reset_failed(self, pk):
        """
            reset failed counter
        :param pk:
        :return:
        """
        TriggerService.objects.filter(consumer__name__id=pk).update(
            consumer_failed=0, provider_failed=0)
        TriggerService.objects.filter(provider__name__id=pk).update(
            consumer_failed=0, provider_failed=0)

    def send_signal(self, trigger_id, title, link=''):

        t = TriggerService.objects.get(id=trigger_id)

        if t.provider.duration != 'n':

            kwargs = {'user': t.user, 'title': title,
                      'link': link, 'duration': t.provider.duration}

            signals.digest_event.send(sender=t.provider.name, **kwargs)


1			# coding: utf-8
2			# Using OAuth1Session
3			from requests_oauthlib import OAuth1Session, OAuth2Session
4
5			# django stuff
6			from django.core.cache import caches
7			from django.core.urlresolvers import reverse
8
9			try:
10			from django.apps import apps
11			get_model = apps.get_model
12			except ImportError:
13			from django.db.models.loading import get_model
14
15			# django_th stuff
16			from django_th import signals
17			from django_th.models import UserService, ServicesActivated, TriggerService
18			from django_th.publishing_limit import PublishingLimit
19			from django_th.html_entities import HtmlEntities
20
21
22			class ServicesMgr(object):
23			"""
24			Main Service Class
25			"""
26			name = ''
27			title = ''
28			body = ''
29			data = {}
30
31			class __ServicesMgr:
32
33			def __init__(self, arg):
34			self.val = arg
35
36			def __str__(self):
37			return repr(self) + self.val
38
39			instance = None
40
41			def __init__(self, arg, **kwargs):
42			base = 'https://www.urltotheserviceapi.com'
43			self.AUTH_URL = '{}/api/rest/v1/oauth/authorize/'.format(base)
44			self.REQ_TOKEN = '{}/api/rest/v1/oauth/request_token/'.format(base)
45			self.ACC_TOKEN = '{}/api/rest/v1/oauth/access_token/'.format(base)
46			self.oauth = 'oauth1'
47			self.token = ''
48			self.service = ''
49			self.scope = ''
50			if not ServicesMgr.instance:
51			ServicesMgr.instance = ServicesMgr.__ServicesMgr(arg)
52			else:
53			ServicesMgr.instance.val = arg
54
55			def __getattr__(self, name):
56			return getattr(self.instance, name)
57
58			def __str__(self):
59			return self.name
60
61			@staticmethod
62			def _get_content(data, which_content):
63			"""
64			get the content that could be hidden
65			in the middle of "content" or "summary detail"
66			from the data of the provider
67			"""
68			content = ''
69			if data.get(which_content):
70			if type(data.get(which_content)) is list or\
71			type(data.get(which_content)) is tuple or\
72			type(data.get(which_content)) is dict:
73			if 'value' in data.get(which_content)[0]:
74			content = data.get(which_content)[0].value
75			else:
76			if type(data.get(which_content)) is str:
77			content = data.get(which_content)
78			else:
79			# if not str or list or tuple
80			# or dict it could be feedparser.FeedParserDict
81			# so get the item value
82			content = data.get(which_content)['value']
83			return content
84
85			@staticmethod
86			def set_title(data):
87			"""
88			handle the title from the data
89			:param data: contains the data from the provider
90			:type data: dict
91			:rtype: string
92			"""
93			title = (data.get('title') if data.get('title') else data.get('link'))
94
95			return title
96
97			def set_content(self, data):
98			"""
99			handle the content from the data
100			:param data: contains the data from the provider
101			:type data: dict
102			:rtype: string
103			"""
104			content = self._get_content(data, 'content')
105
106			if content == '':
107			content = self._get_content(data, 'summary_detail')
108
109			if content == '':
110			if data.get('description'):
111			content = data.get('description')
112
113			return content
114
115			def read_data(self, **kwargs):
116			"""
117			get the data from the service
118
119			:param kwargs: contain keyword args : trigger_id and model name
120			:type kwargs: dict
121			:rtype: model
122			"""
123			model = get_model('django_th', kwargs['model_name'])
124
125			return model.objects.get(trigger_id=kwargs['trigger_id'])
126
127			def process_data(self, **kwargs):
128			"""
129			get the data from the cache
130			:param kwargs: contain keyword args : trigger_id at least
131			:type kwargs: dict
132			"""
133			cache = caches[kwargs.get('cache_stack')]
134			cache_data = cache.get(kwargs.get('cache_stack') + '_' +
135			kwargs.get('trigger_id'))
136			return PublishingLimit.get_data(kwargs.get('cache_stack'),
137			cache_data, kwargs.get('trigger_id'))
138
139			def save_data(self, trigger_id, **data):
140			"""
141			used to save data to the service
142			but first of all
143			make some work about the data to find
144			and the data to convert
145			:param trigger_id: trigger ID from which to save data
146			:param data: the data to check to be used and save
147			:type trigger_id: int
148			:type data: dict
149			:return: the status of the save statement
150			:rtype: boolean
151			"""
152			title = self.set_title(data)
153			title = HtmlEntities(title).html_entity_decode
154			content = self.set_content(data)
155			content = HtmlEntities(content).html_entity_decode
156			if data.get('output_format'):
157			# pandoc to convert tools
158			import pypandoc
159			content = pypandoc.convert(content,
160			data.get('output_format'),
161			format='html')
162			return title, content
163
164			def auth(self, request):
165			"""
166			get the auth of the services
167			:param request: contains the current session
168			:type request: dict
169			:rtype: dict
170			"""
171			request_token = self.get_request_token(request)
172
173			return request_token
174
175			def callback_url(self, request):
176			"""
177			the url to go back after the external service call
178			:param request: contains the current session
179			:type request: dict
180			:rtype: string
181			"""
182			service = self.service.split('Service')[1].lower()
183			return_to = '{service}_callback'.format(service=service)
184			return '%s://%s%s' % (request.scheme, request.get_host(),
185			reverse(return_to))
186
187			def callback(self, request, **kwargs):
188			"""
189			Called from the Service when the user accept to activate it
190			the url to go back after the external service call
191			:param request: contains the current session
192			:param kwargs: keyword args
193			:type request: dict
194			:type kwargs: dict
195			:rtype: string
196			"""
197			if self.oauth == 'oauth1':
198			token = self.callback_oauth1(request, **kwargs)
199			else:
200			token = self.callback_oauth2(request)
201
202			service_name = ServicesActivated.objects.get(name=self.service)
203
204			UserService.objects.filter(user=request.user,
205			name=service_name
206			).update(token=token)
207			back = self.service.split('Service')[1].lower()
208			back_to = '{back_to}/callback.html'.format(back_to=back)
209			return back_to
210
211			def callback_oauth1(self, request, **kwargs):
212			"""
213			Process for oAuth 1
214			:param request: contains the current session
215			:param kwargs: keyword args
216			:type request: dict
217			:type kwargs: dict
218			:rtype: string
219			"""
220			if kwargs.get('access_token') == '' \
221			or kwargs.get('access_token') is None:
222			access_token = self.get_access_token(
223			request.session['oauth_token'],
224			request.session['oauth_token_secret'],
225			request.GET.get('oauth_verifier', '')
226			)
227			print(request.session)
228			else:
229			access_token = kwargs.get('access_token')
230
231			if type(access_token) == str:
232			token = access_token
233			else:
234			token = '#TH#'.join((access_token.get('oauth_token'),
235			access_token.get('oauth_token_secret')))
236			return token
237
238			def callback_oauth2(self, request):
239			"""
240			Process for oAuth 2
241			:param request: contains the current session
242			:return:
243			"""
244			callback_url = self.callback_url(request)
245
246			oauth = OAuth2Session(client_id=self.consumer_key,
247			redirect_uri=callback_url,
248			scope=self.scope)
249			request_token = oauth.fetch_token(self.REQ_TOKEN,
250			code=request.GET.get('code', ''),
251			authorization_response=callback_url,
252			client_secret=self.consumer_secret,
253			scope=self.scope,
254			verify=False)
255			return request_token.get('access_token')
256
257			def get_request_token(self, request):
258			"""
259			request the token to the external service
260			"""
261			if self.oauth == 'oauth1':
262			oauth = OAuth1Session(self.consumer_key,
263			client_secret=self.consumer_secret)
264			request_token = oauth.fetch_request_token(self.REQ_TOKEN)
265			# Save the request token information for later
266			request.session['oauth_token'] = request_token['oauth_token']
267			request.session['oauth_token_secret'] = request_token[
268			'oauth_token_secret']
269			print(request.session)
270			return request_token
271			else:
272			callback_url = self.callback_url(request)
273			oauth = OAuth2Session(client_id=self.consumer_key,
274			redirect_uri=callback_url,
275			scope=self.scope)
276			authorization_url, state = oauth.authorization_url(self.AUTH_URL)
277			return authorization_url
278
279			def get_access_token(self, oauth_token, oauth_token_secret,
280			oauth_verifier):
281			"""
282			get the access token
283			the url to go back after the external service call
284			:param oauth_token: oauth token
285			:param oauth_token_secret: oauth secret token
286			:param oauth_verifier: oauth verifier
287			:type oauth_token: string
288			:type oauth_token_secret: string
289			:type oauth_verifier: string
290			:rtype: dict
291			"""
292			# Using OAuth1Session
293			oauth = OAuth1Session(self.consumer_key,
294			client_secret=self.consumer_secret,
295			resource_owner_key=oauth_token,
296			resource_owner_secret=oauth_token_secret,
297			verifier=oauth_verifier)
298			oauth_tokens = oauth.fetch_access_token(self.ACC_TOKEN)
299
300			return oauth_tokens
301
302			def reset_failed(self, pk):
303			"""
304			reset failed counter
305			:param pk:
306			:return:
307			"""
308			TriggerService.objects.filter(consumer__name__id=pk).update(
309			consumer_failed=0, provider_failed=0)
310			TriggerService.objects.filter(provider__name__id=pk).update(
311			consumer_failed=0, provider_failed=0)
312
313			def send_signal(self, trigger_id, title, link=''):
314
315			t = TriggerService.objects.get(id=trigger_id)
316
317			if t.provider.duration != 'n':
318
319			kwargs = {'user': t.user, 'title': title,
320			'link': link, 'duration': t.provider.duration}
321
322			signals.digest_event.send(sender=t.provider.name, **kwargs)
323

foxmask / django-th

Push — master ( 856bc1...ee6984 )

__ServicesMgr.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ServicesMgr.init__() A