109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
"""样本缴费方案进度汇总（列表「按样本」维度）"""
|
|
|
|
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 _schedule_id(row: Mapping[str, Any]) -> str:
|
|
return str(row.get("id") or row.get("scheduleId") or row.get("schedule_id") or "").strip()
|
|
|
|
|
|
def _transfer_schedule_id(row: Mapping[str, Any]) -> str:
|
|
return str(row.get("scheduleId") or row.get("schedule_id") or "").strip()
|
|
|
|
|
|
def is_schedule_effectively_cleared(
|
|
schedule: Mapping[str, Any],
|
|
*,
|
|
confirmed_paid: Decimal,
|
|
) -> bool:
|
|
"""
|
|
批次已缴清：显式 cleared，或财务确认实缴（yjf）已覆盖应缴。
|
|
兼容历史数据未回写 schedule_status 的情况。
|
|
"""
|
|
if _status(schedule, "scheduleStatus", "schedule_status") == "cleared":
|
|
return True
|
|
due = _dec(schedule.get("dueAmount", schedule.get("due_amount")))
|
|
return confirmed_paid >= due
|
|
|
|
|
|
def compute_sample_payment_progress(
|
|
*,
|
|
schedules: Iterable[Mapping[str, Any]],
|
|
transfers: Iterable[Mapping[str, Any]],
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
progressStatus 优先级：yjq > bfyj > dfj（不含付款确认中）
|
|
已缴仅计 pay_status=yjf；有效期排除 schedule_status=void
|
|
已缴清期：schedule_status=cleared，或该期 yjf 合计 >= 应缴
|
|
canConfirm：无 jfqrz，且存在有效未清期
|
|
"""
|
|
sch_list: List[Mapping[str, Any]] = list(schedules or [])
|
|
tr_list: List[Mapping[str, Any]] = list(transfers or [])
|
|
|
|
paid_by_schedule: Dict[str, Decimal] = {}
|
|
paid = Decimal("0")
|
|
for t in tr_list:
|
|
if _status(t, "payStatus", "pay_status") != "yjf":
|
|
continue
|
|
amount = _dec(t.get("transferAmount", t.get("transfer_amount")))
|
|
paid += amount
|
|
sid = _transfer_schedule_id(t)
|
|
if sid:
|
|
paid_by_schedule[sid] = paid_by_schedule.get(sid, Decimal("0")) + amount
|
|
|
|
active = [
|
|
s
|
|
for s in sch_list
|
|
if _status(s, "scheduleStatus", "schedule_status") != "void"
|
|
]
|
|
active_schedule_count = len(active)
|
|
cleared_flags = [
|
|
is_schedule_effectively_cleared(
|
|
s,
|
|
confirmed_paid=paid_by_schedule.get(_schedule_id(s), Decimal("0")),
|
|
)
|
|
for s in active
|
|
]
|
|
cleared_schedule_count = sum(1 for cleared in cleared_flags if cleared)
|
|
|
|
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 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(not cleared for cleared in cleared_flags)
|
|
|
|
return {
|
|
"progressStatus": progress,
|
|
"paidAmount": paid,
|
|
"dueRemainAmount": due_remain,
|
|
"clearedScheduleCount": cleared_schedule_count,
|
|
"activeScheduleCount": active_schedule_count,
|
|
"canConfirm": can_confirm,
|
|
}
|