PyRIGS/versioning/versioning.py

from diff_match_patch import diff_match_patch
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist
from django.db.models import EmailField, IntegerField, TextField, CharField, BooleanField
from django.utils.functional import cached_property
from reversion.models import Version, VersionQuerySet


class RevisionMixin:
    @property
    def is_first_version(self):
        versions = RIGSVersion.objects.get_for_object(self)
        return len(versions) == 1

    @property
    def current_version(self):
        version = RIGSVersion.objects.get_for_object(self).select_related("revision").first()
        return version

    @property
    def last_edited_at(self):
        version = self.current_version
        if version is None:
            return None
        return version.revision.date_created

    @property
    def last_edited_by(self):
        version = self.current_version
        if version is None:
            return None
        return version.revision.user

    @property
    def current_version_id(self):
        version = self.current_version
        if version is None:
            return None
        return version.display_id

    @property
    def date_created(self):
        return self.current_version.revision.date_created


class FieldComparison:
    def __init__(self, field=None, old=None, new=None):
        self.field = field
        self._old = old
        self._new = new

    def display_value(self, value):
        if (
            isinstance(self.field, (IntegerField, CharField))
            and self.field.choices is not None
            and len(self.field.choices) > 0
        ):
            choice = [x[1] for x in self.field.choices if x[0] == value]
            if len(choice) > 0:
                return choice[0]
        if isinstance(self.field, BooleanField):
            if value:
                return "✓"
            else:
                return "&#10007"
        return value

    @property
    def old(self):
        return self.display_value(self._old)

    @property
    def new(self):
        return self.display_value(self._new)

    @property
    def long(self):
        if isinstance(self.field, EmailField):
            return True
        return False

    @property
    def linebreaks(self):
        if isinstance(self.field, TextField):
            return True
        return False

    @property
    def diff(self):
        oldText = str(self.display_value(self._old)) or ""
        newText = str(self.display_value(self._new)) or ""
        dmp = diff_match_patch()
        diffs = dmp.diff_main(oldText, newText)
        dmp.diff_cleanupSemantic(diffs)

        outputDiffs = []

        for op, data in diffs:
            if op == dmp.DIFF_INSERT:
                outputDiffs.append({"type": "insert", "text": data})
            elif op == dmp.DIFF_DELETE:
                outputDiffs.append({"type": "delete", "text": data})
            elif op == dmp.DIFF_EQUAL:
                outputDiffs.append({"type": "equal", "text": data})
        return outputDiffs


class ModelComparison:
    def __init__(self, old=None, new=None, version=None, follow=False, excluded_keys=["date_joined"]):
        # recieves two objects of the same model, and compares them. Returns an array of FieldCompare objects
        try:
            self.fields = old._meta.get_fields()
        except AttributeError:
            self.fields = new._meta.get_fields()

        self.old = old
        self.new = new
        self.excluded_keys = excluded_keys
        self.version = version
        self.follow = follow

    @cached_property
    def revision(self):
        return self.version.revision

    @cached_property
    def field_changes(self):
        changes = []
        for field in self.fields:
            field_name = field.name
            if field_name not in self.excluded_keys:  # if we're excluding this field, skip over it
                try:
                    oldValue = getattr(self.old, field_name, None)
                except ObjectDoesNotExist:
                    oldValue = None

                try:
                    newValue = getattr(self.new, field_name, None)
                except ObjectDoesNotExist:
                    newValue = None

                bothBlank = (not oldValue) and (not newValue)
                if oldValue != newValue and not bothBlank:
                    comparison = FieldComparison(field, oldValue, newValue)
                    changes.append(comparison)
        return changes

    @cached_property
    def fields_changed(self):
        return len(self.field_changes) > 0

    @cached_property
    def item_changes(self):
        if self.follow and self.version.object is not None:
            from RIGS.models import EventAuthorisation
            from training.models import TrainingLevelQualification, TrainingItemQualification

            item_type = ContentType.objects.get_for_model(self.version.object)
            old_item_versions = (
                self.version.parent.revision.version_set.exclude(content_type=item_type)
                .exclude(content_type=ContentType.objects.get_for_model(TrainingItemQualification))
                .exclude(content_type=ContentType.objects.get_for_model(TrainingLevelQualification))
            )
            new_item_versions = self.version.revision.version_set.exclude(content_type=item_type).exclude(
                content_type=ContentType.objects.get_for_model(EventAuthorisation)
            )

            comparisonParams = {
                "excluded_keys": ["id", "event", "order", "checklist", "level", "_order", "date_joined"]
            }

            # Build some dicts of what we have
            item_dict = {}  # build a list of items, key is the item_pk
            for version in old_item_versions:  # put all the old versions in a list
                if version._model is None:
                    continue
                compare = ModelComparison(old=version._object_version.object, **comparisonParams)
                item_dict[version.object_id] = compare

            for version in new_item_versions:  # go through the new versions
                if version._model is None:
                    continue
                try:
                    compare = item_dict[version.object_id]  # see if there's a matching old version
                    compare.new = version._object_version.object  # then add the new version to the dictionary
                except KeyError:  # there's no matching old version, so add this item to the dictionary by itself
                    compare = ModelComparison(new=version._object_version.object, **comparisonParams)

                item_dict[version.object_id] = compare  # update the dictionary with the changes

            changes = []
            for _, compare in list(item_dict.items()):
                if compare.fields_changed:
                    changes.append(compare)

            return changes

    @cached_property
    def name(self):
        obj = self.new if self.new else self.old

        if hasattr(obj, "activity_feed_string"):
            return obj.activity_feed_string
        else:
            return str(obj)

    @cached_property
    def items_changed(self):
        return self.item_changes is not None and len(self.item_changes) > 0

    @cached_property
    def anything_changed(self):
        return self.fields_changed or self.items_changed


class RIGSVersionManager(VersionQuerySet):
    def get_for_multiple_models(self, model_array):
        content_types = []
        for model in model_array:
            content_types.append(ContentType.objects.get_for_model(model))

        return (
            self.filter(content_type__in=content_types)
            .select_related(
                "revision",
            )
            .order_by("-revision__date_created")
        )


class RIGSVersion(Version):
    class Meta:
        proxy = True

    objects = RIGSVersionManager.as_manager()

    # Gets the most recent previous version
    @cached_property
    def parent(self):
        thisId = self.object_id

        versions = (
            RIGSVersion.objects.get_for_object_reference(self.content_type.model_class(), thisId)
            .select_related("revision", "revision__user")
            .all()
        )

        try:
            previousVersion = versions.filter(revision_id__lt=self.revision_id).latest("revision_id")
        except ObjectDoesNotExist:
            return False

        return previousVersion

    @cached_property
    def changes(self):
        return ModelComparison(
            version=self,
            new=self._object_version.object,
            old=self.parent._object_version.object if self.parent else None,
            follow=True,
        )

    @property
    def display_id(self):
        return f"V{self.pk} | R{self.revision.pk}"

    @property
    def display_name(self):
        if hasattr(self.changes.new, "display_id"):
            id = self.changes.new.display_id
        else:
            id = self.changes.new.pk

        return f"{id} | {self.changes.new.__class__.__name__}"