Add functionality to run formulas on existing records

src/country_workspace/workspaces/admin/__init__.py

@@ -5,6 +5,7 @@
 from .mapping_importer import CountryMappingImporterAdmin
 from .program import CountryProgramAdmin
 from .rdp import CountryRdpAdmin
+from .transformer import CountryTransformerAdmin
 
 __all__ = [
     "CountryBatchAdmin",
@@ -14,4 +15,5 @@
     "CountryMappingImporterAdmin",
     "CountryProgramAdmin",
     "CountryRdpAdmin",
+    "CountryTransformerAdmin",
 ]

src/country_workspace/workspaces/admin/batch/reprocessing.py

@@ -53,6 +53,26 @@ def _preserve_flex_fields(
     ), preserved
 
 
+def _get_batch_from_job(job: AsyncJob) -> tuple[int, Batch]:
+    batch_id = job.config.get("batch_id")
+    if not batch_id:
+        raise ValueError("batch_id is required in job config")
+
+    batch = Batch.objects.select_related("program", "country_office").filter(pk=batch_id).first()
+    if not batch:
+        logger.error("Batch %s not found", batch_id)
+        raise Batch.DoesNotExist(f"Batch {batch_id} not found")
+    return batch_id, batch
+
+
+def _sync_household_refs(batch: Batch) -> None:
+    match batch.source:
+        case Batch.BatchSource.KOBO:
+            _sync_kobo_household_refs(batch)
+        case Batch.BatchSource.RDI:
+            _sync_rdi_household_refs(batch)
+
+
 def _apply_import_processor(
     record: Household | Individual,
     processor: Callable[[Any], dict[str, Any]],
@@ -180,14 +200,6 @@ def _sync_kobo_household_refs(batch: Batch) -> None:
         )
 
 
-def _sync_household_refs(batch: Batch) -> None:
-    match batch.source:
-        case Batch.BatchSource.KOBO:
-            _sync_kobo_household_refs(batch)
-        case Batch.BatchSource.RDI:
-            _sync_rdi_household_refs(batch)
-
-
 def _resolve_config_object[T](
     queryset: QuerySet[T],
     object_id: int | None,
@@ -285,11 +297,7 @@ def _active_records(
 
 
 def reprocess_batch(job: AsyncJob) -> dict[str, Any]:
-    if not (batch_id := job.config.get("batch_id")):
-        raise ValueError("batch_id is required in job config")
-    if not (batch := Batch.objects.select_related("program", "country_office").filter(pk=batch_id).first()):
-        logger.error("Batch %s not found", batch_id)
-        raise Batch.DoesNotExist(f"Batch {batch_id} not found")
+    batch_id, batch = _get_batch_from_job(job)
 
     config = _resolve_reprocess_config(batch, job.config)
     is_master_detail = batch.program.is_master_detail
@@ -378,3 +386,38 @@ def reprocess_batch(job: AsyncJob) -> dict[str, Any]:
 
     logger.info("Batch reprocessing initiated: %s", response)
     return response
+
+
+def apply_batch_transformers(job: AsyncJob) -> dict[str, Any]:
+    batch_id, batch = _get_batch_from_job(job)
+
+    household_transformer_id, _ = _resolve_config_object(
+        batch.country_office.transformers.all(),
+        job.config.get("household_transformer_id"),
+        "Household transformer",
+    )
+    individual_transformer_id, _ = _resolve_config_object(
+        batch.country_office.transformers.all(),
+        job.config.get("individual_transformer_id"),
+        "Individual transformer",
+    )
+    if not household_transformer_id and not individual_transformer_id:
+        raise ValueError("At least one transformer id is required in job config")
+
+    postprocessing_result = run_batch_postprocessing(
+        batch,
+        household_transformer_id=household_transformer_id,
+        individual_transformer_id=individual_transformer_id,
+        sync_household_refs=_sync_household_refs,
+    )
+
+    response = {
+        "batch_id": batch_id,
+        "batch_name": batch.name,
+        "transformed_individuals": postprocessing_result.get("transformed_individuals", 0),
+    }
+    if batch.program.is_master_detail:
+        response["transformed_households"] = postprocessing_result.get("transformed_households", 0)
+
+    logger.info("Batch transformer apply finished: %s", response)
+    return response

src/country_workspace/workspaces/admin/transformer.py

@@ -1,15 +1,71 @@
-from django.contrib import admin
+from admin_extra_buttons.decorators import button
+from django import forms
+from django.contrib import admin, messages
 from django.core.cache import cache
+from django.db import models
 from django.db.models import QuerySet
 from django.forms import ModelForm
-from django.http import HttpRequest
+from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
+from django.shortcuts import render
+from django.urls import reverse
+from django.utils.translation import gettext_lazy as _
+from strategy_field.utils import fqn
 
+from country_workspace.models import AsyncJob, Batch
 from country_workspace.state import state
 from country_workspace.workspaces.models import CountryTransformer
 from country_workspace.workspaces.options import WorkspaceModelAdmin
 from country_workspace.workspaces.sites import workspace
 
 
+class RunTransformerForm(forms.Form):
+    class ApplyToOptions(models.TextChoices):
+        HOUSEHOLDS = "households", _("Households only")
+        INDIVIDUALS = "individuals", _("Individuals only")
+        BOTH = "both", _("Households and Individuals")
+
+    batch = forms.ModelChoiceField(
+        queryset=Batch.objects.none(),
+        label=_("Batch"),
+        help_text=_("Select an existing batch to update records before pushing to HOPE."),
+    )
+    apply_to = forms.ChoiceField(
+        label=_("Apply formula to"),
+        choices=ApplyToOptions.choices,
+        help_text=_("Choose which record type should be updated by this formula."),
+    )
+
+    def __init__(self, *args: object, **kwargs: object) -> None:
+        office = kwargs.pop("office", None)
+        program = kwargs.pop("program", None)
+        super().__init__(*args, **kwargs)
+
+        qs = Batch.objects.order_by("-import_date")
+        if office:
+            qs = qs.filter(country_office=office)
+        if program:
+            qs = qs.filter(program=program)
+        self.fields["batch"].queryset = qs.select_related("program")
+
+        if not program:
+            self.fields["apply_to"].choices = [
+                (self.ApplyToOptions.INDIVIDUALS, self.ApplyToOptions.INDIVIDUALS.label),
+                (self.ApplyToOptions.BOTH, self.ApplyToOptions.BOTH.label),
+            ]
+            return
+
+        if program.is_master_detail:
+            self.fields["apply_to"].choices = [
+                (self.ApplyToOptions.HOUSEHOLDS, self.ApplyToOptions.HOUSEHOLDS.label),
+                (self.ApplyToOptions.INDIVIDUALS, self.ApplyToOptions.INDIVIDUALS.label),
+                (self.ApplyToOptions.BOTH, self.ApplyToOptions.BOTH.label),
+            ]
+        else:
+            self.fields["apply_to"].choices = [
+                (self.ApplyToOptions.INDIVIDUALS, self.ApplyToOptions.INDIVIDUALS.label),
+            ]
+
+
 @admin.register(CountryTransformer, site=workspace)
 class CountryTransformerAdmin(WorkspaceModelAdmin):
     list_display = ("name", "description", "created_by", "created_at")
@@ -63,6 +119,73 @@ def delete_queryset(self, request: HttpRequest, queryset: QuerySet[CountryTransf
         super().delete_queryset(request, queryset)
         self._invalidate_transformer_cache()
 
+    @button(
+        label="Run Formula on Existing Records",
+        change_form=True,
+        permission="country_workspace.reprocess_batch",
+        html_attrs={"title": "Run this formula in Country Workspace without rule commits"},
+    )
+    def run_on_existing_records(self, request: HttpRequest, pk: str) -> HttpResponse:
+        obj = self.get_object(request, pk)
+        if not obj:
+            return HttpResponse("Transformer not found", status=404)
+
+        if request.method == "POST" and "apply" in request.POST:
+            form = RunTransformerForm(request.POST, office=state.tenant, program=state.program)
+            if form.is_valid():
+                batch = form.cleaned_data["batch"]
+                apply_to = form.cleaned_data["apply_to"]
+                if not request.user.has_perm("country_workspace.reprocess_batch", batch.program):  # type: ignore[attr-defined]
+                    self.message_user(
+                        request,
+                        _("You do not have permission to run formulas on this batch."),
+                        messages.ERROR,
+                    )
+                    return HttpResponseRedirect(self.get_change_url(request, obj))
+
+                config: dict[str, int] = {"batch_id": batch.pk}
+                if batch.program.is_master_detail and apply_to in (
+                    RunTransformerForm.ApplyToOptions.HOUSEHOLDS,
+                    RunTransformerForm.ApplyToOptions.BOTH,
+                ):
+                    config["household_transformer_id"] = obj.pk
+                if apply_to in (
+                    RunTransformerForm.ApplyToOptions.INDIVIDUALS,
+                    RunTransformerForm.ApplyToOptions.BOTH,
+                ):
+                    config["individual_transformer_id"] = obj.pk
+
+                job = AsyncJob.objects.create(
+                    description=f"Run formula '{obj.name}' on batch {batch.name}",
+                    type=AsyncJob.JobType.TASK,
+                    owner=request.user,
+                    action=fqn("country_workspace.workspaces.admin.batch.reprocessing.apply_batch_transformers"),
+                    program=batch.program,
+                    batch=batch,
+                    config=config,
+                )
+                job.queue()
+
+                self.message_user(
+                    request,
+                    _("Formula execution has been scheduled for the selected batch."),
+                    messages.SUCCESS,
+                )
+                return HttpResponseRedirect(reverse("workspace:workspaces_countrybatch_changelist"))
+
+            self.message_user(request, _("Please correct the errors below."), messages.ERROR)
+        else:
+            form = RunTransformerForm(office=state.tenant, program=state.program)
+
+        context = self.get_common_context(
+            request,
+            pk=pk,
+            title=_("Run Formula on Existing Records"),
+            form=form,
+            transformer=obj,
+        )
+        return render(request, "workspace/admin_extra_buttons/run_transformer_form.html", context)
+
     def _invalidate_transformer_cache(self) -> None:
         """Invalidate cache keys related to transformers."""
         if state.tenant:

src/country_workspace/workspaces/sites.py

@@ -196,6 +196,7 @@ def _current_modeladmin(self, request: "HttpRequest") -> str | None:
             "workspaces_countryhousehold": "CountryHouseholdAdmin",
             "workspaces_countryindividual": "CountryIndividualAdmin",
             "workspaces_countrybatch": "CountryBatchAdmin",
+            "workspaces_countrytransformer": "CountryTransformerAdmin",
             "workspaces_countrymappingimporter": "CountryMappingImporterAdmin",
             "workspaces_countryrdp": "CountryRdpAdmin",
             "workspaces_countryasyncjob": "CountryJobAdmin",
@@ -293,6 +294,12 @@ def get_menu_items(self, request: "HttpRequest") -> list[dict[str, Any]]:
                         "icon": "icon-loop",
                         "selected": current_admin == "CountryMappingImporterAdmin",
                     },
+                    {
+                        "name": _("Transformers"),
+                        "url": reverse("workspace:workspaces_countrytransformer_changelist"),
+                        "icon": "icon-equalizer",
+                        "selected": current_admin == "CountryTransformerAdmin",
+                    },
                     {
                         "name": apps.get_model("country_workspace", "Rdp")._meta.verbose_name_plural,
                         "url": reverse("workspace:workspaces_countryrdp_changelist"),

src/country_workspace/workspaces/templates/workspace/admin_extra_buttons/run_transformer_form.html

@@ -0,0 +1,69 @@
+{% extends 'workspace/_base.html' %}
+{% load i18n %}
+{% block breadcrumbs %}
+{% endblock breadcrumbs %}
+{% block content %}
+    <div id="content">
+        <div id="content-main">
+            <div class="block">
+                <h1 class="text-2xl">
+                    {% translate 'Run Formula on Existing Records' %}
+                </h1>
+                <div class="block p-5">
+                    <h2>
+                        {% blocktranslate with transformer_name=transformer.name %}
+                            Formula: {{ transformer_name }}
+                        {% endblocktranslate %}
+                    </h2>
+                    <p class="mt-2">
+                        {% translate "This executes the selected formula directly in Country Workspace and updates existing records before they are pushed to HOPE." %}
+                    </p>
+                    <p class="mt-2">
+                        {% translate "No admin rule creation, rule commits, or custom code execution is required." %}
+                    </p>
+                    <form method="post" class="mt-5">
+                        {% csrf_token %}
+                        {% if form.errors %}
+                            <div class="error">
+                                <p class="errornote">
+                                    {% translate "Please correct the errors below." %}
+                                </p>
+                            </div>
+                        {% endif %}
+                        <fieldset class="module aligned">
+                            {% for field in form %}
+                                <div class="form-row {% if field.errors %}errors{% endif %}">
+                                    <div>
+                                        <label for="{{ field.id_for_label }}" class="required">
+                                            {{ field.label }}:
+                                        </label>
+                                        {{ field }}
+                                        {% if field.help_text %}
+                                            <p class="help">
+                                                {{ field.help_text }}
+                                            </p>
+                                        {% endif %}
+                                        {% if field.errors %}
+                                            <ul class="errorlist">
+                                                {% for error in field.errors %}
+                                                    <li>
+                                                        {{ error }}
+                                                    </li>
+                                                {% endfor %}
+                                            </ul>
+                                        {% endif %}
+                                    </div>
+                                </div>
+                            {% endfor %}
+                        </fieldset>
+                        <div class="mt-10">
+                            <input type="hidden" name="apply" value="yes" />
+                            <input type="submit" class="button" value="{% translate 'Run Formula' %}" />
+                            <input type="button" class="button closelink" onclick="javascript:history.back()" value="{% translate 'Go Back' %}" />
+                        </div>
+                    </form>
+                </div>
+            </div>
+        </div>
+    </div>
+{% endblock content %}