Merge pull request #2629 from IFRCGo/feature/localunit-request-changes

Local Unit: Export feature and add validations checks

Author: szabozoltan69

assets

@@ -20,6 +20,82 @@
 
 ContextType = TypeVar("ContextType")
 
+START_ROW = 4
+
+
+# NOTE(IMPORTANT): These mappings should align with templates and serializer fields.
+# Also make sure to update export.py if there are changes here in the headers.
+
+
+BASE_HEADER_MAP = {
+    "Date of Update": "date_of_data",
+    "Local Unit Name (En)": "english_branch_name",
+    "Local Unit Name (Local)": "local_branch_name",
+    "Visibility": "visibility",
+    "Coverage": "level",
+    "Sub-type": "subtype",
+    "Focal Person (En)": "focal_person_en",
+    "Source (En)": "source_en",
+    "Source (Local)": "source_loc",
+    "Focal Person (Local)": "focal_person_loc",
+    "Address (Local)": "address_loc",
+    "Address (En)": "address_en",
+    "Locality (Local)": "city_loc",
+    "Locality (En)": "city_en",
+    "Local Unit Post Code": "postcode",
+    "Local Unit Email": "email",
+    "Local Unit Phone Number": "phone",
+    "Local Unit Website": "link",
+    "Latitude": "latitude",
+    "Longitude": "longitude",
+}
+
+HEALTH_HEADER_MAP = {
+    **BASE_HEADER_MAP,
+    **{
+        "Focal Person Name (En)": "focal_person_en",
+        "Focal Person Name (Local)": "focal_person_loc",
+        "Focal Person Email": "focal_point_email",
+        "Focal Person Phone Number": "focal_point_phone_number",
+        "Focal Person Position": "focal_point_position",
+        "Health Facility Type": "health_facility_type",
+        "Other Facility Type": "other_facility_type",
+        "Affiliation": "affiliation",
+        "Other Affiliation": "other_affiliation",
+        "Functionality": "functionality",
+        "Primary Health Care Center": "primary_health_care_center",
+        "Specialities": "speciality",
+        "Hospital Type": "hospital_type",
+        "Teaching Hospital": "is_teaching_hospital",
+        "In-patient Capacity": "is_in_patient_capacity",
+        "Isolation Rooms": "is_isolation_rooms_wards",
+        "Number of Isolation Beds": "number_of_isolation_rooms",
+        "Warehousing": "is_warehousing",
+        "Cold Chain": "is_cold_chain",
+        "Other Medical Heal": "other_medical_heal",
+        "Maximum Bed Capacity": "maximum_capacity",
+        "General Medical Services": "general_medical_services",
+        "Specialized Medical Services (beyond primary level)": "specialized_medical_beyond_primary_level",
+        "Blood Services": "blood_services",
+        "Other Services": "other_services",
+        "Total Number of Human Resources": "total_number_of_human_resource",
+        "General Practitioners": "general_practitioner",
+        "Resident Doctors": "residents_doctor",
+        "Specialists": "specialist",
+        "Nurses": "nurse",
+        "Nursing Aids": "nursing_aid",
+        "Dentists": "dentist",
+        "Midwife": "midwife",
+        "Pharmacists": "pharmacists",
+        "Other Profiles": "other_profiles",
+        "Other Training Facility": "other_training_facilities",
+        "Professional Training Facilities": "professional_training_facilities",
+        "Ambulance Type A": "ambulance_type_a",
+        "Ambulance Type B": "ambulance_type_b",
+        "Ambulance Type C": "ambulance_type_c",
+    },
+}
+
 
 class BulkUploadError(Exception):
     """Custom exception for bulk upload errors."""
@@ -137,7 +213,7 @@ def delete_existing_local_unit(self):
     def _validate_type(self, fieldnames) -> None:
         pass
 
-    def is_excel_data_empty(self, sheet, data_start_row=4):
+    def is_excel_data_empty(self, sheet, data_start_row=START_ROW):
         """Check if file is empty or not"""
         for row in sheet.iter_rows(values_only=True, min_row=data_start_row):
             if any(cell is not None for cell in row):
@@ -238,7 +314,7 @@ def _finalize_failure(self) -> None:
         self.bulk_upload.status = LocalUnitBulkUpload.Status.FAILED
         self.bulk_upload.save(update_fields=["success_count", "failed_count", "status", "error_file"])
 
-        logger.info(f"[BulkUpload:{self.bulk_upload.pk}] FAILED: " f"{self.success_count} succeeded, {self.failed_count} failed.")
+        logger.info(f"[BulkUpload:{self.bulk_upload.pk}] SUMMARY: " f"{self.success_count} SUCCESS, {self.failed_count} FAILED.")
 
 
 @dataclass(frozen=True)
@@ -250,28 +326,7 @@ class LocalUnitUploadContext:
 
 
 class BaseBulkUploadLocalUnit(BaseBulkUpload[LocalUnitUploadContext]):
-    HEADER_MAP = {
-        "Date of Update": "date_of_data",
-        "Local Unit Name (En)": "english_branch_name",
-        "Local Unit Name (Local)": "local_branch_name",
-        "Visibility": "visibility",
-        "Coverage": "level",
-        "Sub-type": "subtype",
-        "Focal Person (En)": "focal_person_en",
-        "Source (En)": "source_en",
-        "Source (Local)": "source_loc",
-        "Focal Person (Local)": "focal_person_loc",
-        "Address (Local)": "address_loc",
-        "Address (En)": "address_en",
-        "Locality (Local)": "city_loc",
-        "Locality (En)": "city_en",
-        "Local Unit Post Code": "postcode",
-        "Local Unit Email": "email",
-        "Local Unit Phone Number": "phone",
-        "Local Unit Website": "link",
-        "Latitude": "latitude",
-        "Longitude": "longitude",
-    }
+    HEADER_MAP = BASE_HEADER_MAP
 
     def __init__(self, bulk_upload: LocalUnitBulkUpload):
         from local_units.serializers import LocalUnitBulkUploadDetailSerializer
@@ -315,48 +370,8 @@ def _validate_type(self, fieldnames: list[str]) -> None:
 
 class BulkUploadHealthData(BaseBulkUpload[LocalUnitUploadContext]):
     # Local Unit headers + Health Data headers
-    HEADER_MAP = {
-        **BaseBulkUploadLocalUnit.HEADER_MAP,
-        **{
-            "Focal Person Email": "focal_point_email",
-            "Focal Person Phone Number": "focal_point_phone_number",
-            "Focal Person Position": "focal_point_position",
-            "Health Facility Type": "health_facility_type",
-            "Other Facility Type": "other_facility_type",
-            "Affiliation": "affiliation",
-            "Other Affiliation": "other_affiliation",
-            "Functionality": "functionality",
-            "Primary Health Care Center": "primary_health_care_center",
-            "Specialities": "speciality",
-            "Hospital Type": "hospital_type",
-            "Teaching Hospital": "is_teaching_hospital",
-            "In-patient Capacity": "is_in_patient_capacity",
-            "Isolation Rooms": "is_isolation_rooms_wards",
-            "Number of Isolation Beds": "number_of_isolation_rooms",
-            "Warehousing": "is_warehousing",
-            "Cold Chain": "is_cold_chain",
-            "Maximum Bed Capacity": "maximum_capacity",
-            "General Medical Services": "general_medical_services",
-            "Specialized Medical Services (beyond primary level)": "specialized_medical_beyond_primary_level",
-            "Blood Services": "blood_services",
-            "Other Services": "other_services",
-            "Total Number of Human Resources": "total_number_of_human_resource",
-            "General Practitioners": "general_practitioner",
-            "Resident Doctors": "residents_doctor",
-            "Specialists": "specialist",
-            "Nurses": "nurse",
-            "Nursing Aids": "nursing_aid",
-            "Dentists": "dentist",
-            "Midwife": "midwife",
-            "Pharmacists": "pharmacists",
-            "Other Profiles": "other_profiles",
-            "Other Training Facility": "other_training_facilities",
-            "Professional Training Facilities": "professional_training_facilities",
-            "Ambulance Type A": "ambulance_type_a",
-            "Ambulance Type B": "ambulance_type_b",
-            "Ambulance Type C": "ambulance_type_c",
-        },
-    }
+
+    HEADER_MAP = HEALTH_HEADER_MAP
 
     def __init__(self, bulk_upload: LocalUnitBulkUpload):
         from local_units.serializers import LocalUnitBulkUploadDetailSerializer

go-static/files/local_units/Administrative Bulk Import Template - Local Units.xlsx

@@ -0,0 +1,101 @@
+from pathlib import Path
+from typing import Any, Callable
+
+from django.conf import settings
+from django.http import HttpResponse
+from openpyxl import load_workbook
+
+from local_units.bulk_upload import BASE_HEADER_MAP, HEALTH_HEADER_MAP, START_ROW
+from local_units.models import HealthData, LocalUnit
+
+LOCAL_UNIT_EXPORT_HANDLERS: dict[str, Callable[[LocalUnit], Any]] = {
+    "date_of_data": lambda u: u.date_of_data.strftime("%Y-%m-%d") if u.date_of_data else "",
+    "visibility": lambda u: u.get_visibility_display() if u.visibility else "",
+    "level": lambda u: u.level.name if u.level else "",
+    "latitude": lambda u: u.location.y if u.location else "",
+    "longitude": lambda u: u.location.x if u.location else "",
+}
+
+HEALTH_EXPORT_HANDLERS: dict[str, Callable[[HealthData], Any]] = {
+    "affiliation": lambda h: h.affiliation.name if h.affiliation else "",
+    "functionality": lambda h: h.functionality.name if h.functionality else "",
+    "health_facility_type": lambda h: h.health_facility_type.name if h.health_facility_type else "",
+    "hospital_type": lambda h: h.hospital_type.name if h.hospital_type else "",
+    "primary_health_care_center": lambda h: h.primary_health_care_center.name if h.primary_health_care_center else "",
+    "general_medical_services": lambda h: ", ".join(m.name for m in h.general_medical_services.all()),
+    "specialized_medical_beyond_primary_level": lambda h: ", ".join(
+        m.name for m in h.specialized_medical_beyond_primary_level.all()
+    ),
+    "blood_services": lambda h: ", ".join(m.name for m in h.blood_services.all()),
+    "professional_training_facilities": lambda h: ", ".join(m.name for m in h.professional_training_facilities.all()),
+    "other_profiles": lambda h: ", ".join(p.position for p in h.other_profiles.all()),
+    "other_medical_heal": lambda h: "Yes" if h.other_medical_heal else "No",
+    "is_in_patient_capacity": lambda h: "Yes" if h.is_in_patient_capacity else "No",
+    "is_teaching_hospital": lambda h: "Yes" if h.is_teaching_hospital else "No",
+    "is_isolation_rooms_wards": lambda h: "Yes" if h.is_isolation_rooms_wards else "No",
+    "is_warehousing": lambda h: "Yes" if h.is_warehousing else "No",
+    "is_cold_chain": lambda h: "Yes" if h.is_cold_chain else "No",
+}
+
+
+def _resolve_value(unit: LocalUnit, field: str, is_health: bool):
+    if field in LOCAL_UNIT_EXPORT_HANDLERS:
+        return LOCAL_UNIT_EXPORT_HANDLERS[field](unit)
+
+    # First check if there's a specific handler for health fields and fall back to direct attribute access
+    if is_health and unit.health and field in HEALTH_EXPORT_HANDLERS:
+        return HEALTH_EXPORT_HANDLERS[field](unit.health)
+
+    if is_health and unit.health and hasattr(unit.health, field):
+        return getattr(unit.health, field) or ""
+
+    return getattr(unit, field, "") or ""
+
+
+def export_local_units_to_excel(
+    queryset,
+    *,
+    is_health: bool = False,
+    file_name: str | None = None,
+):
+    """
+    Exports local units to an Excel file and returns it as an HTTP response.
+    """
+
+    header_map = HEALTH_HEADER_MAP if is_health else BASE_HEADER_MAP
+
+    if is_health:
+        template = "Health-Care-Bulk-Import-Template-Local-Units.xlsm"
+    else:
+        template = "Administrative-Bulk-Import-Template-Local-Units.xlsx"
+
+    template_path = Path(settings.BASE_DIR) / "go-static/files/local_units" / template
+
+    if not template_path.exists():
+        raise FileNotFoundError(f"Excel template not found: {template_path}")
+
+    wb = load_workbook(template_path, read_only=False)
+    ws = wb.active
+
+    if ws is None:
+        raise ValueError("Worksheet could not be loaded from template.")
+
+    headers = [cell.value for cell in ws[2] if cell.value]
+    start_row = START_ROW
+
+    for row_idx, unit in enumerate(queryset, start=start_row):
+        for col_idx, header in enumerate(headers, start=1):
+            field = header_map.get(str(header).strip())
+            if not field:
+                continue
+            ws.cell(
+                row=row_idx,
+                column=col_idx,
+                value=_resolve_value(unit, field, is_health),
+            )
+
+    response = HttpResponse(content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
+    response["Content-Disposition"] = f'attachment; filename="{file_name or "local_units_export.xlsx"}"'
+
+    wb.save(response)
+    return response

go-static/files/local_units/Administrative-Bulk-Import-Template-Local-Units.xlsx

@@ -0,0 +1,53 @@
+from django.core.management.base import BaseCommand
+from django.db import transaction
+
+from api.models import Country, CountryType
+from local_units.models import ExternallyManagedLocalUnit, LocalUnit, LocalUnitType
+
+
+class Command(BaseCommand):
+    help = "Create Externally Managed status for all local units type and mark all local units as externally managed."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Show how many local units would be updated without making changes.",
+        )
+
+    @transaction.atomic
+    def handle(self, *args, **options):
+        local_unit_qs = LocalUnit.objects.all()
+
+        if options["dry_run"]:
+            self.stdout.write(
+                self.style.WARNING(f"[Dry Run]: {local_unit_qs.count()} local units would be marked as externally managed.")
+            )
+            return
+
+        # Create Country level Externally Managed
+        local_unit_types = LocalUnitType.objects.all()
+        self.stdout.write(self.style.NOTICE("\n Creating/Updating Externally Managed local units"))
+        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():
+                instance, _ = ExternallyManagedLocalUnit.objects.get_or_create(
+                    country=country,
+                    local_unit_type=local_unit_type,
+                )
+                instance.enabled = True
+                instance.save(update_fields=["enabled"])
+                self.stdout.write(self.style.SUCCESS(f"\t--> Created externally managed for {local_unit_type.name}"))
+
+        # Update all local units to Externally managed
+        updated_count = local_unit_qs.update(
+            status=LocalUnit.Status.EXTERNALLY_MANAGED,
+        )
+
+        self.stdout.write(self.style.SUCCESS(f"Successfully marked {updated_count} local units as externally managed."))

go-static/files/local_units/Health-Care-Bulk-Import-Template-Local-Units.xlsm

@@ -442,6 +442,7 @@ class Status(models.IntegerChoices):
     email = models.EmailField(max_length=255, blank=True, null=True, verbose_name=_("Email"))
     link = models.URLField(max_length=255, blank=True, null=True, verbose_name=_("Social link"))
     location = models.PointField(srid=4326, help_text="Local Unit Location")
+    # FIXME: Migrate this to the status instead of boolen!
     is_deprecated = models.BooleanField(default=False, verbose_name=_("Is deprecated?"))
     deprecated_reason = models.IntegerField(
         choices=DeprecateReason.choices, verbose_name=_("deprecated reason"), blank=True, null=True