chg ! bulk records update (#43)

Author: vitali-yanushchyk-valor

docs/src/.pages

@@ -6,4 +6,5 @@ nav:
   - dev_helpers.md
   - flows
   - Interfaces: interfaces.md
+  - Data cleaning and updates: data_cleaning_updates.md
   - Import Data: import_data

docs/src/data_cleaning_updates.md

@@ -0,0 +1,87 @@
+# Data cleaning and updates.
+
+There are several options to clean data for households or individuals in the collector interface.
+
+## Bulk Actions
+
+The system allows users to perform mass operations on multiple records at once. These actions help streamline data management by applying updates to multiple `Household` or `Individual` entities in a single process.
+
+### In-System Bulk Data Modifications
+
+To perform bulk updates, go to:
+```
+Home › Households | Individuals
+```
+Select the records and apply the desired action:
+
+- **Mass update record fields**
+
+    Allows the user to define values and apply them using functions like:
+
+    ```
+    set, set null, upper, lower, toggle, ...
+    ```
+
+- **Update fields using RegEx**
+
+    Allows the user to define a regular expression and a substitution pattern for any field and apply it.
+
+---
+
+### Bulk Data Export, Offline Editing, and Reimport
+
+Users can modify data in bulk by exporting records to a `.xlsx` file, making necessary changes outside the system, and then importing the updated file back.
+
+The `.xlsx` file contains two protected columns: `id` and `version`. **These columns should not be modified during regular editing**, as they are essential for correctly identifying and updating records.
+
+Additionally, the system includes a **concurrency control mechanism** to prevent accidental overwrites of updated data.
+
+#### Process Overview
+
+1. **Export Data**
+
+    To begin, navigate to:
+    ```
+    Home › Households | Individuals
+    ```
+    Select the records and apply the action **"Export records as .xlsx for bulk updates"**.
+    Then, choose the columns to update and press **[Export]**.
+    The export process will be scheduled as an asynchronous task.
+
+2. **Edit the Exported File**
+
+    Once the `.xlsx` file is generated, update the necessary values while ensuring that the `id` and `version` columns remain unchanged.
+
+3. **Import the Updated File Back into the System**
+
+    After making the necessary modifications, navigate to:
+    ```
+    Home › Program
+    ```
+    Press **[Update Records]**, select the file, choose the target (`Household | Individual`), and optionally provide a description.
+    Finally, press **[Import]**. This action will schedule an asynchronous task to process the updates.
+
+#### **Preventing Conflicts with `CONCURRENCY_GUARD`**
+
+To ensure data consistency, the system provides the **`CONCURRENCY_GUARD`** parameter.
+When enabled, it prevents updates if a record has changed after the export. This ensures that newer modifications made by other users or processes are not accidentally overwritten during import.
+If a conflict is detected, the system will reject the update, requiring the user to re-export the latest data before making further changes.
+
+---
+
+## **Single-Entity Operations**
+
+Unlike bulk actions, these operations apply to a single `Household` or `Individual` entry. To access these actions, go to:
+```
+Home › Households | Individuals
+```
+Select a specific `Household` or `Individual`, then use the available buttons:
+
+### Validate
+Ensures that the record meets the required data standards.
+If any issues are detected, the affected fields turn red, and an explanatory error message appears.
+
+### View Raw Data
+
+Allows the user to see the raw, unprocessed data of the selected record.
+This can be useful for debugging or reviewing the exact stored values before applying further modifications.

docs/src/index.md

@@ -14,6 +14,6 @@ management process.
 
 - [Import data](import_data/index.md) from [Kobo](import_data/kobo.md) / [Aurora](import_data/aurora.md) / [XLS](import_data/xls.md) (RDI format)
 - Data validation
-- Data cleaning and updates
+- [Data cleaning and updates](data_cleaning_updates.md)
 - Push data to HOPE
 - Export/Amend/Import process

src/country_workspace/config/fragments/constance.py

@@ -50,6 +50,13 @@
     "KOBO_API_URL": ("", "Kobo API Server address", str),
     "CACHE_TIMEOUT": (86400, "Cache Redis TTL", int),
     "CACHE_BY_VERSION": (False, "Invalidate Cache on CW version change", bool),
+    "CONCURRENCY_GUARD": (
+        True,
+        "Prevent updates if data has changed after export. When enabled, the system will reject updates to records"
+        " that were modified after they were exported. This helps maintain data consistency and prevents accidental"
+        " overwrites of newer information.",
+        bool,
+    ),
 }
 
 CONSTANCE_CONFIG_FIELDSETS = {
@@ -63,6 +70,7 @@
         "KOBO_API_TOKEN",
         "KOBO_API_URL",
     ),
+    "Data consistency": ("CONCURRENCY_GUARD",),
 }
 
 # Mapping of config keys to masked default display values in the Constance admin UI.

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

@@ -1,12 +1,13 @@
 from typing import TYPE_CHECKING
 
 from django.contrib import admin, messages
-from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
-from django.shortcuts import render
+from django.http import HttpRequest, HttpResponse
+from django.shortcuts import render, redirect
 from django.utils.translation import gettext as _
 from strategy_field.utils import fqn
 
-from .bulk_update import BulkUpdateForm, bulk_update_export_template
+from country_workspace.workspaces.admin.forms import BulkUpdateExportForm
+from .bulk_update import bulk_update_export_template
 from .calculate_checksum import calculate_checksum_impl
 from .mass_update import MassUpdateForm, mass_update_impl
 from .regex import RegexUpdateForm, regex_update_impl
@@ -29,7 +30,7 @@ def validate_records(
 ) -> None:
     opts = queryset.model._meta
     job = AsyncJob.objects.create(
-        description="Validate Queryset records for updates",
+        description=validate_records.short_description,
         type=AsyncJob.JobType.ACTION,
         owner=state.request.user,
         action=fqn(validate_queryset),
@@ -47,7 +48,7 @@ def mass_update(
     request: HttpRequest,
     queryset: "QuerySet[Beneficiary]",
 ) -> "HttpResponse":
-    ctx = model_admin.get_common_context(request, title=_("Mass update"))
+    ctx = model_admin.get_common_context(request, title=_(mass_update.short_description))
     ctx["checker"] = checker = model_admin.get_checker(request)
     ctx["preserved_filters"] = model_admin.get_preserved_filters(request)
     form = MassUpdateForm(request.POST, checker=checker)
@@ -57,7 +58,7 @@ def mass_update(
         opts = queryset.model._meta
 
         job = AsyncJob.objects.create(
-            description="Mass update record fields",
+            description=mass_update.short_description,
             type=AsyncJob.JobType.ACTION,
             owner=state.request.user,
             action=fqn(mass_update_impl),
@@ -82,7 +83,7 @@ def regex_update(
     request: "HttpRequest",
     queryset: "QuerySet[Beneficiary]",
 ) -> HttpResponse:
-    ctx = model_admin.get_common_context(request, title=_("Regex update"))
+    ctx = model_admin.get_common_context(request, title=_(regex_update.short_description))
     ctx["checker"] = checker = model_admin.get_checker(request)
     ctx["queryset"] = queryset
     ctx["opts"] = model_admin.model._meta
@@ -96,9 +97,8 @@ def regex_update(
         form = RegexUpdateForm(request.POST, checker=checker)
         if form.is_valid():
             opts = queryset.model._meta
-
             job = AsyncJob.objects.create(
-                description="Mass update record fields",
+                description=regex_update.short_description,
                 type=AsyncJob.JobType.ACTION,
                 owner=state.request.user,
                 action=fqn(regex_update_impl),
@@ -125,22 +125,22 @@ def regex_update(
     return render(request, "workspace/actions/regex.html", ctx)
 
 
-@admin.action(description="Create XLS template for bulk updates", permissions=["export"])
+@admin.action(description="Export records as .xlsx for bulk updates", permissions=["export"])
 def bulk_update_export(
     model_admin: "BeneficiaryBaseAdmin",
     request: HttpRequest,
     queryset: "QuerySet[Beneficiary]",
 ) -> HttpResponse:
-    ctx = model_admin.get_common_context(request, title=_("Export data for bulk update"))
+    ctx = model_admin.get_common_context(request, title=_(bulk_update_export.short_description))
     ctx["checker"] = checker = model_admin.get_checker(request)
     ctx["preserved_filters"] = model_admin.get_preserved_filters(request)
-    form = BulkUpdateForm(request.POST, checker=checker)
+    form = BulkUpdateExportForm(request.POST, checker=checker)
     ctx["form"] = form
     if "_export" in request.POST and form.is_valid():
-        columns = {"fields": ["id"] + sorted(form.cleaned_data["fields"])}
+        columns = ["id", "version"] + sorted(form.cleaned_data["fields"])
         opts = queryset.model._meta
         job = AsyncJob.objects.create(
-            description="Mass update record fields",
+            description=bulk_update_export.short_description,
             type=AsyncJob.JobType.TASK,
             owner=state.request.user,
             action=fqn(bulk_update_export_template),
@@ -153,7 +153,7 @@ def bulk_update_export(
         )
         job.queue()
         model_admin.message_user(request, "Task scheduled", messages.SUCCESS)
-        return HttpResponseRedirect(".")
+        return redirect(".")
 
     return render(request, "workspace/actions/bulk_update_export.html", ctx)
 
@@ -166,7 +166,7 @@ def calculate_checksum(
 ) -> HttpResponse:
     opts = queryset.model._meta
     job = AsyncJob.objects.create(
-        description="Calculate record checksum",
+        description=calculate_checksum.short_description,
         type=AsyncJob.JobType.ACTION,
         owner=state.request.user,
         action=fqn(calculate_checksum_impl),
@@ -178,4 +178,4 @@ def calculate_checksum(
     )
     job.queue()
     model_admin.message_user(request, "Task scheduled", messages.SUCCESS)
-    return HttpResponseRedirect(".")
+    return redirect(".")

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

@@ -1,35 +1,27 @@
 import io
 from io import BytesIO
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Callable
 
 from django import forms
 from django.apps import apps
 from django.core.exceptions import ObjectDoesNotExist
 from django.core.files.storage import default_storage
 from xlsxwriter import Workbook
 
+from constance import config as constance_config
+
 from hope_flex_fields.models import DataChecker, FlexField
 from hope_flex_fields.xlsx import get_format_for_field
 from hope_smart_import.readers import open_xls
 
 from country_workspace.models import AsyncJob, Program
-from country_workspace.workspaces.admin.cleaners.base import BaseActionForm
 
 if TYPE_CHECKING:
     from django.db.models import QuerySet
 
     from country_workspace.types import Beneficiary
 
 
-class BulkUpdateForm(BaseActionForm):
-    fields = forms.MultipleChoiceField(choices=[], widget=forms.CheckboxSelectMultiple())
-
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
-        checker: "DataChecker" = kwargs.pop("checker")
-        super().__init__(*args, **kwargs)
-        self.fields["fields"].choices = [(name, name) for name, fld in checker.get_form()().fields.items()]
-
-
 """
 # class Criteria:
 #     pass
@@ -130,12 +122,11 @@ def create_xls_importer(
     queryset: "QuerySet[Beneficiary]",
     program: Program,
     columns: list[str],
-) -> [io.BytesIO, Workbook]:
+) -> tuple[io.BytesIO, Workbook]:
     out = BytesIO()
     dc: DataChecker = program.get_checker_for(queryset.model)
 
     workbook = Workbook(out, {"in_memory": True, "default_date_format": "yyyy/mm/dd"})
-
     header_format = workbook.add_format(
         {
             "bold": False,
@@ -147,14 +138,13 @@ def create_xls_importer(
             "indent": 1,
         },
     )
-
     header_format.set_bg_color("#DDDDDD")
     header_format.set_locked(True)
     header_format.set_align("center")
     header_format.set_bottom_color("black")
     worksheet = workbook.add_worksheet()
     worksheet.protect()
-    worksheet.unprotect_range("B1:ZZ999", None)
+    worksheet.unprotect_range("C1:ZZ999", None)
 
     for i, fld_name in enumerate(columns):
         fld = dc_get_field(dc, fld_name)
@@ -179,39 +169,62 @@ def create_xls_importer(
     return out, workbook
 
 
-# def bulk_update_export_template(queryset, program_pk: str, columns: list[str], filename: str) -> bytes:
 def bulk_update_export_template(job: AsyncJob) -> bytes:
     model = apps.get_model(job.config["model_name"])
     queryset = model.objects.filter(pk__in=job.config["pks"])
     filename = "bulk_update_export_template/%s/%s/%s.xlsx" % (job.program.pk, job.owner.pk, job.config["model_name"])
-    out, __ = create_xls_importer(queryset.all(), job.program, job.config["columns"])
+    out, __ = create_xls_importer(queryset, job.program, job.config["columns"])
     path = default_storage.save(filename, out)
     job.file = path
     job.save()
     return path
 
 
-def bulk_update_individual(job: AsyncJob) -> dict[str, Any]:
-    ret = {"not_found": []}
-    for e in open_xls(io.BytesIO(job.file.read()), start_at=0):
+def bulk_update_collection(job: AsyncJob, collection_getter: Callable[[int], Any]) -> dict[str, Any]:
+    result: dict[str, Any] = {"not_found": []}
+    version_check = constance_config.CONCURRENCY_GUARD
+    if version_check:
+        result["version_mismatch"] = []
+    errors = {}
+
+    file_data = job.file.read()
+    rows = open_xls(io.BytesIO(file_data), start_at=0)
+    for line_number, row in enumerate(rows, start=1):
         try:
-            _id = e.pop("id")
-            ind = job.program.individuals.get(id=_id)
-            ind.flex_fields.update(**e)
-            ind.save()
+            _id = int(row.pop("id"))
+            entity = collection_getter(_id)
+        except (KeyError, ValueError):
+            errors.setdefault("Invalid or missing 'id' on line", []).append(line_number)
+            continue
         except ObjectDoesNotExist:
-            ret["not_found"].append(_id)
-    return ret
+            result["not_found"].append(_id)
+            continue
+
+        if version_check:
+            try:
+                _version = int(row.pop("version"))
+            except (KeyError, ValueError):
+                errors.setdefault("Invalid or missing 'version' on line", []).append(line_number)
+                continue
+
+            if entity.version != _version:
+                result["version_mismatch"].append(_id)
+                continue
+
+        entity.flex_fields.update(**row)
+        entity.save()
+
+    if errors:
+        result["errors"] = errors
+
+    return result
+
+
+def bulk_update_individual(job: AsyncJob) -> dict[str, Any]:
+    program = job.program
+    return bulk_update_collection(job, lambda _id: program.individuals.get(id=_id))
 
 
 def bulk_update_household(job: AsyncJob) -> dict[str, Any]:
-    ret = {"not_found": []}
-    for e in open_xls(io.BytesIO(job.file.read()), start_at=0):
-        try:
-            _id = e.pop("id")
-            ind = job.program.households.get(id=_id)
-            ind.flex_fields.update(**e)
-            ind.save()
-        except ObjectDoesNotExist:
-            ret["not_found"].append(_id)
-    return ret
+    program = job.program
+    return bulk_update_collection(job, lambda _id: program.households.get(id=_id))