PayrollManagementSystem/backend/app/services/coefficient_service.py

"""系数表是参数版本的编辑视图，不保留第二套工资公式或独立系数数据。"""

from datetime import date
from decimal import ROUND_HALF_UP, Decimal
from fractions import Fraction
from io import BytesIO
from zipfile import ZipFile

from openpyxl import Workbook, load_workbook
from pydantic import ValidationError
from sqlalchemy import select

from app.core.exceptions import BadRequestError
from app.schemas.coefficient import CoefficientRowIn
from app.schemas.param import ParamChangeIn, ParamChangeItemIn
from app.services import param_service
from app.services.param_registry import get_spec, validate_and_normalize
from app.services.rate_rules import decide, personal_values
from app.services.tplus.workbench import digest

CATEGORIES = ("municipal_water", "rail_transit")
HEADERS = ("product_category", "customer_type", "stage", "value", "slope", "boundary")


def _coefficient(rate):
    return str(1 / rate) if rate else None


def _average_text(value):
    if value is None:
        return None
    return format(
        param_service.to_decimal(value).quantize(Decimal("0.000001"), rounding=ROUND_HALF_UP), "f"
    )


def _revision(session, lock=False):
    if session is None:
        return []
    from app.models import ParamChangeBatch

    query = select(
        ParamChangeBatch.id, ParamChangeBatch.effective_date, ParamChangeBatch.version_no
    ).order_by(ParamChangeBatch.id)
    if lock:
        query = query.with_for_update()
    return [tuple(row) for row in session.execute(query)]


def catalog(session, as_of=None, revision=None):
    day = as_of or date.today()
    snap = param_service.resolve_snapshot(session, day)
    rows = []
    for category in CATEGORIES:
        values = personal_values(snap, category)
        params = {**values, "enterprise": snap.get("enterprise.rate", product_category=category)}
        rates = [
            decide(
                params, category=category, customer_type="personal", role_code="salesperson", days=d
            )[0]
            for d in range(1, 361)
        ]
        average = sum((1 / r for r in rates if r), Fraction(0)) / 360 if all(rates) else None
        for stage, field, days in (
            ("first", "first", 1),
            ("linear", "start", 2),
            ("after", "floor", int(Fraction(values["boundary"])) + 1),
        ):
            rate = decide(
                params,
                category=category,
                customer_type="personal",
                role_code="salesperson",
                days=days,
            )[0]
            rows.append(
                {
                    "product_category": category,
                    "customer_type": "personal",
                    "stage": stage,
                    "value": values[field],
                    "slope": values["slope"] if stage == "linear" else None,
                    "boundary": int(Fraction(values["boundary"])) if stage == "linear" else None,
                    "rate": str(rate),
                    "system_coefficient": _coefficient(rate),
                    "average_coefficient_360": _average_text(average),
                }
            )
        rate = Fraction(params["enterprise"]) / 1000
        rows.append(
            {
                "product_category": category,
                "customer_type": "enterprise",
                "stage": "fixed",
                "value": params["enterprise"],
                "slope": None,
                "boundary": None,
                "rate": str(rate),
                "system_coefficient": _coefficient(rate),
                "average_coefficient_360": _average_text(1 / rate if rate else None),
            }
        )
        rows.append(
            {
                "product_category": category,
                "customer_type": "manager_personal",
                "stage": "excluded",
                "value": "0",
                "rate": "0",
                "system_coefficient": None,
                "average_coefficient_360": None,
                "readonly": True,
            }
        )
    return {
        "as_of": str(day),
        "rows": rows,
        "fingerprint": digest(
            {
                "as_of": day,
                "rows": rows,
                "revision": _revision(session) if revision is None else revision,
            }
        ),
        "average_rule": "第1—360天每日系统系数的算术平均，展示保留6位小数；存在零计提日时不适用。经理个人户不单笔计提。",
    }


def items_for(rows):
    items = []
    seen = set()
    for row in rows:
        key = (row.product_category, row.customer_type, row.stage)
        if key in seen:
            raise BadRequestError("存在重复的产品／户型／阶段")
        seen.add(key)
        if row.customer_type == "enterprise":
            if row.stage != "fixed":
                raise BadRequestError("企业户使用固定阶段")
            fields = [("enterprise.rate", row.value)]
        elif row.stage == "fixed":
            raise BadRequestError("个人户请选择首日、线性或超期阶段")
        elif row.stage == "linear":
            if row.slope is None or row.boundary is None:
                raise BadRequestError("线性阶段须填写日斜率和截止天数")
            try:
                endpoint = Fraction(row.value) - row.boundary * Fraction(row.slope)
            except (ValueError, ZeroDivisionError) as exc:
                raise BadRequestError("比例及日斜率须为有效数值或分数") from exc
            if endpoint < 0:
                raise BadRequestError("线性阶段末比例不能为负数")
            fields = [
                ("coefficient.personal.start", row.value),
                ("coefficient.personal.slope", row.slope),
                ("coefficient.personal.boundary", str(row.boundary)),
            ]
        else:
            fields = [
                (
                    "coefficient.personal.first"
                    if row.stage == "first"
                    else "coefficient.personal.floor",
                    row.value,
                )
            ]
        for code, value in fields:
            normalized = validate_and_normalize(get_spec(code), value)
            items.append(
                ParamChangeItemIn(
                    param_code=code, dimension_value=row.product_category, value=normalized
                )
            )
    return items


def save(session, payload, operator, meta=None):
    from app.models import ParamDefinition

    list(session.scalars(select(ParamDefinition).order_by(ParamDefinition.id).with_for_update()))
    revision = _revision(session, lock=True)
    if payload.fingerprint != catalog(session, payload.base_date, revision)["fingerprint"]:
        raise BadRequestError("系数表已变化，请重新查询或重新导入预览")
    return param_service.submit_change(
        session,
        ParamChangeIn(
            change_type=2,
            effective_date=payload.effective_date,
            remark=payload.remark,
            items=items_for(payload.rows),
        ),
        operator=operator,
        meta=meta,
    )


def workbook(session, as_of=None):
    data = catalog(session, as_of)
    book = Workbook()
    sheet = book.active
    sheet.title = "系数参数"
    sheet.append([*HEADERS, "system_coefficient", "average_coefficient_360"])
    for row in data["rows"]:
        if row.get("readonly"):
            continue
        sheet.append(
            [row.get(k) for k in (*HEADERS, "system_coefficient", "average_coefficient_360")]
        )
    note = book.create_sheet("填写说明")
    note.append(["基准日期", data["as_of"]])
    for line in (
        "value为‰数；线性阶段value为公式起点，slope为每日递减‰数，可填写分数。",
        "产品：municipal_water市政水利／rail_transit轨道交通；户型personal个人户／enterprise企业户。",
        "阶段：first首日／linear线性／after超期／fixed企业固定；经理个人户不单笔计提，不导入。",
        data["average_rule"],
        "系统系数及平均系数是只读派生列，导入时可留空，填写则必须与参数一致。",
    ):
        note.append([line])
    for column in sheet.columns:
        sheet.column_dimensions[column[0].column_letter].width = 28
    sheet.freeze_panes = "A2"
    output = BytesIO()
    book.save(output)
    return output.getvalue()


def import_preview(session, content):
    if len(content) > 2 * 1024 * 1024:
        raise BadRequestError("系数文件不能超过2MB")
    try:
        with ZipFile(BytesIO(content)) as archive:
            if sum(item.file_size for item in archive.infolist()) > 20 * 1024 * 1024:
                raise BadRequestError("工作簿解压后超过20MB，请使用系数模板")
        book = load_workbook(BytesIO(content), read_only=True, data_only=False)
        sheet = book["系数参数"]
        if sheet.max_row > 30 or sheet.max_column > 8:
            raise BadRequestError("系数表最多16条数据")
        values = list(sheet.iter_rows(values_only=True))
        if tuple(values[0][:6]) != HEADERS:
            raise BadRequestError("请使用系数表模板，保留前六列列名")
        rows = []
        errors = []
        for number, value in enumerate(values[1:], 2):
            if not any(v is not None for v in value):
                continue
            try:
                raw = {
                    k: str(v) if v is not None else None
                    for k, v in zip(HEADERS, value[:6], strict=True)
                }
                raw["boundary"] = int(raw["boundary"]) if raw["boundary"] else None
                row = CoefficientRowIn.model_validate(raw)
                items_for([row])
                rows.append(row)
            except (ValidationError, ValueError, BadRequestError) as exc:
                errors.append({"row": number, "message": str(exc)[:180]})
        if not rows and not errors:
            errors.append({"row": 2, "message": "文件没有可导入数据"})
        if rows:
            items_for(rows)
        # 派生列只做校验，不作为第二套输入。用临时参数快照复算。
        base = date.fromisoformat(str(book["填写说明"].cell(1, 2).value))
        snap = param_service.resolve_snapshot(session, base)
        temporary = {code: dict(dims) for code, dims in snap._values.items()}
        for item in items_for(rows):
            temporary.setdefault(item.param_code, {})[item.dimension_value] = item.value
        candidate = param_service.ParamSnapshot(base, temporary, {})
        for number, raw in enumerate(values[1:], 2):
            if len(raw) < 7:
                continue
            matching = next(
                (
                    r
                    for r in rows
                    if (r.product_category, r.customer_type, r.stage) == tuple(raw[:3])
                ),
                None,
            )
            if matching is None:
                continue
            pv = personal_values(candidate, matching.product_category)
            day = {"first": 1, "linear": 2, "after": int(Fraction(pv["boundary"])) + 1, "fixed": 1}[
                matching.stage
            ]
            rate = decide(
                {
                    **pv,
                    "enterprise": candidate.get(
                        "enterprise.rate", product_category=matching.product_category
                    ),
                },
                category=matching.product_category,
                customer_type=matching.customer_type,
                role_code="salesperson",
                days=day,
            )[0]
            if raw[6] not in (None, "") and (rate == 0 or Fraction(str(raw[6])) != 1 / rate):
                errors.append(
                    {"row": number, "message": "系统系数与公式参数不一致，请清空派生列或重新导出"}
                )
            if len(raw) > 7 and raw[7] not in (None, ""):
                if matching.customer_type == "enterprise":
                    average = 1 / rate if rate else None
                else:
                    daily = [
                        decide(
                            {**pv, "enterprise": "0"},
                            category=matching.product_category,
                            customer_type="personal",
                            role_code="salesperson",
                            days=d,
                        )[0]
                        for d in range(1, 361)
                    ]
                    average = sum((1 / r for r in daily), Fraction(0)) / 360 if all(daily) else None
                if average is None or Decimal(str(raw[7])) != Decimal(_average_text(average)):
                    errors.append({"row": number, "message": "360天平均系数与公式参数不一致"})
        return {
            "rows": [r.model_dump() for r in rows],
            "errors": errors,
            "can_apply": not errors,
            "fingerprint": catalog(session, base)["fingerprint"],
            "base_date": str(base),
        }
    except BadRequestError:
        raise
    except Exception as exc:
        raise BadRequestError("无法读取系数表，请检查工作表名称和文件格式") from exc
    finally:
        if "book" in locals():
            book.close()