konova/codelist/management/commands/sync_codelist.py

"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: michel.peltriaux@sgdnord.rlp.de
Created on: 31.05.22

"""

from django.db import transaction

from codelist.models import KonovaCode
from compensation.models import CompensationAction, CompensationState
from intervention.models import Legal, Handler, Responsibility
from konova.management.commands.setup import BaseKonovaCommand


class Command(BaseKonovaCommand):
    help = "Updates internal codelist by external API"

    def handle(self, *args, **options):
        try:
            with transaction.atomic():
                self.sync_codelist()
        except KeyboardInterrupt:
            self._break_line()
            exit(-1)

    def __get_newest_code(self, code):
        code = KonovaCode.objects.filter(
            atom_id=code.atom_id,
            parent=code.parent,
            code_lists__in=code.code_lists.all(),
        ).order_by(
            "-id"
        ).first()
        return code

    def __migrate_compensation_action_codes(self):
        all_actions = CompensationAction.objects.all()
        used_codes = []
        for action in all_actions:
            stored_codes = action.action_type.all()
            codes = []
            for code in stored_codes:
                codes.append(self.__get_newest_code(code))
            action.action_type.set(codes)
            used_codes += codes

            stored_codes = action.action_type_details.all()
            codes = []
            for code in stored_codes:
                codes.append(self.__get_newest_code(code))
            action.action_type_details.set(codes)
            used_codes += codes

            action.save()
        return used_codes

    def __migrate_compensation_state_codes(self):
        all_states = CompensationState.objects.all()
        used_codes = []
        for state in all_states:
            code = state.biotope_type
            if code is not None:
                new_code = self.__get_newest_code(code)
                state.biotope_type = new_code
                used_codes.append(new_code)

            stored_codes = state.biotope_type_details.all()
            codes = []
            for code in stored_codes:
                codes.append(self.__get_newest_code(code))
            state.biotope_type_details.set(codes)

            used_codes += codes
            state.save()
        return used_codes

    def __migrate_legal_codes(self):
        all_legal = Legal.objects.all()
        used_codes = []
        for legal in all_legal:
            code = legal.process_type
            if code is not None:
                new_code = self.__get_newest_code(code)
                legal.process_type = new_code
                used_codes.append(new_code)

            stored_codes = legal.laws.all()
            codes = []
            for code in stored_codes:
                codes.append(self.__get_newest_code(code))
            legal.laws.set(codes)

            used_codes += codes
            legal.save()
        return used_codes

    def __migrate_handler_codes(apps):
        all_handlers = Handler.objects.all()
        used_codes = []
        for handler in all_handlers:
            code = handler.type
            if code is None:
                continue
            new_code = apps.__get_newest_code(code)
            handler.type = new_code
            used_codes.append(new_code)
            handler.save()
        return used_codes

    def __migrate_responsibility_codes(apps):
        all_resps = Responsibility.objects.all()
        used_codes = []
        for responsibility in all_resps:
            code = responsibility.registration_office
            if code is not None:
                new_code = apps.__get_newest_code(code)
                responsibility.registration_office = new_code
                used_codes.append(new_code)

            code = responsibility.conservation_office
            if code is not None:
                new_code = apps.__get_newest_code(code)
                responsibility.conservation_office = new_code
                used_codes.append(new_code)

            responsibility.save()
        return used_codes

    def sync_codelist(self):
        """ Due to issues on the external codelist app there can be multiple entries of the same code
        (atom_id, parent, list) but with different identifiers.

        These issues have been resolved but already

        Returns:

        """
        self._write_warning("Sync codes in usage and replace by newest entries...")
        used_codes = []
        used_codes += self.__migrate_compensation_action_codes()
        used_codes += self.__migrate_compensation_state_codes()
        used_codes += self.__migrate_legal_codes()
        used_codes += self.__migrate_handler_codes()
        used_codes += self.__migrate_responsibility_codes()
        self._write_success(f"Synced {len(used_codes)} code usages!")

        all_codes = KonovaCode.objects.all()
        newest_code_ids = []
        for code in all_codes:
            newest_code = self.__get_newest_code(code)
            newest_code_ids.append(newest_code.id)

        code_ids_to_keep = set(newest_code_ids)
        self._write_warning(f"Of {all_codes.count()} KonovaCodes there are {len(code_ids_to_keep)} to keep as newest versions...")

        deletable_codes = KonovaCode.objects.all().exclude(
            id__in=code_ids_to_keep
        )
        deletable_codes_count = deletable_codes.count()
        self._write_warning(f"{deletable_codes_count} found which are obsolet...")
        if deletable_codes_count > 0:
            deletable_codes.delete()
            self._write_success("Obsolete codes deleted!")