#!/usr/bin/env python # -*- coding: utf-8 -*- # Copyright (C) 2012-2017 Étienne Loks # 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 . # See the file COPYING for details. import datetime from django.conf import settings from django.contrib.gis.db import models from django.db.models import Q from django.db.models.signals import post_save, post_delete from django.template.defaultfilters import slugify from django.utils.translation import ugettext_lazy as _, pgettext_lazy, \ activate, deactivate from ishtar_common.data_importer import post_importer_action from ishtar_common.models import Document, GeneralType, get_external_id, \ LightHistorizedItem, OwnPerms, Address, Person, post_save_cache, \ ImageModel, DashboardFormItem, ExternalIdManager, ShortMenuItem from ishtar_common.utils import cached_label_changed class WarehouseType(GeneralType): class Meta: verbose_name = _(u"Warehouse type") verbose_name_plural = _(u"Warehouse types") ordering = ('label',) post_save.connect(post_save_cache, sender=WarehouseType) post_delete.connect(post_save_cache, sender=WarehouseType) class Warehouse(Address, DashboardFormItem, OwnPerms, ShortMenuItem): SLUG = 'warehouse' SHOW_URL = 'show-warehouse' TABLE_COLS = ['name', 'warehouse_type'] BASE_SEARCH_VECTORS = ['name', 'warehouse_type__label', "external_id", "town", "comment"] EXTRA_REQUEST_KEYS = {} # alternative names of fields for searches ALT_NAMES = { 'name': ( pgettext_lazy("key for text search", u"name"), 'name__iexact' ), 'warehouse_type': ( pgettext_lazy("key for text search", u"type"), 'warehouse_type__label__iexact' ), } for v in ALT_NAMES.values(): for language_code, language_lbl in settings.LANGUAGES: activate(language_code) EXTRA_REQUEST_KEYS[unicode(v[0])] = v[1] deactivate() objects = ExternalIdManager() name = models.CharField(_(u"Name"), max_length=200) warehouse_type = models.ForeignKey(WarehouseType, verbose_name=_(u"Warehouse type")) person_in_charge = models.ForeignKey( Person, on_delete=models.SET_NULL, related_name='warehouse_in_charge', verbose_name=_(u"Person in charge"), null=True, blank=True) comment = models.TextField(_(u"Comment"), null=True, blank=True) associated_divisions = models.ManyToManyField( 'WarehouseDivision', verbose_name=_("Divisions"), blank=True, through='WarehouseDivisionLink' ) documents = models.ManyToManyField( Document, related_name='warehouses', verbose_name=_(u"Documents"), blank=True) external_id = models.TextField(_(u"External ID"), blank=True, null=True) auto_external_id = models.BooleanField( _(u"External ID is set automatically"), default=False) class Meta: verbose_name = _(u"Warehouse") verbose_name_plural = _(u"Warehouses") permissions = ( ("view_warehouse", u"Can view all Warehouses"), ("view_own_warehouse", u"Can view own Warehouse"), ("add_own_warehouse", u"Can add own Warehouse"), ("change_own_warehouse", u"Can change own Warehouse"), ("delete_own_warehouse", u"Can delete own Warehouse"), ) def __unicode__(self): return self.name def naturel_key(self): return (self.external_id, ) def _get_base_image_path(self): return u"{}/{}".format(self.SLUG, self.external_id) @property def location_types(self): return [ wd.division.label for wd in WarehouseDivisionLink.objects.filter( warehouse=self).order_by('order').all() ] @property def associated_filename(self): return datetime.date.today().strftime('%Y-%m-%d') + '-' + \ slugify(unicode(self)) @classmethod def get_query_owns(cls, ishtaruser): return cls._construct_query_own( '', cls._get_query_owns_dicts(ishtaruser)) @classmethod def _get_query_owns_dicts(cls, ishtaruser): return [{'person_in_charge__ishtaruser': ishtaruser}] @property def number_of_finds(self): from archaeological_finds.models import Find return Find.objects.filter(container__responsible=self).count() @property def number_of_finds_hosted(self): from archaeological_finds.models import Find return Find.objects.filter(container__location=self).count() @property def number_of_containers(self): return Container.objects.filter(location=self).count() def _get_divisions(self, current_path, remaining_division, depth=0): if not remaining_division: return [current_path] current_division = remaining_division.pop(0) q = ContainerLocalisation.objects.filter( division=current_division, ) for div, ref in current_path: q = q.filter( container__division__division=div, container__division__reference=ref ) res = [] old_ref = None if not q.count(): return [current_path] for ref in q.values('reference').order_by('reference').all(): if ref['reference'] == old_ref: continue old_ref = ref['reference'] cpath = current_path[:] cpath.append((current_division, ref['reference'])) for r in self._get_divisions(cpath, remaining_division[:], depth + 1): res.append(r) return res @property def available_division_tuples(self): """ :return: ordered list of available paths. Each path is a list of tuple with the WarehouseDivisionLink and the reference. """ divisions = list( WarehouseDivisionLink.objects.filter(warehouse=self ).order_by('order').all()) return self._get_divisions([], divisions) def _number_of_items_by_place(self, model, division_key='division'): res = {} paths = self.available_division_tuples[:] for path in paths: q = model.objects cpath = [] for division, ref in path: cpath.append(ref) attrs = { division_key + "__division": division, division_key + "__reference": ref } q = q.filter(**attrs) if tuple(cpath) not in res: res[tuple(cpath)] = q.count() res = [(k, res[k]) for k in res] final_res, current_res, depth = [], [], 1 len_divisions = WarehouseDivisionLink.objects.filter( warehouse=self).count() for path, nb in sorted(res, key=lambda x: (len(x[0]), x[0])): if depth != len(path): final_res.append(current_res[:]) current_res = [] depth = len(path) if path[-1] == '-': continue path = list(path) + ['' for idx in range(len_divisions - len(path))] current_res.append((path, nb)) final_res.append(current_res[:]) return final_res def _number_of_finds_by_place(self): from archaeological_finds.models import Find return self._number_of_items_by_place( Find, division_key='container__division') @property def number_of_finds_by_place(self, update=False): return self._get_or_set_stats('_number_of_finds_by_place', update, settings.CACHE_SMALLTIMEOUT) def _number_of_containers_by_place(self): return self._number_of_items_by_place(Container) @property def number_of_containers_by_place(self, update=False): return self._get_or_set_stats('_number_of_containers_by_place', update, settings.CACHE_SMALLTIMEOUT) def save(self, *args, **kwargs): self.update_search_vector() super(Warehouse, self).save(*args, **kwargs) for container in self.containers.all(): cached_label_changed(Container, instance=container) self.skip_history_when_saving = True if not self.external_id or self.auto_external_id: external_id = get_external_id('warehouse_external_id', self) if external_id != self.external_id: updated = True self.auto_external_id = True self.external_id = external_id self._cached_label_checked = False self.save() return class Collection(LightHistorizedItem): name = models.CharField(_(u"Name"), max_length=200, null=True, blank=True) description = models.TextField(_(u"Description"), null=True, blank=True) warehouse = models.ForeignKey(Warehouse, verbose_name=_(u"Warehouse"), related_name='collections') class Meta: verbose_name = _(u"Collection") verbose_name_plural = _(u"Collection") ordering = ('name',) def __unicode__(self): return self.name class WarehouseDivision(GeneralType): class Meta: verbose_name = _(u"Warehouse division type") verbose_name_plural = _(u"Warehouse division types") post_save.connect(post_save_cache, sender=WarehouseDivision) post_delete.connect(post_save_cache, sender=WarehouseDivision) class WarehouseDivisionLink(models.Model): RELATED_SET_NAME = "divisions" RELATED_ATTRS = ["order"] RELATIVE_MODELS = {Warehouse: 'warehouse'} warehouse = models.ForeignKey(Warehouse, related_name='divisions') division = models.ForeignKey(WarehouseDivision) order = models.IntegerField(_("Order"), default=10) class Meta: ordering = ('warehouse', 'order') unique_together = ('warehouse', 'division') def __unicode__(self): return u"{} - {}".format(self.warehouse, self.division) class ContainerType(GeneralType): length = models.IntegerField(_(u"Length (mm)"), blank=True, null=True) width = models.IntegerField(_(u"Width (mm)"), blank=True, null=True) height = models.IntegerField(_(u"Height (mm)"), blank=True, null=True) volume = models.FloatField(_(u"Volume (l)"), blank=True, null=True) reference = models.CharField(_(u"Ref."), max_length=30) class Meta: verbose_name = _(u"Container type") verbose_name_plural = _(u"Container types") ordering = ('label',) post_save.connect(post_save_cache, sender=ContainerType) post_delete.connect(post_save_cache, sender=ContainerType) class Container(LightHistorizedItem, ImageModel): TABLE_COLS = ['reference', 'container_type__label', 'cached_location', 'cached_division', 'old_reference'] IMAGE_PREFIX = 'containers/' BASE_SEARCH_VECTORS = ['reference', 'container_type__label', 'cached_location'] M2M_SEARCH_VECTORS = ['division__reference', 'division__division__division__label'] # search parameters EXTRA_REQUEST_KEYS = { 'location': 'location__pk', 'location_id': 'location__pk', 'responsible_id': 'responsible__pk', 'container_type': 'container_type__pk', 'reference': 'reference__icontains', 'old_reference': 'old_reference__icontains', 'finds__base_finds__context_record__operation': 'finds__base_finds__context_record__operation', 'finds__base_finds__context_record': 'finds__base_finds__context_record', 'finds': 'finds', 'container_type__label': 'container_type__label', } SHOW_URL = 'show-container' COL_LABELS = { 'cached_location': _(u"Location - index"), 'cached_division': _(u"Precise localisation"), 'container_type__label': _(u"Type") } CACHED_LABELS = ['cached_division', 'cached_label', 'cached_location', ] # alternative names of fields for searches ALT_NAMES = { 'location_name': ( pgettext_lazy("key for text search", u"location"), 'location__name__iexact' ), 'responsible_name': ( pgettext_lazy("key for text search", u"responsible-warehouse"), 'responsible__name__iexact' ), 'container_type': ( pgettext_lazy("key for text search", u"type"), 'container_type__label__iexact' ), 'reference': ( pgettext_lazy("key for text search", u"reference"), 'reference__iexact' ), } for v in ALT_NAMES.values(): for language_code, language_lbl in settings.LANGUAGES: activate(language_code) EXTRA_REQUEST_KEYS[unicode(v[0])] = v[1] deactivate() objects = ExternalIdManager() # fields location = models.ForeignKey( Warehouse, verbose_name=_(u"Location (warehouse)"), related_name='containers') responsible = models.ForeignKey( Warehouse, verbose_name=_(u"Responsible warehouse"), related_name='owned_containers') container_type = models.ForeignKey(ContainerType, verbose_name=_("Container type")) reference = models.TextField(_(u"Container ref.")) comment = models.TextField(_(u"Comment"), null=True, blank=True) cached_label = models.TextField(_(u"Localisation"), null=True, blank=True, db_index=True) cached_location = models.TextField(_(u"Cached location"), null=True, blank=True, db_index=True) cached_division = models.TextField(_(u"Cached division"), null=True, blank=True, db_index=True) index = models.IntegerField(u"Container ID", default=0) old_reference = models.TextField(_(u"Old reference"), blank=True, null=True) external_id = models.TextField(_(u"External ID"), blank=True, null=True) auto_external_id = models.BooleanField( _(u"External ID is set automatically"), default=False) class Meta: verbose_name = _(u"Container") verbose_name_plural = _(u"Containers") ordering = ('cached_label',) unique_together = ('index', 'responsible') permissions = ( ("view_container", u"Can view all Containers"), ("view_own_container", u"Can view own Container"), ("add_own_container", u"Can add own Container"), ("change_own_container", u"Can change own Container"), ("delete_own_container", u"Can delete own Container"), ) def __unicode__(self): return self.cached_label def natural_key(self): return (self.external_id, ) def _generate_cached_label(self): items = [self.reference, self.precise_location] cached_label = u" | ".join(items) return cached_label def _generate_cached_location(self): items = [self.location.name, unicode(self.index)] cached_label = u" - ".join(items) return cached_label def _generate_cached_division(self): locas = [ u"{} {}".format(loca.division.division, loca.reference) for loca in ContainerLocalisation.objects.filter( container=self) ] return u" | ".join(locas) @classmethod def get_query_owns(cls, ishtaruser): return Q(history_creator=ishtaruser.user_ptr) | \ Q(location__person_in_charge__ishtaruser=ishtaruser) | \ Q(responsible__person_in_charge__ishtaruser=ishtaruser) @property def associated_filename(self): filename = datetime.date.today().strftime('%Y-%m-%d') filename += u'-' + self.reference filename += u"-" + self.location.name filename += u"-" + unicode(self.index) if self.cached_division is None: self.skip_history_when_saving = True self.save() if self.cached_division: filename += u"-" + self.cached_division return slugify(filename) @property def precise_location(self): location = self.location.name if self.cached_division: location += self.cached_division return location def get_localisations(self): """ Get precise localisation of the container in the warehouse. :return: tuple of strings with localisations """ return tuple(( loca.reference for loca in ContainerLocalisation.objects.filter( container=self).order_by('division__order') )) def get_localisation(self, place): locas = self.get_localisations() if len(locas) < (place + 1): return "" return locas[place] @property def localisation_1(self): return self.get_localisation(0) @property def localisation_2(self): return self.get_localisation(1) @property def localisation_3(self): return self.get_localisation(2) @property def localisation_4(self): return self.get_localisation(3) @property def localisation_5(self): return self.get_localisation(4) @property def localisation_6(self): return self.get_localisation(5) @property def localisation_7(self): return self.get_localisation(6) @property def localisation_8(self): return self.get_localisation(7) @property def localisation_9(self): return self.get_localisation(8) def set_localisation(self, place, value): """ Set the reference for the localisation number "place" (starting from 0) :param place: the number of the localisation :param value: the reference to be set :return: the container location object or None if the place does not exist """ q = WarehouseDivisionLink.objects.filter( warehouse=self.location).order_by('order') for idx, division_link in enumerate(q.all()): if idx == place: break else: return dct = {'container': self, 'division': division_link} if not value: if ContainerLocalisation.objects.filter(**dct).count(): c = ContainerLocalisation.objects.filter(**dct).all()[0] c.delete() return dct['defaults'] = {'reference': value} obj, created = ContainerLocalisation.objects.update_or_create(**dct) return obj @post_importer_action def set_localisation_1(self, context, value): return self.set_localisation(0, value) @post_importer_action def set_localisation_2(self, context, value): return self.set_localisation(1, value) @post_importer_action def set_localisation_3(self, context, value): return self.set_localisation(2, value) @post_importer_action def set_localisation_4(self, context, value): return self.set_localisation(3, value) @post_importer_action def set_localisation_5(self, context, value): return self.set_localisation(4, value) @post_importer_action def set_localisation_6(self, context, value): return self.set_localisation(5, value) @post_importer_action def set_localisation_7(self, context, value): return self.set_localisation(6, value) @post_importer_action def set_localisation_8(self, context, value): return self.set_localisation(7, value) @post_importer_action def set_localisation_9(self, context, value): return self.set_localisation(8, value) def pre_save(self): if not self.index: q = Container.objects.filter(responsible=self.responsible).order_by( '-index') if q.count(): self.index = q.all()[0].index + 1 else: self.index = 1 def save(self, *args, **kwargs): super(Container, self).save(*args, **kwargs) updated = False if not self.index: self.skip_history_when_saving = True q = Container.objects.filter(responsible=self.responsible).exclude( pk=self.pk).order_by('-index') if q.count(): self.index = q.all()[0].index + 1 else: self.index = 1 updated = True self.skip_history_when_saving = True if not self.external_id or self.auto_external_id: external_id = get_external_id('container_external_id', self) if external_id != self.external_id: updated = True self.auto_external_id = True self.external_id = external_id if updated: self._cached_label_checked = False self.save() return # remove old location in warehouse q = ContainerLocalisation.objects.filter(container=self).exclude( division__warehouse=self.location) for loca in q.all(): loca.delete() post_save.connect(cached_label_changed, sender=Container) class ContainerLocalisation(models.Model): container = models.ForeignKey(Container, verbose_name=_(u"Container"), related_name='division') division = models.ForeignKey(WarehouseDivisionLink, verbose_name=_(u"Division")) reference = models.CharField(_(u"Reference"), max_length=200, default='') class Meta: verbose_name = _(u"Container localisation") verbose_name_plural = _(u"Container localisations") unique_together = ('container', 'division') ordering = ('container', 'division__order') def __unicode__(self): lbl = u" - ".join((unicode(self.container), unicode(self.division), self.reference)) return lbl def save(self, *args, **kwargs): super(ContainerLocalisation, self).save(*args, **kwargs) cached_label_changed(Container, instance=self.container)