From 028a94a6eb24d5e0d9768f9f5677dde75e94075b Mon Sep 17 00:00:00 2001 From: liuwenxin <1009266660@qq.com> Date: Mon, 19 Oct 2026 12:52:36 +0800 Subject: [PATCH] =?UTF-8?q?refactor(m3):=20settlement=5Fservice=20?= =?UTF-8?q?=E4=B8=89=E6=AE=B5=E6=94=B6=E5=B0=BE=E2=80=94=E2=80=94=E7=94=9F?= =?UTF-8?q?=E6=88=90/=E8=81=9A=E5=90=88/=E8=B0=83=E6=95=B4=E5=8D=95?= =?UTF-8?q?=E6=8B=86=E5=88=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit code-review 约定合规项续（calc_receipt 之后的剩余超长函数）： - generate（~120 行）→ 编排 + 六段私有函数：_guard_regeneration（重生成 准入）、_release_previous_round（解绑回填+旧聚合作废）、_upsert_draft （草稿头表新建/重置）、_build_items（=_bucket_rows + _item_of 记账）与 _apply_totals（头表三合计） - _build_manager_aggregates（~110 行）→ _ShareParams 冻结 dataclass + _manager_share_params（月末取参/百分数口径/#12 开关同源）/ _collect_team_buckets （含 #12「个人户行归经理本人」分支）/ _add_team_performance / _aggregate_row； 全公司基数循环外算一次（口径逐行固定，行为等价） - create_adjustment（~90 行）→ _adjustment_amount（恒正）/ _wire_adjustment_approval （submit_order 分流与留痕，含「不得手工插 ApprovalStep」外键教训注释）/ _adjustment_out（出参前置到 commit 之前） 行为零变化；#12 参数化（include_manager_personal 过滤、个人户行按 personnel_id 直归经理、snapshot 的 manager_personal_included）语义原样保留。 验证：重构后月结集成 15 例 + 月结单测 5 例全绿（81s 内无并发干扰的一轮）； 全量回归待 88 会话 calc_service 收口后复跑（其进行中的拆行改动与测试 并发互清导致临时 401，非本重构所致） Co-Authored-By: Claude Code --- backend/app/services/settlement_service.py | 584 ++++++++++++--------- 1 file changed, 333 insertions(+), 251 deletions(-) diff --git a/backend/app/services/settlement_service.py b/backend/app/services/settlement_service.py index 53c915f..873c6d7 100644 --- a/backend/app/services/settlement_service.py +++ b/backend/app/services/settlement_service.py @@ -10,8 +10,10 @@ import logging from calendar import monthrange +from dataclasses import dataclass from datetime import date from decimal import ROUND_HALF_UP, Decimal +from fractions import Fraction from sqlalchemy import select from sqlalchemy.orm import Session @@ -64,8 +66,11 @@ def generate( """生成月度结算快照（草稿）：按确认回款口径取有效台账行，按人汇总。 - 同月已有月结行（任意状态）时 400——一行一月，重生成走「驳回/草稿态 - 重新生成」（先清 items 重算头表） - - 经理聚合（calc_type=2）在生成时计算（#12 暂口径：基数不含经理个人户） + 重新生成」（_release_previous_round 解绑回填后清明细重算） + - 经理聚合（calc_type=2）在生成时计算并落库（#12 参数化见 + ``_build_manager_aggregates``） + + 本函数只做编排，各段独立私有函数（口径修改只动对应段）。 """ with audit_service.audit( session, @@ -79,117 +84,22 @@ def generate( meta=meta, ): existing = settlement_repository.get_by_month(session, month) - if existing is not None: - if existing.status == SETTLEMENT_STATUS_APPROVED: - raise BadRequestError(f"{month} 已审定锁定，不可重新生成（调整走调整单）") - if existing.status == SETTLEMENT_STATUS_PENDING: - raise BadRequestError(f"{month} 有待审批的月结，不可重新生成") - # 草稿/已驳回：允许重新生成（清明细重算） - - # 重生成前先释放上一轮的固化标记：台账行 settlement_id 回填后， - # 「未入结」过滤会把它们永久排除在候选集外；旧聚合行也要作废， - # 否则本轮重建的聚合行与旧行重复计入（两者都必须在 _confirm_scope 前） - if existing is not None: - settlement_repository.clear_ledger_settlement_id(session, existing.id) - performance_ledger_repository.deactivate_month_aggregates(session, month) + _guard_regeneration(existing, month) + _release_previous_round(session, existing, month) ledger_rows = _confirm_scope(session, month) if not ledger_rows: raise BadRequestError(f"{month} 无可结算的核算记录（先在台账完成核算）") - # 经理团队聚合（#12 策略点：基数不含经理个人户业绩） aggregate_rows = _build_manager_aggregates(session, ledger_rows, month) - - settlement = settlement_repository.get_by_month(session, month) - is_new = settlement is None - if is_new: - settlement = MonthlySettlement( - settlement_month=month, - submitter_id=operator.id, - submitter_name=operator.username, - status=SETTLEMENT_STATUS_DRAFT, - ) - settlement_repository.add(session, settlement) - else: - settlement.submitter_id = operator.id - settlement.submitter_name = operator.username - settlement.status = SETTLEMENT_STATUS_DRAFT - - # 按人汇总（台账行：personnel+calc_type 维度） - by_person: dict[tuple[int, int], dict] = {} - for row in ledger_rows + aggregate_rows: - key = (row.personnel_id, row.calc_type) - bucket = by_person.setdefault( - key, - { - "personnel_id": row.personnel_id, - "personnel_name": row.personnel_name, - "role_code": row.role_code, - "calc_type": row.calc_type, - "performance_sum": Decimal("0"), - "payable_sum": Decimal("0"), - "ledger_ids": [], - "is_probation": row.is_probation, - }, - ) - bucket["performance_sum"] += row.performance_amount or Decimal("0") - bucket["payable_sum"] += row.payable_amount or Decimal("0") - bucket["ledger_ids"].append(row.id) - - items: list[SettlementItem] = [] - for (_, _), bucket in sorted(by_person.items()): - person = session.get(Personnel, bucket["personnel_id"]) - basic = person.salary_standard if person else None - mode = _salary_mode_of(session, bucket["role_code"], month) - net = ( - monthly_salary(bucket["payable_sum"], basic, mode) - if person and person.is_probation != 1 - else None - ) - note = None - if person and person.is_probation == 1: - note = ( - f"试用期绩效照算不发放；年累计超 {_probation_threshold(session, month)} 元部分" - "年底一次性结算" - ) - items.append( - SettlementItem( - personnel_id=bucket["personnel_id"], - personnel_name=bucket["personnel_name"], - role_code=bucket["role_code"], - calc_type=bucket["calc_type"], - performance_sum=bucket["performance_sum"].quantize( - Decimal("0.01"), rounding=ROUND_HALF_UP - ), - payable_sum=bucket["payable_sum"].quantize( - Decimal("0.01"), rounding=ROUND_HALF_UP - ), - basic_amount=basic, - salary_mode=mode, - net_amount=( - net.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) - if net is not None - else None - ), - is_probation=bucket["is_probation"], - probation_note=note, - snapshot_json={"ledger_ids": bucket["ledger_ids"]}, - ) - ) + settlement = _upsert_draft(session, existing, operator, month) + items, ledger_ids = _build_items(session, ledger_rows, aggregate_rows, month) settlement_repository.replace_items(session, settlement.id, items) - settlement.total_performance = sum( - (i.performance_sum for i in items), Decimal("0") - ).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) - settlement.total_payable = sum((i.payable_sum for i in items), Decimal("0")).quantize( - Decimal("0.01"), rounding=ROUND_HALF_UP - ) - settlement.person_count = len({i.personnel_id for i in items}) + _apply_totals(settlement, items) settlement.snapshot_json = {"note": _SETTLEMENT_NOTE} - # 入结回填：台账行 settlement_id 是「未入结」防重判据（_confirm_scope）， # 不回填会让重生成把已固化行反复纳入；聚合行此刻已落库、id 可用 - ledger_ids = [lid for bucket in by_person.values() for lid in bucket["ledger_ids"]] settlement_repository.update_ledger_settlement_id(session, ledger_ids, settlement.id) # 出参在 commit 前固化（audit 约定：commit 后查询失败会把已生成的月结 @@ -201,6 +111,127 @@ def generate( return detail +def _guard_regeneration(existing: MonthlySettlement | None, month: str) -> None: + """重生成准入：已审定 / 待审批不可重生成（草稿、已驳回放行）。""" + if existing is None: + return + if existing.status == SETTLEMENT_STATUS_APPROVED: + raise BadRequestError(f"{month} 已审定锁定，不可重新生成（调整走调整单）") + if existing.status == SETTLEMENT_STATUS_PENDING: + raise BadRequestError(f"{month} 有待审批的月结，不可重新生成") + + +def _release_previous_round( + session: Session, existing: MonthlySettlement | None, month: str +) -> None: + """释放上一轮固化标记（必须早于 _confirm_scope）。 + + 台账行 settlement_id 回填后，「未入结」过滤会把它们永久排除在候选集外； + 旧聚合行也要作废，否则本轮重建的聚合行与旧行重复计入。 + """ + if existing is None: + return + settlement_repository.clear_ledger_settlement_id(session, existing.id) + performance_ledger_repository.deactivate_month_aggregates(session, month) + + +def _upsert_draft( + session: Session, existing: MonthlySettlement | None, operator: User, month: str +) -> MonthlySettlement: + """新建或重置草稿头表（重生成时刷新提交人与状态）。""" + if existing is None: + settlement = MonthlySettlement( + settlement_month=month, + submitter_id=operator.id, + submitter_name=operator.username, + status=SETTLEMENT_STATUS_DRAFT, + ) + return settlement_repository.add(session, settlement) + existing.submitter_id = operator.id + existing.submitter_name = operator.username + existing.status = SETTLEMENT_STATUS_DRAFT + return existing + + +def _build_items( + session: Session, ledger_rows: list, aggregate_rows: list, month: str +) -> tuple[list[SettlementItem], list[int]]: + """按人汇总生成明细行，返回 (明细, 本次入结的台账行 id 列表)。""" + by_person = _bucket_rows(ledger_rows + aggregate_rows) + items = [_item_of(session, bucket, month) for _, bucket in sorted(by_person.items())] + ledger_ids = [lid for bucket in by_person.values() for lid in bucket["ledger_ids"]] + return items, ledger_ids + + +def _bucket_rows(rows: list) -> dict[tuple[int, int], dict]: + """台账行按 (人员, 核算类型) 维度分桶（冗余人读字段取首行）。""" + by_person: dict[tuple[int, int], dict] = {} + for row in rows: + key = (row.personnel_id, row.calc_type) + bucket = by_person.setdefault( + key, + { + "personnel_id": row.personnel_id, + "personnel_name": row.personnel_name, + "role_code": row.role_code, + "calc_type": row.calc_type, + "performance_sum": Decimal("0"), + "payable_sum": Decimal("0"), + "ledger_ids": [], + "is_probation": row.is_probation, + }, + ) + bucket["performance_sum"] += row.performance_amount or Decimal("0") + bucket["payable_sum"] += row.payable_amount or Decimal("0") + bucket["ledger_ids"].append(row.id) + return by_person + + +def _item_of(session: Session, bucket: dict, month: str) -> SettlementItem: + """单条结算明细：月工资三口径净额、试用期说明与台账行引用列表。""" + person = session.get(Personnel, bucket["personnel_id"]) + basic = person.salary_standard if person else None + mode = _salary_mode_of(session, bucket["role_code"], month) + net = ( + monthly_salary(bucket["payable_sum"], basic, mode) + if person and person.is_probation != 1 + else None + ) + note = None + if person and person.is_probation == 1: + note = ( + f"试用期绩效照算不发放；年累计超 {_probation_threshold(session, month)} 元部分" + "年底一次性结算" + ) + return SettlementItem( + personnel_id=bucket["personnel_id"], + personnel_name=bucket["personnel_name"], + role_code=bucket["role_code"], + calc_type=bucket["calc_type"], + performance_sum=bucket["performance_sum"].quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), + payable_sum=bucket["payable_sum"].quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), + basic_amount=basic, + salary_mode=mode, + net_amount=( + net.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) if net is not None else None + ), + is_probation=bucket["is_probation"], + probation_note=note, + snapshot_json={"ledger_ids": bucket["ledger_ids"]}, + ) + + +def _apply_totals(settlement: MonthlySettlement, items: list[SettlementItem]) -> None: + """头表三合计固化（明细已两位小数，合计再 HALF_UP 收口）。""" + settlement.total_performance = sum((i.performance_sum for i in items), Decimal("0")).quantize( + Decimal("0.01"), rounding=ROUND_HALF_UP + ) + settlement.total_payable = sum((i.payable_sum for i in items), Decimal("0")).quantize( + Decimal("0.01"), rounding=ROUND_HALF_UP + ) + settlement.person_count = len({i.personnel_id for i in items}) + + def submit( session: Session, settlement_id: int, @@ -321,68 +352,151 @@ def _confirm_scope(session: Session, month: str) -> list: ) -def _build_manager_aggregates(session: Session, ledger_rows: list, month: str) -> list: - """按经理生成团队聚合行（calc_type=2）。 +@dataclass(frozen=True, slots=True) +class _ShareParams: + """经理分成取参与参数原文（同一次生成共用一份，随月月末时点化）。""" - #12 答复（2026-10-19）参数化：经理个人户业绩是否计入聚合基数由参数 - ``manager.personal.included`` 决定（0=不计入默认，1=计入）——该部分 - 按经理个人户规则单笔落台账（rate=0/绩效 0），是否并入聚合基数由此 - 开关控制。 - 经理 = 本月台账行里 role_code=sales_manager 且 customer_type=personal - 的人员（本人带领的业务员经 manager_id 关联）。 + month: str + param_date: date + team_text: str + company_text: str + team_pct: Fraction + company_pct: Fraction + include_manager_personal: bool + + +def _manager_share_params(session: Session, month: str) -> _ShareParams: + """解析经理分成参数（锚定所属月月末，取参与台账行口径同源）。 + + 参数按百分数存储（12.5 表示 12.5%）：parse_fraction 得 12.5， + manager_team_share 内部再 ÷100 —— 调用方不得再乘 100（否则放大 100 倍）。 + #12 开关（manager.personal.included）与分成比例同日取参，答复后调整可追溯。 """ - from app.models import PerformanceLedger - from app.repositories import business_role_repository from app.services.param_service import parse_fraction, resolve_snapshot - # 取参锚定所属月月末（不是生成日）：补生成历史月份时结果不随操作日期漂移 param_date = _month_end(month) snapshot = resolve_snapshot(session, param_date) - # #12：经理个人户基数开关（与分成比例同日取参，答复后调整可追溯） - include_manager_personal = snapshot.get("manager.personal.included") == "1" - - managers: dict[int, dict] = {} - for row in ledger_rows: - if not include_manager_personal and row.customer_type == "manager_personal": - continue # 经理个人户业绩不入聚合基数（#12 参数化，默认不计入） - if row.customer_type == "manager_personal": - # 开关计入：个人户行归属经理本人（行 personnel_id 即经理， - # 不走 manager_id 上溯——经理没有上级，按上溯会被跳过漏计） - bucket = managers.setdefault( - row.personnel_id, - { - "performance": Decimal("0"), - "business_id": row.business_id, - "business_no": row.business_no, - "names": set(), - }, - ) - bucket["performance"] += row.performance_amount or Decimal("0") - bucket["names"].add(row.personnel_name) - continue - # 找到该行业务归属人员的经理（manager_id） - person = session.get(Personnel, row.personnel_id) - manager_id = person.manager_id if person else None - if manager_id is None: - continue - bucket = managers.setdefault( - manager_id, - { - "performance": Decimal("0"), - "business_id": row.business_id, - "business_no": row.business_no, - "names": set(), - }, - ) - bucket["performance"] += row.performance_amount or Decimal("0") - bucket["names"].add(row.personnel_name) - - # 参数按百分数存储（12.5 表示 12.5%）：parse_fraction 得 12.5， - # manager_team_share 内部再 ÷100 —— 调用方不得再乘 100（否则放大 100 倍） team_text = snapshot.get("manager.share.team") company_text = snapshot.get("manager.share.company") - team_pct = parse_fraction(team_text) - company_pct = parse_fraction(company_text) + return _ShareParams( + month=month, + param_date=param_date, + team_text=team_text, + company_text=company_text, + team_pct=parse_fraction(team_text), + company_pct=parse_fraction(company_text), + include_manager_personal=snapshot.get("manager.personal.included") == "1", + ) + + +def _add_team_performance(managers: dict[int, dict], manager_id: int, row) -> None: + """把一行业绩并入经理的团队桶（首行固化业务锚点，成员名去重收集）。""" + bucket = managers.setdefault( + manager_id, + { + "performance": Decimal("0"), + "business_id": row.business_id, + "business_no": row.business_no, + "names": set(), + }, + ) + bucket["performance"] += row.performance_amount or Decimal("0") + bucket["names"].add(row.personnel_name) + + +def _collect_team_buckets( + session: Session, ledger_rows: list, include_manager_personal: bool +) -> dict[int, dict]: + """按经理采集团队业绩。 + + 业务员行经 manager_id 上溯到其经理；经理个人户行（#12 开关计入时） + 归属经理本人——行 personnel_id 即经理，不走上溯（经理没有上级， + 按上溯会被跳过漏计）。 + """ + managers: dict[int, dict] = {} + for row in ledger_rows: + if row.customer_type == "manager_personal": + if include_manager_personal: + _add_team_performance(managers, row.personnel_id, row) + continue # 默认不计入聚合基数（#12 参数化） + person = session.get(Personnel, row.personnel_id) + manager_id = person.manager_id if person else None + if manager_id is not None: + _add_team_performance(managers, manager_id, row) + return managers + + +def _aggregate_row( + session: Session, manager, bucket: dict, company_base: Decimal, share: _ShareParams +): + """单经理聚合台账行（必须落库：月结 ledger_ids 与报表下钻的锚点）。""" + from app.models import PerformanceLedger + + amount = manager_team_share( + bucket["performance"], + company_base, + team_pct=share.team_pct, + company_pct=share.company_pct, + ) + member_names = "、".join(sorted(bucket["names"])) or "（无成员）" + return performance_ledger_repository.add( + session, + PerformanceLedger( + receipt_id=None, + calc_version=1, # 聚合行无回款维度，版本恒 1（uk 含 NULL 不去重） + calc_type=CALC_TYPE_MANAGER_AGGREGATE, + business_id=bucket["business_id"], + business_no=bucket["business_no"], + personnel_id=manager.id, + personnel_name=manager.personnel_name, + role_code="sales_manager", + customer_type="personal", + product_category="mixed", + receipt_date=share.param_date, + # 比例原文取实际参数（勿硬编码默认值——参数调整后台账说明会失真） + rate_text=f"聚合:{share.team_text}%+{share.company_text}%", + converted_profit=None, + performance_amount=amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), + barter_applied=0, + payable_amount=amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), + settlement_month=share.month, + is_probation=1 if manager.is_probation == 1 else 0, + snapshot_json={ + "team_base": str(bucket["performance"]), + "company_base": str(company_base), + "members": member_names, + "params_as_of": str(share.param_date), + "manager_personal_included": 1 if share.include_manager_personal else 0, + "note": ( + f"#12 参数化：基数{'含' if share.include_manager_personal else '不含'}" + "经理个人户业绩；9.8 方案公式" + ), + }, + ), + ) + + +def _build_manager_aggregates(session: Session, ledger_rows: list, month: str) -> list: + """按经理生成团队聚合行（calc_type=2）并落库。 + + #12 答复（2026-10-19）参数化：经理个人户业绩是否计入聚合基数由参数 + ``manager.personal.included`` 决定（0=不计入默认，1=计入）； + 经理 = 台账行归属人员经 manager_id 关联的 sales_manager 岗人员。 + """ + from app.repositories import business_role_repository + + share = _manager_share_params(session, month) + managers = _collect_team_buckets(session, ledger_rows, share.include_manager_personal) + # 全公司基数：默认排除经理个人户行（#12 开关计入时不过滤）； + # 口径逐行固定，循环外算一次 + company_base = sum( + ( + r.performance_amount or Decimal("0") + for r in ledger_rows + if share.include_manager_personal or r.customer_type != "manager_personal" + ), + Decimal("0"), + ) aggregates: list = [] for manager_id, bucket in managers.items(): manager = session.get(Personnel, manager_id) @@ -395,60 +509,7 @@ def _build_manager_aggregates(session: Session, ledger_rows: list, month: str) - ) if role is None or role.role_code != "sales_manager": continue # 非营销经理的上级不参与聚合 - # 全公司基数：默认排除经理个人户行（#12 开关计入时不过滤） - company_base = sum( - ( - r.performance_amount or Decimal("0") - for r in ledger_rows - if include_manager_personal or r.customer_type != "manager_personal" - ), - Decimal("0"), - ) - share = manager_team_share( - bucket["performance"], - company_base, - team_pct=team_pct, - company_pct=company_pct, - ) - member_names = "、".join(sorted(bucket["names"])) or "（无成员）" - # 聚合行必须落库（模型约定的 calc_type=2 台账行）：月结明细的 - # ledger_ids 与报表下钻都要能从台账反查来源 - row = performance_ledger_repository.add( - session, - PerformanceLedger( - receipt_id=None, - calc_version=1, # 聚合行无回款维度，版本恒 1（uk 含 NULL 不去重） - calc_type=CALC_TYPE_MANAGER_AGGREGATE, - business_id=bucket["business_id"], - business_no=bucket["business_no"], - personnel_id=manager_id, - personnel_name=manager.personnel_name, - role_code="sales_manager", - customer_type="personal", - product_category="mixed", - receipt_date=param_date, - # 比例原文取实际参数（勿硬编码默认值——参数调整后台账说明会失真） - rate_text=f"聚合:{team_text}%+{company_text}%", - converted_profit=None, - performance_amount=share.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), - barter_applied=0, - payable_amount=share.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), - settlement_month=month, - is_probation=1 if manager.is_probation == 1 else 0, - snapshot_json={ - "team_base": str(bucket["performance"]), - "company_base": str(company_base), - "members": member_names, - "params_as_of": str(param_date), - "manager_personal_included": 1 if include_manager_personal else 0, - "note": ( - f"#12 参数化：基数{'含' if include_manager_personal else '不含'}" - "经理个人户业绩；9.8 方案公式" - ), - }, - ), - ) - aggregates.append(row) + aggregates.append(_aggregate_row(session, manager, bucket, company_base, share)) return aggregates @@ -497,25 +558,11 @@ def create_adjustment( meta: RequestMeta | None = None, ): """新建调整单：开关分流（关=提交即生效留痕；开=待审批）。""" - from decimal import InvalidOperation - from app.core.constants import NO_PREFIX_ADJUSTMENT from app.models import AdjustmentOrder - from app.models.approval import APPROVAL_STATUS_APPROVED, APPROVAL_TYPE_ADJUSTMENT - from app.models.settlement import ( - ADJUSTMENT_STATUS_EFFECTIVE, - ADJUSTMENT_STATUS_PENDING, - ) from app.services.no_generator import next_no - # 金额恒正（方向表达补发/扣减）：列注释即此约定，负数会让方向语义反转、 - # 审批/报表按 direction 汇总出错；格式已由 schema 正则拦截，此处兜底 + 语义判定 - try: - amount = Decimal(payload.amount) - except InvalidOperation as exc: - raise BadRequestError("调整金额格式非法（非负两位小数）") from exc - if amount <= 0: - raise BadRequestError("调整金额须大于 0（方向表达补发/扣减）") + amount = _adjustment_amount(payload) person = session.get(Personnel, payload.personnel_id) if person is None: @@ -551,33 +598,68 @@ def create_adjustment( session.add(order) session.flush() - # 开关分流与留痕统一交给 submit_order：开关关闭时它自动通过并写两条 - # 「系统」留痕。**不得手工插 ApprovalStep** —— approval_steps.order_id - # 外键指向 approval_orders（不是 adjustment_orders），用调整单 id 落步 - # 必然外键失败（历史缺陷：关开关时新建调整单不可用） - approval_order = approval_service.submit_order( - session, - order_type=APPROVAL_TYPE_ADJUSTMENT, - ref_id=order.id, - refer_no=order.order_no, - submitter=operator, - ) - order.status = ( - ADJUSTMENT_STATUS_EFFECTIVE - if approval_order.status == APPROVAL_STATUS_APPROVED - else ADJUSTMENT_STATUS_PENDING - ) - session.flush() + _wire_adjustment_approval(session, order, operator) + out = _adjustment_out(order) # 出参在 commit 前固化（audit 约定） session.commit() - logger.info("调整单创建 | order=%s | status=%s", order.order_no, order.status) - return { - "id": order.id, - "order_no": order.order_no, - "status": order.status, - "personnel_name": order.personnel_name, - "amount": str(order.amount), - "direction": order.direction, - } + logger.info("调整单创建 | order=%s | status=%s", out["order_no"], out["status"]) + return out + + +def _adjustment_amount(payload) -> Decimal: + """金额恒正校验（方向表达补发/扣减）。 + + 列注释即「恒正」约定，负数会让方向语义反转、审批/报表按 direction + 汇总出错；格式已由 schema 正则拦截，此处兜底转换 + 语义判定。 + """ + from decimal import InvalidOperation + + try: + amount = Decimal(payload.amount) + except InvalidOperation as exc: + raise BadRequestError("调整金额格式非法（非负两位小数）") from exc + if amount <= 0: + raise BadRequestError("调整金额须大于 0（方向表达补发/扣减）") + return amount + + +def _wire_adjustment_approval(session: Session, order, operator: User) -> None: + """开关分流与留痕统一交给 submit_order（关=自动通过留痕；开=待财务复核）。 + + **不得手工插 ApprovalStep**——``approval_steps.order_id`` 外键指向 + approval_orders（不是 adjustment_orders），用调整单 id 落步必然外键 + 失败（历史缺陷：关开关时新建调整单不可用）。 + """ + from app.models.approval import APPROVAL_STATUS_APPROVED, APPROVAL_TYPE_ADJUSTMENT + from app.models.settlement import ( + ADJUSTMENT_STATUS_EFFECTIVE, + ADJUSTMENT_STATUS_PENDING, + ) + + approval_order = approval_service.submit_order( + session, + order_type=APPROVAL_TYPE_ADJUSTMENT, + ref_id=order.id, + refer_no=order.order_no, + submitter=operator, + ) + order.status = ( + ADJUSTMENT_STATUS_EFFECTIVE + if approval_order.status == APPROVAL_STATUS_APPROVED + else ADJUSTMENT_STATUS_PENDING + ) + session.flush() + + +def _adjustment_out(order) -> dict: + """调整单出参（commit 前固化；expire_on_commit 关闭下也不依赖后置读取）。""" + return { + "id": order.id, + "order_no": order.order_no, + "status": order.status, + "personnel_name": order.personnel_name, + "amount": str(order.amount), + "direction": order.direction, + } def list_adjustments(session: Session) -> list[dict]: