zkxw/backend/app/services/sample_payment_period_summary.py

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 ""


_DISPLAY_PAID_STATUSES = ("yjf",)


def summarize_period(
    *,
    due_amount: Any,
    transfers: Iterable[Mapping[str, Any]],
) -> Dict[str, Any]:
    """
    displayPaidAmount: 已缴费金额仅计付款完成(yjf)
    arrearsAmount: max(due - displayPaidAmount, 0)
    financeAudit: jfqrz > (confirmed>=due 且无确认中 → yjf) > (0<confirmed → partial) > dfj
    payDate: 首笔 transferTime 的日期部分（调用方应按 create_time ASC 传入）
    payTypeSummary: 0 笔空；1 笔用该笔 type；多笔「多笔」
    """
    trs: List[Mapping[str, Any]] = list(transfers or [])
    display_paid = sum(
        (
            _dec(t.get("transferAmount", t.get("transfer_amount")))
            for t in trs
            if _status(t, "payStatus", "pay_status") in _DISPLAY_PAID_STATUSES
        ),
        Decimal("0"),
    )
    due = _dec(due_amount)
    arrears = due - display_paid
    if arrears < 0:
        arrears = Decimal("0")

    has_jfqrz = any(_status(t, "payStatus", "pay_status") == "jfqrz" for t in trs)
    confirmed = sum(
        (
            _dec(t.get("transferAmount", t.get("transfer_amount")))
            for t in trs
            if _status(t, "payStatus", "pay_status") == "yjf"
        ),
        Decimal("0"),
    )
    if has_jfqrz:
        audit = "jfqrz"
    elif due > 0 and confirmed >= due:
        audit = "yjf"
    elif confirmed > 0:
        audit = "partial"
    else:
        audit = "dfj"

    pay_date = ""
    if trs:
        raw = str(trs[0].get("transferTime") or trs[0].get("transfer_time") or "")
        pay_date = raw[:10] if len(raw) >= 10 else raw

    types = [
        _status(t, "transferType", "transfer_type")
        for t in trs
        if _status(t, "transferType", "transfer_type")
    ]
    if len(types) == 0:
        pay_type = ""
    elif len(set(types)) == 1 and len(types) == 1:
        pay_type = types[0]
    else:
        pay_type = "多笔"

    return {
        "displayPaidAmount": display_paid,
        "arrearsAmount": arrears,
        "payDate": pay_date,
        "payTypeSummary": pay_type,
        "financeAudit": audit,
    }