diff --git a/backend/app/api/v1/calc.py b/backend/app/api/v1/calc.py index 9162e19..3948b44 100644 --- a/backend/app/api/v1/calc.py +++ b/backend/app/api/v1/calc.py @@ -8,9 +8,10 @@ from typing import Annotated from fastapi import APIRouter, Depends, Path, Query +from pydantic import BaseModel, Field from app.api.deps import DbSession, RequestMetaDep, require_roles -from app.core.constants import ROLE_ADMIN, ROLE_MARKETING_OPS +from app.core.constants import ROLE_ADMIN, ROLE_FINANCE, ROLE_MARKETING_OPS from app.core.response import ApiResponse, PageResult, ok from app.models import User from app.schemas.calc import CalcReceiptOut, LedgerOut, LedgerQuery @@ -19,6 +20,8 @@ from app.services import calc_service router = APIRouter(prefix="/calc", tags=["绩效工资 · 绩效核算"]) _CalcWriter = Annotated[User, Depends(require_roles(ROLE_ADMIN, ROLE_MARKETING_OPS))] +# 重算是授权动作：管理员/财务（营销后勤不可重算） +_RecalcUser = Annotated[User, Depends(require_roles(ROLE_ADMIN, ROLE_FINANCE))] @router.get("/ledger", response_model=ApiResponse[PageResult[LedgerOut]], summary="绩效台账") @@ -55,3 +58,50 @@ def calc_receipt( calc_service.calc_receipt(session, receipt_id, operator=current_user, meta=meta), message="核算完成", ) + + +class RecalcIn(BaseModel): + """重算入参：原因必填（重算授权留痕）。""" + + reason: str = Field(min_length=1, max_length=255, description="重算原因") + + +class BatchIn(BaseModel): + """批量核算入参。""" + + receipt_ids: list[int] = Field(min_length=1, description="回款 ID 列表") + + +@router.post( + "/receipts/{receipt_id}/recalc", + response_model=ApiResponse[CalcReceiptOut], + summary="重算回款", +) +def recalc_receipt( + receipt_id: Annotated[int, Path(ge=1)], + payload: RecalcIn, + session: DbSession, + current_user: _RecalcUser, + meta: RequestMetaDep, +) -> ApiResponse[CalcReceiptOut]: + """重算（管理员/财务）：作废旧版本、按当前参数落新版本并留痕。""" + return ok( + calc_service.recalc_receipt( + session, receipt_id, payload.reason, operator=current_user, meta=meta + ), + message="重算完成", + ) + + +@router.post("/batch", response_model=ApiResponse[list[dict]], summary="批量核算") +def batch_calc( + payload: BatchIn, + session: DbSession, + current_user: _CalcWriter, + meta: RequestMetaDep, +) -> ApiResponse[list[dict]]: + """批量核算：逐笔独立事务，返回逐笔成败清单（失败不中断其它笔）。""" + return ok( + calc_service.batch_calc(session, payload.receipt_ids, operator=current_user, meta=meta), + message="批量核算完成", + ) diff --git a/backend/app/repositories/performance_ledger_repository.py b/backend/app/repositories/performance_ledger_repository.py index 1ca9115..137dbe9 100644 --- a/backend/app/repositories/performance_ledger_repository.py +++ b/backend/app/repositories/performance_ledger_repository.py @@ -74,13 +74,13 @@ def mark_inactive(session: Session, row: PerformanceLedger) -> None: def next_version(session: Session, receipt_id: int) -> int: - """某回款的下一个核算版本号（max+1；无行则 1）。""" + """某回款的下一个核算版本号（max+1；无行时 1）。""" current = session.execute( select(func.max(PerformanceLedger.calc_version)).where( PerformanceLedger.receipt_id == receipt_id ) ).scalar_one() - return int(current) + 1 + return int(current or 0) + 1 def list_ledger(session: Session, query: LedgerQuery) -> list[PerformanceLedger]: diff --git a/backend/app/services/calc_service.py b/backend/app/services/calc_service.py index dd05dfe..7c880eb 100644 --- a/backend/app/services/calc_service.py +++ b/backend/app/services/calc_service.py @@ -74,31 +74,42 @@ def calc_receipt( *, operator: User, meta: RequestMeta | None = None, + recalc_reason: str | None = None, ) -> CalcReceiptOut: - """单笔回款核算：回款 → 发货批次 → 业务 → 明细行逐行折算 → 台账固化。 + """单笔回款核算 / 重算（recalc_reason 非 None 时为重算模式）。 - 防重前两层在本函数（FOR UPDATE 锁行 + settlement_status=0 校验）， - 第三层是 CAS UPDATE（rowcount=0 即并发撞车 400）。全程一个 audit - 事务（MODULE_CALC），出参在 commit 前固化。 + 防重前两层在本函数（FOR UPDATE 锁行 + settlement_status 校验），第三层 + 是落台账 uk(receipt_id, calc_version)。全程一个 audit 事务，出参在 + commit 前固化。 - :raises BadRequestError: 已核算 / 未指定发货批次 / 归属未指定 / - 无明细行 / 并发撞车 + 首算模式：回款必须未核算；业务级一次性计提（#25）——已有生效台账行时 + 仅登记进度（repeated=True）。 + 重算模式：回款必须已核算过；作废旧版本行（is_active=0），按当前参数 + 落 calc_version+1 新行，calc_reason 留痕。 """ + is_recalc = recalc_reason is not None + action = "重算" if is_recalc else "核算" with audit_service.audit( session, operator=operator, record=AuditRecord( module=MODULE_CALC, - action="核算", - summary=f"核算回款 #{receipt_id}", - request_data={"receipt_id": receipt_id}, + action=action, + summary=f"{action}回款 #{receipt_id}", + request_data={ + "receipt_id": receipt_id, + **({"reason": recalc_reason} if is_recalc else {}), + }, ), meta=meta, ): receipt = session.get(Receipt, receipt_id, with_for_update=True) if receipt is None: raise BadRequestError("回款不存在") - if receipt.settlement_status == SETTLEMENT_DONE: + if is_recalc: + if receipt.settlement_status == SETTLEMENT_PENDING: + raise BadRequestError("该回款尚未核算过（无旧版本可重算，走首次核算）") + elif receipt.settlement_status == SETTLEMENT_DONE: raise BadRequestError("该回款已核算，禁止重复核算（重算请走重算入口）") business = session.get(Business, receipt.business_id, with_for_update=True) @@ -107,11 +118,19 @@ def calc_receipt( if business.owner_personnel_id is None: raise BadRequestError("业务未指定负责营销人员，无法核算（先在业务归属中指定）") - # 核算粒度 #25（一期口径）：业务级一次性计提——已有生效台账行时， - # 本笔回款仅置已核算状态（支付进度），不重复计提 - repeated = performance_ledger_repository.has_active_for_business( - session, business.id, calc_type=CALC_TYPE_RECEIPT - ) + # 重算：先作废旧版本行（is_active=0），折算体随后按当前参数重算 + if is_recalc: + old = performance_ledger_repository.get_active_by_receipt(session, receipt_id) + if old is not None: + performance_ledger_repository.mark_inactive(session, old) + + # 核算粒度 #25（一期口径，仅首算模式）：业务级一次性计提——已有生效 + # 台账行时，本笔回款仅置已核算状态（支付进度），不重复计提 + repeated = False + if not is_recalc: + repeated = performance_ledger_repository.has_active_for_business( + session, business.id, calc_type=CALC_TYPE_RECEIPT + ) if repeated: _mark_settled(session, receipt) row = performance_ledger_repository.get_active_by_receipt_of_business( @@ -232,6 +251,7 @@ def calc_receipt( session, PerformanceLedger( receipt_id=receipt.id, + calc_version=performance_ledger_repository.next_version(session, receipt.id), calc_type=CALC_TYPE_RECEIPT, business_id=business.id, business_no=business.business_no, @@ -254,9 +274,13 @@ def calc_receipt( is_probation=1 if person.is_probation == 1 else 0, missing_cost=1 if missing_cost else 0, snapshot_json=snapshot_json, + calc_reason=recalc_reason, ), ) - _mark_settled(session, receipt) + # CAS 仅首算执行：重算模式状态本就是已核算（UPDATE ... WHERE status=0 + # 会 rowcount=0 误报并发撞车） + if not is_recalc: + _mark_settled(session, receipt) session.flush() out = CalcReceiptOut(ledger=_to_out(row), repeated=False) session.commit() @@ -358,6 +382,59 @@ def _row_profit(revenue: Decimal, cost: Decimal) -> Decimal: return revenue - cost +def recalc_receipt( + session: Session, + receipt_id: int, + reason: str, + *, + operator: User, + meta: RequestMeta | None = None, +) -> CalcReceiptOut: + """重算（授权动作，管理员/财务）：作废旧版本行、按当前参数重算新版本。 + + 薄委托 calc_receipt(recalc_reason=...)——折算体只有一份；所属月已审定 + 的拦截在月结（T5）接入后补入本函数。 + """ + if not reason or not reason.strip(): + raise BadRequestError("重算必须填写原因") + return calc_receipt( + session, receipt_id, operator=operator, meta=meta, recalc_reason=reason.strip() + ) + + +def batch_calc( + session: Session, + receipt_ids: list[int], + *, + operator: User, + meta: RequestMeta | None = None, +) -> list[dict]: + """批量核算：逐笔独立事务（各笔自带 audit），返回逐笔成败清单。 + + **不开外层 audit**——外层包 audit 会违反嵌套禁令，且单笔失败不应 + 回滚其它笔的核算结果。 + """ + results: list[dict] = [] + for rid in receipt_ids: + try: + out = calc_receipt(session, rid, operator=operator, meta=meta) + results.append( + { + "receipt_id": rid, + "ok": True, + "repeated": out.repeated, + "ledger_id": out.ledger.id, + } + ) + except Exception as exc: + results.append( + {"receipt_id": rid, "ok": False, "error": f"{type(exc).__name__}: {exc}"} + ) + ok_count = sum(1 for r in results if r["ok"]) + logger.info("批量核算 | 总数=%s 成功=%s", len(results), ok_count) + return results + + def list_ledger_page(session: Session, query): """分页查询台账（api 层转发）。""" rows = performance_ledger_repository.list_ledger(session, query) diff --git a/backend/tests/integration/test_calc.py b/backend/tests/integration/test_calc.py index d4597cb..06b3865 100644 --- a/backend/tests/integration/test_calc.py +++ b/backend/tests/integration/test_calc.py @@ -119,6 +119,14 @@ def _calc(client: TestClient, headers: dict, receipt_id: int): return client.post(f"{_CALC}/receipts/{receipt_id}/calc", headers=headers) +@pytest.fixture +def finance(make_user, auth_headers) -> dict: + """财务角色账号（重算授权角色之一）。""" + user = make_user(["finance"]) + user["headers"] = auth_headers(user) + return user + + _GOLD_ITEMS = [ { "product_name": "t_市政_金标", @@ -141,7 +149,7 @@ class TestGoldenStandard: """金标准：双类别混合业务复算到分（Decimal 精确断言）。""" def test_复算到分_全字段固化(self, client: TestClient, admin: dict) -> None: - biz, receipt = _create_full_chain( + _biz, receipt = _create_full_chain( client, admin["headers"], items=_GOLD_ITEMS, @@ -185,7 +193,7 @@ class TestGoldenStandard: assert detail["settlement_status"] == 1 def test_已核算回款再核算400(self, client: TestClient, admin: dict) -> None: - biz, receipt = _create_full_chain( + _biz, receipt = _create_full_chain( client, admin["headers"], items=_GOLD_ITEMS, @@ -288,7 +296,7 @@ class TestGuards: "cost_amount": None, } ] - biz, receipt = _create_full_chain( + _biz, receipt = _create_full_chain( client, admin["headers"], items=items, @@ -315,7 +323,7 @@ class TestGuards: "cost_amount": "9000.00", } ] - biz, receipt = _create_full_chain( + _biz, receipt = _create_full_chain( client, admin["headers"], items=items, @@ -333,7 +341,7 @@ class TestGuards: def test_准入_normal403_marketing_ops可用( self, client: TestClient, admin: dict, normal: dict, make_user, auth_headers ) -> None: - biz, receipt = _create_full_chain( + _biz, receipt = _create_full_chain( client, admin["headers"], items=_GOLD_ITEMS, @@ -368,3 +376,132 @@ class TestGuards: assert resp.status_code == 200 rows = resp.json()["data"]["list"] assert rows and rows[0]["business_no"] == biz["business_no"] + + +class TestRecalc: + """重算：授权动作，作废旧版本、落新版本并留痕。""" + + def test_重算_新版本生效_旧版本保留( + self, client: TestClient, admin: dict, finance: dict + ) -> None: + _biz, receipt = _create_full_chain( + client, + admin["headers"], + items=_GOLD_ITEMS, + ship_date="2026-10-01", + receipt_date="2026-10-20", + amount="20000.00", + ) + assert _calc(client, admin["headers"], receipt["id"]).status_code == 200 + + fin = finance + resp = client.post( + f"{_CALC}/receipts/{receipt['id']}/recalc", + json={"reason": "成本录入修正"}, + headers=fin["headers"], + ) + assert resp.status_code == 200, resp.text + + factory = get_session_factory() + with factory() as session: + rows = ( + session.execute( + select(PerformanceLedger) + .where(PerformanceLedger.receipt_id == receipt["id"]) + .order_by(PerformanceLedger.calc_version) + ) + .scalars() + .all() + ) + assert [(r.calc_version, r.is_active) for r in rows] == [(1, 0), (2, 1)] + assert rows[1].calc_reason == "成本录入修正" + # 详情回显的是新版本 + assert ( + client.get( + f"{_CALC}/ledger", + params={"receipt_id": receipt["id"], "is_active": 1}, + headers=admin["headers"], + ).json()["data"]["list"][0]["calc_version"] + == 2 + ) + + def test_重算_原因缺失400(self, client: TestClient, admin: dict) -> None: + _biz, receipt = _create_full_chain( + client, + admin["headers"], + items=_GOLD_ITEMS, + ship_date="2026-10-01", + receipt_date="2026-10-20", + amount="100.00", + ) + assert _calc(client, admin["headers"], receipt["id"]).status_code == 200 + resp = client.post( + f"{_CALC}/receipts/{receipt['id']}/recalc", + json={"reason": " "}, + headers=admin["headers"], + ) + assert resp.status_code == 400 + + def test_未核算过_重算400(self, client: TestClient, admin: dict) -> None: + _biz, receipt = _create_full_chain( + client, + admin["headers"], + items=_GOLD_ITEMS, + ship_date="2026-10-01", + receipt_date="2026-10-20", + amount="100.00", + ) + resp = client.post( + f"{_CALC}/receipts/{receipt['id']}/recalc", + json={"reason": "无旧版本"}, + headers=admin["headers"], + ) + assert resp.status_code == 400 + assert "尚未核算" in resp.json()["message"] + + def test_重算_普通用户与营销后勤403( + self, client: TestClient, normal: dict, make_user, auth_headers + ) -> None: + ops = make_user(["marketing_ops"]) + ops["headers"] = auth_headers(ops) + for headers in (normal["headers"], ops["headers"]): + resp = client.post(f"{_CALC}/receipts/1/recalc", json={"reason": "x"}, headers=headers) + assert resp.status_code == 403 + + +class TestBatch: + """批量核算：逐笔独立事务，失败不中断。""" + + def test_批量_逐笔成败清单(self, client: TestClient, admin: dict) -> None: + biz, r1 = _create_full_chain( + client, + admin["headers"], + items=_GOLD_ITEMS, + ship_date="2026-10-01", + receipt_date="2026-10-20", + amount="20000.00", + ) + # 同业务第二笔回款：核算走 #25 重复分支（ok=True、repeated=True） + resp = client.post( + _RECEIPTS, + json={ + "business_id": biz["id"], + "receipt_date": "2026-10-25", + "amount": "100.00", + }, + headers=admin["headers"], + ) + assert resp.status_code == 200, resp.text + r2 = resp.json()["data"] + + resp = client.post( + f"{_CALC}/batch", + json={"receipt_ids": [r1["id"], r2["id"], 999999]}, + headers=admin["headers"], + ) + assert resp.status_code == 200, resp.text + results = resp.json()["data"] + by_id = {r["receipt_id"]: r for r in results} + assert by_id[r1["id"]]["ok"] is True and by_id[r1["id"]]["repeated"] is False + assert by_id[r2["id"]]["ok"] is True and by_id[r2["id"]]["repeated"] is True + assert by_id[999999]["ok"] is False