diff --git a/backend/app/services/calc_service.py b/backend/app/services/calc_service.py index 13424f7..1c386d3 100644 --- a/backend/app/services/calc_service.py +++ b/backend/app/services/calc_service.py @@ -78,14 +78,17 @@ def calc_receipt( ) -> CalcReceiptOut: """单笔回款核算 / 重算（recalc_reason 非 None 时为重算模式）。 - 防重前两层在本函数（FOR UPDATE 锁行 + settlement_status 校验），第三层 - 是落台账 uk(receipt_id, calc_version)。全程一个 audit 事务，出参在 - commit 前固化。 + 防重前两层在 ``_load_and_guard``（FOR UPDATE 锁行 + settlement_status + 与月结锁定校验），第三层是落台账 uk(receipt_id, calc_version)。全程 + 一个 audit 事务，出参在 commit 前固化。 首算模式：回款必须未核算；业务级一次性计提（#25）——已有生效台账行时 - 仅登记进度（repeated=True）。 + 仅登记进度（repeated=True，``_prepare_mode`` 内提前返回）。 重算模式：回款必须已核算过；作废旧版本行（is_active=0），按当前参数 落 calc_version+1 新行，calc_reason 留痕。 + + 本函数只做编排：锁行校验 → 模式预处理 → 输入装载 → 逐行折算 → + 计提组成 → 落库，各段独立私有函数（口径修改只动对应段）。 """ is_recalc = recalc_reason is not None action = "重算" if is_recalc else "核算" @@ -103,212 +106,321 @@ def calc_receipt( ), meta=meta, ): - receipt = session.get(Receipt, receipt_id, with_for_update=True) - if receipt is None: - raise BadRequestError("回款不存在") - if is_recalc: - if receipt.settlement_status == SETTLEMENT_PENDING: - raise BadRequestError("该回款尚未核算过（无旧版本可重算，走首次核算）") - elif receipt.settlement_status == SETTLEMENT_DONE: - raise BadRequestError("该回款已核算，禁止重复核算（重算请走重算入口）") + receipt, business = _load_and_guard(session, receipt_id, is_recalc=is_recalc) + repeated_out = _prepare_mode(session, receipt, business, is_recalc=is_recalc) + if repeated_out is not None: + return repeated_out - # 月结锁定拦截：已审定月份的回款不得再核算/重算——金额已固化， - # 变动唯一通道是调整单（否则重算会绕开审批改写已锁定月份的应付） - from app.services import settlement_service + shipment = _load_shipment(session, receipt) + person = _load_person(session, business) + snapshot = resolve_snapshot(session, receipt.receipt_date) + items = _load_items(session, business) - settlement_service.ensure_month_not_locked(session, receipt.receipt_date.strftime("%Y-%m")) - - 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("业务未指定负责营销人员，无法核算（先在业务归属中指定）") - - # 重算：先作废旧版本行（is_active=0），折算体随后按当前参数重算。 - # 无生效行 = 该回款是「进度登记」（业务级一次性计提 #25 的第二笔及 - # 以后）——本无独立计提，继续重算会凭空落一行生效行，同业务出现 - # 两行生效行导致双重计提 → 一律 400 - if is_recalc: - old = performance_ledger_repository.get_active_by_receipt(session, receipt_id) - if old is None: - raise BadRequestError( - "该回款为进度登记（业务已按一次性计提核算），无独立台账行可重算" - ) - 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( - 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="核算归属" + conversion = _convert_items(snapshot, person, items) + amounts = _compose_amounts( + session, snapshot, person, business, receipt, shipment, conversion ) - 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 对整数值无小数位， - # 快照/台账口径需与列一致；显式 HALF_UP——Python 默认 - # 银行家舍入会让 x.xx5 少 1 分） - "converted": str( - row_converted.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) - ), - } - ) - - # 天数只依赖发货/回款日期，先于比例分派算出（业务员递减公式的输入） - 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/序列化三方一致。 - - 显式传 rounding——quantize 缺省是上下文 ROUND_HALF_EVEN（银行家 - 舍入），x.xx5 会舍成偶数值，与业务四舍五入口径差 1 分。 - """ - if value is None: - return None - return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) - - row = performance_ledger_repository.add( + row = _persist_ledger( 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, - 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, - calc_reason=recalc_reason, - ), + receipt=receipt, + business=business, + person=person, + shipment=shipment, + items=items, + conversion=conversion, + amounts=amounts, + recalc_reason=recalc_reason, + is_recalc=is_recalc, ) - # 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() - logger.info("核算完成 | receipt=%s | ledger=%s | payable=%s", receipt_id, row.id, payable) + logger.info( + "核算完成 | receipt=%s | ledger=%s | payable=%s", + receipt_id, + row.id, + amounts.payable, + ) return out +def _load_and_guard( + session: Session, receipt_id: int, *, is_recalc: bool +) -> tuple[Receipt, Business]: + """锁行 + 模式校验 + 月结锁定拦截 + 业务归属校验（核算/重算共用）。""" + receipt = session.get(Receipt, receipt_id, with_for_update=True) + if receipt is None: + raise BadRequestError("回款不存在") + if is_recalc: + if receipt.settlement_status == SETTLEMENT_PENDING: + raise BadRequestError("该回款尚未核算过（无旧版本可重算，走首次核算）") + elif receipt.settlement_status == SETTLEMENT_DONE: + raise BadRequestError("该回款已核算，禁止重复核算（重算请走重算入口）") + + # 月结锁定拦截：已审定月份的回款不得再核算/重算——金额已固化， + # 变动唯一通道是调整单（否则重算会绕开审批改写已锁定月份的应付） + from app.services import settlement_service + + settlement_service.ensure_month_not_locked(session, receipt.receipt_date.strftime("%Y-%m")) + + 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("业务未指定负责营销人员，无法核算（先在业务归属中指定）") + return receipt, business + + +def _prepare_mode( + session: Session, receipt: Receipt, business: Business, *, is_recalc: bool +) -> CalcReceiptOut | None: + """模式预处理：重算作废旧版本；首算判业务级一次性计提。 + + 返回非 None = 本笔仅登记进度（repeated=True 的出参，已 commit）， + 调用方直接返回。 + """ + # 重算：先作废旧版本行（is_active=0），折算体随后按当前参数重算。 + # 无生效行 = 该回款是「进度登记」（业务级一次性计提 #25 的第二笔及 + # 以后）——本无独立计提，继续重算会凭空落一行生效行，同业务出现 + # 两行生效行导致双重计提 → 一律 400 + if is_recalc: + old = performance_ledger_repository.get_active_by_receipt(session, receipt.id) + if old is None: + raise BadRequestError("该回款为进度登记（业务已按一次性计提核算），无独立台账行可重算") + performance_ledger_repository.mark_inactive(session, old) + return None + + # 核算粒度 #25（一期口径，仅首算模式）：业务级一次性计提——已有生效 + # 台账行时，本笔回款仅置已核算状态（支付进度），不重复计提 + if not performance_ledger_repository.has_active_for_business( + session, business.id, calc_type=CALC_TYPE_RECEIPT + ): + return None + _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 + + +def _load_shipment(session: Session, receipt: Receipt) -> Shipment: + """核销基准：发货批次存在且带发货日（回款天数公式的输入）。""" + 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("发货批次缺少发货日期，无法核算") + return shipment + + +def _load_person(session: Session, business: Business): + """核算归属：在职 + 绑定岗位角色（计提公式锚点）。""" + person = personnel_service.ensure_active(session, business.owner_personnel_id, label="核算归属") + if person.business_role_id is None: + raise BadRequestError("归属人员未绑定岗位角色，无法确定计提公式") + return person + + +def _load_items(session: Session, business: Business) -> list[BusinessItem]: + """业务产品明细行（逐行折算的输入；空明细无法核算）。""" + items = business_repository.list_items(session, business.id) + if not items: + raise BadRequestError("业务无产品明细行，无法核算") + return items + + +@dataclass(frozen=True, slots=True) +class _Conversion: + """逐行折算结果（行级明细入台账快照，汇总值驱动计提与起付判定）。""" + + total_revenue: Decimal + total_profit: Decimal + converted_total: Decimal + missing_cost: bool + line_details: list[dict[str, Any]] + + +def _convert_items(snapshot, person, items: list[BusinessItem]) -> _Conversion: + """逐行折算（行级 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(_q2(row_converted)), + } + ) + return _Conversion(total_revenue, total_profit, converted_total, missing_cost, line_details) + + +@dataclass(frozen=True, slots=True) +class _Amounts: + """计提组成结果（比例分派 → 绩效 → 抵顶 → 支付权重 → 应付）。""" + + days: int + decision: _RateDecision + converted: Decimal + performance: Decimal + barter_applied: bool + weight_r: int + weight: Fraction + payable: Decimal + start_ok: bool + + +def _compose_amounts( + session: Session, + snapshot, + person, + business: Business, + receipt: Receipt, + shipment: Shipment, + conversion: _Conversion, +) -> _Amounts: + """按比例分派与支付权重组成应付（与 param_service.preview 同公式路径）。""" + # 天数只依赖发货/回款日期，先于比例分派算出（业务员递减公式的输入） + days = payment_days(shipment.ship_date, receipt.receipt_date) + decision = _personal_rate(session, snapshot, person, business, days) + + if decision.base_override is not None: + converted = decision.base_override # 信息员：基数=本人销售额 + else: + converted = conversion.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, conversion.total_revenue, start_ratio) + weight_r = clamp_weight_r(received_total, conversion.total_revenue) + weight = payment_weight(weight_r) + payable = param_service.to_decimal(Fraction(base_amount) * weight) + return _Amounts( + days=days, + decision=decision, + converted=converted, + performance=performance, + barter_applied=barter_applied, + weight_r=weight_r, + weight=weight, + payable=payable, + start_ok=start_ok, + ) + + +def _persist_ledger( + session: Session, + *, + receipt: Receipt, + business: Business, + person, + shipment: Shipment, + items: list[BusinessItem], + conversion: _Conversion, + amounts: _Amounts, + recalc_reason: str | None, + is_recalc: bool, +) -> PerformanceLedger: + """落台账行（版本号自增；金额两位 HALF_UP）+ 首算 CAS 置已核算。""" + receipt_date = receipt.receipt_date + snapshot_json: dict[str, Any] = { + "params_as_of": str(receipt_date), + "line_details": conversion.line_details, + "start_payment_ok": amounts.start_ok, + "rate_note": amounts.decision.rate_note, + "revenue_total": str(conversion.total_revenue), + "profit_total": str(conversion.total_profit), + } + if amounts.decision.manager_pending: + snapshot_json["strategy_note"] = _NOTE_MANAGER_PENDING + + row = performance_ledger_repository.add( + 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, + 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=amounts.days, + rate_text=str(amounts.decision.rate), + converted_profit=_q2(amounts.converted), + performance_amount=_q2(amounts.performance), + barter_applied=1 if amounts.barter_applied else 0, + weight_r=amounts.weight_r, + payment_weight=param_service.to_decimal(amounts.weight), + payable_amount=_q2(amounts.payable), + settlement_month=receipt_date.strftime("%Y-%m"), + is_probation=1 if person.is_probation == 1 else 0, + missing_cost=1 if conversion.missing_cost else 0, + snapshot_json=snapshot_json, + calc_reason=recalc_reason, + ), + ) + # CAS 仅首算执行：重算模式状态本就是已核算（UPDATE ... WHERE status=0 + # 会 rowcount=0 误报并发撞车） + if not is_recalc: + _mark_settled(session, receipt) + return row + + +def _q2(value: Decimal | None) -> Decimal | None: + """金额统一两位小数（HALF_UP）：内存/DB/序列化三方一致。 + + 显式传 rounding——quantize 缺省是上下文 ROUND_HALF_EVEN（银行家舍入）， + x.xx5 会舍成偶数值，与业务四舍五入口径差 1 分。 + """ + if value is None: + return None + return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) + + def _personal_rate( session: Session, snapshot, diff --git a/backend/tests/unit/test_calc_helpers.py b/backend/tests/unit/test_calc_helpers.py new file mode 100644 index 0000000..d383415 --- /dev/null +++ b/backend/tests/unit/test_calc_helpers.py @@ -0,0 +1,28 @@ +"""核算落库辅助纯函数测试。 + +守的是：金额两位固化的舍入模式 —— quantize 缺省是上下文 ROUND_HALF_EVEN +（银行家舍入），x.xx5 会舍成偶数尾数，与业务四舍五入口径差 1 分 +（docstring 声称 HALF_UP，必须用测试锁死实际行为）。 +""" + +from decimal import Decimal + +from app.services.calc_service import _q2 + + +class TestQ2: + """金额两位小数（HALF_UP）。""" + + def test_半进位一律入(self) -> None: + # 缺省 ROUND_HALF_EVEN 会舍成 2.34（4 是偶数） + assert _q2(Decimal("2.345")) == Decimal("2.35") + + def test_负数半进位远离零(self) -> None: + assert _q2(Decimal("-2.345")) == Decimal("-2.35") + + def test_普通进位不变(self) -> None: + assert _q2(Decimal("39.111111")) == Decimal("39.11") + assert _q2(Decimal("39.115000")) == Decimal("39.12") + + def test_None直通(self) -> None: + assert _q2(None) is None