Source code for akvo.rsr.management.commands.eutf_hierarchy_check

# -*- coding: utf-8 -*-

# Akvo Reporting is covered by the GNU Affero General Public License.
# See more details in the license.txt file located at the root folder of the Akvo RSR module.
# For additional details on the GNU license please see < http://www.gnu.org/licenses/agpl.html >.


import tablib
from django.conf import settings

from django.core.management.base import BaseCommand

from akvo.rsr.models import IndicatorPeriod
from akvo.utils import rsr_send_mail
from ...models import Project, Result, Indicator

INDICATORS = 'indicators'
PERIODS = 'periods'


[docs]class Command(BaseCommand): args = '' help = ('Script that checks the EUTF results framework "core indicators" for errors') def __init__(self): super(Command, self).__init__() self.root = Project.objects.get(pk=settings.EUTF_ROOT_PROJECT) self.all_projects = self.root.descendants() self.eutf_results = Result.objects.filter(project=self.root) self.dec_results = Result.objects.filter(parent_result__in=self.eutf_results) self.ctr_results = Result.objects.filter(parent_result__in=self.dec_results) self.missing_results = [] self.indicator_orphans = [] self.period_orphans = []
[docs] @staticmethod def find_indicator_orphans(results, results_above): orphans = Indicator.objects.none() if Indicator.objects.filter(result__in=results, parent_indicator__isnull=True).exists(): orphans = Indicator.objects.filter(result__in=results, parent_indicator__isnull=True) if Indicator.objects.filter(parent_indicator__result__in=results_above, result__isnull=True).exists(): orphans = orphans | Indicator.objects.filter(parent_indicator__result__in=results_above, result__isnull=True) return orphans.distinct()
[docs] @staticmethod def find_period_orphans(results, results_above): orphans = IndicatorPeriod.objects.none() if IndicatorPeriod.objects.filter(indicator__result__in=results, parent_period__isnull=True).exists(): orphans = IndicatorPeriod.objects.filter(indicator__result__in=results, parent_period__isnull=True) if IndicatorPeriod.objects.filter(parent_period__indicator__result__in=results_above, indicator__isnull=True).exists(): orphans = orphans | IndicatorPeriod.objects.filter( parent_period__indicator__result__in=results_above, indicator__isnull=True) return orphans.distinct()
[docs] @staticmethod def create_indicator_report(orphans): problem_indicators = tablib.Dataset() problem_indicators.headers = [ 'Indicator ID', 'Result ID', 'Result parent result ID', 'Project ID', 'Parent result project ID', 'Indicator title', 'Result title', 'Project title', 'Parent result project title', ] for orphan in orphans: pk = orphan.pk title = orphan.title, result_pk = orphan.result.pk if orphan.result else None result_title = orphan.result.title if result_pk else None project_pk = orphan.result.project.pk if result_pk and orphan.result.project else None project_title = orphan.result.project.title if project_pk else None result_parent_result_pk = (orphan.result.parent_result.pk if result_pk and orphan.result.parent_result else None) result_parent_result_project_pk = (orphan.result.parent_result.project.pk if result_parent_result_pk and orphan.result.parent_result.project else None) result_parent_result_project_title = (orphan.result.parent_result.project.title if result_parent_result_pk and orphan.result.parent_result.project else None) problem_indicators.append([ pk, result_pk, result_parent_result_pk, project_pk, result_parent_result_project_pk, title, result_title, project_title, result_parent_result_project_title, ]) return problem_indicators
[docs] @staticmethod def create_period_report(orphans): problem_periods = tablib.Dataset() problem_periods.headers = [ 'Period ID', 'Parent period ID', 'Indicator ID', 'Indicator parent indicator ID', 'Parent period indicator ID', 'Result ID', 'Project ID', 'Project title', 'Result title', 'Indicator title', 'Period start and end', ] for orphan in orphans: pk = orphan.pk parent_period_pk = orphan.parent_period.pk if orphan.parent_period else None indicator_pk = orphan.indicator.pk if orphan.indicator else None indicator_parent_indicator_pk = (orphan.indicator.parent_indicator.pk if indicator_pk and orphan.indicator.parent_indicator else None) parent_period_indicator_pk = (orphan.parent_period.indicator.pk if parent_period_pk and orphan.parent_period.indicator else None) result_pk = (orphan.indicator.result.pk if indicator_pk and orphan.indicator.result else None) project_pk = (orphan.indicator.result.project.pk if result_pk and orphan.indicator.result.project else None) project_title = orphan.indicator.result.project.title if project_pk else None result_title = (orphan.indicator.result.title if indicator_pk and orphan.indicator.result else None) indicator_title = orphan.indicator.title if indicator_pk else None problem_periods.append([ pk, parent_period_pk, indicator_pk, indicator_parent_indicator_pk, parent_period_indicator_pk, result_pk, project_pk, project_title, result_title, indicator_title, "{} : {}".format(orphan.period_start, orphan.period_end), ]) return problem_periods
[docs] def analyze_indicators(self): self.indicator_orphans = self.find_indicator_orphans(self.dec_results, self.eutf_results) self.indicator_orphans = self.indicator_orphans | self.find_indicator_orphans( self.ctr_results, self.dec_results )
[docs] def analyze_periods(self): self.period_orphans = self.find_period_orphans(self.dec_results, self.eutf_results) self.period_orphans = self.period_orphans | self.find_period_orphans( self.ctr_results, self.dec_results )
[docs] @staticmethod def send_eutf_error_email(error_type, attachments): recipients = getattr( settings, 'EUTF_HIERARCHY_ERROR_RECIPIENTS', ['rsr@akvo.org', ] ) rsr_send_mail( recipients, subject='results_framework/eutf_hierarchy_error_subject.txt', message='results_framework/eutf_hierarchy_error_message.txt', msg_context={'error_type': error_type}, html_message='results_framework/eutf_hierarchy_error_message.html', attachments=attachments )
[docs] def handle(self, *args, **options): self.analyze_indicators() if self.indicator_orphans: problem_indicators = self.create_indicator_report(self.indicator_orphans) content = "The following problem indicators were detected\n\n{}".format( problem_indicators.export('tsv')) attachment = [{ 'filename': 'eutf_hierarchy_indicator_errors.tsv', 'content': content, 'mimetype': 'text/tab-separated-values' }] self.send_eutf_error_email(INDICATORS, attachment) return self.analyze_periods() if self.period_orphans: problem_periods = self.create_period_report(self.period_orphans) content = "The following problem periods were detected\n\n{}".format( problem_periods.export('tsv')) attachment = [{ 'filename': 'eutf_hierarchy_period_errors.tsv', 'content': content, 'mimetype': 'text/tab-separated-values' }] self.send_eutf_error_email(PERIODS, attachment)