299978: Refactor role fields to use ID suffixes for household data co…

src/country_workspace/contrib/hope/push/config.py

@@ -11,7 +11,7 @@
 
 # Matches tags like: IND-25-0000.0051
 IND_TAG_RE = re.compile(r"^IND(?:-\d+)+\.\d+$")
-ROLE_FIELDS: Final[tuple[str, ...]] = ("head_of_household", "primary_collector", "alternate_collector")
+ROLE_FIELDS: Final[tuple[str, ...]] = ("head_of_household_id", "primary_collector_id", "alternate_collector_id")
 
 
 class SelectionConfig(TypedDict):

src/country_workspace/contrib/kobo/sync.py

@@ -212,13 +212,13 @@ def _is_head_of_household(individual: Individual) -> bool:
 
 def set_roles_and_relationships(household: Household, individuals: list[Individual]) -> None:
     if primary_collector := next(filter(_is_primary_collector, individuals), None):
-        household.flex_fields["primary_collector"] = getattr(primary_collector, "id", None)
+        household.flex_fields["primary_collector_id"] = getattr(primary_collector, "id", None)
 
     if alternate_collector := next(filter(_is_alternate_collector, individuals), None):
-        household.flex_fields["alternate_collector"] = getattr(alternate_collector, "id", None)
+        household.flex_fields["alternate_collector_id"] = getattr(alternate_collector, "id", None)
 
     if head_of_household := next(filter(_is_head_of_household, individuals), None):
-        household.flex_fields["head_of_household"] = getattr(head_of_household, "id", None)
+        household.flex_fields["head_of_household_id"] = getattr(head_of_household, "id", None)
 
     household.save(update_fields=["flex_fields"])
 

src/country_workspace/datasources/rdi/processors.py

@@ -246,10 +246,14 @@ def _sync_ind_pks(households_mapping: dict, individuals_mapping: dict) -> None:
 
     for v in households_mapping.values():
         hh_flex_fields = v.flex_fields.copy()
-        hh_flex_fields["head_of_household"] = pk_mapping.get(v.flex_fields.get("head_of_household"))
-        hh_flex_fields["primary_collector"] = pk_mapping.get(v.flex_fields.get("primary_collector"))
-        if alt_id := v.flex_fields.get("alternate_collector"):  # is optional
-            hh_flex_fields["alternate_collector"] = pk_mapping.get(alt_id)
+        hh_flex_fields["head_of_household_id"] = pk_mapping.get(
+            v.flex_fields.get("head_of_household_id", v.flex_fields.get("head_of_household"))
+        )
+        hh_flex_fields["primary_collector_id"] = pk_mapping.get(
+            v.flex_fields.get("primary_collector_id", v.flex_fields.get("primary_collector"))
+        )
+        if alt_id := v.flex_fields.get("alternate_collector_id", v.flex_fields.get("alternate_collector")):
+            hh_flex_fields["alternate_collector_id"] = pk_mapping.get(alt_id)
 
         v.flex_fields = hh_flex_fields
         v.save(update_fields=["flex_fields"])

src/country_workspace/models/household.py

@@ -71,12 +71,12 @@ def validate_with_checker(self, fail_if_alien: bool = False) -> bool:
 
     @property
     def head(self) -> "QuerySet[Individual]":
-        return self.flex_fields.get("head_of_household")
+        return self.flex_fields.get("head_of_household_id", self.flex_fields.get("head_of_household"))
 
     @property
     def primary_collector(self) -> "QuerySet[Individual]":
-        return self.flex_fields.get("primary_collector")
+        return self.flex_fields.get("primary_collector_id", self.flex_fields.get("primary_collector"))
 
     @property
     def alternate_collector(self) -> "QuerySet[Individual]":
-        return self.flex_fields.get("alternate_collector")
+        return self.flex_fields.get("alternate_collector_id", self.flex_fields.get("alternate_collector"))

src/country_workspace/utils/gen_rdi.py

@@ -90,7 +90,7 @@ def get_fields(self, row: dict) -> list[str]:
 HOUSEHOLDS_SPEC = SheetSpec(
     name=SheetName.HOUSEHOLDS,
     postfix="_h_c",
-    plain_fields=("household_id", "head_of_household", "primary_collector", "alternate_collector"),
+    plain_fields=("household_id", "head_of_household_id", "primary_collector_id", "alternate_collector_id"),
     id_key="household_id",
     parent_id_key=None,
     exclude_from_export=("individuals_start", "individuals_count"),
@@ -378,7 +378,7 @@ def generate_households_data(hh_form: FlexForm, config: GeneratorConfig, fake: F
             {
                 "household_id": hh_id,
                 "size": individuals_count,
-                "head_of_household": ind_counter + rng.randint(0, individuals_count - 1),
+                "head_of_household_id": ind_counter + rng.randint(0, individuals_count - 1),
                 "individuals_start": ind_counter,
                 "individuals_count": individuals_count,
             }
@@ -394,16 +394,16 @@ def update_collectors(households: list[dict], total_individuals: int, rng: Rando
         return
 
     for hh_data in households:
-        if "primary_collector" in hh_data:
-            hh_data["primary_collector"] = rng.randint(1, total_individuals)
+        if "primary_collector_id" in hh_data:
+            hh_data["primary_collector_id"] = rng.randint(1, total_individuals)
 
-        if "alternate_collector" in hh_data:
-            primary = hh_data.get("primary_collector")
+        if "alternate_collector_id" in hh_data:
+            primary = hh_data.get("primary_collector_id")
             if total_individuals >= 2 and rng.randint(0, 1) == 1:
                 alt = rng.randint(1, total_individuals - 1)
-                hh_data["alternate_collector"] = alt if (primary is None or alt < primary) else alt + 1
+                hh_data["alternate_collector_id"] = alt if (primary is None or alt < primary) else alt + 1
             else:
-                hh_data["alternate_collector"] = None
+                hh_data["alternate_collector_id"] = None
 
 
 def write_cell(ws: "Worksheet", row: int, col: int, value: Any, *, date_fmt: Any) -> None:

src/country_workspace/versioning/checkers.py

@@ -49,9 +49,12 @@ def create_hope_checkers() -> None:
         ("consent", defs["bool"], {}),
         ("country", defs["h_country"], {"label": "Country", "required": True}),
         ("country_origin", defs["h_country"], {}),
+        ("head_of_household_id", defs["char"], {"label": "Head of Household ID"}),
         ("household_id", defs["char"], {"label": "Household ID"}),
         ("name_enumerator", defs["char"], {"label": "Enumerator"}),
         ("org_enumerator", defs["char"], {}),
+        ("primary_collector_id", defs["char"], {"label": "Primary Collector ID"}),
+        ("alternate_collector_id", defs["char"], {"label": "Alternate Collector ID"}),
         ("registration_method", defs["char"], {}),
         ("residence_status", defs["h_residence"], {}),
         ("size", defs["int"], {}),
@@ -83,7 +86,6 @@ def create_hope_checkers() -> None:
 
     individual_fields_spec: list[FieldSpec] = [
         ("address", defs["char"], {}),
-        ("alternate_collector_id", defs["char"], {"label": "Alternative Collector for"}),
         ("birth_date", defs["date"], {"label": "Birth Date", "required": True}),
         ("disability", defs["i_disability"], {"label": "Disability"}),
         ("estimated_birth_date", defs["bool"], {"label": "Estimated Birth Date", "required": False}),
@@ -96,7 +98,6 @@ def create_hope_checkers() -> None:
         ("national_id_no", defs["char"], {}),
         ("national_id_photo", defs["char"], {}),
         ("phone_no", defs["char"], {}),
-        ("primary_collector_id", defs["char"], {"label": "Primary Collector for"}),
         ("relationship", defs["i_relationship"], {"label": "Relationship", "required": True}),
         ("role", defs["i_role"], {"label": "Role"}),
     ]

src/country_workspace/versioning/scripts/0032_role_fields_use_id_suffix.py

@@ -0,0 +1,81 @@
+# Generated by HCW 0.1.0 on 2026 02 18
+from packaging.version import Version
+from django.db import transaction
+from django.utils.text import slugify
+from hope_flex_fields.models import FieldDefinition, FlexField
+
+_script_for_version = Version("0.1.0")
+
+FLEX_FIELDS_FWD = {
+    "alternate_collector": "alternate_collector_id",
+    "primary_collector": "primary_collector_id",
+    "head_of_household": "head_of_household_id",
+}
+
+FIELD_DEFINITIONS_FWD = {
+    "Alternate Collector Reference ID": "alternate_collector_id",
+    "Primary Collector Reference ID": "primary_collector_id",
+    "Head of Household ID": "head_of_household_id",
+}
+
+ROLE_LABELS = {
+    "alternate_collector_id": "Alternate Collector ID",
+    "primary_collector_id": "Primary Collector ID",
+    "head_of_household_id": "Head of Household ID",
+}
+
+
+def _rename_flex_fields(name_map: dict[str, str]) -> None:
+    fields = FlexField.objects.filter(name__in=name_map).select_related("fieldset")
+    for field in fields:
+        new_name = name_map[field.name]
+        if field.name == new_name:
+            continue
+        if FlexField.objects.filter(fieldset=field.fieldset, name=new_name).exclude(pk=field.pk).exists():
+            continue
+        field.name = new_name
+        field.slug = slugify(new_name)
+        field.save(update_fields=["name", "slug"])
+
+
+def _rename_field_definitions(name_map: dict[str, str]) -> None:
+    for old_name, new_name in name_map.items():
+        fd = FieldDefinition.objects.filter(name=old_name).first()
+        if not fd:
+            continue
+        if FieldDefinition.objects.filter(name=new_name).exclude(pk=fd.pk).exists():
+            continue
+        fd.name = new_name
+        fd.slug = slugify(new_name)
+        fd.save(update_fields=["name", "slug"])
+
+
+def _apply_role_labels() -> None:
+    to_update = []
+    for field in FlexField.objects.filter(name__in=ROLE_LABELS):
+        attrs = dict(field.attrs or {})
+        if attrs.get("label") == ROLE_LABELS[field.name]:
+            continue
+        attrs["label"] = ROLE_LABELS[field.name]
+        field.attrs = attrs
+        to_update.append(field)
+    if to_update:
+        FlexField.objects.bulk_update(to_update, ["attrs"])
+
+
+@transaction.atomic
+def forward() -> None:
+    _rename_flex_fields(FLEX_FIELDS_FWD)
+    _rename_field_definitions(FIELD_DEFINITIONS_FWD)
+    _apply_role_labels()
+
+
+@transaction.atomic
+def backward() -> None:
+    _rename_flex_fields({v: k for k, v in FLEX_FIELDS_FWD.items()})
+    _rename_field_definitions({v: k for k, v in FIELD_DEFINITIONS_FWD.items()})
+
+
+class Scripts:
+    requires = []
+    operations = [(forward, backward)]

src/country_workspace/workspaces/admin/cleaners/bulk_update.py

@@ -412,8 +412,8 @@ def _validate_integer(value: str, field: str, line_number: int, errors: dict) ->
 
 
 def validate_individual_reference_ids(row_data: dict, line_number: int, errors: dict) -> None:
-    required_fields = ("head_of_household", "primary_collector")
-    optional_fields = ("alternate_collector",)
+    required_fields = ("head_of_household_id", "primary_collector_id")
+    optional_fields = ("alternate_collector_id",)
 
     sheet_fields = row_data.keys()
     for field in required_fields: