Official stats prototype

api/cases/enums.py

@@ -417,11 +417,13 @@ class LicenceDecisionType:
     ISSUED = "issued"
     REFUSED = "refused"
     REVOKED = "revoked"
+    ISSUED_ON_APPEAL = "issued_on_appeal"
 
     choices = [
         (ISSUED, "issued"),
         (REFUSED, "refused"),
         (REVOKED, "revoked"),
+        (ISSUED_ON_APPEAL, "issued_on_appeal"),
     ]
 
     decision_map = {
@@ -439,8 +441,13 @@ def templates(cls):
             cls.ISSUED: SIEL_LICENCE_TEMPLATE_ID,
             cls.REFUSED: SIEL_REFUSAL_TEMPLATE_ID,
             cls.REVOKED: None,
+            cls.ISSUED_ON_APPEAL: None,
         }
 
     @classmethod
     def advice_type_to_decision(cls, advice_type):
         return cls.decision_map[advice_type]
+
+    @classmethod
+    def get_template(cls, decision):
+        return cls.templates()[decision]

api/cases/migrations/0069_licencedecision_excluded_from_statistics_reason.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.16 on 2024-11-14 16:36
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cases", "0068_populate_licence_decisions"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="licencedecision",
+            name="excluded_from_statistics_reason",
+            field=models.TextField(blank=True, default=None, null=True),
+        ),
+    ]

api/cases/migrations/0070_attach_licence_to_licence_decision.py

@@ -0,0 +1,96 @@
+# Generated by Django 4.2.16 on 2024-11-13 17:19
+import functools
+import operator
+
+from django.contrib.postgres.aggregates import ArrayAgg
+from django.db import migrations, transaction
+from django.db.models import Case as DBCase, Q, TextField, Value, When
+from django.db.models.functions import Cast
+
+from api.audit_trail.enums import AuditType
+from api.cases.enums import AdviceType, LicenceDecisionType
+
+
+@transaction.atomic
+def attach_licence_to_licence_decisions(apps, schema_editor):
+    LicenceDecision = apps.get_model("cases", "LicenceDecision")
+
+    Audit = apps.get_model("audit_trail", "Audit")
+    GeneratedCaseDocument = apps.get_model("generated_documents", "GeneratedCaseDocument")
+    LicenceDecision = apps.get_model("cases", "LicenceDecision")
+    Licence = apps.get_model("licences", "Licence")
+
+    licence_decisions_to_update = []
+
+    final_decision_qs = Audit.objects.filter(verb=AuditType.CREATED_FINAL_RECOMMENDATION).order_by("-created_at")
+
+    document_qs = (
+        GeneratedCaseDocument.objects.filter(
+            template_id__in=LicenceDecisionType.templates().values(),
+            advice_type=AdviceType.APPROVE,
+            visible_to_exporter=True,
+            safe=True,
+        )
+        .annotate(template_ids=ArrayAgg(Cast("template_id", output_field=TextField()), distinct=True))
+        .filter(
+            functools.reduce(
+                operator.or_,
+                [Q(template_ids=[template_id]) for template_id in LicenceDecisionType.templates().values()],
+            )
+        )
+        .annotate(
+            decision=DBCase(
+                *[
+                    When(template_ids=[template_id], then=Value(decision))
+                    for decision, template_id in LicenceDecisionType.templates().items()
+                ]
+            )
+        )
+    )
+
+    # When running tests audit entries are not available so filtering documents
+    # the audit log created date earlier fails
+    if final_decision_qs:
+        earliest_audit_log = final_decision_qs.last()
+        document_qs = document_qs.filter(
+            created_at__date__lt=earliest_audit_log.created_at.date(),
+        )
+
+    for audit_log in final_decision_qs:
+        advice_type = audit_log.payload["decision"]
+        if advice_type != AdviceType.APPROVE:
+            continue
+
+        decision = LicenceDecisionType.advice_type_to_decision(advice_type)
+        obj = LicenceDecision.objects.get(
+            case_id=str(audit_log.target_object_id),
+            decision=decision,
+            created_at=audit_log.created_at,
+        )
+        obj.licence = Licence.objects.get(reference_code=audit_log.payload["licence_reference"])
+        licence_decisions_to_update.append(obj)
+
+    for document in document_qs:
+        obj = LicenceDecision.objects.get(
+            case_id=str(document.case_id),
+            decision=document.decision,
+            created_at=document.created_at,
+        )
+        obj.licence = document.licence
+        licence_decisions_to_update.append(obj)
+
+    LicenceDecision.objects.bulk_update(licence_decisions_to_update, ["licence"])
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cases", "0069_licencedecision_excluded_from_statistics_reason"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            attach_licence_to_licence_decisions,
+            migrations.RunPython.noop,
+        ),
+    ]

api/cases/migrations/0071_licencedecision_previous_decision.py

@@ -0,0 +1,79 @@
+# Generated by Django 4.2.16 on 2024-11-18 13:14
+
+from django.db import migrations, models, transaction
+from django.db.models import Count
+import django.db.models.deletion
+
+
+@transaction.atomic
+def populate_previous_decisions(apps, schema_editor):
+    Case = apps.get_model("cases", "Case")
+    LicenceDecision = apps.get_model("cases", "LicenceDecision")
+
+    licence_decisions_to_update = []
+
+    # We only need to update cases where there are multiple decisions
+    # In case of single decision then previous_decision field of licence decision is
+    # not set by default so nothing to update
+    case_qs = (
+        Case.objects.all()
+        .annotate(
+            num_decisions=Count("licence_decisions"),
+        )
+        .filter(
+            num_decisions__gt=1,
+        )
+    )
+
+    # By default previous_decision is not set for all decisions.
+    # Previous_decision is reset whenever there is a change in decision otherwise it
+    # points to the decision in previous instance.
+    #
+    # When exporting we just filter all decisions where previous_decision is None which
+    # gives us the earliest decision time as required
+    # ---------------------------------------------------
+    #  [ld1, ld2, ..., ldn]       | [previous_decision field value]
+    # ---------------------------------------------------
+    #  [ISSUED, ISSUED]           | [None, ld1]
+    #  [ISSUED, ISSUED, ISSUED]   | [None, ld1, ld2]
+    #  [ISSUED, REFUSED]          | [None, None]
+    #  [REFUSED, ISSUED]          | [None, None]
+    #  [ISSUED, REVOKED]          | [None, None]
+    #  [ISSUED, REVOKED, ISSUED]  | [None, None, None]
+    # ---------------------------------------------------
+    #
+    for case in case_qs:
+        previous_decision = None
+        for item in case.licence_decisions.order_by("created_at"):
+            if previous_decision and item.decision == previous_decision.decision:
+                item.previous_decision = previous_decision
+                licence_decisions_to_update.append(item)
+
+            previous_decision = item
+
+    LicenceDecision.objects.bulk_update(licence_decisions_to_update, ["previous_decision"])
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cases", "0070_attach_licence_to_licence_decision"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="licencedecision",
+            name="previous_decision",
+            field=models.ForeignKey(
+                default=None,
+                null=True,
+                on_delete=django.db.models.deletion.DO_NOTHING,
+                related_name="previous_decisions",
+                to="cases.licencedecision",
+            ),
+        ),
+        migrations.RunPython(
+            populate_previous_decisions,
+            migrations.RunPython.noop,
+        ),
+    ]

api/cases/migrations/0072_licencedecision_denial_reasons.py

@@ -0,0 +1,19 @@
+# Generated by Django 4.2.16 on 2024-11-21 12:52
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("denial_reasons", "0006_populate_uuid_field"),
+        ("cases", "0071_licencedecision_previous_decision"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="licencedecision",
+            name="denial_reasons",
+            field=models.ManyToManyField(to="denial_reasons.denialreason"),
+        ),
+    ]

api/cases/migrations/0073_update_licencedecision_denial_reasons.py

@@ -0,0 +1,56 @@
+# Generated by Django 4.2.16 on 2024-11-21 13:24
+
+from django.db import migrations
+
+from api.audit_trail.enums import AuditType
+
+
+def update_licencedecision_denial_reasons(apps, schema_editor):
+    LicenceDecision = apps.get_model("cases", "LicenceDecision")
+    Advice = apps.get_model("cases", "Advice")
+
+    denial_reasons = (
+        Advice.objects.filter(
+            case__licence_decisions__decision="refused",
+            team_id="58e77e47-42c8-499f-a58d-94f94541f8c6",  # Just care about LU advice
+        )
+        .only("denial_reasons__display_value", "case__licence_decisions__id")
+        .exclude(denial_reasons__display_value__isnull=True)  # This removes refusals without any criteria
+        .values_list("denial_reasons__display_value", "case__licence_decisions__id")
+        .order_by()  # We need to remove the order_by to make sure the distinct works
+        .distinct()
+    )
+
+    updated_cases = set()
+    for denial_reason, licence_decision_id in denial_reasons:
+        licence_decision = LicenceDecision.objects.get(pk=licence_decision_id)
+        licence_decision.denial_reasons.add(denial_reason)
+        updated_cases.add(licence_decision.case.pk)
+
+    Audit = apps.get_model("audit_trail", "Audit")
+    Case = apps.get_model("cases", "Case")
+    refusal_criteria_audits = Audit.objects.exclude(target_object_id__in=updated_cases).filter(
+        verb=AuditType.CREATE_REFUSAL_CRITERIA
+    )
+    for audit in refusal_criteria_audits:
+        case = Case.objects.get(pk=audit.target_object_id)
+        refusal_licence_decisions = case.licence_decisions.filter(decision="refused")
+        if not refusal_licence_decisions.exists():
+            continue
+        for licence_decision in refusal_licence_decisions:
+            denial_reasons = audit.payload["additional_text"].replace(".", "").replace(" ", "").split(",")
+            licence_decision.denial_reasons.set(denial_reasons)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cases", "0072_licencedecision_denial_reasons"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            update_licencedecision_denial_reasons,
+            migrations.RunPython.noop,
+        ),
+    ]

api/cases/migrations/0074_alter_licencedecision_options_and_more.py

@@ -0,0 +1,66 @@
+# Generated by Django 4.2.16 on 2024-11-21 23:21
+
+from django.db import migrations, models, transaction
+
+from django.db.models import Count
+
+from api.cases.enums import LicenceDecisionType
+
+
+@transaction.atomic
+def populate_issued_on_appeal(apps, schema_editor):
+    Case = apps.get_model("cases", "Case")
+    LicenceDecision = apps.get_model("cases", "LicenceDecision")
+
+    licence_decisions_to_update = []
+
+    # Filter cases that have two decisions because for appeals
+    # the first decision will be refused which is issued on appeal later
+    case_qs = (
+        Case.objects.all()
+        .annotate(
+            num_decisions=Count("licence_decisions"),
+        )
+        .filter(
+            num_decisions=2,
+        )
+    )
+
+    for case in case_qs:
+        prev, current = case.licence_decisions.all()
+        if prev.decision == LicenceDecisionType.REFUSED and current.decision == LicenceDecisionType.ISSUED:
+            current.decision = LicenceDecisionType.ISSUED_ON_APPEAL
+            licence_decisions_to_update.append(current)
+
+    LicenceDecision.objects.bulk_update(licence_decisions_to_update, ["decision"])
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cases", "0073_update_licencedecision_denial_reasons"),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name="licencedecision",
+            options={"ordering": ("created_at",)},
+        ),
+        migrations.AlterField(
+            model_name="licencedecision",
+            name="decision",
+            field=models.CharField(
+                choices=[
+                    ("issued", "issued"),
+                    ("refused", "refused"),
+                    ("revoked", "revoked"),
+                    ("issued_on_appeal", "issued_on_appeal"),
+                ],
+                max_length=50,
+            ),
+        ),
+        migrations.RunPython(
+            populate_issued_on_appeal,
+            migrations.RunPython.noop,
+        ),
+    ]

api/cases/migrations/tests/test_0070_attach_licence_to_licence_decision.py

@@ -0,0 +1,17 @@
+import pytest
+
+from api.cases.enums import LicenceDecisionType
+
+INITIAL_MIGRATION = "0069_licencedecision_excluded_from_statistics_reason"
+MIGRATION_UNDER_TEST = "0070_attach_licence_to_licence_decision"
+
+
+@pytest.mark.django_db()
+def test_attach_licence_to_licence_decisions(migrator):
+
+    old_state = migrator.apply_initial_migration(("cases", INITIAL_MIGRATION))
+    new_state = migrator.apply_tested_migration(("cases", MIGRATION_UNDER_TEST))
+
+    LicenceDecision = new_state.apps.get_model("cases", "LicenceDecision")
+
+    assert LicenceDecision.objects.filter(decision=LicenceDecisionType.ISSUED, licence__isnull=True).count() == 0