diff --git a/backend/app/api/router.py b/backend/app/api/router.py index 7e57bfb..94e25fd 100644 --- a/backend/app/api/router.py +++ b/backend/app/api/router.py @@ -15,6 +15,7 @@ from app.api.v1.business import businesses as business_businesses from app.api.v1.business import ratings as business_ratings from app.api.v1.business import receipts as business_receipts from app.api.v1.business import shipments as business_shipments +from app.api.v1.calc import router as calc_router from app.api.v1.system import approvals as system_approvals from app.api.v1.system import business_roles as system_business_roles from app.api.v1.system import logs as system_logs @@ -59,4 +60,7 @@ business_router.include_router(business_receipts.router) business_router.include_router(business_ratings.router) v1_router.include_router(business_router) +# 绩效核算（M3 ④）：/api/v1/calc/*，读登录即可、写限管理员与营销后勤 +v1_router.include_router(calc_router) + api_router.include_router(v1_router) diff --git a/backend/app/api/v1/calc.py b/backend/app/api/v1/calc.py new file mode 100644 index 0000000..9162e19 --- /dev/null +++ b/backend/app/api/v1/calc.py @@ -0,0 +1,57 @@ +"""绩效核算接口（M3 ④）：台账查询与单笔核算触发。 + +核算动作 = 把回款固化为台账行（每笔回款仅一次；重算走独立入口并授权）。 +准入：读=登录即可；写（核算/重算/批量）=管理员与营销后勤 +（权限清单待澄清 #6，答复后细化）。 +""" + +from typing import Annotated + +from fastapi import APIRouter, Depends, Path, Query + +from app.api.deps import DbSession, RequestMetaDep, require_roles +from app.core.constants import ROLE_ADMIN, ROLE_MARKETING_OPS +from app.core.response import ApiResponse, PageResult, ok +from app.models import User +from app.schemas.calc import CalcReceiptOut, LedgerOut, LedgerQuery +from app.services import calc_service + +router = APIRouter(prefix="/calc", tags=["绩效工资 · 绩效核算"]) + +_CalcWriter = Annotated[User, Depends(require_roles(ROLE_ADMIN, ROLE_MARKETING_OPS))] + + +@router.get("/ledger", response_model=ApiResponse[PageResult[LedgerOut]], summary="绩效台账") +def list_ledger( + session: DbSession, + query: Annotated[LedgerQuery, Query()], +) -> ApiResponse[PageResult[LedgerOut]]: + """分页查询绩效核算台账（默认仅生效行，可筛选历史版本）。""" + return ok(calc_service.list_ledger_page(session, query)) + + +@router.get("/ledger/{ledger_id}", response_model=ApiResponse[LedgerOut], summary="台账详情") +def get_ledger( + session: DbSession, + ledger_id: Annotated[int, Path(ge=1)], +) -> ApiResponse[LedgerOut]: + """台账行详情（含计算快照）。""" + return ok(calc_service.get_ledger(session, ledger_id)) + + +@router.post( + "/receipts/{receipt_id}/calc", + response_model=ApiResponse[CalcReceiptOut], + summary="核算回款", +) +def calc_receipt( + receipt_id: Annotated[int, Path(ge=1)], + session: DbSession, + current_user: _CalcWriter, + meta: RequestMetaDep, +) -> ApiResponse[CalcReceiptOut]: + """把回款固化为台账行（每笔回款仅一次；业务级一次性计提，重复回款仅登记进度）。""" + return ok( + calc_service.calc_receipt(session, receipt_id, operator=current_user, meta=meta), + message="核算完成", + ) diff --git a/backend/app/repositories/performance_ledger_repository.py b/backend/app/repositories/performance_ledger_repository.py new file mode 100644 index 0000000..1ca9115 --- /dev/null +++ b/backend/app/repositories/performance_ledger_repository.py @@ -0,0 +1,114 @@ +"""绩效核算台账仓储：只读写数据并 flush；commit/rollback 在服务层。""" + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.models import PerformanceLedger +from app.schemas.calc import LedgerQuery + + +def get_by_id(session: Session, ledger_id: int) -> PerformanceLedger | None: + """按主键取台账行。""" + return session.get(PerformanceLedger, ledger_id) + + +def get_active_by_receipt( + session: Session, receipt_id: int, *, calc_type: int = 1 +) -> PerformanceLedger | None: + """取某回款的当前生效核算行（is_active=1；无则 None）。""" + return session.execute( + select(PerformanceLedger).where( + PerformanceLedger.receipt_id == receipt_id, + PerformanceLedger.calc_type == calc_type, + PerformanceLedger.is_active == 1, + ) + ).scalar_one_or_none() + + +def has_active_for_business(session: Session, business_id: int, *, calc_type: int = 1) -> bool: + """该业务是否已有生效核算行（核算粒度 #25：业务级一次性计提的判定）。""" + return ( + session.execute( + select(func.count()) + .select_from(PerformanceLedger) + .where( + PerformanceLedger.business_id == business_id, + PerformanceLedger.calc_type == calc_type, + PerformanceLedger.is_active == 1, + ) + ).scalar_one() + > 0 + ) + + +def get_active_by_receipt_of_business( + session: Session, business_id: int, *, calc_type: int = 1 +) -> PerformanceLedger: + """取该业务当前生效的核算行（重复回款登记进度时回显既有台账）。""" + return ( + session.execute( + select(PerformanceLedger) + .where( + PerformanceLedger.business_id == business_id, + PerformanceLedger.calc_type == calc_type, + PerformanceLedger.is_active == 1, + ) + .order_by(PerformanceLedger.id.desc()) + ) + .scalars() + .first() + ) + + +def add(session: Session, row: PerformanceLedger) -> PerformanceLedger: + """落台账行（flush 取 id；commit 在服务层）。""" + session.add(row) + session.flush() + return row + + +def mark_inactive(session: Session, row: PerformanceLedger) -> None: + """重算时把旧版本行翻为已替代（全表唯一的受控 UPDATE 例外；不 commit）。""" + row.is_active = 0 + session.flush() + + +def next_version(session: Session, receipt_id: int) -> int: + """某回款的下一个核算版本号（max+1；无行则 1）。""" + current = session.execute( + select(func.max(PerformanceLedger.calc_version)).where( + PerformanceLedger.receipt_id == receipt_id + ) + ).scalar_one() + return int(current) + 1 + + +def list_ledger(session: Session, query: LedgerQuery) -> list[PerformanceLedger]: + """分页查询台账（筛选：月/人员/业务/生效/类型）。""" + stmt = select(PerformanceLedger).where(*_filters(query)) + stmt = stmt.order_by(PerformanceLedger.id.desc()).offset(query.offset).limit(query.page_size) + return list(session.execute(stmt).scalars()) + + +def count_ledger(session: Session, query: LedgerQuery) -> int: + """台账计数（与 list_ledger 同筛选）。""" + return int( + session.execute( + select(func.count()).select_from(PerformanceLedger).where(*_filters(query)) + ).scalar_one() + ) + + +def _filters(query: LedgerQuery) -> tuple: + filters = [] + if query.settlement_month: + filters.append(PerformanceLedger.settlement_month == query.settlement_month) + if query.personnel_id is not None: + filters.append(PerformanceLedger.personnel_id == query.personnel_id) + if query.business_id is not None: + filters.append(PerformanceLedger.business_id == query.business_id) + if query.is_active is not None: + filters.append(PerformanceLedger.is_active == query.is_active) + if query.calc_type is not None: + filters.append(PerformanceLedger.calc_type == query.calc_type) + return tuple(filters) diff --git a/backend/app/schemas/calc.py b/backend/app/schemas/calc.py new file mode 100644 index 0000000..597961a --- /dev/null +++ b/backend/app/schemas/calc.py @@ -0,0 +1,62 @@ +"""绩效核算的请求/响应模型（M3 ④）。""" + +from datetime import date, datetime +from decimal import Decimal + +from pydantic import BaseModel, ConfigDict, Field + +from app.core.response import PageQuery + + +class LedgerQuery(PageQuery): + """台账列表查询条件。""" + + settlement_month: str | None = Field(default=None, max_length=7, description="归属月 YYYY-MM") + personnel_id: int | None = Field(default=None, ge=1, description="归属人员筛选") + business_id: int | None = Field(default=None, ge=1, description="业务筛选") + is_active: int | None = Field(default=None, ge=0, le=1, description="生效标记筛选") + calc_type: int | None = Field(default=None, ge=1, le=2, description="核算类型筛选") + + +class LedgerOut(BaseModel): + """台账行响应（金额/比例一律 str）。""" + + model_config = ConfigDict(from_attributes=True) + + id: int + receipt_id: int | None = None + calc_version: int + is_active: int + calc_type: int + business_id: int + business_no: str + personnel_id: int + personnel_name: str + role_code: str + customer_type: str + product_category: str + ship_date: date | None = None + receipt_date: date + payment_days: int | None = None + rate_text: str + converted_profit: Decimal | None = None + performance_amount: Decimal | None = None + barter_applied: int + weight_r: int | None = None + payment_weight: Decimal | None = None + payable_amount: Decimal | None = None + settlement_month: str + is_probation: int + missing_cost: int + snapshot_json: dict | None = None + settlement_id: int | None = None + calc_reason: str | None = None + create_time: datetime + + +class CalcReceiptOut(BaseModel): + """单笔核算结果：台账行 + 重复计提标记（核算粒度 #25 一期口径）。""" + + ledger: LedgerOut + #: True=该业务此前已核算过，本笔回款仅登记进度、未重复计提（#25 一期口径） + repeated: bool = False diff --git a/backend/app/services/calc_service.py b/backend/app/services/calc_service.py new file mode 100644 index 0000000..dd05dfe --- /dev/null +++ b/backend/app/services/calc_service.py @@ -0,0 +1,391 @@ +"""绩效核算服务（M3 ④）：回款 → 台账的固化流水线。 + +单笔核算（calc_receipt）与计算预览（param_service.preview）**共用同一 +公式路径与取值防线**：resolve_snapshot 按回款到账日取参（裁决 #11）、按 +岗位模板分派计提比例（‰ 存储口径 ÷1000 转纯比例）、行级折算（混合业务 +按行汇总，对照表 #14）。两者的任何口径修改必须同步——预览给业务对数、 +核算才是算钱的法定固化点。 + +策略点（待业务答复，全部集中为单函数，答复后只改函数 + 按回款批量重算）： +- ``_personal_rate``：#12 经理个人户单笔落台账 rate=0/绩效 0（12.5%+5% + 聚合口径在月结时由 manager_team_share 计算）；#20 在 ``_row_profit`` +- 核算粒度 #25：业务级一次性计提——首个触发核算的回款计算全额应发， + 后续回款仅置 settlement_status=1（repeated=True），不重复计提 +""" + +import logging +from dataclasses import dataclass +from decimal import Decimal +from fractions import Fraction +from typing import Any + +from sqlalchemy import update +from sqlalchemy.orm import Session + +from app.core.constants import MODULE_CALC +from app.core.exceptions import BadRequestError +from app.models import ( + Business, + BusinessItem, + Customer, + PerformanceLedger, + Receipt, + Shipment, + User, +) +from app.models.performance_ledger import CALC_TYPE_RECEIPT +from app.models.receipt import SETTLEMENT_DONE, SETTLEMENT_PENDING +from app.repositories import business_repository, performance_ledger_repository +from app.schemas.calc import CalcReceiptOut +from app.services import audit_service, param_service, personnel_service +from app.services.audit_service import AuditRecord, RequestMeta +from app.services.formula import ( + apply_barter, + clamp_weight_r, + converted_profit, + payment_days, + payment_weight, + performance_amount, + profit_rate, + salesperson_rate, + start_payment_ok, +) +from app.services.param_service import parse_fraction, resolve_snapshot + +logger = logging.getLogger(__name__) + +# snapshot_json 里的策略点注记（答复回填时的检索锚点） +_NOTE_MANAGER_PENDING = "经理个人户单笔不计提（#12 待答复）：聚合口径在月结时由团队聚合计算" + + +@dataclass(frozen=True, slots=True) +class _RateDecision: + """单笔计提比例的分派结果。""" + + rate: Fraction # 纯比例（str() 产出分数原文如 88/9000，直接落台账 rate_text） + rate_note: str # 人读说明（入 snapshot） + base_override: Decimal | None # 非 None 时绩效基数用它（信息员=销售额） + manager_pending: bool # True=经理聚合口径，单笔落 0（#12 策略点） + + +def calc_receipt( + session: Session, + receipt_id: int, + *, + operator: User, + meta: RequestMeta | None = None, +) -> CalcReceiptOut: + """单笔回款核算：回款 → 发货批次 → 业务 → 明细行逐行折算 → 台账固化。 + + 防重前两层在本函数（FOR UPDATE 锁行 + settlement_status=0 校验）， + 第三层是 CAS UPDATE（rowcount=0 即并发撞车 400）。全程一个 audit + 事务（MODULE_CALC），出参在 commit 前固化。 + + :raises BadRequestError: 已核算 / 未指定发货批次 / 归属未指定 / + 无明细行 / 并发撞车 + """ + with audit_service.audit( + session, + operator=operator, + record=AuditRecord( + module=MODULE_CALC, + action="核算", + summary=f"核算回款 #{receipt_id}", + request_data={"receipt_id": receipt_id}, + ), + meta=meta, + ): + receipt = session.get(Receipt, receipt_id, with_for_update=True) + if receipt is None: + raise BadRequestError("回款不存在") + if receipt.settlement_status == SETTLEMENT_DONE: + raise BadRequestError("该回款已核算，禁止重复核算（重算请走重算入口）") + + business = session.get(Business, receipt.business_id, with_for_update=True) + if business is None: + raise BadRequestError("回款关联的业务不存在") + 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 + ) + if repeated: + _mark_settled(session, receipt) + row = performance_ledger_repository.get_active_by_receipt_of_business( + session, business.id + ) + session.flush() + out = CalcReceiptOut(ledger=_to_out(row), repeated=True) + session.commit() + logger.info( + "回款登记进度（业务已核算）| receipt=%s | business=%s", receipt_id, business.id + ) + return out + + if receipt.shipment_id is None: + raise BadRequestError("回款未指定发货批次，无法确定发货日期（先指定批次再核算）") + shipment = session.get(Shipment, receipt.shipment_id) + if shipment is None or shipment.ship_date is None: + raise BadRequestError("发货批次缺少发货日期，无法核算") + + person = personnel_service.ensure_active( + session, business.owner_personnel_id, label="核算归属" + ) + if person.business_role_id is None: + raise BadRequestError("归属人员未绑定岗位角色，无法确定计提公式") + + receipt_date = receipt.receipt_date + snapshot = resolve_snapshot(session, receipt_date) + items = business_repository.list_items(session, business.id) + if not items: + raise BadRequestError("业务无产品明细行，无法核算") + + # 逐行折算（行级 product_category 阈值；成本缺失行毛利按 0 计并标记） + total_revenue = Decimal("0") + total_profit = Decimal("0") + converted_total = Decimal("0") + missing_cost = False + line_details: list[dict[str, Any]] = [] + for item in items: + row_revenue = item.sale_amount + if item.cost_amount is None: + missing_cost = True + row_profit = Decimal("0") + else: + row_profit = _row_profit(row_revenue, item.cost_amount) + total_revenue += row_revenue + total_profit += row_profit + threshold = parse_fraction( + snapshot.get( + "product.threshold", + role_code=role_code_of(person), + product_category=item.product_category, + ) + ) + row_converted = converted_profit( + row_revenue, profit_rate(row_profit, row_revenue), threshold + ) + converted_total += row_converted + line_details.append( + { + "item_id": item.id, + "product_category": item.product_category, + "revenue": str(row_revenue), + "profit": str(row_profit), + # 行级折算同样固化为两位（to_decimal 对整数值无小数位， + # 快照/台账口径需与列一致） + "converted": str(row_converted.quantize(Decimal("0.01"))), + } + ) + + # 天数只依赖发货/回款日期，先于比例分派算出（业务员递减公式的输入） + days = payment_days(shipment.ship_date, receipt_date) + decision = _personal_rate(session, snapshot, person, business, days) + + if decision.base_override is not None: + converted = decision.base_override # 信息员：基数=本人销售额 + else: + converted = converted_total + + performance = ( + Decimal("0") + if decision.manager_pending + else performance_amount(converted, decision.rate) + ) + barter_applied = receipt.is_barter == 1 + base_amount = performance + if barter_applied: + discount = parse_fraction( + snapshot.get("barter.discount", role_code=role_code_of(person)) + ) + base_amount = apply_barter(performance, discount) + + received_total = business.total_received_amount or Decimal("0") + start_ratio = parse_fraction( + snapshot.get("payment.start.ratio", role_code=role_code_of(person)) + ) + start_ok = start_payment_ok(received_total, total_revenue, start_ratio) + weight_r = clamp_weight_r(received_total, total_revenue) + weight = payment_weight(weight_r) + payable = param_service.to_decimal(Fraction(base_amount) * weight) + + settlement_month = receipt_date.strftime("%Y-%m") + snapshot_json: dict[str, Any] = { + "params_as_of": str(receipt_date), + "line_details": line_details, + "start_payment_ok": start_ok, + "rate_note": decision.rate_note, + "revenue_total": str(total_revenue), + "profit_total": str(total_profit), + } + if decision.manager_pending: + snapshot_json["strategy_note"] = _NOTE_MANAGER_PENDING + + def q2(value: Decimal | None) -> Decimal | None: + """金额统一两位小数（HALF_UP）：内存/DB/序列化三方一致。""" + return value.quantize(Decimal("0.01")) if value is not None else None + + row = performance_ledger_repository.add( + session, + PerformanceLedger( + receipt_id=receipt.id, + calc_type=CALC_TYPE_RECEIPT, + business_id=business.id, + business_no=business.business_no, + personnel_id=person.id, + personnel_name=person.personnel_name, + role_code=role_code_of(person), + customer_type=_customer_type(session, business), + product_category=_category_label(items), + ship_date=shipment.ship_date, + receipt_date=receipt_date, + payment_days=days, + rate_text=str(decision.rate), + converted_profit=q2(converted), + performance_amount=q2(performance), + barter_applied=1 if barter_applied else 0, + weight_r=weight_r, + payment_weight=param_service.to_decimal(weight), + payable_amount=q2(payable), + settlement_month=settlement_month, + is_probation=1 if person.is_probation == 1 else 0, + missing_cost=1 if missing_cost else 0, + snapshot_json=snapshot_json, + ), + ) + _mark_settled(session, receipt) + session.flush() + out = CalcReceiptOut(ledger=_to_out(row), repeated=False) + session.commit() + logger.info("核算完成 | receipt=%s | ledger=%s | payable=%s", receipt_id, row.id, payable) + return out + + +def _personal_rate( + session: Session, + snapshot, + person, + business: Business, + days: int, +) -> _RateDecision: + """按岗位模板分派单笔计提比例（与 param_service._preview_rate 同构）。 + + #12 策略点：经理个人户（owner 岗位绑 TPL_MANAGER_SHARE）单笔不计提 + —— 12.5%+5% 是月度团队聚合口径，单笔落 rate=0/绩效 0 的台账行保事实。 + """ + role = _role_of(session, person) + customer_type = _customer_type(session, business) + + if customer_type == "enterprise": + # 存储口径 ‰ → 纯比例（漏换算放大 1000 倍，与预览同防线） + rate = parse_fraction(snapshot.get("enterprise.rate")) / 1000 + return _RateDecision(rate, "企业户固定计提比例", None, False) + + template = role.calc_template if role else "TPL_SALESPERSON_DECAY" + if template == "TPL_MANAGER_SHARE": + return _RateDecision( + Decimal("0"), + "经理个人户：12.5%+5% 聚合计提（月结时计算），单笔落 0", + None, + True, + ) + if template == "TPL_INFO_SALES_FIXED": + # 信息员基数=本人销售额（覆盖折算与阈值），‰ 换算同防线 + rate = parse_fraction(snapshot.get("info.rate")) / 1000 + return _RateDecision(rate, "信息员：本人销售额 × 计提比例", None, False) + if template == "TPL_SALESPERSON_DECAY": + start = parse_fraction(snapshot.get("salesperson.rate.start")) + slope = parse_fraction(snapshot.get("salesperson.rate.slope")) + floor = parse_fraction(snapshot.get("salesperson.rate.floor")) + boundary = int(snapshot.get("salesperson.days.boundary")) + rate = salesperson_rate(days, start=start, slope=slope, floor=floor, boundary_days=boundary) + note = f"业务员递减：{start}‰ 起，{days} 天（分界 {boundary} 天，超界取 {floor}‰）" + return _RateDecision(rate, note, None, False) + raise BadRequestError(f"公式模板 {template} 暂不支持核算") + + +def _role_of(session: Session, person): + from app.repositories import business_role_repository + + if person.business_role_id is None: + return None + return business_role_repository.get_by_id(session, person.business_role_id) + + +def role_code_of(person) -> str: + """人员岗位编码（台账冗余固化用；无岗位为空串——前置校验已拦）。""" + from app.repositories import business_role_repository + + if person.business_role_id is None: + return "" + session = Session.object_session(person) + role = business_role_repository.get_by_id(session, person.business_role_id) + return role.role_code if role else "" + + +def _customer_type(session: Session, business: Business) -> str: + customer = session.get(Customer, business.customer_id) + return customer.customer_type if customer else "personal" + + +def _category_label(items: list[BusinessItem]) -> str: + """单类别直出；混合业务记 mixed（行级明细在 snapshot_json）。""" + categories = {item.product_category for item in items} + return categories.pop() if len(categories) == 1 else "mixed" + + +def _mark_settled(session: Session, receipt: Receipt) -> None: + """CAS 置已核算：并发下另一事务先行置 1 时 rowcount=0 → 400。""" + result = session.execute( + update(Receipt) + .where(Receipt.id == receipt.id, Receipt.settlement_status == SETTLEMENT_PENDING) + .values(settlement_status=SETTLEMENT_DONE) + ) + if result.rowcount == 0: + raise BadRequestError("该回款已被并发核算，请刷新后查看台账") + + +def _row_profit(revenue: Decimal, cost: Decimal) -> Decimal: + """行毛利 = 行销售额 − 行成本。 + + #20 策略点（待业务答复，2026-10-16 暂定口径）：金额暂按**不含税**原样 + 相减；答复若为含税口径，只改本函数（含税换算/取不含税金额列）+ 按回款 + 批量重算回填。 + """ + return revenue - cost + + +def list_ledger_page(session: Session, query): + """分页查询台账（api 层转发）。""" + rows = performance_ledger_repository.list_ledger(session, query) + total = performance_ledger_repository.count_ledger(session, query) + from app.core.response import PageResult + from app.schemas.calc import LedgerOut + + return PageResult[LedgerOut]( + records=[LedgerOut.model_validate(r) for r in rows], + total=total, + page_num=query.page_num, + page_size=query.page_size, + ) + + +def get_ledger(session: Session, ledger_id: int): + """台账详情（api 层转发）。""" + row = performance_ledger_repository.get_by_id(session, ledger_id) + if row is None: + from app.core.exceptions import NotFoundError + + raise NotFoundError("台账记录不存在") + from app.schemas.calc import LedgerOut + + return LedgerOut.model_validate(row) + + +def _to_out(row: PerformanceLedger): + from app.schemas.calc import LedgerOut + + return LedgerOut.model_validate(row) diff --git a/backend/tests/integration/conftest.py b/backend/tests/integration/conftest.py index 43ad0e6..670f445 100644 --- a/backend/tests/integration/conftest.py +++ b/backend/tests/integration/conftest.py @@ -101,6 +101,14 @@ def _purge_test_data() -> None: session.execute(delete(UserRole).where(UserRole.role_id.in_(role_ids))) session.execute(delete(Role).where(Role.id.in_(role_ids))) + # ---- M3-core：核算台账 / 月结 / 调整单（必须先于回款/业务/人员删除—— + # ledger.receipt_id/personnel_id/settlement_id 外键 RESTRICT 会挡住后面 + # 的单据与人员清理；仅测试写入，全量清）---- + session.execute(delete(PerformanceLedger)) + session.execute(delete(SettlementItem)) + session.execute(delete(MonthlySettlement)) + session.execute(delete(AdjustmentOrder).where(AdjustmentOrder.order_no.like("TZ%"))) + # ---- M2 业务数据（T11）：业务/发货/回款三类单据 ---- # 编号系统生成、无 t_ 前缀可用，按业务编号前缀全量清（三类单据表目前 # 只有测试会写）。外键均为 RESTRICT，删除顺序不能倒：回款 → 发货 @@ -182,12 +190,6 @@ def _purge_test_data() -> None: # 审批单无 t_ 前缀可用，按 AP 编号前缀全量清（steps 外键 CASCADE 随删； # ref_id 固化无外键，业务已在上一步删除，无悬挂影响） session.execute(delete(ApprovalOrder).where(ApprovalOrder.order_no.like("AP%"))) - # ---- M3-core：核算台账 / 月结 / 调整单（仅测试写入，全量清； - # 删除顺序：台账→明细→月结头→调整单，TZ 前缀全量清）---- - session.execute(delete(PerformanceLedger)) - session.execute(delete(SettlementItem)) - session.execute(delete(MonthlySettlement)) - session.execute(delete(AdjustmentOrder).where(AdjustmentOrder.order_no.like("TZ%"))) # 开关参数恢复种子基线：测试提交的变更批次条目 id 均大于种子终值 22 # （V3 十九条 + V6 两条 + V7 调整单开关一条），删新条目与新批次后解析 # 回退到种子值（开关=0） diff --git a/backend/tests/integration/test_calc.py b/backend/tests/integration/test_calc.py new file mode 100644 index 0000000..d4597cb --- /dev/null +++ b/backend/tests/integration/test_calc.py @@ -0,0 +1,370 @@ +"""单笔回款核算集成测试（M3 ④：金标准复算到分）。 + +金标准场景（双类别混合业务，覆盖行级折算与汇总）： +- 行1 市政水利：100×100.00 = 销售 10000，成本 3000 → 毛利率 0.7，阈值 30% + → 折算 10000×(0.7−0.3) = 4000 +- 行2 轨道交通：50×200.00 = 销售 10000，成本 8000 → 毛利率 0.2，阈值 20% + → 折算 10000×(0.2−0.2) = 0 +- 折算合计 4000；发货 10-01 → 回款 10-20，days=20 +- 率 = (10 − 20×4/360)‰ = 88/9000；绩效 = 4000×88/9000 ≈ 39.11 +- 回款 20000（=销售额 100%）→ R=100、权重 1 → 应付 = 39.11 + +守的还有：核算粒度 #25（业务级一次性计提——第二笔回款仅登记进度）、 +已核算再核算 400、未指定批次 400、成本缺失行毛利按 0 计、负绩效照算、 +准入（normal 403 / marketing_ops 可用）。 +""" + +import uuid +from decimal import Decimal + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import select + +from app.core.db import get_session_factory +from app.models import PerformanceLedger, Receipt +from app.models.receipt import SETTLEMENT_DONE +from tests.integration.factories import business_payload, customer + +pytestmark = pytest.mark.integration + +_CALC = "/api/v1/calc" +_BUSINESSES = "/api/v1/business/businesses" +_SHIPMENTS = "/api/v1/business/shipments" +_RECEIPTS = "/api/v1/business/receipts" +_PERSONNEL = "/api/v1/base/personnel" + + +def _person(client: TestClient, headers: dict) -> dict: + resp = client.post( + _PERSONNEL, + json={ + "personnel_name": f"t_核算_{uuid.uuid4().hex[:8]}", + "business_role_id": _role_id("salesperson"), + "salary_standard": "3000.00", + }, + headers=headers, + ) + assert resp.status_code == 200, resp.text + return resp.json()["data"] + + +def _role_id(code: str) -> int: + from sqlalchemy import select + + from app.core.constants import BUSINESS_ROLE_SALESPERSON + from app.core.db import get_session_factory + from app.models import BusinessRole + + factory = get_session_factory() + with factory() as session: + role_code = BUSINESS_ROLE_SALESPERSON if code == "salesperson" else code + return int( + session.execute( + select(BusinessRole.id).where( + BusinessRole.role_code == role_code, BusinessRole.status == 1 + ) + ).scalar_one() + ) + + +def _create_full_chain( + client: TestClient, + headers: dict, + *, + items: list[dict], + ship_date: str, + receipt_date: str, + amount: str, +) -> tuple[dict, dict]: + """客户 → 人员 → 业务（双行明细+归属）→ 发货 → 回款（指定批次）。""" + cust = customer(client, headers) + person = _person(client, headers) + resp = client.post( + _BUSINESSES, + json=business_payload(cust["id"], owner_personnel_id=person["id"], items=items), + headers=headers, + ) + assert resp.status_code == 200, resp.text + biz = resp.json()["data"] + + resp = client.post( + _SHIPMENTS, + json={ + "business_id": biz["id"], + "ship_date": ship_date, + "items": [{"product_name": "t_发货物", "quantity": "1", "sale_price": "1.00"}], + }, + headers=headers, + ) + assert resp.status_code == 200, resp.text + shipment = resp.json()["data"] + + resp = client.post( + _RECEIPTS, + json={ + "business_id": biz["id"], + "receipt_date": receipt_date, + "amount": amount, + "shipment_id": shipment["id"], + }, + headers=headers, + ) + assert resp.status_code == 200, resp.text + receipt = resp.json()["data"] + return biz, receipt + + +def _calc(client: TestClient, headers: dict, receipt_id: int): + return client.post(f"{_CALC}/receipts/{receipt_id}/calc", headers=headers) + + +_GOLD_ITEMS = [ + { + "product_name": "t_市政_金标", + "product_category": "municipal_water", + "quantity": "100", + "sale_price": "100.00", + "cost_amount": "3000.00", + }, + { + "product_name": "t_轨交_金标", + "product_category": "rail_transit", + "quantity": "50", + "sale_price": "200.00", + "cost_amount": "8000.00", + }, +] + + +class TestGoldenStandard: + """金标准：双类别混合业务复算到分（Decimal 精确断言）。""" + + def test_复算到分_全字段固化(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="20000.00", + ) + + resp = _calc(client, admin["headers"], receipt["id"]) + assert resp.status_code == 200, resp.text + out = resp.json()["data"] + assert out["repeated"] is False + + ledger = out["ledger"] + assert ledger["payment_days"] == 20 + assert ledger["settlement_month"] == "2026-10" + # 率 = (10 − 20×4/360)‰ = 88/9000，Fraction 约分原文 11/1125 + assert ledger["rate_text"] == "11/1125" + assert ledger["weight_r"] == 100 + assert ledger["missing_cost"] == 0 + assert ledger["snapshot_json"]["line_details"][0]["converted"] == "4000.00" + + # 金额锚定 DB 事实源（NUMERIC(14,2) 读回恒两位；响应 JSON 的 Decimal + # 序列化 scale 不作断言——归前端展示问题） + factory = get_session_factory() + with factory() as session: + row = session.execute( + select(PerformanceLedger).where( + PerformanceLedger.receipt_id == receipt["id"], + PerformanceLedger.is_active == 1, + ) + ).scalar_one() + assert row.converted_profit == Decimal("4000.00") + # 绩效 = 4000 × 88/9000 = 39.111… → 落库两位 39.11 + assert row.performance_amount == Decimal("39.11") + # 应付 = 绩效 × 权重 1 = 39.11 + assert row.payable_amount == Decimal("39.11") + + # 回款状态被 CAS 置已核算 + detail = client.get(f"{_RECEIPTS}/{receipt['id']}", headers=admin["headers"]).json()["data"] + assert detail["settlement_status"] == 1 + + 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 = _calc(client, admin["headers"], receipt["id"]) + assert resp.status_code == 400 + assert "已核算" in resp.json()["message"] + + def test_业务级一次性计提_第二笔仅登记进度(self, client: TestClient, admin: dict) -> None: + """核算粒度 #25 一期口径：同业务第二笔回款不重复计提。""" + biz, first = _create_full_chain( + client, + admin["headers"], + items=_GOLD_ITEMS, + ship_date="2026-10-01", + receipt_date="2026-10-20", + amount="10000.00", + ) + assert _calc(client, admin["headers"], first["id"]).status_code == 200 + + resp = client.post( + _RECEIPTS, + json={ + "business_id": biz["id"], + "receipt_date": "2026-10-25", + "amount": "5000.00", + "shipment_id": None, + }, + headers=admin["headers"], + ) + assert resp.status_code == 200, resp.text + second = resp.json()["data"] + + resp = _calc(client, admin["headers"], second["id"]) + assert resp.status_code == 200, resp.text + out = resp.json()["data"] + assert out["repeated"] is True + + factory = get_session_factory() + with factory() as session: + rows = ( + session.execute( + select(PerformanceLedger).where( + PerformanceLedger.business_id == biz["id"], + PerformanceLedger.is_active == 1, + ) + ) + .scalars() + .all() + ) + assert len(rows) == 1, "第二笔回款不得新增台账行（一次性计提）" + receipts = ( + session.execute(select(Receipt).where(Receipt.business_id == biz["id"])) + .scalars() + .all() + ) + assert all(r.settlement_status == SETTLEMENT_DONE for r in receipts) + + +class TestGuards: + """校验链与策略点。""" + + def test_未指定批次400(self, client: TestClient, admin: dict) -> None: + cust = customer(client, admin["headers"]) + person = _person(client, admin["headers"]) + resp = client.post( + _BUSINESSES, + json=business_payload(cust["id"], owner_personnel_id=person["id"]), + headers=admin["headers"], + ) + assert resp.status_code == 200 + biz = resp.json()["data"] + resp = client.post( + _RECEIPTS, + json={ + "business_id": biz["id"], + "receipt_date": "2026-10-20", + "amount": "100.00", + }, + headers=admin["headers"], + ) + assert resp.status_code == 200 + receipt = resp.json()["data"] + + resp = _calc(client, admin["headers"], receipt["id"]) + assert resp.status_code == 400 + assert "发货批次" in resp.json()["message"] + + def test_成本缺失行_毛利按0计并标记(self, client: TestClient, admin: dict) -> None: + items = [ + { + "product_name": "t_无成本_市政", + "product_category": "municipal_water", + "quantity": "100", + "sale_price": "100.00", + "cost_amount": None, + } + ] + biz, receipt = _create_full_chain( + client, + admin["headers"], + items=items, + ship_date="2026-10-01", + receipt_date="2026-10-20", + amount="10000.00", + ) + resp = _calc(client, admin["headers"], receipt["id"]) + assert resp.status_code == 200, resp.text + ledger = resp.json()["data"]["ledger"] + # 成本缺失 → 毛利按 0 → 折算 = 销售额×(0−30%) = **−3000**（保守少算 + # 方向：缺成本的行宁少算不多发），missing_cost=1 + assert ledger["missing_cost"] == 1 + assert float(ledger["converted_profit"]) == -3000.0 + assert float(ledger["performance_amount"]) == pytest.approx(-29.33, abs=0.01) + + def test_负绩效照算不归零(self, client: TestClient, admin: dict) -> None: + items = [ + { + "product_name": "t_亏损_市政", + "product_category": "municipal_water", + "quantity": "100", + "sale_price": "100.00", + "cost_amount": "9000.00", + } + ] + biz, receipt = _create_full_chain( + client, + admin["headers"], + items=items, + ship_date="2026-10-01", + receipt_date="2026-10-20", + amount="10000.00", + ) + resp = _calc(client, admin["headers"], receipt["id"]) + assert resp.status_code == 200, resp.text + ledger = resp.json()["data"]["ledger"] + # 毛利 1000、率 0.1 < 阈值 30% → 折算 10000×(0.1−0.3) = −2000 → 负绩效 + assert float(ledger["converted_profit"]) == -2000.0 + assert float(ledger["performance_amount"]) < 0 + + def test_准入_normal403_marketing_ops可用( + self, client: TestClient, admin: dict, normal: dict, make_user, auth_headers + ) -> 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 = _calc(client, normal["headers"], receipt["id"]) + assert resp.status_code == 403 + + ops = make_user(["marketing_ops"]) + ops["headers"] = auth_headers(ops) + resp = _calc(client, ops["headers"], receipt["id"]) + assert resp.status_code == 200, resp.text + + def test_台账列表与筛选(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", + ) + _calc(client, admin["headers"], receipt["id"]) + + resp = client.get( + f"{_CALC}/ledger", + params={"settlement_month": "2026-10", "is_active": 1}, + headers=admin["headers"], + ) + assert resp.status_code == 200 + rows = resp.json()["data"]["list"] + assert rows and rows[0]["business_no"] == biz["business_no"]