feat(command): add script to generate permissions by local unit type (#2517)

* feat(command): add script to generate permissions by local unit type
* fix(api): update email notification sending logic on local unit creation
* fix(api): update permission on local unit update and validate api
* fix(permission): update permission creation logic
* chore(filter): add filter by country id on externally managed api
* chore(local-units): Update local unit permission script
- Add validation check for unique managed local units
* chore(script): add script to delete old global validator permission of local unit
* fix(serializer): add local unit permission in user me serializer
* fix(api): add is new local unit field and update userme serializer
* fix(models): make is new local unit field default false
---------

Co-authored-by: Sushil Tiwari <[email protected]>

Author: sudip-khanal

api/serializers.py

@@ -1,4 +1,5 @@
 import json
+from collections import defaultdict
 from typing import List, Union
 
 from django.conf import settings
@@ -11,6 +12,7 @@
 
 # from api.utils import pdf_exporter
 from api.tasks import generate_url
+from api.utils import CountryValidator, RegionValidator
 from deployments.models import EmergencyProject, Personnel, PersonnelDeployment
 from dref.models import Dref, DrefFinalReport, DrefOperationalUpdate
 from lang.models import String
@@ -1771,6 +1773,10 @@ class UserMeSerializer(UserSerializer):
     user_countries_regions = serializers.SerializerMethodField()
     limit_access_to_guest = serializers.BooleanField(read_only=True, source="profile.limit_access_to_guest")
 
+    local_unit_country_validators = serializers.SerializerMethodField()
+    local_unit_region_validators = serializers.SerializerMethodField()
+    local_unit_global_validators = serializers.SerializerMethodField()
+
     class Meta:
         model = User
         fields = UserSerializer.Meta.fields + (
@@ -1782,6 +1788,9 @@ class Meta:
             "is_per_admin_for_countries",
             "user_countries_regions",
             "limit_access_to_guest",
+            "local_unit_country_validators",
+            "local_unit_region_validators",
+            "local_unit_global_validators",
         )
 
     @staticmethod
@@ -1845,6 +1854,49 @@ def get_user_countries_regions(user):
         qs = UserCountry.objects.filter(user=user).distinct("country")
         return UserCountrySerializer(qs, many=True).data
 
+    @staticmethod
+    def get_local_unit_global_validators(user) -> List[int]:
+        permission_codenames = Permission.objects.filter(
+            codename__startswith="local_unit_global_validator", group__user=user
+        ).values_list("codename", flat=True)
+        global_validators = {int(code.split("_")[-1]) for code in permission_codenames}
+        return list(global_validators)
+
+    @staticmethod
+    def get_local_unit_region_validators(user) -> list[RegionValidator]:
+        permission_codenames = Permission.objects.filter(
+            codename__startswith="local_unit_region_validator",
+            group__user=user,
+        ).values_list("codename", flat=True)
+        region_validators = defaultdict(set)
+        for code in permission_codenames:
+            parts = code.split("_")[-2:]
+            if len(parts) == 2 and all(p.isdigit() for p in parts):
+                local_unit_type_id = int(parts[0])
+                region_id = int(parts[1])
+                region_validators[region_id].add(local_unit_type_id)
+        return [
+            {"region": region, "local_unit_types": list(local_unit_types)}
+            for region, local_unit_types in region_validators.items()
+        ]
+
+    @staticmethod
+    def get_local_unit_country_validators(user) -> list[CountryValidator]:
+        permission_codenames = Permission.objects.filter(
+            codename__startswith="local_unit_country_validator", group__user=user
+        ).values_list("codename", flat=True)
+        country_validator = defaultdict(set)
+        for code in permission_codenames:
+            parts = code.split("_")[-2:]
+            if len(parts) == 2 and all(p.isdigit() for p in parts):
+                local_unit_type_id = int(parts[0])
+                country_id = int(parts[1])
+                country_validator[country_id].add(local_unit_type_id)
+        return [
+            {"country": country, "local_unit_types": list(local_unit_types)}
+            for country, local_unit_types in country_validator.items()
+        ]
+
 
 class ActionSerializer(ModelSerializer):
     class Meta:

api/utils.py

@@ -1,6 +1,6 @@
 import base64
 import typing
-from typing import Optional
+from typing import Optional, TypedDict
 
 from django.core.exceptions import ValidationError
 from django.db import models
@@ -146,3 +146,13 @@ def generate_field_report_title(
     else:
         summary = f"{country.iso3}: {dtype.name} - {start_date} - {title} {suffix}"
     return summary
+
+
+class RegionValidator(TypedDict):
+    region: int
+    local_unit_types: list[int]
+
+
+class CountryValidator(TypedDict):
+    country: int
+    local_unit_types: list[int]

local_units/admin.py

@@ -56,6 +56,7 @@ class LocalUnitAdmin(CompareVersionAdmin, admin.OSMGeoAdmin):
     readonly_fields = (
         "validated",
         "is_locked",
+        "is_new_local_unit",
     )
     list_filter = (
         AutocompleteFilterFactory("Country", "country"),

local_units/filterset.py

@@ -35,4 +35,5 @@ class Meta:
             "country__name": ["exact", "in"],
             "country__iso3": ["exact", "in"],
             "country__iso": ["exact", "in"],
+            "country__id": ["exact", "in"],
         }

local_units/management/commands/make_global_validator_permission.py

@@ -0,0 +1,72 @@
+from django.contrib.auth.models import Group, Permission
+from django.contrib.contenttypes.models import ContentType
+from django.core.management.base import BaseCommand
+from django.db import transaction
+
+from api.models import Country, CountryType, Region
+from local_units.models import LocalUnit, LocalUnitType
+
+
+class Command(BaseCommand):
+    help = "Create local unit validator permissions and groups for each local unit type"
+
+    @transaction.atomic
+    def handle(self, *args, **options):
+        content_type = ContentType.objects.get_for_model(LocalUnit)
+        local_unit_types = LocalUnitType.objects.all()
+
+        # Global level Permissions and Groups
+        self.stdout.write(self.style.NOTICE("\n--- Creating/Updating GLOBAL validator permissions and groups"))
+        for local_unit_type in local_unit_types.iterator():
+            codename = f"local_unit_global_validator_{local_unit_type.id}"
+            name = f"Global validator for {local_unit_type.name}"
+            permission, _ = Permission.objects.get_or_create(
+                codename=codename,
+                content_type=content_type,
+                defaults=dict(name=name),
+            )
+            group_name = f"Local unit global validator for {local_unit_type.name}"
+            group, _ = Group.objects.get_or_create(name=group_name)
+            group.permissions.add(permission)
+            self.stdout.write(self.style.SUCCESS(f"\t--> {local_unit_type.name}"))
+
+        # Region level Permissions and Groups
+        self.stdout.write(self.style.NOTICE("\n--- Creating/Updating REGION-LEVEL validator permissions and groups"))
+        regions = Region.objects.all()
+        for region in regions.iterator():
+            region_name = region.get_name_display()
+            self.stdout.write(self.style.NOTICE(f"--> Region: {region_name}"))
+            for local_unit_type in local_unit_types.iterator():
+                codename = f"local_unit_region_validator_{local_unit_type.id}_{region.id}"
+                name = f"Local unit validator for {local_unit_type.name} {region_name}"
+                permission, _ = Permission.objects.get_or_create(
+                    codename=codename,
+                    content_type=content_type,
+                    defaults=dict(name=name),
+                )
+                group_name = f"Local unit validator for {local_unit_type.name} {region_name}"
+                group, _ = Group.objects.get_or_create(name=group_name)
+                group.permissions.add(permission)
+                self.stdout.write(self.style.SUCCESS(f"\t--> {local_unit_type.name}"))
+
+        # Country level Permissions and Groups
+        self.stdout.write(self.style.NOTICE("\n--- Creating/Updating COUNTRY-LEVEL validator permissions and groups"))
+        countries = Country.objects.filter(
+            is_deprecated=False, independent=True, iso3__isnull=False, record_type=CountryType.COUNTRY
+        )
+        for country in countries.iterator():
+            self.stdout.write(self.style.NOTICE(f"--> Country: {country.name}"))
+            for local_unit_type in local_unit_types.iterator():
+                codename = f"local_unit_country_validator_{local_unit_type.id}_{country.id}"
+                name = f"Local unit validator for {local_unit_type.name} {country.name}"
+                permission, _ = Permission.objects.get_or_create(
+                    codename=codename,
+                    content_type=content_type,
+                    defaults=dict(name=name),
+                )
+                group_name = f"Local unit validator for {local_unit_type.name} {country.name}"
+                group, _ = Group.objects.get_or_create(name=group_name)
+                group.permissions.add(permission)
+                self.stdout.write(self.style.SUCCESS(f"\t--> {local_unit_type.name}"))
+
+        self.stdout.write(self.style.SUCCESS("\n All permissions and groups created/updated successfully\n"))

local_units/management/commands/make_local_unit_validator_permissions.py

@@ -8,12 +8,11 @@
 from local_units.models import LocalUnit, Validator
 from local_units.utils import (
     get_email_context,
-    get_global_validators,
-    get_region_admins,
+    get_global_validators_by_type,
+    get_region_validators_by_type,
 )
 from main.sentry import SentryMonitor
-
-# from notifications.notification import send_notification
+from notifications.notification import send_notification
 
 
 class Command(BaseCommand):
@@ -28,15 +27,15 @@ def handle(self, *args, **options):
             validated=False,
             is_deprecated=False,
             last_sent_validator_type=Validator.LOCAL,
-            created_at__lte=timezone.now() - timedelta(days=14),
+            created_at__lte=timezone.now() - timedelta(days=7),
         )
 
         # Global Validators: 28 days
         queryset_for_global_validators = LocalUnit.objects.filter(
             validated=False,
             is_deprecated=False,
             last_sent_validator_type=Validator.REGIONAL,
-            created_at__lte=timezone.now() - timedelta(days=28),
+            created_at__lte=timezone.now() - timedelta(days=14),
         )
 
         for local_unit in queryset_for_regional_validators:
@@ -46,11 +45,11 @@ def handle(self, *args, **options):
             email_subject = "Action Required: Local Unit Pending Validation"
             email_type = "Local Unit"
 
-            for region_admin_validator in get_region_admins(local_unit):
+            for region_admin_validator in get_region_validators_by_type(local_unit):
                 try:
                     email_context["full_name"] = region_admin_validator.get_full_name()
                     email_body = render_to_string("email/local_units/local_unit.html", email_context)
-                    print(email_subject, region_admin_validator.email, email_body, email_type)  # send_notification disabling
+                    send_notification(email_subject, region_admin_validator.email, email_body, email_type)
                     local_unit.last_sent_validator_type = Validator.REGIONAL
                     local_unit.save(update_fields=["last_sent_validator_type"])
                 except Exception as e:
@@ -68,11 +67,11 @@ def handle(self, *args, **options):
             email_subject = "Action Required: Local Unit Pending Validation"
             email_type = "Local Unit"
 
-            for global_validator in get_global_validators():
+            for global_validator in get_global_validators_by_type(local_unit):
                 try:
                     email_context["full_name"] = global_validator.get_full_name()
                     email_body = render_to_string("email/local_units/local_unit.html", email_context)
-                    print(email_subject, global_validator.email, email_body, email_type)  # send_notification disabling
+                    send_notification(email_subject, global_validator.email, email_body, email_type)
                     local_unit.last_sent_validator_type = Validator.GLOBAL
                     local_unit.save(update_fields=["last_sent_validator_type"])
                 except Exception as e:

local_units/management/commands/notify_validators.py

@@ -0,0 +1,30 @@
+from django.contrib.auth.models import Group, Permission
+from django.contrib.contenttypes.models import ContentType
+from django.core.management.base import BaseCommand
+from django.db import transaction
+
+from local_units.models import LocalUnit
+
+
+class Command(BaseCommand):
+    help = "Delete old permission and group for Local Unit Global Validator"
+
+    @transaction.atomic
+    def handle(self, *args, **options):
+        self.stdout.write(self.style.NOTICE(" Deleting old permission and group for local unit global validator"))
+        codename = "local_unit_global_validator"
+        group_name = "Local Unit Global Validators"
+        content_type = ContentType.objects.get_for_model(LocalUnit)
+
+        permission = Permission.objects.filter(codename=codename, content_type=content_type)
+        if permission.exists():
+            permission.delete()
+            self.stdout.write(self.style.SUCCESS(f"\t--> Deleted Permission {codename}"))
+        else:
+            self.stdout.write(self.style.WARNING(f"\t--> Permission {codename} not found"))
+        group = Group.objects.filter(name=group_name)
+        if group.exists():
+            group.delete()
+            self.stdout.write(self.style.SUCCESS(f"\t--> Deleted Group {group_name}"))
+        else:
+            self.stdout.write(self.style.WARNING(f"\t--> Group {group_name} not found"))

local_units/management/commands/remove_old_global_validator_permission.py

@@ -0,0 +1,19 @@
+# Generated by Django 4.2.19 on 2025-07-25 11:24
+
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("local_units", "0021_localunit_update_reason_overview_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="localunit",
+            name="is_new_local_unit",
+            field=models.BooleanField(default=False, verbose_name="Is New Local Unit?"),
+        ),
+    ]

local_units/migrations/0022_localunit_is_new_local_unit.py

@@ -371,6 +371,8 @@ class DeprecateReason(models.IntegerChoices):
         default=Validator.LOCAL,
     )
 
+    is_new_local_unit = models.BooleanField(default=False, verbose_name=("Is New Local Unit?"))
+
     def __str__(self):
         branch_name = self.local_branch_name or self.english_branch_name
         return f"{branch_name} ({self.country.name})"