Payment Accrual Code Improvements

commcare_connect/form_receiver/processor.py

@@ -318,7 +318,7 @@ def process_deliver_unit(user, xform: XForm, app: CommCareApp, opportunity: Oppo
             if completed_work_needs_save:
                 completed_work.save()
 
-    update_payment_accrued(opportunity, [user.id])
+    update_payment_accrued(opportunity, [user.id], incremental=True)
     transaction.on_commit(partial(download_user_visit_attachments.delay, user_visit.id))
 
 

commcare_connect/opportunity/tests/test_models.py

@@ -118,5 +118,5 @@ def test_access_visit_count(opportunity: Opportunity):
     UserVisitFactory(
         completed_work=completed_work, deliver_unit=deliver_unit, user=access.user, opportunity=access.opportunity
     )
-    update_payment_accrued(opportunity, [access.user.id])
+    update_payment_accrued(opportunity, [access.user])
     assert access.visit_count == 1

commcare_connect/opportunity/tests/test_visit_import.py

@@ -146,7 +146,7 @@ def test_payment_accrued(opportunity: Opportunity):
                     opportunity_access=access,
                     completed_work=completed_work,
                 )
-    update_payment_accrued(opportunity, {mobile_user.id for mobile_user in mobile_users})
+    update_payment_accrued(opportunity, [mobile_user.id for mobile_user in mobile_users])
     for access in access_objects:
         access.refresh_from_db()
         assert access.payment_accrued == sum(payment_unit.amount for payment_unit in payment_units)
@@ -175,7 +175,7 @@ def test_duplicate_payment(opportunity: Opportunity, mobile_user: User):
         opportunity_access=access,
         completed_work=completed_work,
     )
-    update_payment_accrued(opportunity, {mobile_user.id})
+    update_payment_accrued(opportunity, [mobile_user.id])
     access.refresh_from_db()
     assert access.payment_accrued == payment_unit.amount * 2
     _validate_saved_fields(access)
@@ -211,7 +211,7 @@ def test_payment_accrued_optional_deliver_units(opportunity: Opportunity):
                 status=VisitValidationStatus.approved.value,
                 completed_work=completed_work,
             )
-    update_payment_accrued(opportunity, {access.user.id for access in access_objects})
+    update_payment_accrued(opportunity, [access.user.id for access in access_objects])
     for access in access_objects:
         access.refresh_from_db()
         assert access.payment_accrued == sum(payment_unit.amount for payment_unit in payment_units)
@@ -245,7 +245,7 @@ def test_payment_accrued_asymmetric_optional_deliver_units(opportunity: Opportun
         status=VisitValidationStatus.approved.value,
         completed_work=completed_work,
     )
-    update_payment_accrued(opportunity, {mobile_user.id})
+    update_payment_accrued(opportunity, [mobile_user.id])
     access.refresh_from_db()
     assert access.payment_accrued == payment_unit.amount * 1
     optional_deliver_unit_2 = DeliverUnitFactory(payment_unit=payment_unit, app=opportunity.deliver_app, optional=True)
@@ -257,7 +257,7 @@ def test_payment_accrued_asymmetric_optional_deliver_units(opportunity: Opportun
         status=VisitValidationStatus.approved.value,
         completed_work=completed_work,
     )
-    update_payment_accrued(opportunity, {mobile_user.id})
+    update_payment_accrued(opportunity, [mobile_user.id])
     access.refresh_from_db()
     assert access.payment_accrued == payment_unit.amount * 2
     _validate_saved_fields(access)
@@ -670,7 +670,7 @@ def test_review_completed_work_status(
         deliver_unit=deliver_unit,
     )
     assert access.payment_accrued == 0
-    update_payment_accrued(opportunity, {mobile_user_with_connect_link.id})
+    update_payment_accrued(opportunity, [mobile_user_with_connect_link.id])
     completed_works = CompletedWork.objects.filter(opportunity_access=access)
     payment_accrued = 0
     for cw in completed_works:

commcare_connect/opportunity/utils/completed_work.py

@@ -1,84 +1,179 @@
+from collections import defaultdict
+
+from django.db.models import Sum
+
 from commcare_connect.opportunity.models import (
     CompletedWork,
     CompletedWorkStatus,
+    DeliverUnit,
     OpportunityAccess,
     Payment,
+    PaymentUnit,
     VisitReviewStatus,
     VisitValidationStatus,
 )
 
 
+def get_completed_work_completed_approved_count(completed_works):
+    counts = defaultdict(lambda: {"completed": 0, "approved": 0})
+
+    completed_works = completed_works.prefetch_related("uservisit_set")
+
+    payment_units = PaymentUnit.objects.filter(
+        id__in=completed_works.values_list("payment_unit_id", flat=True)
+    ).values("id", "parent_payment_unit")
+    deliver_units = DeliverUnit.objects.filter(
+        payment_unit__in=completed_works.values_list("payment_unit_id", flat=True)
+    ).values("id", "optional", "payment_unit_id")
+
+    parent_child_payment_unit_map = defaultdict(list)
+    for payment_unit in payment_units:
+        parent_id = payment_unit["parent_payment_unit"]
+        parent_child_payment_unit_map[parent_id].append(payment_unit["id"])
+
+    deliver_unit_map = defaultdict(list)
+    for deliver_unit in deliver_units:
+        pu_id = deliver_unit["payment_unit_id"]
+        du_id = deliver_unit["id"]
+        optional = deliver_unit.get("optional", False)
+        deliver_unit_map[pu_id].append((du_id, optional))
+
+    for completed_work in completed_works:
+        if completed_work.id in counts:
+            continue
+
+        unit_counts = defaultdict(int)
+        approved_unit_counts = defaultdict(int)
+
+        for user_visit in completed_work.uservisit_set.all():
+            unit_counts[user_visit.deliver_unit_id] += 1
+            if user_visit.status == VisitValidationStatus.approved.value:
+                approved_unit_counts[user_visit.deliver_unit_id] += 1
+
+        required_deliver_units = [
+            du_id for du_id, optional in deliver_unit_map[completed_work.payment_unit_id] if not optional
+        ]
+        optional_deliver_units = [
+            du_id for du_id, optional in deliver_unit_map[completed_work.payment_unit_id] if optional
+        ]
+
+        number_completed = min([unit_counts[deliver_id] for deliver_id in required_deliver_units], default=0)
+        number_approved = min([approved_unit_counts[deliver_id] for deliver_id in required_deliver_units], default=0)
+
+        if optional_deliver_units:
+            optional_completed = sum(unit_counts[deliver_id] for deliver_id in optional_deliver_units)
+            number_completed = min(number_completed, optional_completed)
+
+            optional_approved = sum(approved_unit_counts[deliver_id] for deliver_id in optional_deliver_units)
+            number_approved = min(number_approved, optional_approved)
+
+        child_payment_units = parent_child_payment_unit_map[completed_work.payment_unit_id]
+        if child_payment_units:
+            child_completed_works = CompletedWork.objects.filter(
+                opportunity_access=completed_work.opportunity_access,
+                payment_unit__in=child_payment_units,
+                entity_id=completed_work.entity_id,
+            )
+            child_completed_work_count = child_approved_work_count = 0
+            for child_completed_work in child_completed_works:
+                if child_completed_work.id not in counts:
+                    counts[child_completed_work.id]["approved"] = child_completed_work.approved_count
+                    counts[child_completed_work.id]["completed"] = child_completed_work.completed_count
+                child_approved_work_count += counts[child_completed_work.id]["approved"]
+                child_completed_work_count += counts[child_completed_work.id]["completed"]
+            number_completed = min(number_completed, child_completed_work_count)
+            number_approved = min(number_approved, child_approved_work_count)
+
+        counts[completed_work.id]["approved"] = number_approved
+        counts[completed_work.id]["completed"] = number_completed
+    return counts
+
+
 def update_status(completed_works, opportunity_access, compute_payment=True):
     """
     Updates the status of completed works and optionally calculates & update total payment_accrued.
 
     If compute_payment is True, the saved fields related to completed/approved work and payments
     earned will also be saved against the model.
     """
-    payment_accrued = 0
-    for completed_work in completed_works:
-        payment_accrued += _update_status_set_saved_fields_and_get_payment_accrued(
-            completed_work, opportunity_access, compute_payment
-        )
-
+    update_status_and_set_saved_fields(completed_works, opportunity_access.opportunity, compute_payment)
     if compute_payment:
-        opportunity_access.payment_accrued = payment_accrued
+        opportunity_access.payment_accrued = (
+            CompletedWork.objects.filter(opportunity_access=opportunity_access)
+            .aggregate(payment_accrued=Sum("saved_payment_accrued"))
+            .get("payment_accrued", 0)
+            or 0
+        )
         opportunity_access.save()
 
 
-def _update_status_set_saved_fields_and_get_payment_accrued(completed_work, opportunity_access, compute_payment):
-    completed_count = completed_work.completed_count
-    if completed_count < 1:
-        return 0
-
-    amount_accrued = 0
-    made_changes = False
-    if opportunity_access.opportunity.auto_approve_payments:
-        visits = completed_work.uservisit_set.values_list("status", "reason", "review_status")
-        if any(status == VisitValidationStatus.rejected for status, *_ in visits):
-            completed_work.status = CompletedWorkStatus.rejected
-            completed_work.reason = "\n".join(reason for _, reason, _ in visits if reason)
-        elif all(status == VisitValidationStatus.approved for status, *_ in visits):
-            completed_work.status = CompletedWorkStatus.approved
-
-        if (
-            opportunity_access.opportunity.managed
-            and not all(review_status == VisitReviewStatus.agree for *_, review_status in visits)
-            and completed_work.status == CompletedWorkStatus.approved
-        ):
-            completed_work.status = CompletedWorkStatus.pending
-
-        made_changes = True
-
-    if compute_payment:
-        approved_count = completed_work.approved_count
-
-        amount_accrued = amount_accrued_usd = org_amount_accrued = org_amount_accrued_usd = 0
-        if approved_count > 0 and completed_work.status == CompletedWorkStatus.approved:
-            from commcare_connect.opportunity.visit_import import get_exchange_rate
+def update_status_and_set_saved_fields(completed_works, opportunity, compute_payment):
+    to_update = []
+    completed_approved_count = get_completed_work_completed_approved_count(completed_works)
+    for completed_work in completed_works:
+        completed_count = completed_approved_count[completed_work.id]["completed"]
+        if completed_count < 1:
+            continue
 
-            amount_accrued = approved_count * completed_work.payment_unit.amount
-            exchange_rate = get_exchange_rate(
-                opportunity_access.opportunity.currency, completed_work.status_modified_date
-            )
-            amount_accrued_usd = amount_accrued / exchange_rate
-            # if it's a managed opportunity we also need to update the org payment amounts
-            if opportunity_access.opportunity.managed:
-                org_amount_accrued = approved_count * opportunity_access.managed_opportunity.org_pay_per_visit
-                org_amount_accrued_usd = org_amount_accrued / exchange_rate
-
-        completed_work.saved_completed_count = completed_count
-        completed_work.saved_approved_count = approved_count
-        completed_work.saved_payment_accrued = amount_accrued
-        completed_work.saved_payment_accrued_usd = amount_accrued_usd
-        completed_work.saved_org_payment_accrued = org_amount_accrued
-        completed_work.saved_org_payment_accrued_usd = org_amount_accrued_usd
-        made_changes = True
-
-    if made_changes:
-        completed_work.save()
-
-    return amount_accrued
+        amount_accrued = 0
+        updated = False
+        if opportunity.auto_approve_payments:
+            visits = completed_work.uservisit_set.values_list("status", "reason", "review_status")
+            if any(status == VisitValidationStatus.rejected for status, *_ in visits):
+                completed_work.status = CompletedWorkStatus.rejected
+                completed_work.reason = "\n".join(reason for _, reason, _ in visits if reason)
+            elif all(status == VisitValidationStatus.approved for status, *_ in visits):
+                completed_work.status = CompletedWorkStatus.approved
+
+            if (
+                opportunity.managed
+                and not all(review_status == VisitReviewStatus.agree for *_, review_status in visits)
+                and completed_work.status == CompletedWorkStatus.approved
+            ):
+                completed_work.status = CompletedWorkStatus.pending
+
+            updated = True
+
+        if compute_payment:
+            approved_count = completed_approved_count[completed_work.id]["approved"]
+
+            amount_accrued = amount_accrued_usd = org_amount_accrued = org_amount_accrued_usd = 0
+            if approved_count > 0 and completed_work.status == CompletedWorkStatus.approved:
+                from commcare_connect.opportunity.visit_import import get_exchange_rate
+
+                amount_accrued = approved_count * completed_work.payment_unit.amount
+                exchange_rate = get_exchange_rate(opportunity.currency, completed_work.status_modified_date)
+                amount_accrued_usd = amount_accrued / exchange_rate
+                # if it's a managed opportunity we also need to update the org payment amounts
+                if opportunity.managed:
+                    org_amount_accrued = approved_count * opportunity.org_pay_per_visit
+                    org_amount_accrued_usd = org_amount_accrued / exchange_rate
+
+            completed_work.saved_completed_count = completed_count
+            completed_work.saved_approved_count = approved_count
+            completed_work.saved_payment_accrued = amount_accrued
+            completed_work.saved_payment_accrued_usd = amount_accrued_usd
+            completed_work.saved_org_payment_accrued = org_amount_accrued
+            completed_work.saved_org_payment_accrued_usd = org_amount_accrued_usd
+            updated = True
+
+        if updated:
+            to_update.append(completed_work)
+
+    CompletedWork.objects.bulk_update(
+        to_update,
+        fields=[
+            "reason",
+            "status",
+            "status_modified_date",
+            "saved_completed_count",
+            "saved_approved_count",
+            "saved_payment_accrued",
+            "saved_payment_accrued_usd",
+            "saved_org_payment_accrued",
+            "saved_org_payment_accrued_usd",
+        ],
+    )
 
 
 def update_work_payment_date(access: OpportunityAccess):

commcare_connect/opportunity/views.py

@@ -775,7 +775,7 @@ def user_profile(request, org_slug=None, opp_id=None, pk=None):
         [
             cw
             for cw in CompletedWork.objects.filter(opportunity_access=access, status=CompletedWorkStatus.pending)
-            if cw.approved_count
+            if cw.saved_approved_count
         ]
     )
     user_catchment_data = [
@@ -928,7 +928,7 @@ def approve_visit(request, org_slug=None, pk=None):
                 user_visit.justification = justification
 
         user_visit.save()
-        update_payment_accrued(opportunity=user_visit.opportunity, users=[user_visit.user])
+        update_payment_accrued(opportunity=user_visit.opportunity, users=[user_visit.user], incremental=True)
 
     if user_visit.opportunity.managed:
         return redirect("opportunity:user_visit_review", org_slug, opp_id)
@@ -947,7 +947,7 @@ def reject_visit(request, org_slug=None, pk=None):
     user_visit.reason = reason
     user_visit.save()
     access = OpportunityAccess.objects.get(user_id=user_visit.user_id, opportunity_id=user_visit.opportunity_id)
-    update_payment_accrued(opportunity=access.opportunity, users=[access.user])
+    update_payment_accrued(opportunity=access.opportunity, users=[access.user], incremental=True)
     return redirect("opportunity:user_visits_list", org_slug=org_slug, opp_id=user_visit.opportunity_id, pk=access.id)
 
 
@@ -1182,7 +1182,9 @@ def user_visit_review(request, org_slug, opp_id):
         user_visits = UserVisit.objects.filter(pk__in=updated_reviews)
         if review_status in [VisitReviewStatus.agree.value, VisitReviewStatus.disagree.value]:
             user_visits.update(review_status=review_status)
-            update_payment_accrued(opportunity=opportunity, users=[visit.user for visit in user_visits])
+            update_payment_accrued(
+                opportunity=opportunity, users=[visit.user for visit in user_visits], incremental=True
+            )
     RequestConfig(request, paginate={"per_page": 15}).configure(table)
     return render(
         request,

commcare_connect/opportunity/visit_import.py

@@ -186,14 +186,25 @@ def get_missing_justification_message(visits_ids):
     return f"Justification is required for flagged visits: {id_list}"
 
 
-def update_payment_accrued(opportunity: Opportunity, users):
-    """Updates payment accrued for completed and approved CompletedWork instances."""
+def update_payment_accrued(opportunity: Opportunity, users: list, incremental=False):
+    """Updates payment accrued for completed and approved CompletedWork instances.
+    Skips already processed completed works when incremental is true."""
+
     access_objects = OpportunityAccess.objects.filter(user__in=users, opportunity=opportunity, suspended=False)
+    filter_kwargs = {}
+    exclude_status = [CompletedWorkStatus.rejected]
+    if incremental:
+        exclude_status.append(CompletedWorkStatus.approved)
+        filter_kwargs["saved_approved_count"] = 0
+
     for access in access_objects:
         with cache.lock(f"update_payment_accrued_lock_{access.id}", timeout=900):
-            completed_works = access.completedwork_set.exclude(status=CompletedWorkStatus.rejected).select_related(
-                "payment_unit"
+            completed_works = (
+                access.completedwork_set.filter(**filter_kwargs)
+                .exclude(status__in=exclude_status)
+                .select_related("payment_unit")
             )
+
             update_status(completed_works, access, compute_payment=True)