zkxw/backend/app/services/protocol_payment_installment.py

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

from __future__ import annotations

import calendar
import json
from decimal import Decimal, ROUND_DOWN, ROUND_HALF_UP
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(
            (frac * Decimal(DAYS_PER_MONTH)).to_integral_value(
                rounding=ROUND_HALF_UP
            )
        )
        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[Decimal]:
    if not raw:
        return []
    try:
        data = json.loads(raw)
        return [Decimal(str(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 = [Decimal(str(v)) for v in values]
    total = sum(nums)
    mx = max(nums)
    if contract_amount is not None and abs(total - contract_amount) < Decimal('0.02'):
        return False
    return mx <= Decimal('100') and abs(total - Decimal('100')) < Decimal('0.05')


def _money_to_cents(value: Decimal) -> int:
    return int(
        (Decimal(str(value)) * Decimal('100')).to_integral_value(
            rounding=ROUND_HALF_UP
        )
    )


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 = _money_to_cents(contract_amount)
    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 = _money_to_cents(contract_amount)
    amounts: List[Decimal] = []
    sum_cents = 0
    for i in range(count):
        if i == count - 1:
            cents = total_cents - sum_cents
        else:
            ratio = Decimal(str(ratios[i]))
            cents = int(
                (Decimal(total_cents) * ratio / Decimal('100')).to_integral_value(
                    rounding=ROUND_DOWN
                )
            )
            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 = [Decimal(str(ratios[i])) for i in unpaid_indices]
    ratio_sum = sum(subset)
    if ratio_sum <= 0:
        return []
    total_cents = _money_to_cents(remaining)
    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(
                (Decimal(total_cents) * subset[j] / ratio_sum).to_integral_value(
                    rounding=ROUND_DOWN
                )
            )
            sum_cents += cents
        amounts.append(Decimal(cents) / Decimal(100))
    return amounts


def resolve_locked_remaining_due_amounts(
    contract_amount: Decimal,
    locked_due_amount: Decimal,
    stored: Sequence[Number],
    unpaid_indices: Sequence[int],
) -> List[Decimal]:
    """按剩余合同金额分配未缴期次；有期次权重时按权重，无则均摊。"""
    contract = Decimal(str(contract_amount))
    locked = Decimal(str(locked_due_amount))
    remaining = contract - locked
    if remaining < 0:
        raise ValueError("合同金额不能小于已缴或锁定明细金额")
    indices = [int(index) for index in unpaid_indices]
    if not indices:
        return []
    values = [Decimal(str(value)) for value in stored]
    if (
        values
        and max(indices) < len(values)
        and all(values[index] > 0 for index in indices)
    ):
        weighted = calc_unpaid_amounts_by_ratios(remaining, values, indices)
        if weighted:
            return weighted
    return _calc_installment_due_amounts_inline(remaining, len(indices))