Merge pull request 'master' (#287) from master into Docker

Reviewed-on: SGD-Nord/konova#287
This commit is contained in:
mpeltriaux 2023-02-01 14:10:04 +01:00
commit 951f810ce5

View File

@ -0,0 +1,69 @@
"""
Author: Michel Peltriaux
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
Contact: ksp-servicestelle@sgdnord.rlp.de
Created on: 01.02.23
"""
from compensation.models import Compensation, EcoAccount
from compensation.utils.quality import CompensationQualityChecker, EcoAccountQualityChecker
from ema.models import Ema
from ema.utils.quality import EmaQualityChecker
from intervention.models import Intervention
from intervention.utils.quality import InterventionQualityChecker
from konova.management.commands.setup import BaseKonovaCommand
class Command(BaseKonovaCommand):
help = "Runs quality check on certain entries"
__interventions = []
__compensations = []
__ecoaccount = []
__ema = []
identifier_like = None
title_like = None
def add_arguments(self, parser):
try:
parser.add_argument("--identifier-like", type=str)
parser.add_argument("--title-like", type=str)
except ValueError as e:
self._write_error(f"Argument error: {e}")
exit(-1)
def handle(self, *args, **options):
self.__handle_arguments(options)
self.__get_objects()
self.perform_quality_check()
def __handle_arguments(self, options):
self.identifier_like = options["identifier_like"] or ""
self.title_like = options["title_like"] or ""
def __get_objects(self):
_filter = {
"identifier__icontains": self.identifier_like,
"title__icontains": self.title_like,
}
self.__interventions = Intervention.objects.filter(**_filter)
self.__compensations = Compensation.objects.filter(**_filter)
self.__ecoaccount = EcoAccount.objects.filter(**_filter)
self.__ema = Ema.objects.filter(**_filter)
def perform_quality_check(self):
# Interventions
_runs = [
(self.__interventions, InterventionQualityChecker),
(self.__compensations, CompensationQualityChecker),
(self.__ecoaccount, EcoAccountQualityChecker),
(self.__ema, EmaQualityChecker),
]
for run in _runs:
entries = run[0]
CheckerClass = run[1]
for entry in entries:
checker = CheckerClass(entry)
checker.run_check()
if not checker.valid:
self._write_error(f"{entry.identifier};{';'.join(checker.messages)}")