zkxw/backend/app/services/sample_payment_progress.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 ""


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,
    }