__ServicesMgr.__init__() - Code Metrics - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

ServicesMgr.init__() A
last analyzed 2017-10-09 22:02 UTC

↳ 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
from django.conf import settings

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:  # NOQA

        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(kwargs['app_label'], 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['django_th']
        cache_data = cache.get(kwargs.get('cache_stack') + '_' +
                               kwargs.get('trigger_id'))
        return PublishingLimit.get_data(kwargs.get('cache_stack'),
                                        cache_data,
                                        int(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_digest_event(self, trigger_id, title, link=''):
        """
        handling of the signal of digest
        :param trigger_id:
        :param title:
        :param link:
        :return:
        """
        if settings.DJANGO_TH.get('digest_event'):

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

foxmask / django-th

__ServicesMgr.__init__() A last analyzed 2017-10-09 22:02 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ServicesMgr.init__() A
last analyzed 2017-10-09 22:02 UTC