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

Digest.str() A
last analyzed 2017-10-09 22:02 UTC

↳ Parent: Digest

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

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

# coding: utf-8
import arrow

from django.conf import settings
from django.contrib.auth.models import User
from django.core.exceptions import ObjectDoesNotExist
from django.db import models
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.utils.timezone import now
from django.utils.translation import ugettext_lazy as _

from logging import getLogger

from django_th.signals import digest_event
from django_th.tools import warn_user_and_admin

logger = getLogger('django_th.trigger_happy')


class ServicesActivated(models.Model):

    """
        Services Activated from the admin
    """
    name = models.CharField(max_length=200, unique=True)
    status = models.BooleanField(default=False)
    auth_required = models.BooleanField(default=True)
    self_hosted = models.BooleanField(default=False)
    description = models.CharField(max_length=200)

    class Meta:
        verbose_name = 'Services'
        verbose_name_plural = 'Services'

    def show(self):
        """

        :return: string representing object
        """
        return "Service Activated %s %s %s %s" % (self.name, self.status,
                                                  self.auth_required,
                                                  self.description)

    def __str__(self):
        return self.name


class UserService(models.Model):

    """
        UserService a model to link service and user
    """
    DAY = 'd'
    WEEK = 'w'
    MONTH = 'm'
    NONE = 'n'
    DURATION = (
        (DAY, _('Day')),
        (WEEK, _('Week')),
        (MONTH, _('Month')),
        (NONE, _('None'))
    )

    user = models.ForeignKey(User)
    token = models.CharField(max_length=255, blank=True)
    name = models.ForeignKey(
        ServicesActivated, to_field='name', related_name='+')
    username = models.CharField(
        _('username'), max_length=255, default='', blank=True)
    password = models.CharField(
        _('password'), max_length=128, default='', blank=True)
    host = models.CharField(_('host'), max_length=255, default='', blank=True)
    client_id = models.CharField(
        _('client id'), max_length=255, default='', blank=True)
    client_secret = models.CharField(
        _('client secret'), max_length=255, default='', blank=True)
    duration = models.CharField(max_length=1, choices=DURATION, default=NONE)

    counter_ok = models.IntegerField(default=0)
    counter_ko = models.IntegerField(default=0)

    def show(self):
        """

        :return: string representing object
        """
        return "User Service %s %s %s" % (self.user, self.token, self.name)

    def __str__(self):
        return self.name.name


class TriggerService(models.Model):

    """
        TriggerService
    """
    provider = models.ForeignKey(UserService, related_name='+', blank=True)
    consumer = models.ForeignKey(UserService, related_name='+', blank=True)
    description = models.CharField(max_length=200)
    user = models.ForeignKey(User)
    date_created = models.DateField(auto_now_add=True)
    date_triggered = models.DateTimeField(null=True)
    status = models.BooleanField(default=False)
    result = models.CharField(max_length=255, default='')
    date_result = models.DateTimeField(auto_now=True, null=True)
    provider_failed = models.IntegerField(db_index=True, default=0)
    consumer_failed = models.IntegerField(db_index=True, default=0)

    counter_ok = models.IntegerField(default=0)
    counter_ko = models.IntegerField(default=0)

    def show(self):
        """

        :return: string representing object
        """
        return "My Service %s - %s - %s - %s" % (self.user,
                                                 self.provider.name,
                                                 self.consumer.name,
                                                 self.description)

    def __str__(self):
        return "%s - %s - %s - %s" % (self.user,
                                      self.provider.name,
                                      self.consumer.name,
                                      self.description)


class Digest(models.Model):
    """
    Digest service to store the data from other service
    """
    user = models.ForeignKey(User)
    title = models.CharField(max_length=600)
    link = models.URLField()
    duration = models.CharField(max_length=1)
    date_end = models.DateField()
    provider = models.CharField(max_length=40)

    def show(self):
        """

        :return: string representing object
        """
        return "Digest %s - %s - %s - %s - %s - %s" % (
            self.user, self.provider, self.title, self.link, self.duration,
            self.date_end)

    def __str__(self):
        return "%s - %s - %s - %s - %s - %s" % (
            self.user, self.provider, self.title, self.link, self.duration,
            self.date_end)


def update_result(trigger_id, msg, status):
    """
    :param trigger_id: trigger id
    :param msg: result msg
    :param status: status of the handling of the current trigger
    :return:
    """
    # if status is True, reset *_failed counter
    if status:
        service = TriggerService.objects.get(id=trigger_id)

        TriggerService.objects.filter(id=trigger_id).\
            update(result=msg,
                   date_result=now(),
                   provider_failed=0,
                   consumer_failed=0,
                   counter_ok=service.counter_ok + 1)
        UserService.objects.filter(user=service.user,
                                   name=service.consumer.name).update(
            counter_ok=service.counter_ok + 1)
    # otherwise, add 1 to the consumer_failed
    else:
        service = TriggerService.objects.get(id=trigger_id)
        failed = service.consumer_failed + 1

        UserService.objects.filter(user=service.user, name=service.consumer.name
                                   ).update(counter_ko=service.counter_ko + 1)

        if failed > settings.DJANGO_TH.get('failed_tries', 5):
            TriggerService.objects.filter(id=trigger_id).\
                update(result=msg, date_result=now(), status=False,
                       counter_ko=service.counter_ko + 1)
        else:
            TriggerService.objects.filter(id=trigger_id).\
                update(result=msg, date_result=now(), consumer_failed=failed,
                       counter_ko=service.counter_ko + 1)

        warn_user_and_admin('consumer', service)


def th_create_user_profile(sender, instance, created, **kwargs):
    # create the default service that does not
    # need any third party auth
    user = instance
    if user.last_login is None:
        services = ('ServiceRss', 'ServicePelican', )
        for service in services:
            if any(service in s for s in settings.TH_SERVICES):
                try:
                    sa = ServicesActivated.objects.get(name=service)
                    UserService.objects.get_or_create(user=user, name=sa)
                except ObjectDoesNotExist:
                    logger.debug("A new user %s has been connected but %s "
                                 "could not be added to his services because "
                                 "the service is present in TH_SERVICES but not"
                                 " activated from the Admin Panel" %
                                 (user, service))


post_save.connect(th_create_user_profile, sender=User,
                  dispatch_uid="create_user_profile")


@receiver(digest_event)
def digest_save(sender, **kwargs):
    """

    :param sender:
    :param kwargs:
    :return:
    """
    # set the deadline of the publication of the digest data
    duration = kwargs.get('duration')
    if duration not in ('d', 'w', 'm'):
        return
    # get the current date
    now = arrow.utcnow().to(settings.TIME_ZONE)

    # set the deadline
    if duration == 'd':
        # set tomorrow
        tomorrow = now.shift(days=+1)
        date_end = tomorrow.date()  # noqa extrat the date part
    elif duration == 'w':
        # set next week
        next_week = now.shift(weeks=+1)
        date_end = next_week.date()
    else:
        # set next month
        next_month = now.shift(months=+1)
        date_end = next_month.date()

    Digest.objects.create(user=kwargs.get('user'),
                          title=kwargs.get('title'),
                          link=kwargs.get('link'),
                          duration=duration,
                          date_end=str(date_end),
                          provider=sender)


1			# coding: utf-8
2			import arrow
3
4			from django.conf import settings
5			from django.contrib.auth.models import User
6			from django.core.exceptions import ObjectDoesNotExist
7			from django.db import models
8			from django.db.models.signals import post_save
9			from django.dispatch import receiver
10			from django.utils.timezone import now
11			from django.utils.translation import ugettext_lazy as _
12
13			from logging import getLogger
14
15			from django_th.signals import digest_event
16			from django_th.tools import warn_user_and_admin
17
18			logger = getLogger('django_th.trigger_happy')
19
20
21			class ServicesActivated(models.Model):
22
23			"""
24			Services Activated from the admin
25			"""
26			name = models.CharField(max_length=200, unique=True)
27			status = models.BooleanField(default=False)
28			auth_required = models.BooleanField(default=True)
29			self_hosted = models.BooleanField(default=False)
30			description = models.CharField(max_length=200)
31
32			class Meta:
33			verbose_name = 'Services'
34			verbose_name_plural = 'Services'
35
36			def show(self):
37			"""
38
39			:return: string representing object
40			"""
41			return "Service Activated %s %s %s %s" % (self.name, self.status,
42			self.auth_required,
43			self.description)
44
45			def __str__(self):
46			return self.name
47
48
49			class UserService(models.Model):
50
51			"""
52			UserService a model to link service and user
53			"""
54			DAY = 'd'
55			WEEK = 'w'
56			MONTH = 'm'
57			NONE = 'n'
58			DURATION = (
59			(DAY, _('Day')),
60			(WEEK, _('Week')),
61			(MONTH, _('Month')),
62			(NONE, _('None'))
63			)
64
65			user = models.ForeignKey(User)
66			token = models.CharField(max_length=255, blank=True)
67			name = models.ForeignKey(
68			ServicesActivated, to_field='name', related_name='+')
69			username = models.CharField(
70			_('username'), max_length=255, default='', blank=True)
71			password = models.CharField(
72			_('password'), max_length=128, default='', blank=True)
73			host = models.CharField(_('host'), max_length=255, default='', blank=True)
74			client_id = models.CharField(
75			_('client id'), max_length=255, default='', blank=True)
76			client_secret = models.CharField(
77			_('client secret'), max_length=255, default='', blank=True)
78			duration = models.CharField(max_length=1, choices=DURATION, default=NONE)
79
80			counter_ok = models.IntegerField(default=0)
81			counter_ko = models.IntegerField(default=0)
82
83			def show(self):
84			"""
85
86			:return: string representing object
87			"""
88			return "User Service %s %s %s" % (self.user, self.token, self.name)
89
90			def __str__(self):
91			return self.name.name
92
93
94			class TriggerService(models.Model):
95
96			"""
97			TriggerService
98			"""
99			provider = models.ForeignKey(UserService, related_name='+', blank=True)
100			consumer = models.ForeignKey(UserService, related_name='+', blank=True)
101			description = models.CharField(max_length=200)
102			user = models.ForeignKey(User)
103			date_created = models.DateField(auto_now_add=True)
104			date_triggered = models.DateTimeField(null=True)
105			status = models.BooleanField(default=False)
106			result = models.CharField(max_length=255, default='')
107			date_result = models.DateTimeField(auto_now=True, null=True)
108			provider_failed = models.IntegerField(db_index=True, default=0)
109			consumer_failed = models.IntegerField(db_index=True, default=0)
110
111			counter_ok = models.IntegerField(default=0)
112			counter_ko = models.IntegerField(default=0)
113
114			def show(self):
115			"""
116
117			:return: string representing object
118			"""
119			return "My Service %s - %s - %s - %s" % (self.user,
120			self.provider.name,
121			self.consumer.name,
122			self.description)
123
124			def __str__(self):
125			return "%s - %s - %s - %s" % (self.user,
126			self.provider.name,
127			self.consumer.name,
128			self.description)
129
130
131			class Digest(models.Model):
132			"""
133			Digest service to store the data from other service
134			"""
135			user = models.ForeignKey(User)
136			title = models.CharField(max_length=600)
137			link = models.URLField()
138			duration = models.CharField(max_length=1)
139			date_end = models.DateField()
140			provider = models.CharField(max_length=40)
141
142			def show(self):
143			"""
144
145			:return: string representing object
146			"""
147			return "Digest %s - %s - %s - %s - %s - %s" % (
148			self.user, self.provider, self.title, self.link, self.duration,
149			self.date_end)
150
151			def __str__(self):
152			return "%s - %s - %s - %s - %s - %s" % (
153			self.user, self.provider, self.title, self.link, self.duration,
154			self.date_end)
155
156
157			def update_result(trigger_id, msg, status):
158			"""
159			:param trigger_id: trigger id
160			:param msg: result msg
161			:param status: status of the handling of the current trigger
162			:return:
163			"""
164			# if status is True, reset *_failed counter
165			if status:
166			service = TriggerService.objects.get(id=trigger_id)
167
168			TriggerService.objects.filter(id=trigger_id).\
169			update(result=msg,
170			date_result=now(),
171			provider_failed=0,
172			consumer_failed=0,
173			counter_ok=service.counter_ok + 1)
174			UserService.objects.filter(user=service.user,
175			name=service.consumer.name).update(
176			counter_ok=service.counter_ok + 1)
177			# otherwise, add 1 to the consumer_failed
178			else:
179			service = TriggerService.objects.get(id=trigger_id)
180			failed = service.consumer_failed + 1
181
182			UserService.objects.filter(user=service.user, name=service.consumer.name
183			).update(counter_ko=service.counter_ko + 1)
184
185			if failed > settings.DJANGO_TH.get('failed_tries', 5):
186			TriggerService.objects.filter(id=trigger_id).\
187			update(result=msg, date_result=now(), status=False,
188			counter_ko=service.counter_ko + 1)
189			else:
190			TriggerService.objects.filter(id=trigger_id).\
191			update(result=msg, date_result=now(), consumer_failed=failed,
192			counter_ko=service.counter_ko + 1)
193
194			warn_user_and_admin('consumer', service)
195
196
197			def th_create_user_profile(sender, instance, created, **kwargs):
198			# create the default service that does not
199			# need any third party auth
200			user = instance
201			if user.last_login is None:
202			services = ('ServiceRss', 'ServicePelican', )
203			for service in services:
204			if any(service in s for s in settings.TH_SERVICES):
205			try:
206			sa = ServicesActivated.objects.get(name=service)
207			UserService.objects.get_or_create(user=user, name=sa)
208			except ObjectDoesNotExist:
209			logger.debug("A new user %s has been connected but %s "
210			"could not be added to his services because "
211			"the service is present in TH_SERVICES but not"
212			" activated from the Admin Panel" %
213			(user, service))
214
215
216			post_save.connect(th_create_user_profile, sender=User,
217			dispatch_uid="create_user_profile")
218
219
220			@receiver(digest_event)
221			def digest_save(sender, **kwargs):
222			"""
223
224			:param sender:
225			:param kwargs:
226			:return:
227			"""
228			# set the deadline of the publication of the digest data
229			duration = kwargs.get('duration')
230			if duration not in ('d', 'w', 'm'):
231			return
232			# get the current date
233			now = arrow.utcnow().to(settings.TIME_ZONE)
234
235			# set the deadline
236			if duration == 'd':
237			# set tomorrow
238			tomorrow = now.shift(days=+1)
239			date_end = tomorrow.date() # noqa extrat the date part
240			elif duration == 'w':
241			# set next week
242			next_week = now.shift(weeks=+1)
243			date_end = next_week.date()
244			else:
245			# set next month
246			next_month = now.shift(months=+1)
247			date_end = next_month.date()
248
249			Digest.objects.create(user=kwargs.get('user'),
250			title=kwargs.get('title'),
251			link=kwargs.get('link'),
252			duration=duration,
253			date_end=str(date_end),
254			provider=sender)
255

foxmask / django-th

Digest.__str__() A last analyzed 2017-10-09 22:02 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

Digest.str() A
last analyzed 2017-10-09 22:02 UTC