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

"""费率配置读取服务。

从 payroll_config 表读取指定地区的费率；地区未单独插入时回退到全国默认。
"""

from __future__ import annotations

import json
from typing import Any

from db.connection import get_conn
from util.logging_config import get_skill_logger

log = get_skill_logger()


def get_config_value(config_key: str, region: str = "default") -> float | str | None:
    """读取单个配置项的值。

    优先取 (config_key, region)；若该地区无记录，回退到 (config_key, 'default')。
    """
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            "SELECT config_value FROM payroll_config WHERE config_key=? AND region=?",
            (config_key, region),
        )
        row = cur.fetchone()
        if row is None and region != "default":
            cur.execute(
                "SELECT config_value FROM payroll_config WHERE config_key=? AND region='default'",
                (config_key,),
            )
            row = cur.fetchone()
        if row is None:
            return None
        return row[0]
    finally:
        conn.close()


def get_tax_brackets(region: str = "default") -> list[dict[str, Any]]:
    """读取 7 级累计预扣税率表。"""
    raw = get_config_value("tax_brackets_json", region)
    if raw is None or not isinstance(raw, str):
        return []
    try:
        brackets = json.loads(raw)
        return brackets if isinstance(brackets, list) else []
    except (json.JSONDecodeError, TypeError):
        return []


def get_region_config(region: str = "default") -> dict[str, Any]:
    """读取一个地区的全部费率配置，返回 dict。"""
    keys = [
        "social_pension_rate",
        "social_medical_rate",
        "social_unemployment_rate",
        "social_base_min",
        "social_base_max",
        "housing_fund_rate",
        "housing_fund_base_min",
        "housing_fund_base_max",
        "tax_threshold",
    ]
    result: dict[str, Any] = {"region": region}
    for key in keys:
        result[key] = get_config_value(key, region)
    result["tax_brackets"] = get_tax_brackets(region)
    return result


def list_all_config(region: str | None = None) -> list[dict[str, Any]]:
    """列出费率配置（供 view-config Action）。

    region 为 None 时列出全部地区；否则只列该地区（含回退默认值）。
    """
    conn = get_conn()
    try:
        cur = conn.cursor()
        if region is None:
            cur.execute(
                "SELECT config_key, region, config_value, display_name, description FROM payroll_config ORDER BY region, config_key"
            )
        else:
            cur.execute(
                "SELECT config_key, region, config_value, display_name, description FROM payroll_config WHERE region=? ORDER BY config_key",
                (region,),
            )
        rows = cur.fetchall()
        return [
            {
                "config_key": r[0],
                "region": r[1],
                "config_value": r[2],
                "display_name": r[3],
                "description": r[4],
            }
            for r in rows
        ]
    finally:
        conn.close()


def update_config_value(config_key: str, region: str, config_value: float) -> dict[str, Any]:
    """更新单个费率配置（供 edit-config Action）。

    返回 {ok, skill, config_key, region, old_value, new_value}。
    """
    log.info("config_update key=%s region=%s value=%s", config_key, region, config_value)
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            "SELECT config_value FROM payroll_config WHERE config_key=? AND region=?",
            (config_key, region),
        )
        row = cur.fetchone()
        if row is None:
            return {
                "ok": False,
                "skill": "query-payslip-default",
                "error_code": "CONFIG_NOT_FOUND",
                "error": f"配置项 {config_key}（地区 {region}）不存在",
            }
        old_value = row[0]
        cur.execute(
            """
            UPDATE payroll_config
            SET config_value=?, updated_at=strftime('%s','now')
            WHERE config_key=? AND region=?
            """,
            (config_value, config_key, region),
        )
        conn.commit()
        return {
            "ok": True,
            "skill": "query-payslip-default",
            "config_key": config_key,
            "region": region,
            "old_value": old_value,
            "new_value": config_value,
        }
    finally:
        conn.close()