PayrollManagementSystem/backend/app/services/salary_completion.py

"""完整工资按人汇总：基本工资只计一次，试用期缓发及年度释放分列。"""

from decimal import ROUND_HALF_UP, Decimal

from sqlalchemy import select

from app.models import BusinessRole, Personnel, ProbationPayment, SettlementItem
from app.services import formula, probation_service
from app.services.manager_share import month_end

ZERO = Decimal("0")


def build_items(session, rows, month):
    from app.services import personnel_salary
    from app.services.settlement_service import _salary_mode_of

    end = month_end(month)
    # 人员锁同时串行化不同工资月份对同一年度缓发余额的结清。
    people = {
        p.id: p for p in session.scalars(select(Personnel).order_by(Personnel.id).with_for_update())
    }
    roles = {r.id: r.role_code for r in session.scalars(select(BusinessRole))}
    profiles = {pid: personnel_salary.resolve(person, end) for pid, person in people.items()}
    buckets = {}

    def bucket(person):
        return buckets.setdefault(
            person.id,
            {
                "person": person,
                "performance": ZERO,
                "normal": ZERO,
                "deferred": ZERO,
                "ledger_ids": [],
                "annual": [],
                "details": [],
            },
        )

    for person in people.values():
        if (
            personnel_salary.employed(person, profiles[person.id], end)
            and profiles[person.id].salary_standard is not None
        ):
            bucket(person)
    for row in rows:
        person = people.get(row.personnel_id)
        if person is None:
            continue
        item = bucket(person)
        item["performance"] += row.performance_amount or ZERO
        key = "deferred" if row.is_probation == 1 else "normal"
        item[key] += row.payable_amount or ZERO
        item["ledger_ids"].append(row.id)
        item["details"].append(
            {
                "ledger_id": row.id,
                "calc_type": row.calc_type,
                "performance": str(row.performance_amount or ZERO),
                "payable": str(row.payable_amount or ZERO),
            }
        )
        probation_service.record(session, row)
    session.flush()
    preview = probation_service.preview(session, month, lock=True)
    for annual in preview["rows"]:
        person = people.get(annual["personnel_id"])
        if person is None or not Decimal(annual["delta"]):
            continue
        later = session.scalars(
            select(ProbationPayment).where(
                ProbationPayment.personnel_id == person.id,
                ProbationPayment.year == annual["year"],
                ProbationPayment.month > month,
            )
        )
        if any(probation_service.payment_state(session, payment, True) != "ignored" for payment in later):
            from app.core.exceptions import BadRequestError

            raise BadRequestError("该年度已有后续月份结清记录，请按月份顺序处理差额")
        bucket(person)["annual"].append(annual)
    result = []
    for pid, values in sorted(buckets.items()):
        person = values["person"]
        profile = profiles[pid]
        role = roles.get(profile.business_role_id, "salesperson")
        mode = _salary_mode_of(session, role, month)
        annual = sum((Decimal(r["delta"]) for r in values["annual"]), ZERO)
        normal = values["normal"]
        basic = profile.salary_standard if personnel_salary.employed(person, profile, end) else ZERO
        trial = probation_service.on_date(profile, end)
        net = (
            ((basic or ZERO) + normal + annual)
            if trial
            else formula.monthly_salary(normal + annual, basic, mode)
        )
        result.append(
            SettlementItem(
                personnel_id=pid,
                personnel_name=person.personnel_name,
                role_code=role,
                calc_type=2 if role == "sales_manager" else 1,
                performance_sum=values["performance"].quantize(
                    Decimal("0.01"), rounding=ROUND_HALF_UP
                ),
                payable_sum=(normal + annual).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP),
                normal_payable_sum=normal,
                deferred_sum=values["deferred"],
                probation_payment_sum=annual,
                basic_amount=basic,
                salary_mode="INCLUDE" if trial else mode,
                net_amount=net.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
                if net is not None
                else None,
                is_probation=int(trial),
                probation_note="试用期绩效进入缓发台账；本月仅发基本工资及符合条件的年度补发"
                if trial
                else None,
                snapshot_json={
                    "ledger_ids": values["ledger_ids"],
                    "calculation_details": values["details"],
                    "probation_settlements": values["annual"],
                },
            )
        )
    return result, [row.id for row in rows]


def frozen_rows(session, month, personnel_ids=None):
    from app.models import MonthlySettlement

    statement = (
        select(SettlementItem)
        .join(MonthlySettlement)
        .where(MonthlySettlement.settlement_month == month)
    )
    if personnel_ids is not None:
        statement = statement.where(SettlementItem.personnel_id.in_(personnel_ids))
    return [
        {
            "personnel_id": item.personnel_id,
            "personnel_name": item.personnel_name,
            "basic_amount": str(item.basic_amount) if item.basic_amount is not None else None,
            "normal_payable_sum": str(item.normal_payable_sum),
            "deferred_sum": str(item.deferred_sum),
            "probation_payment_sum": str(item.probation_payment_sum),
            "payable_sum": str(item.payable_sum),
            "net_amount": str(item.net_amount) if item.net_amount is not None else None,
        }
        for item in session.scalars(statement)
    ]


def export_settlement(session, month):
    from io import BytesIO

    from openpyxl import Workbook

    book = Workbook()
    sheet = book.active
    sheet.title = "工资分项"
    sheet.append(
        ["人员", "基本工资", "正常绩效", "试用期缓发", "年终及续付差额", "本月实付绩效", "净工资"]
    )
    for row in frozen_rows(session, month):
        sheet.append(
            [
                row["personnel_name"],
                *[
                    Decimal(row[key]) if row[key] is not None else None
                    for key in (
                        "basic_amount",
                        "normal_payable_sum",
                        "deferred_sum",
                        "probation_payment_sum",
                        "payable_sum",
                        "net_amount",
                    )
                ],
            ]
        )
        if sheet.cell(sheet.max_row, 1).data_type == "f":
            sheet.cell(sheet.max_row, 1).data_type = "s"
    sheet.freeze_panes = "A2"
    for col in sheet.columns:
        sheet.column_dimensions[col[0].column_letter].width = 24
    output = BytesIO()
    book.save(output)
    return output.getvalue()