query-payslip-default/scripts/service/tax_service.py

"""个税累计预扣计算服务。

按"累计预扣法"计算当月个税：
  累计预扣预缴应纳税所得额 = 累计收入 - 累计减除费用 - 累计专项扣除 - 累计专项附加扣除 - 累计依法确定的其他扣除
  当月应预扣预缴税额 = (累计预扣预缴应纳税所得额 × 预扣率 - 速算扣除数) - 已预扣预缴税额

简化假设：
- 专项附加扣除与其他扣除暂不计入（数据不可得）
- 累计减除费用 = 5000 × 当年至本月份数
- 累计专项扣除 = 累计社保+公积金个人部分（从历史 snapshots 读取）
"""

from __future__ import annotations

import time
from typing import Any

from service.config_service import get_config_value, get_tax_brackets
from db.connection import get_conn
from util.logging_config import get_skill_logger

log = get_skill_logger()


def _month_index(period: str) -> int:
    """从 'YYYY-MM' 提取月份数（1-12）。"""
    try:
        return int(period.split("-")[1])
    except (ValueError, IndexError):
        return 1


def _year_of_period(period: str) -> str:
    """从 'YYYY-MM' 提取年份。"""
    try:
        return period.split("-")[0]
    except (ValueError, IndexError):
        return "1970"


def get_cumulative_deduction_before(period: str) -> dict[str, float]:
    """读取同一年的历史 snapshots，累计社保/公积金个人部分。

    period 之前的同一年 snapshots（按 period 字符串排序 < 当前 period）。
    返回 {social_cum, housing_fund_cum, tax_cum}。
    """
    year = _year_of_period(period)
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            """
            SELECT detail_json FROM payslip_snapshots
            WHERE period LIKE ? AND period < ?
            ORDER BY period ASC
            """,
            (f"{year}-%", period),
        )
        import json
        social_cum = 0.0
        housing_fund_cum = 0.0
        tax_cum = 0.0
        for (detail_str,) in cur.fetchall():
            if not detail_str:
                continue
            try:
                detail = json.loads(detail_str)
            except (json.JSONDecodeError, TypeError):
                continue
            ded = detail.get("deduction", {}) if isinstance(detail, dict) else {}
            social_cum += float(ded.get("social", {}).get("personal_total", 0) or 0)
            housing_fund_cum += float(ded.get("housing_fund", {}).get("personal", 0) or 0)
            tax_cum += float(ded.get("tax", {}).get("actual", 0) or 0)
        return {"social_cum": social_cum, "housing_fund_cum": housing_fund_cum, "tax_cum": tax_cum}
    finally:
        conn.close()


def compute_cumulative_tax(
    *,
    current_gross: float,
    period: str,
    current_social: float,
    current_housing_fund: float,
    region: str = "default",
) -> dict[str, Any]:
    """按累计预扣法计算当月个税。

    返回 {
        actual_tax: float,       # 本次应预扣个税
        cumulative_income: float,
        cumulative_deduction: float,
        cumulative_taxable: float,
        cumulative_tax: float,   # 累计应预扣
        prior_tax_paid: float,   # 之前已预扣
        brackets_used: dict,     # 命中的税率档
        has_history: bool,       # 是否有历史累计数据
        note: str,
    }
    """
    log.info(
        "tax_compute_start period=%s gross=%s social=%s housing=%s region=%s",
        period, current_gross, current_social, current_housing_fund, region,
    )
    threshold = get_config_value("tax_threshold", region) or 5000
    brackets = get_tax_brackets(region)

    month_idx = _month_index(period)
    cumulative_threshold = float(threshold) * month_idx

    prior = get_cumulative_deduction_before(period)
    has_history = prior["social_cum"] > 0 or prior["tax_cum"] > 0

    cumulative_income = current_gross + prior.get("income_cum", 0.0) if has_history else current_gross
    # 简化：累计专项扣除 = 累计社保+公积金个人
    cumulative_special_deduction = (
        prior["social_cum"] + prior["housing_fund_cum"]
        + current_social + current_housing_fund
    )

    cumulative_taxable = cumulative_income - cumulative_threshold - cumulative_special_deduction
    if cumulative_taxable < 0:
        cumulative_taxable = 0.0

    cumulative_tax = 0.0
    bracket_hit = None
    if brackets and cumulative_taxable > 0:
        for bracket in brackets:
            upper = bracket.get("upper")
            rate = float(bracket.get("rate", 0))
            quick = float(bracket.get("quick_deduction", 0))
            if upper is None or cumulative_taxable <= float(upper):
                cumulative_tax = cumulative_taxable * rate - quick
                bracket_hit = bracket
                break
        if bracket_hit is None and brackets:
            bracket_hit = brackets[-1]
            rate = float(bracket_hit.get("rate", 0))
            quick = float(bracket_hit.get("quick_deduction", 0))
            cumulative_tax = cumulative_taxable * rate - quick

    cumulative_tax = max(0.0, cumulative_tax)
    prior_tax_paid = prior["tax_cum"] if has_history else 0.0
    actual_tax = max(0.0, cumulative_tax - prior_tax_paid)

    note = "仅供参考，非官方数据"
    if not has_history:
        note = "仅供参考（无累计历史数据，累计按本月单月计算），非官方数据"

    return {
        "actual_tax": round(actual_tax, 2),
        "cumulative_income": round(cumulative_income, 2),
        "cumulative_threshold": round(cumulative_threshold, 2),
        "cumulative_special_deduction": round(cumulative_special_deduction, 2),
        "cumulative_taxable": round(cumulative_taxable, 2),
        "cumulative_tax": round(cumulative_tax, 2),
        "prior_tax_paid": round(prior_tax_paid, 2),
        "brackets_used": bracket_hit,
        "has_history": has_history,
        "note": note,
    }