from django.db import models from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.fields import GenericForeignKey from django.utils.translation import gettext_lazy as _ from django.utils import timezone from datetime import datetime, timedelta class AssetHistoryQuerySet(models.QuerySet): """Custom QuerySet for AssetHistory with filtering helpers.""" def for_asset(self, asset): """Filter history for a specific asset.""" content_type = ContentType.objects.get_for_model(asset) return self.filter(content_type=content_type, object_id=asset.pk) def before_date(self, date): """Filter history before a specific date.""" if isinstance(date, str): date = datetime.fromisoformat(date) return self.filter(timestamp__lte=date) def after_date(self, date): """Filter history after a specific date.""" if isinstance(date, str): date = datetime.fromisoformat(date) return self.filter(timestamp__gte=date) def between_dates(self, start_date, end_date): """Filter history between two dates.""" return self.after_date(start_date).before_date(end_date) def status_changes(self): """Filter only status change entries.""" return self.filter(action_type='status_change') def creations(self): """Filter only creation entries.""" return self.filter(action_type='create') def deletions(self): """Filter only deletion entries.""" return self.filter(action_type='delete') class AssetHistoryManager(models.Manager): """Custom manager for AssetHistory.""" def get_queryset(self): return AssetHistoryQuerySet(self.model, using=self._db) def for_asset(self, asset): return self.get_queryset().for_asset(asset) def get_content_type_for_model(self, model_name, app_label=None): """ Get ContentType for a model, handling cases where multiple apps have the same model name. Returns the ContentType or raises ContentType.DoesNotExist if not found. If multiple ContentTypes exist and no app_label specified, returns the first one. """ if app_label: return ContentType.objects.get(app_label=app_label, model=model_name.lower()) content_types = list(ContentType.objects.filter(model=model_name.lower())) if not content_types: raise ContentType.DoesNotExist(f"Model '{model_name}' not found") # If multiple, return the first one (usually 'assets' if it exists) # User can specify app_label to disambiguate return content_types[0] def get_asset_status_at_date(self, asset, date): """ Get the status of an asset at a specific date. Returns the status value or None if asset didn't exist or was deleted. """ if isinstance(date, str): date = datetime.fromisoformat(date) content_type = ContentType.objects.get_for_model(asset) # Check if asset was created by the date creation = self.filter( content_type=content_type, object_id=asset.pk, action_type='create', timestamp__lte=date ).first() if not creation: return None # Asset didn't exist yet # Check if asset was deleted by the date deletion = self.filter( content_type=content_type, object_id=asset.pk, action_type='delete', timestamp__lte=date ).first() if deletion: return None # Asset was deleted # Get the last status change before the date last_status_change = self.filter( content_type=content_type, object_id=asset.pk, field_name='status', timestamp__lte=date ).order_by('-timestamp').first() if last_status_change: return last_status_change.new_value # If no status change, try to get the current status from the asset itself # This should not happen in normal cases, but it's a fallback try: return asset.status except: return None def get_assets_active_at_date(self, asset_model_class, date): """ Get all assets of a given type that were active (status='active') at a specific date. """ if isinstance(date, str): date = datetime.fromisoformat(date) content_type = ContentType.objects.get_for_model(asset_model_class) # Get all created assets by that date created_by_date = self.filter( content_type=content_type, action_type='create', timestamp__lte=date ).values_list('object_id', flat=True).distinct() if not created_by_date: return [] # Get deleted assets by that date deleted_by_date = self.filter( content_type=content_type, action_type='delete', timestamp__lte=date ).values_list('object_id', flat=True).distinct() # Get asset IDs with last status != 'active' before the date non_active_ids = [] for asset_id in created_by_date: if asset_id not in deleted_by_date: last_status = self.filter( content_type=content_type, object_id=asset_id, field_name='status', timestamp__lte=date ).order_by('-timestamp').first() if last_status and last_status.new_value != 'active': non_active_ids.append(asset_id) # Return active assets (created, not deleted, status='active') active_ids = [aid for aid in created_by_date if aid not in deleted_by_date and aid not in non_active_ids] return asset_model_class.objects.filter(pk__in=active_ids) class AssetHistory(models.Model): """ Model to track the history of changes to assets. Uses ContentType to work with any asset type (generic asset tracking). This model stores: - Who made the change (user) - When the change was made (timestamp) - What changed (field_name, old_value, new_value) - Which asset was modified (generic relation) This design is efficient for querying history across all assets for timeline views. """ # Generic relation to any asset content_type = models.ForeignKey( ContentType, on_delete=models.CASCADE, verbose_name=_("Type d'asset") ) object_id = models.PositiveIntegerField( verbose_name=_("ID de l'asset") ) asset = GenericForeignKey('content_type', 'object_id') # Tracking information user = models.ForeignKey( 'auth.User', on_delete=models.SET_NULL, null=True, blank=True, related_name='asset_changes', verbose_name=_("Utilisateur") ) timestamp = models.DateTimeField( default=timezone.now, db_index=True, verbose_name=_("Date de modification") ) # Change details field_name = models.CharField( max_length=100, verbose_name=_("Champ modifié") ) old_value = models.TextField( null=True, blank=True, verbose_name=_("Ancienne valeur") ) new_value = models.TextField( null=True, blank=True, verbose_name=_("Nouvelle valeur") ) # Optional: action type for special events action_type = models.CharField( max_length=50, default='update', choices=[ ('create', _('Création')), ('update', _('Modification')), ('delete', _('Suppression')), ('status_change', _('Changement de statut')), ], verbose_name=_("Type d'action") ) objects = AssetHistoryManager() class Meta: verbose_name = _("Historique d'asset") verbose_name_plural = _("Historiques d'assets") ordering = ['-timestamp'] indexes = [ models.Index(fields=['content_type', 'object_id', '-timestamp']), models.Index(fields=['-timestamp']), models.Index(fields=['user', '-timestamp']), models.Index(fields=['action_type', '-timestamp']), models.Index(fields=['field_name', '-timestamp']), models.Index(fields=['content_type', 'action_type']), ] def __str__(self): return f"{self.get_action_type_display()} - {self.field_name} - {self.timestamp}" def get_display_old_value(self): """Get a human-readable display of the old value""" return self._format_value(self.old_value, self.field_name) def get_display_new_value(self): """Get a human-readable display of the new value""" return self._format_value(self.new_value, self.field_name) def _format_value(self, value, field_name): """Format a value for display based on field type""" if value is None or value == '' or value == 'None': return _("(vide)") # Special handling for common field types if field_name in ['installation_date', 'uninstallation_date', 'last_inspection_date', 'last_preventive_date', 'last_corrective_date', 'last_survey_date', 'marked_as_removed_date']: try: from django.utils.dateparse import parse_datetime dt = parse_datetime(value) if dt: from django.utils.formats import date_format return date_format(dt, "SHORT_DATETIME_FORMAT") except: pass # For status field, try to get the display value if field_name == 'status': from .core import ASSET_STATUS_CHOICES status_dict = dict(ASSET_STATUS_CHOICES) return str(status_dict.get(value, value)) # For Foreign Key fields, try to get the object's code if field_name in ['location', 'structure', 'model', 'category', 'replaced_by']: try: # Value is stored as the object's PK (ID) pk_value = int(value) # Get the asset to determine the correct model class if self.asset: related_object = None # Try to get the related object from the asset if hasattr(self.asset, field_name): related_field = getattr(self.asset.__class__, field_name, None) if related_field and hasattr(related_field, 'field'): related_model = related_field.field.related_model try: related_object = related_model.objects.get(pk=pk_value) except: pass # Format the display if related_object: if hasattr(related_object, 'code'): return str(related_object.code) elif hasattr(related_object, 'name_fr'): return str(related_object.name_fr) elif hasattr(related_object, 'name'): return str(related_object.name) else: return str(related_object) except (ValueError, TypeError, AttributeError): pass return str(value) def get_field_display_name(self): """Get a human-readable field name""" field_names = { 'code': _('Code'), 'name_fr': _('Nom (FR)'), 'name_nl': _('Nom (NL)'), 'status': _('Statut'), 'installation_date': _("Date d'installation"), 'uninstallation_date': _('Date de désinstallation'), 'serial_number': _('Numéro de série'), 'warranty_duration': _('Durée de garantie'), 'last_inspection_date': _('Dernière inspection'), 'last_preventive_date': _('Dernier entretien préventif'), 'last_corrective_date': _('Dernier entretien correctif'), 'last_survey_date': _('Dernier relevé'), 'category': _('Catégorie'), 'brand': _('Marque'), 'model': _('Modèle'), 'location': _('Localisation'), 'structure': _('Structure'), 'lon': _('Longitude'), 'lat': _('Latitude'), 'geom': _('Géométrie'), 'geojson': _('GeoJSON'), } return field_names.get(self.field_name, self.field_name)