"""绩效核算服务(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 ROUND_HALF_UP, 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, recalc_reason: str | None = None, ) -> CalcReceiptOut: """单笔回款核算 / 重算(recalc_reason 非 None 时为重算模式)。 防重前两层在 ``_load_and_guard``(FOR UPDATE 锁行 + settlement_status 与月结锁定校验),第三层是落台账 uk(receipt_id, calc_version)。全程 一个 audit 事务,出参在 commit 前固化。 首算模式:回款必须未核算;业务级一次性计提(#25)——已有生效台账行时 仅登记进度(repeated=True,``_prepare_mode`` 内提前返回)。 重算模式:回款必须已核算过;作废旧版本行(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=action, summary=f"{action}回款 #{receipt_id}", request_data={ "receipt_id": receipt_id, **({"reason": recalc_reason} if is_recalc else {}), }, ), meta=meta, ): 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 shipment = _load_shipment(session, receipt) person = _load_person(session, business) snapshot = resolve_snapshot(session, receipt.receipt_date) items = _load_items(session, business) conversion = _convert_items(snapshot, person, items) amounts = _compose_amounts( session, snapshot, person, business, receipt, shipment, conversion ) row = _persist_ledger( session, receipt=receipt, business=business, person=person, shipment=shipment, items=items, conversion=conversion, amounts=amounts, recalc_reason=recalc_reason, is_recalc=is_recalc, ) session.flush() out = CalcReceiptOut(ledger=_to_out(row), repeated=False) session.commit() 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, 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")) # int(参数原文) 对合法入库的 '360.0' 会抛 ValueError 使整月核算 500; # 经 Fraction 转换再取整(INT 型参数入库时已校验分母为 1,取整无损) boundary = int(parse_fraction(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 recalc_receipt( session: Session, receipt_id: int, reason: str, *, operator: User, meta: RequestMeta | None = None, ) -> CalcReceiptOut: """重算(授权动作,管理员/财务):作废旧版本行、按当前参数重算新版本。 薄委托 calc_receipt(recalc_reason=...)——折算体只有一份;所属月已审定 的拦截在 calc_receipt 内(核算与重算同口径)。 """ 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) 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)