diff --git a/backend/app/services/sample_payment_progress.py b/backend/app/services/sample_payment_progress.py new file mode 100644 index 00000000..7a1fb345 --- /dev/null +++ b/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, + } diff --git a/backend/tests/test_sample_payment_progress.py b/backend/tests/test_sample_payment_progress.py new file mode 100644 index 00000000..3401a9c9 --- /dev/null +++ b/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()