path: root/ishtar_common/models.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (C) 2010-2012 Étienne Loks  <etienne.loks_AT_peacefrogsDOTnet>

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

# See the file COPYING for details.

"""
Models description
"""
import datetime

from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist, ValidationError
from django.core.validators import validate_slug
from django.utils.translation import ugettext_lazy as _, ugettext
from django.utils.safestring import SafeUnicode, mark_safe
from django.core.urlresolvers import reverse, NoReverseMatch
from django.db.utils import DatabaseError
from django.db.models import Q, Max, Count
from django.db.models.signals import post_save

from django.contrib.auth.models import User
from django.contrib.gis.db import models
from django.contrib import admin

from simple_history.models import HistoricalRecords as BaseHistoricalRecords

def post_save_user(sender, **kwargs):
    user = kwargs['instance']
    ishtaruser = None
    q = IshtarUser.objects.filter(username=user.username)
    if not q.count():
        ishtaruser = IshtarUser.create_from_user(user)
    else:
        ishtaruser = q.all()[0]
    if ishtaruser.is_superuser \
       and ishtaruser.person.person_type.txt_idx != 'administrator':
        ishtaruser.person.person_type = PersonType.objects.get(
                                                    txt_idx='administrator')
        ishtaruser.person.save()
post_save.connect(post_save_user, sender=User)

# HistoricalRecords enhancement: don't save identical versions
class HistoricalRecords(BaseHistoricalRecords):
    def create_historical_record(self, instance, type):
        manager = getattr(instance, self.manager_name)
        attrs = {}
        for field in instance._meta.fields:
            attrs[field.attname] = getattr(instance, field.attname)
        history = instance.history.all()
        if not history:
            manager.create(history_type=type, **attrs)
            return
        old_instance = history[0]
        for field in instance._meta.fields:
            if getattr(old_instance, field.attname) != attrs[field.attname]:
                manager.create(history_type=type, **attrs)
                return

# valid ID validator for models
def valid_id(cls):
    def func(value):
        try:
            cls.objects.get(pk=value)
        except ObjectDoesNotExist:
            raise ValidationError(_(u"Not a valid item."))
    return func

def valid_ids(cls):
    def func(value):
        if "," in value:
            value = value.split(",")
        for v in value:
            try:
                cls.objects.get(pk=v)
            except ObjectDoesNotExist:
                raise ValidationError(
                                   _(u"An item selected is not a valid item."))
    return func

# unique validator for models
def is_unique(cls, field):
    def func(value):
        query = {field:value}
        try:
            assert cls.objects.filter(**query).count() == 0
        except AssertionError:
            raise ValidationError(_(u"This item already exist."))
    return func

class OwnPerms:
    """
    Manage special permissions for object's owner
    """
    @classmethod
    def get_query_owns(cls, user):
        """
        Query object to get own items
        """
        return None # implement for each object

    def is_own(self, user):
        """
        Check if the current object is owned by the user
        """
        query = self.get_query_owns(user)
        if not query:
            return False
        query = query & Q(pk=self.pk)
        return cls.objects.filter(query).count()

    @classmethod
    def has_item_of(cls, user):
        """
        Check if the user own some items
        """
        query = cls.get_query_owns(user)
        if not query:
            return False
        return cls.objects.filter(query).count()

    @classmethod
    def get_owns(cls, user):
        """
        Get Own items
        """
        if isinstance(user, User):
            user = IshtarUser.objects.get(user_ptr=user)
        if user.is_anonymous():
            return []
        query = cls.get_query_owns(user)
        if not query:
            return []
        return cls.objects.filter(query).order_by(*cls._meta.ordering).all()

class GeneralType(models.Model):
    """
    Abstract class for "types"
    """
    label = models.CharField(_(u"Label"), max_length=100)
    txt_idx = models.CharField(_(u"Textual ID"),
                         validators=[validate_slug], max_length=30, unique=True)
    comment = models.TextField(_(u"Comment"), blank=True, null=True)
    available = models.BooleanField(_(u"Available"))
    HELP_TEXT = u""

    class Meta:
        abstract = True

    def __unicode__(self):
        return self.label

    def short_label(self):
        return self.label

    @classmethod
    def get_help(cls, dct={}, exclude=[]):
        help_text = cls.HELP_TEXT
        c_rank = -1
        help_items = u"\n"
        for item in cls.get_types(dct=dct, instances=True, exclude=exclude):
            if hasattr(item, '__iter__'):
                # TODO: manage multiple levels
                continue
            if not item.comment:
                continue
            if c_rank > item.rank:
                help_items += u"</dl>\n"
            elif c_rank < item.rank:
                help_items += u"<dl>\n"
            c_rank = item.rank
            help_items += u"<dt>%s</dt><dd>%s</dd>" % (item.label,
                                u"<br/>".join(item.comment.split('\n')))
        c_rank += 1
        if c_rank:
            help_items += c_rank*u"</dl>"
        if help_text or help_items != u'\n':
            return mark_safe(help_text + help_items)
        return u""

    @classmethod
    def get_types(cls, dct={}, instances=False, exclude=[]):
        base_dct = dct.copy()
        if hasattr(cls, 'parent'):
            return cls._get_parent_types(base_dct, instances, exclude=exclude)
        return cls._get_types(base_dct, instances, exclude=exclude)

    @classmethod
    def _get_types(cls, dct={}, instances=False, exclude=[]):
        dct['available'] = True
        if not instances:
            yield ('', '--')
        items = cls.objects.filter(**dct)
        if exclude:
            items = items.exclude(txt_idx__in=exclude)
        for item in items.order_by(*cls._meta.ordering).all():
            if instances:
                item.rank = 0
                yield item
            else:
                yield (item.pk, _(unicode(item)))

    PREFIX = "&rsaquo; "

    @classmethod
    def _get_childs(cls, item, dct, prefix=0, instances=False, exclude=[]):
        prefix += 1
        dct['parent'] = item
        childs = cls.objects.filter(**dct)
        if exclude:
            childs = childs.exclude(txt_idx__in=exclude)
        if hasattr(cls, 'order'):
            childs = childs.order_by('order')
        for child in childs.all():
            if instances:
                child.rank = prefix
                yield child
            else:
                yield (child.pk, SafeUnicode(prefix*cls.PREFIX + \
                                         unicode(_(unicode(child))) ))
            for sub_child in cls._get_childs(child, dct, prefix, instances,
                                             exclude=exclude):
                yield sub_child

    @classmethod
    def _get_parent_types(cls, dct={}, instances=False, exclude=[]):
        dct['available'] = True
        if not instances:
            yield ('', '--')
        dct['parent'] = None
        items = cls.objects.filter(**dct)
        if exclude:
            items = items.exclude(txt_idx__in=exclude)
        if hasattr(cls, 'order'):
            items = items.order_by('order')
        for item in items.all():
            if instances:
                item.rank = 0
                yield item
            else:
                yield (item.pk, unicode(item))
            for child in cls._get_childs(item, dct, instances, exclude=exclude):
                yield child

class HistoryError(Exception):
    def __init__(self, value):
        self.value = value
    def __str__(self):
        return repr(self.value)

class BaseHistorizedItem(models.Model):
    history_modifier = models.ForeignKey(User, related_name='+',
                                      verbose_name=_(u"Last editor"))
    class Meta:
        abstract = True

    def save(self, *args, **kwargs):
        assert hasattr(self, 'history_modifier') == True
        super(BaseHistorizedItem, self).save(*args, **kwargs)
        return True

    def get_previous(self, step=None, date=None, strict=True):
        """
        Get a "step" previous state of the item
        """
        assert step or date
        historized = self.history.all()
        item = None
        if step:
            assert len(historized) > step
            item = historized[step]
        else:
            for step, item in enumerate(historized):
                if item.history_date == date:
                    break
            # ended with no match
            if item.history_date != date:
                return
        item._step = step
        if len(historized) != (step + 1):
            item._previous = historized[step + 1].history_date
        else:
            item._previous = None
        if step > 0:
            item._next = historized[step - 1].history_date
        else:
            item._next = None
        item.history_date = historized[step].history_date
        model = self.__class__
        for k in model._meta.get_all_field_names():
            field = model._meta.get_field_by_name(k)[0]
            if hasattr(field, 'rel') and field.rel:
                if not hasattr(item, k+'_id'):
                    setattr(item, k, getattr(self, k))
                    continue
                val = getattr(item, k+'_id')
                if not val:
                    setattr(item, k, None)
                    continue
                try:
                    val = field.rel.to.objects.get(pk=val)
                    setattr(item, k, val)
                except ObjectDoesNotExist:
                    if strict:
                        raise HistoryError(u"The class %s has no pk %d" % (
                                           unicode(field.rel.to), val))
                    setattr(item, k, None)
        item.pk = self.pk
        return item

    def rollback(self, date):
        """
        Rollback to a previous state
        """
        to_del, new_item = [], None
        for item in self.history.all():
            to_del.append(item)
            if item.history_date == date:
                new_item = item
                break
        if not new_item:
            raise HistoryError(u"The date to rollback to doesn't exist.")
        try:
            for f in self._meta.fields:
                k = f.name
                if k != 'id' and hasattr(self, k):
                    if not hasattr(new_item, k):
                        k = k + "_id"
                    setattr(self, k, getattr(new_item, k))
            self.save()
        except:
            raise HistoryError(u"The rollback has failed.")
        # clean the obsolete history
        for historized_item in to_del:
            historized_item.delete()

    def values(self):
        values = {}
        for f in self._meta.fields:
            k = f.name
            if k != 'id':
                values[k] = getattr(self, k)
        return values

    def get_show_url(self):
        try:
            return reverse('show-'+self.__class__.__name__.lower(),
                                            args=[self.pk, ''])
        except NoReverseMatch:
            return

class LightHistorizedItem(BaseHistorizedItem):
    history_date = models.DateTimeField(default=datetime.datetime.now)
    class Meta:
        abstract = True

    def save(self, *args, **kwargs):
        super(LightHistorizedItem, self).save(*args, **kwargs)
        return True

class Wizard(models.Model):
    url_name = models.CharField(_(u"URL name"), max_length=128, unique=True)
    class Meta:
        verbose_name = _(u"Wizard")
        ordering = ['url_name']

    def __unicode__(self):
        return unicode(self.url_name)

class WizardStep(models.Model):
    order = models.IntegerField(_(u"Order"))
    wizard = models.ForeignKey(Wizard, verbose_name=_(u"Wizard"))
    url_name = models.CharField(_(u"URL name"), max_length=128)
    name = models.CharField(_(u"Label"), max_length=128)
    class Meta:
        verbose_name = _(u"Wizard step")
        ordering = ['wizard', 'order']

    def __unicode__(self):
        return u"%s » %s" % (unicode(self.wizard), unicode(self.name))

class UserDashboard:
    def __init__(self):
        types = IshtarUser.objects.values('person__person_type',
                                          'person__person_type__label')
        self.types = types.annotate(number=Count('pk'))\
                          .order_by('person__person_type')

class Dashboard:
    def __init__(self, model):
        self.model = model
        self.total_number = model.get_total_number()
        history_model = self.model.history.model
        # last edited - created
        self.recents, self.lasts = [], []
        for last_lst, modif_type in ((self.lasts, '+'), (self.recents, '~')):
            last_ids = history_model.objects.values('id')\
                                            .annotate(hd=Max('history_date'))
            last_ids = last_ids.filter(history_type=modif_type)
            from archaeological_finds.models import Find
            if self.model == Find:
                last_ids = last_ids.filter(downstream_treatment_id__isnull=True)
                if modif_type == '+':
                   last_ids = last_ids.filter(upstream_treatment_id__isnull=True)
            last_ids = last_ids.order_by('-hd').distinct().all()[:5]
            for idx in last_ids:
                try:
                    obj = self.model.objects.get(pk=idx['id'])
                except:
                    # deleted object are always referenced in history
                    continue
                obj.history_date = idx['hd']
                last_lst.append(obj)
        # years
        self.years = model.get_years()
        self.years.sort()
        if not self.total_number or not self.years:
            return
        self.values = [('year', _(u"Year"), reversed(self.years))]
        # numbers
        self.numbers = [model.get_by_year(year).count() for year in self.years]
        self.values += [('number', _(u"Number"), reversed(self.numbers))]
        # calculate
        self.average = self.get_average()
        self.variance = self.get_variance()
        self.standard_deviation = self.get_standard_deviation()
        self.median = self.get_median()
        self.mode = self.get_mode()
        # by operation
        if not hasattr(model, 'get_by_operation'):
            return
        operations = model.get_operations()
        operation_numbers = [model.get_by_operation(op).count()
                             for op in operations]
        # calculate
        self.operation_average = self.get_average(operation_numbers)
        self.operation_variance = self.get_variance(operation_numbers)
        self.operation_standard_deviation = self.get_standard_deviation(
                                                       operation_numbers)
        self.operation_median = self.get_median(operation_numbers)
        operation_mode_pk = self.get_mode(dict(zip(operations,
                                                     operation_numbers)))
        if operation_mode_pk:
            from archaeological_operations.models import Operation
            self.operation_mode = unicode(Operation.objects.get(
                                                       pk=operation_mode_pk))

    def get_average(self, vals=[]):
        if not vals:
            vals = self.numbers
        return sum(vals)/len(vals)

    def get_variance(self, vals=[]):
        if not vals:
            vals = self.numbers
        avrg = self.get_average(vals)
        return self.get_average([(x-avrg)**2 for x in vals])

    def get_standard_deviation(self, vals=[]):
        if not vals:
            vals = self.numbers
        return round(self.get_variance(vals)**0.5, 3)

    def get_median(self, vals=[]):
        if not vals:
            vals = self.numbers
        len_vals = len(vals)
        vals.sort()
        if (len_vals % 2) == 1:
            return vals[len_vals/2]
        else:
            return (vals[len_vals/2-1] + vals[len_vals/2])/2.0

    def get_mode(self, vals={}):
        if not vals:
            vals = dict(zip(self.years, self.numbers))
        mx = max(vals.values())
        for v in vals:
            if vals[v] == mx:
                return v

class Department(models.Model):
    label = models.CharField(_(u"Label"), max_length=30)
    number = models.CharField(_(u"Number"), unique=True, max_length=3)

    class Meta:
        verbose_name = _(u"Department")
        verbose_name_plural = _(u"Departments")
        ordering = ['number']

    def __unicode__(self):
        return unicode(self.number) + settings.JOINT + self.label

class Address(BaseHistorizedItem):
    address = models.TextField(_(u"Address"), null=True, blank=True)
    address_complement = models.TextField(_(u"Address complement"), null=True,
                                          blank=True)
    postal_code = models.CharField(_(u"Postal code"), max_length=10, null=True,
                                   blank=True)
    town = models.CharField(_(u"Town"), max_length=30, null=True, blank=True)
    country = models.CharField(_(u"Country"), max_length=30, null=True,
                               blank=True)
    phone = models.CharField(_(u"Phone"), max_length=18, null=True, blank=True)
    mobile_phone = models.CharField(_(u"Mobile phone"), max_length=18,
                                    null=True, blank=True)
    history = HistoricalRecords()

    class Meta:
        abstract = True

class OrganizationType(GeneralType):
    class Meta:
        verbose_name = _(u"Organization type")
        verbose_name_plural = _(u"Organization types")

class Organization(Address, OwnPerms):
    name = models.CharField(_(u"Name"), max_length=100)
    organization_type = models.ForeignKey(OrganizationType,
                                          verbose_name=_(u"Type"))
    history = HistoricalRecords()
    class Meta:
        verbose_name = _(u"Organization")
        verbose_name_plural = _(u"Organizations")
        permissions = (
          ("view_own_organization", ugettext(u"Can view own Organization")),
          ("add_own_organization", ugettext(u"Can add own Organization")),
          ("change_own_organization", ugettext(u"Can change own Organization")),
          ("delete_own_organization", ugettext(u"Can delete own Organization")),
        )

    def __unicode__(self):
        return self.name

class PersonType(GeneralType):
    rights = models.ManyToManyField(WizardStep, verbose_name=_(u"Rights"))
    class Meta:
        verbose_name = _(u"Person type")
        verbose_name_plural = _(u"Person types")

class Person(Address, OwnPerms) :
    TYPE = (('Mr', _(u'Mr')),
            ('Ms', _(u'Miss')),
            ('Md', _(u'Mrs')),
            ('Dr', _(u'Doctor')),
            )
    title = models.CharField(_(u"Title"), max_length=2, choices=TYPE)
    surname = models.CharField(_(u"Surname"), max_length=20, blank=True,
                               null=True)
    name = models.CharField(_(u"Name"), max_length=30)
    email = models.CharField(_(u"Email"), max_length=40, blank=True, null=True)
    person_type = models.ForeignKey(PersonType, verbose_name=_(u"Type"))
    attached_to = models.ForeignKey('Organization',
                       verbose_name=_(u"Is attached to"), blank=True, null=True)

    class Meta:
        verbose_name = _(u"Person")
        verbose_name_plural = _(u"Persons")
        permissions = (
            ("view_person", ugettext(u"Can view Person")),
            ("view_own_person", ugettext(u"Can view own Person")),
            ("add_own_person", ugettext(u"Can add own Person")),
            ("change_own_person", ugettext(u"Can change own Person")),
            ("delete_own_person", ugettext(u"Can delete own Person")),
        )

    def __unicode__(self):
        lbl = u"%s %s" % (self.name, self.surname)
        if self.attached_to:
            lbl += settings.JOINT + unicode(self.attached_to)
        elif self.email:
            lbl += settings.JOINT + self.email
        return lbl

    def full_label(self):
        values = []
        if self.title:
            values = [unicode(_(self.title))]
        values += [unicode(getattr(self, attr))
                   for attr in ('surname', 'name', 'attached_to')
                               if getattr(self, attr)]
        return u" ".join(values)

class IshtarUser(User):
    person = models.ForeignKey(Person, verbose_name=_(u"Person"), unique=True)

    class Meta:
        verbose_name = _(u"Ishtar user")
        verbose_name_plural = _(u"Ishtar users")

    @classmethod
    def create_from_user(cls, user):
        default = user.username
        surname = user.first_name or default
        name = user.last_name or default
        email = user.email
        person_type = None
        if user.is_superuser:
            person_type = PersonType.objects.get(txt_idx='administrator')
        else:
            person_type = PersonType.objects.get(txt_idx='public_access')
        person = Person.objects.create(title='Mr', surname=surname,
                                       name=name, email=email,
                                       person_type=person_type,
                                       history_modifier=user)
        return IshtarUser.objects.create(user_ptr=user, person=person)

class AuthorType(GeneralType):
    class Meta:
        verbose_name = _(u"Author type")
        verbose_name_plural = _(u"Author types")

class Author(models.Model):
    person = models.ForeignKey(Person, verbose_name=_(u"Person"))
    author_type = models.ForeignKey(AuthorType, verbose_name=_(u"Author type"))

    class Meta:
        verbose_name = _(u"Author")
        verbose_name_plural = _(u"Authors")

    def __unicode__(self):
        return unicode(self.person) + settings.JOINT + unicode(self.author_type)

class SourceType(GeneralType):
    class Meta:
        verbose_name = _(u"Source type")
        verbose_name_plural = _(u"Source types")

class Source(models.Model):
    title = models.CharField(_(u"Title"), max_length=200)
    source_type = models.ForeignKey(SourceType, verbose_name=_(u"Type"))
    authors = models.ManyToManyField(Author, verbose_name=_(u"Authors"))
    associated_url = models.URLField(verify_exists=False, blank=True, null=True,
                           verbose_name=_(u"Numerical ressource (web address)"))
    receipt_date = models.DateField(blank=True, null=True,
                                    verbose_name=_(u"Receipt date"))
    creation_date = models.DateField(blank=True, null=True,
                                    verbose_name=_(u"Creation date"))
    TABLE_COLS = ['title', 'source_type', 'authors',]

    class Meta:
        abstract = True

    def __unicode__(self):
        return self.title

if settings.COUNTRY == 'fr':
    class Arrondissement(models.Model):
        name = models.CharField(u"Nom", max_length=30)
        department = models.ForeignKey(Department, verbose_name=u"Département")

        def __unicode__(self):
            return settings.JOINT.join((self.name, unicode(self.department)))

    class Canton(models.Model):
        name = models.CharField(u"Nom", max_length=30)
        arrondissement = models.ForeignKey(Arrondissement,
                                           verbose_name=u"Arrondissement")
        def __unicode__(self):
            return settings.JOINT.join((self.name, unicode(self.arrondissement)))

class Town(models.Model):
    name = models.CharField(_(u"Name"), max_length=100)
    surface = models.IntegerField(_(u"Surface (m²)"), blank=True, null=True)
    center = models.PointField(_(u"Localisation"), srid=settings.SRID,
                               blank=True, null=True)
    if settings.COUNTRY == 'fr':
        numero_insee = models.CharField(u"Numéro INSEE", max_length=6,
                                        unique=True)
        departement = models.ForeignKey(Department, verbose_name=u"Département",
                                       null=True, blank=True)
        canton = models.ForeignKey(Canton, verbose_name=u"Canton", null=True,
                                   blank=True)
    objects = models.GeoManager()

    class Meta:
        verbose_name = _(u"Town")
        verbose_name_plural = _(u"Towns")
        if settings.COUNTRY == 'fr':
            ordering = ['numero_insee']

    def __unicode__(self):
        if settings.COUNTRY == "fr":
            return u"%s (%s)" % (self.name, self.numero_insee)
        return self.name