add ! import mapping rules, profiles

Author: vitali-yanushchyk-valor

pyproject.toml

@@ -46,6 +46,7 @@ dependencies = [
   "hope-flex-fields>=0.6.2",
   "hope-smart-export>=0.3",
   "hope-smart-import>=0.3",
+  "jmespath>=1.0.1",
   "openpyxl>=3.1.5",
   "pillow>=11.2.1",
   "psycopg2-binary>=2.9.9",

src/country_workspace/admin/__init__.py

@@ -12,6 +12,7 @@
 from .individual import IndividualAdmin
 from .job import AsyncJobAdmin
 from .locations import AreaAdmin, AreaTypeAdmin, CountryAdmin
+from .mapping import FieldMappingRuleAdmin, MappingProfileAdmin
 from .office import OfficeAdmin
 from .program import ProgramAdmin
 from .rdp import RdpAdmin
@@ -36,8 +37,10 @@
     "BeneficiaryGroupAdmin",
     "ConstanceAdmin",
     "CountryAdmin",
+    "FieldMappingRuleAdmin",
     "HouseholdAdmin",
     "IndividualAdmin",
+    "MappingProfileAdmin",
     "OfficeAdmin",
     "ProgramAdmin",
     "RdpAdmin",

src/country_workspace/admin/mapping.py

@@ -0,0 +1,70 @@
+from jsoneditor.forms import JSONEditor
+from django.contrib import admin
+from django.db import models
+from django.http import HttpRequest
+
+from ..models import MappingProfile, FieldMappingRule
+from .base import BaseModelAdmin
+
+
+@admin.register(MappingProfile)
+class MappingProfileAdmin(BaseModelAdmin):
+    list_display = ("name", "source_type", "import_schema", "is_active", "inheritance")
+    list_filter = ("source_type", "import_schema", "program", "is_active")
+    search_fields = ("name", "description")
+    filter_horizontal = ("program",)
+    fields = (
+        "name",
+        "description",
+        "parent",
+        "source_type",
+        "import_schema",
+        "program",
+        "is_active",
+        "created_by",
+        "created_at",
+        "updated_at",
+    )
+    readonly_fields = ("created_by", "created_at", "updated_at")
+
+    @admin.display(description="Inheritance Chain")
+    def inheritance(self, obj: MappingProfile) -> str:
+        return obj.get_inheritance_chain()
+
+    def get_queryset(self, request: HttpRequest) -> models.QuerySet[MappingProfile]:
+        return super().get_queryset(request).select_related("parent", "created_by").prefetch_related("program")
+
+
+@admin.register(FieldMappingRule)
+class FieldMappingRuleAdmin(BaseModelAdmin):
+    formfield_overrides = {
+        models.JSONField: {
+            "widget": JSONEditor(
+                init_options={"mode": "code", "modes": ["text", "code", "tree"]},
+                ace_options={"readOnly": False},
+            ),
+        }
+    }
+    list_display = ("name", "profile", "source_fields", "target_fields", "order", "is_active")
+    list_filter = ("profile", "is_active")
+    search_fields = ("name", "description", "profile__name")
+    ordering = ("profile", "order", "name")
+    fields = (
+        "name",
+        "description",
+        "profile",
+        "source_fields",
+        "target_fields",
+        "expression",
+        "default_values",
+        "order",
+        "is_active",
+        "created_by",
+        "created_at",
+        "updated_at",
+    )
+
+    readonly_fields = ("created_by", "created_at", "updated_at")
+
+    def get_queryset(self, request: HttpRequest) -> models.QuerySet[FieldMappingRule]:
+        return super().get_queryset(request).select_related("profile__parent", "created_by")

src/country_workspace/admin/program.py

@@ -64,6 +64,12 @@ def population(self, btn: "LinkButton") -> None:
         obj = btn.context["original"]
         btn.href = f"{base}?program__exact={obj.pk}&country_office__exact={obj.country_office.pk}"
 
+    @link(change_list=False)
+    def mapping_profiles(self, btn: "LinkButton") -> None:
+        obj = btn.context["original"]
+        base = reverse("admin:country_workspace_mappingprofile_changelist")
+        btn.href = f"{base}?program__id__exact={obj.pk}"
+
     @button()
     def zap(self, request: HttpRequest, pk: str) -> None:
         obj: Program = self.get_object(request, pk)

src/country_workspace/datasources/rdi.py

@@ -2,7 +2,8 @@
 from base64 import b64encode
 from collections import defaultdict
 from collections.abc import Iterable, Generator
-from typing import Any, Mapping, cast
+from contextlib import suppress
+from typing import Any, Mapping, cast, NotRequired
 
 import openpyxl
 from PIL import Image
@@ -11,7 +12,7 @@
 from openpyxl.drawing.image import Image as RDIImage
 
 from country_workspace.contrib.kobo.api.data.helpers import VALUE_FORMAT
-from country_workspace.models import AsyncJob, Batch, Household
+from country_workspace.models import AsyncJob, Batch, Household, MappingProfile
 from country_workspace.utils.config import BatchNameConfig, FailIfAlienConfig
 from country_workspace.utils.fields import Record, clean_field_names
 from country_workspace.validators.beneficiaries import validate_beneficiaries
@@ -31,6 +32,7 @@ class Config(BatchNameConfig, FailIfAlienConfig):
     household_pk_col: str
     master_column_label: str
     detail_column_label: str
+    mapping_profile_id: NotRequired[int]
 
 
 class ColumnConfigurationError(Exception):
@@ -74,7 +76,23 @@ def postprocess_cell(sheets: MultiSheet) -> MultiSheet:
         yield sheet_idx, formated_rows
 
 
-def process_households(sheet: Sheet, job: AsyncJob, batch: Batch, config: Config) -> Mapping[int, Household]:
+def apply_mapping_rules(row: Record, mapping_profile: MappingProfile | None) -> dict[str, Any]:
+    result = clean_field_names(row)
+    if not mapping_profile:
+        return result
+
+    rules = mapping_profile.get_all_rules()
+    for rule in sorted(rules, key=lambda r: r.order):
+        if rule.is_active:
+            mapped_data = rule.apply(result)
+            result.update(mapped_data)
+
+    return result
+
+
+def process_households(
+    sheet: Sheet, job: AsyncJob, batch: Batch, config: Config, mapping_profile: MappingProfile | None = None
+) -> Mapping[int, Household]:
     mapping = {}
 
     for i, row in enumerate(sheet, 1):
@@ -87,7 +105,7 @@ def process_households(sheet: Sheet, job: AsyncJob, batch: Batch, config: Config
                 job.program.households.create(
                     batch=batch,
                     name=label,
-                    flex_fields=clean_field_names(row),
+                    flex_fields=apply_mapping_rules(row, mapping_profile),
                 ),
             )
         except Exception as e:
@@ -104,22 +122,25 @@ def full_name_column(row: Record) -> str | None:
 
 
 def process_individuals(
-    sheet: Sheet, household_mapping: Mapping[int, Household], job: AsyncJob, batch: Batch, config: Config
+    sheet: Sheet,
+    household_mapping: Mapping[int, Household],
+    job: AsyncJob,
+    mapping_profile: MappingProfile | None = None,
 ) -> int:
     processed = 0
 
     for i, row in enumerate(sheet, 1):
         name_column = full_name_column(row)
         name = get_value(row, name_column) if name_column else None
-        household_key = get_value(row, config["master_column_label"])
+        household_key = get_value(row, job.config["master_column_label"])
         household = household_mapping.get(household_key)
 
         try:
             job.program.individuals.create(
-                batch=batch,
+                batch=job.batch,
                 name=name,
                 household_id=household.pk,
-                flex_fields=clean_field_names(row),
+                flex_fields=apply_mapping_rules(row, mapping_profile),
             )
         except Exception as e:
             raise SheetProcessingError(INDIVIDUAL, i) from e
@@ -173,6 +194,13 @@ def import_from_rdi(job: AsyncJob) -> dict[str, int]:
     with atomic():
         config: Config = job.config
         rdi = job.file
+
+        config["mapping_profile_id"] = 1
+        mapping_profile = None
+        if config.get("mapping_profile_id"):
+            with suppress(MappingProfile.DoesNotExist):
+                mapping_profile = MappingProfile.objects.get(id=config["mapping_profile_id"], is_active=True)
+
         batch = Batch.objects.create(
             name=config["batch_name"],
             program=job.program,
@@ -183,8 +211,10 @@ def import_from_rdi(job: AsyncJob) -> dict[str, int]:
 
         household_sheet, individual_sheet = read_sheets(config, rdi, 0, 1)
 
-        household_mapping = process_households(household_sheet, job, batch, config)
-        individuals_number = process_individuals(individual_sheet, household_mapping, job, batch, config)
+        household_mapping = process_households(household_sheet, job, batch, config, mapping_profile)
+        individuals_number = process_individuals(
+            individual_sheet, household_mapping, job, batch, config, mapping_profile
+        )
 
         validate_beneficiaries(config, household_mapping)