zkxw/backend/app/services/protocol_payment_installment.py

"""分期缴费：应缴日计算（整数月按日历递增，小数月按 30 天/月换算）"""

from __future__ import annotations

import calendar
import json
from decimal import Decimal, ROUND_DOWN
from typing import List, Optional, Sequence, Union

Number = Union[int, float, Decimal, str]

DAYS_PER_MONTH = 30
DEFAULT_INSTALLMENT_ADVANCE_DAYS = 30


def add_days_yyyymmdd(date_str: str, days: int) -> str:
    raw = (date_str or "").strip().replace("-", "")
    if len(raw) != 8:
        return ""
    from datetime import date, timedelta

    y = int(raw[0:4])
    m = int(raw[4:6])
    d = int(raw[6:8])
    dt = date(y, m, d) + timedelta(days=days)
    return f"{dt.year:04d}{dt.month:02d}{dt.day:02d}"


def add_months_yyyymmdd(date_str: str, months: int) -> str:
    raw = (date_str or "").strip().replace("-", "")
    if len(raw) != 8:
        return ""
    y = int(raw[0:4])
    m = int(raw[4:6])
    d = int(raw[6:8])
    month_index = m - 1 + months
    ny = y + month_index // 12
    nm = month_index % 12 + 1
    last_day = calendar.monthrange(ny, nm)[1]
    nd = min(d, last_day)
    return f"{ny:04d}{nm:02d}{nd:02d}"


def add_interval_months_yyyymmdd(date_str: str, total_months: Number) -> str:
    """整数部分按日历加月，小数部分按 30 天/月换算为天数。"""
    raw = (date_str or "").strip().replace("-", "")
    if len(raw) != 8:
        return ""
    total = Decimal(str(total_months))
    if total == 0:
        return raw
    whole = int(total.to_integral_value(rounding=ROUND_DOWN))
    frac = total - Decimal(whole)
    result = raw
    if whole:
        result = add_months_yyyymmdd(result, whole)
    if frac:
        days = int(round(float(frac) * DAYS_PER_MONTH))
        if days:
            result = add_days_yyyymmdd(result, days)
    return result


def calc_fq_installment_due_date(
    start_date: str,
    index: int,
    interval_months: Number,
    advance_days: int = DEFAULT_INSTALLMENT_ADVANCE_DAYS,
) -> str:
    """第 1 期=开始时间；第 2 期起在间隔递增基础上提前 advance_days 天。"""
    total_months = Decimal(str(index)) * Decimal(str(interval_months))
    base = add_interval_months_yyyymmdd(start_date, total_months)
    if index >= 1:
        return add_days_yyyymmdd(base, -int(advance_days or 0))
    return base


def parse_installment_ratios(raw: Optional[str]) -> List[float]:
    if not raw:
        return []
    try:
        data = json.loads(raw)
        return [float(x) for x in data] if isinstance(data, list) else []
    except (json.JSONDecodeError, TypeError, ValueError):
        return []


def is_legacy_installment_ratios(
    values: Sequence[Number],
    contract_amount: Optional[Decimal] = None,
) -> bool:
    if not values:
        return False
    nums = [float(v) for v in values]
    total = sum(nums)
    mx = max(nums)
    if contract_amount is not None and abs(total - float(contract_amount)) < 0.02:
        return False
    return mx <= 100 and abs(total - 100) < 0.05


def resolve_installment_due_amounts(
    contract_amount: Optional[Decimal],
    stored: Sequence[Number],
    count: int,
) -> List[Decimal]:
    if stored and len(stored) == count:
        if is_legacy_installment_ratios(stored, contract_amount):
            return calc_installment_due_amounts_by_ratios(contract_amount, stored)
        return [Decimal(str(x)) for x in stored]
    return _calc_installment_due_amounts_inline(contract_amount, count)


def _calc_installment_due_amounts_inline(
    contract_amount: Optional[Decimal],
    count: int,
) -> List[Decimal]:
    if contract_amount is None or count < 1:
        return []
    total_cents = int(round(float(contract_amount) * 100))
    base_cents = total_cents // count
    amounts: List[Decimal] = []
    sum_cents = 0
    for i in range(count):
        if i == count - 1:
            cents = total_cents - sum_cents
        else:
            cents = base_cents
            sum_cents += base_cents
        amounts.append(Decimal(cents) / Decimal(100))
    return amounts


def calc_installment_due_amounts_by_ratios(
    contract_amount: Optional[Decimal],
    ratios: Sequence[Number],
) -> List[Decimal]:
    """全额分配：前 n-1 期 floor 到分，末位补尾差。"""
    if contract_amount is None or not ratios:
        return []
    count = len(ratios)
    total_cents = int(round(float(contract_amount) * 100))
    amounts: List[Decimal] = []
    sum_cents = 0
    for i in range(count):
        if i == count - 1:
            cents = total_cents - sum_cents
        else:
            cents = int(total_cents * float(ratios[i]) / 100.0)
            sum_cents += cents
        amounts.append(Decimal(cents) / Decimal(100))
    return amounts


def calc_unpaid_amounts_by_ratios(
    remaining: Decimal,
    ratios: Sequence[Number],
    unpaid_indices: Sequence[int],
) -> List[Decimal]:
    """部分已缴：按未缴期次比例归一化分配 remaining。"""
    if remaining <= 0 or not unpaid_indices:
        return []
    subset = [float(ratios[i]) for i in unpaid_indices]
    ratio_sum = sum(subset)
    if ratio_sum <= 0:
        return []
    total_cents = int(round(float(remaining) * 100))
    amounts: List[Decimal] = []
    sum_cents = 0
    for j, _idx in enumerate(unpaid_indices):
        if j == len(unpaid_indices) - 1:
            cents = total_cents - sum_cents
        else:
            cents = int(total_cents * subset[j] / ratio_sum)
            sum_cents += cents
        amounts.append(Decimal(cents) / Decimal(100))
    return amounts