feat(backend): add sample payment progress aggregation helper

Pure function for per-sample list progressStatus, paid/due totals, and canConfirm.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/sample_payment_progress.py

@@ -0,0 +1,82 @@
+"""样本缴费方案进度汇总（列表「按样本」维度）"""
+
+from __future__ import annotations
+
+from decimal import Decimal
+from typing import Any, Dict, Iterable, List, Mapping
+
+
+def _dec(v: Any) -> Decimal:
+    if v is None or v == "":
+        return Decimal("0")
+    return Decimal(str(v))
+
+
+def _status(row: Mapping[str, Any], *keys: str) -> str:
+    for k in keys:
+        if k in row and row[k] is not None:
+            return str(row[k]).strip()
+    return ""
+
+
+def compute_sample_payment_progress(
+    *,
+    schedules: Iterable[Mapping[str, Any]],
+    transfers: Iterable[Mapping[str, Any]],
+) -> Dict[str, Any]:
+    """
+    progressStatus 优先级：jfqrz > yjq > bfyj > dfj
+    已缴仅计 pay_status=yjf；有效期排除 schedule_status=void
+    canConfirm：无 jfqrz，且存在有效未清期（欠缴>0）或可编辑的 dfj 草稿
+    """
+    sch_list: List[Mapping[str, Any]] = list(schedules or [])
+    tr_list: List[Mapping[str, Any]] = list(transfers or [])
+
+    active = [
+        s
+        for s in sch_list
+        if _status(s, "scheduleStatus", "schedule_status") != "void"
+    ]
+    active_schedule_count = len(active)
+    cleared_schedule_count = sum(
+        1
+        for s in active
+        if _status(s, "scheduleStatus", "schedule_status") == "cleared"
+    )
+
+    paid = sum(
+        (
+            _dec(t.get("transferAmount", t.get("transfer_amount")))
+            for t in tr_list
+            if _status(t, "payStatus", "pay_status") == "yjf"
+        ),
+        Decimal("0"),
+    )
+    due_total = sum((_dec(s.get("dueAmount", s.get("due_amount"))) for s in active), Decimal("0"))
+    due_remain = due_total - paid
+    if due_remain < 0:
+        due_remain = Decimal("0")
+
+    has_jfqrz = any(_status(t, "payStatus", "pay_status") == "jfqrz" for t in tr_list)
+
+    if has_jfqrz:
+        progress = "jfqrz"
+    elif active_schedule_count > 0 and cleared_schedule_count == active_schedule_count:
+        progress = "yjq"
+    elif paid > 0:
+        progress = "bfyj"
+    else:
+        progress = "dfj"
+
+    can_confirm = (not has_jfqrz) and any(
+        _status(s, "scheduleStatus", "schedule_status") != "cleared" for s in active
+    )
+
+    return {
+        "progressStatus": progress,
+        "paidAmount": paid,
+        "dueRemainAmount": due_remain,
+        "clearedScheduleCount": cleared_schedule_count,
+        "activeScheduleCount": active_schedule_count,
+        "canConfirm": can_confirm,
+    }

backend/tests/test_sample_payment_progress.py

@@ -0,0 +1,84 @@
+"""样本缴费进度：状态码与已缴/待缴汇总"""
+
+from decimal import Decimal
+import unittest
+
+from app.services.sample_payment_progress import (
+    compute_sample_payment_progress,
+)
+
+
+class TestSamplePaymentProgress(unittest.TestCase):
+    def test_pending_when_no_transfers(self):
+        r = compute_sample_payment_progress(
+            schedules=[
+                {"scheduleStatus": "open", "dueAmount": Decimal("1000")},
+            ],
+            transfers=[],
+        )
+        self.assertEqual(r["progressStatus"], "dfj")
+        self.assertEqual(r["paidAmount"], Decimal("0"))
+        self.assertEqual(r["dueRemainAmount"], Decimal("1000"))
+        self.assertEqual(r["clearedScheduleCount"], 0)
+        self.assertEqual(r["activeScheduleCount"], 1)
+        self.assertTrue(r["canConfirm"])
+
+    def test_confirming_wins_over_partial(self):
+        r = compute_sample_payment_progress(
+            schedules=[
+                {"id": "s1", "scheduleStatus": "open", "dueAmount": Decimal("1000")},
+            ],
+            transfers=[
+                {"scheduleId": "s1", "payStatus": "yjf", "transferAmount": Decimal("200")},
+                {"scheduleId": "s1", "payStatus": "jfqrz", "transferAmount": Decimal("100")},
+            ],
+        )
+        self.assertEqual(r["progressStatus"], "jfqrz")
+        self.assertEqual(r["paidAmount"], Decimal("200"))
+        self.assertFalse(r["canConfirm"])
+
+    def test_partial_paid(self):
+        r = compute_sample_payment_progress(
+            schedules=[
+                {"id": "s1", "scheduleStatus": "cleared", "dueAmount": Decimal("500")},
+                {"id": "s2", "scheduleStatus": "open", "dueAmount": Decimal("500")},
+            ],
+            transfers=[
+                {"scheduleId": "s1", "payStatus": "yjf", "transferAmount": Decimal("500")},
+            ],
+        )
+        self.assertEqual(r["progressStatus"], "bfyj")
+        self.assertEqual(r["clearedScheduleCount"], 1)
+        self.assertEqual(r["activeScheduleCount"], 2)
+        self.assertTrue(r["canConfirm"])
+
+    def test_all_cleared(self):
+        r = compute_sample_payment_progress(
+            schedules=[
+                {"id": "s1", "scheduleStatus": "cleared", "dueAmount": Decimal("500")},
+                {"id": "s2", "scheduleStatus": "void", "dueAmount": Decimal("500")},
+            ],
+            transfers=[
+                {"scheduleId": "s1", "payStatus": "yjf", "transferAmount": Decimal("500")},
+            ],
+        )
+        self.assertEqual(r["progressStatus"], "yjq")
+        self.assertEqual(r["activeScheduleCount"], 1)
+        self.assertEqual(r["clearedScheduleCount"], 1)
+        self.assertFalse(r["canConfirm"])
+        self.assertEqual(r["dueRemainAmount"], Decimal("0"))
+
+    def test_void_schedules_ignored(self):
+        r = compute_sample_payment_progress(
+            schedules=[
+                {"scheduleStatus": "void", "dueAmount": Decimal("9999")},
+                {"scheduleStatus": "open", "dueAmount": Decimal("100")},
+            ],
+            transfers=[],
+        )
+        self.assertEqual(r["activeScheduleCount"], 1)
+        self.assertEqual(r["dueRemainAmount"], Decimal("100"))
+
+
+if __name__ == "__main__":
+    unittest.main()