"""业务管理业务编排（需求 2.2.1/2.2.4）。 事务边界在本层：repository 只 flush，commit/rollback 只写在这里。 - 业务编号自动生成（no_generator.next_no）：MAX+FOR UPDATE 锁当日序号区间， 残余竞态以唯一键（1062）冲突重试一次兜底 - 五项汇总冗余（total_sale/cost/ship/received/freight）由 ``_recalc_totals`` 单点重算 —— 明细/发货/回款任何写路径收口到此函数，禁止各处手写 - 状态机：1执行中 → 2已完成 / 3已终止；终态不可再变更（一期口径，开放问题 #8）； 终止必填原因；每次流转写 business_status_logs（创建写首条 old=new=1） - 归属变更（负责/协助）写 ownership_change_logs（owner_type=2），同事务一次 commit；只影响变更后核算，不回溯历史（对照表 #19） - 毛利润视图为只读派生值（复用 formula.profit_rate/converted_profit）， 阈值按 asOf 取 M1 参数快照 —— 落库即漂移，M3 台账才是固化点 """ import logging from datetime import date from decimal import ROUND_HALF_UP, Decimal from fractions import Fraction from sqlalchemy import Row, func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.core.constants import ( ACTION_CREATE, ACTION_DELETE, ACTION_UPDATE, MODULE_BUSINESS, NO_PREFIX_BUSINESS, ) from app.core.db_errors import is_duplicate_entry from app.core.exceptions import BadRequestError, NotFoundError from app.core.response import PageQuery, PageResult from app.models import ( ApprovalOrder, Business, BusinessItem, BusinessStatusLog, Customer, OwnershipChangeLog, Personnel, Receipt, SecondaryAllocation, Shipment, User, ) from app.models.approval import APPROVAL_TYPE_SECONDARY_ALLOCATION from app.models.business import ( BUSINESS_STATUS_ACTIVE, BUSINESS_STATUS_DONE, ) from app.models.customer import OWNER_TYPE_BUSINESS from app.repositories import business_repository, customer_repository, personnel_repository from app.schemas.business import ( AllocationVersionOut, BusinessCreateIn, BusinessDetailOut, BusinessItemIn, BusinessItemOut, BusinessOut, BusinessOwnershipIn, BusinessQuery, BusinessStatusIn, BusinessStatusLogOut, BusinessUpdateIn, GrossProfitCategoryOut, GrossProfitOut, GrossProfitRowOut, OwnershipChangeLogOut, SecondaryAllocationIn, SecondaryAllocationOut, ) from app.services import approval_service, audit_service, param_service, personnel_service from app.services.audit_service import AuditRecord, RequestMeta from app.services.formula import converted_profit, profit_rate from app.services.no_generator import next_no logger = logging.getLogger(__name__) # 编号唯一键冲突后的重试次数：MAX+FOR UPDATE 已覆盖绝大多数竞态， # 重试是为极端场景（锁等待超时后重入）兜底 —— 无限重试反而掩盖死锁类缺陷 _NO_RETRY = 1 def _to_out(row: tuple) -> BusinessOut: """(business, company_name, owner_name, assistant_name) → 响应模型。 金额冗余列出参 str；汇总列可能为 NULL（server_default 只对 INSERT 生效， 存量行可能为 NULL），统一转 "0" 保持契约稳定。 """ business, company_name, owner_name, assistant_name = row return BusinessOut( id=business.id, business_no=business.business_no, customer_id=business.customer_id, customer_name=company_name, business_date=business.business_date, status=business.status, owner_personnel_id=business.owner_personnel_id, owner_personnel_name=owner_name, assistant_personnel_id=business.assistant_personnel_id, assistant_personnel_name=assistant_name, is_special=business.is_special, special_note=business.special_note, total_sale_amount=_str_or_zero(business.total_sale_amount), total_cost_amount=_str_or_zero(business.total_cost_amount), total_ship_amount=_str_or_zero(business.total_ship_amount), total_received_amount=_str_or_zero(business.total_received_amount), total_freight_amount=_str_or_zero(business.total_freight_amount), remark=business.remark, ) def _str_or_zero(value: Decimal | None) -> str: return str(value) if value is not None else "0" def _item_to_out(item: BusinessItem) -> BusinessItemOut: return BusinessItemOut( id=item.id, product_id=item.product_id, product_name=item.product_name, spec_model=item.spec_model, unit=item.unit, product_category=item.product_category, quantity=str(item.quantity), sale_price=str(item.sale_price) if item.sale_price is not None else None, sale_amount=str(item.sale_amount), cost_amount=str(item.cost_amount) if item.cost_amount is not None else None, guided_price=str(item.guided_price) if item.guided_price is not None else None, remark=item.remark, ) def _build_item_rows(business_id: int, items: list[BusinessItemIn]) -> list[BusinessItem]: """入参明细行转 ORM 行：sale_amount 服务层按 数量×单价 计算固化。 不信任前端传值（schema 亦无 sale_amount 入参）——单价可空时销售额记 0， 由用户在编辑中补价后重算（毛利润视图中销售额 0 的行全部为 0，防除零）。 乘积显式 quantize 两位固化：数量(2位)×单价(4位) 可产 6 位小数，直写 DECIMAL(14,2) 会被 MySQL 静默舍入且回显漂移（创建响应读内存值、复查 读库值前后不一致）；HALF_UP 与 MySQL 的 DECIMAL 舍入方向一致。 """ rows: list[BusinessItem] = [] for item in items: quantity = Decimal(item.quantity) sale_price = Decimal(item.sale_price) if item.sale_price is not None else None sale_amount = ( (quantity * sale_price).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) if sale_price is not None else Decimal("0.00") ) rows.append( BusinessItem( business_id=business_id, product_id=item.product_id, product_name=item.product_name, spec_model=item.spec_model or "", unit=item.unit, product_category=item.product_category, quantity=quantity, sale_price=sale_price, sale_amount=sale_amount, cost_amount=Decimal(item.cost_amount) if item.cost_amount is not None else None, guided_price=Decimal(item.guided_price) if item.guided_price is not None else None, remark=item.remark, ) ) return rows def _recalc_totals(session: Session, business: Business) -> None: """重算业务的五项汇总冗余（明细/发货/回款写路径统一收口）。 五条独立聚合查询（各表单列 SUM，走 business_id 索引）——业务单据量级 为单业务数行，不值得为省查询做单大 JOIN。 """ sale = session.execute( _sum(BusinessItem.sale_amount, BusinessItem.business_id, business.id) ).scalar_one() cost = session.execute( _sum(BusinessItem.cost_amount, BusinessItem.business_id, business.id) ).scalar_one() ship = session.execute(_sum_ship_items(business.id)).scalar_one() freight = session.execute( _sum(Shipment.freight_amount, Shipment.business_id, business.id) ).scalar_one() received = session.execute(_sum(Receipt.amount, Receipt.business_id, business.id)).scalar_one() business.total_sale_amount = sale or Decimal("0") business.total_cost_amount = cost or Decimal("0") business.total_ship_amount = ship or Decimal("0") business.total_freight_amount = freight or Decimal("0") business.total_received_amount = received or Decimal("0") def _sum(column, match_column, value) -> object: return select(func.sum(column)).where(match_column == value) def _sum_ship_items(business_id: int): """发货明细行金额合计（经发货单关联到业务）。""" from app.models import ShipmentItem return ( select(func.sum(ShipmentItem.amount)) .select_from(ShipmentItem) .join(Shipment, Shipment.id == ShipmentItem.shipment_id) .where(Shipment.business_id == business_id) ) def _generate_no(session: Session) -> str: """生成业务编号（MAX+FOR UPDATE；1062 兜底重试一次，见模块 docstring）。""" return next_no(session, model=Business, column_name="business_no", prefix=NO_PREFIX_BUSINESS) def _ensure_personnel_active(session: Session, personnel_id: int, label: str) -> Personnel: """归属人员须存在且在职（共享校验收口于 personnel_service.ensure_active）。""" return personnel_service.ensure_active(session, personnel_id, label=label) def _ensure_customer(session: Session, customer_id: int) -> Customer: customer = customer_repository.get_by_id(session, customer_id) if customer is None: raise NotFoundError("客户不存在") return customer def _is_duplicate_entry(exc: IntegrityError) -> bool: return is_duplicate_entry(exc) def list_businesses(session: Session, query: BusinessQuery) -> PageResult[BusinessOut]: """分页查询业务（附客户名与归属人名）。""" rows = business_repository.list_businesses( session, offset=query.offset, limit=query.page_size, keyword=query.keyword, status=query.status, customer_id=query.customer_id, owner_personnel_id=query.owner_personnel_id, date_from=query.date_from, date_to=query.date_to, ) total = business_repository.count_businesses( session, keyword=query.keyword, status=query.status, customer_id=query.customer_id, owner_personnel_id=query.owner_personnel_id, date_from=query.date_from, date_to=query.date_to, ) return PageResult[BusinessOut]( records=[_to_out(row) for row in rows], total=total, page_num=query.page_num, page_size=query.page_size, ) def get_detail(session: Session, business_id: int) -> BusinessDetailOut: """业务详情（附明细行与二次分配方案，供编辑抽屉回显）。 :raises NotFoundError: 业务不存在 """ row = business_repository.get_with_names(session, business_id) if row is None: raise NotFoundError("业务不存在") out = BusinessDetailOut(**_to_out(row).model_dump()) out.items = [ _item_to_out(item) for item in business_repository.list_items(session, business_id) ] allocations = business_repository.list_allocations(session, business_id) out.secondary_allocations = [ SecondaryAllocationOut( personnel_id=a.personnel_id, personnel_name=a.personnel_name, ratio=str(a.ratio) ) for a in allocations ] out.allocation_hint = _allocation_hint(allocations) out.allocation_history = [ AllocationVersionOut( personnel_id=a.personnel_id, personnel_name=a.personnel_name, ratio=str(a.ratio), version_no=a.version_no, is_active=a.is_active, create_time=a.create_time, ) for a in business_repository.list_allocation_versions(session, business_id) ] return out def _build_allocation_rows( session: Session, business_id: int, items: list[SecondaryAllocationIn], *, old_ids: set[int], ) -> list[SecondaryAllocation]: """构造二次分配明细行：人员校验 + 姓名冗余固化 + 比例转 Decimal。 :param old_ids: 编辑前已参与的人员 ID 集合（create 传空集）。这些人的 行是**历史事实**——personnel_name 冗余固化的设计本就允许离职后回显， 编辑业务（哪怕只改备注）不应被历史参与人的离职状态卡死；仅对 **新加入**的参与人做在职校验。 :raises BadRequestError: 新增参与人员不存在或已离职 / 比例合计超 100% / 参与人含合同负责人（口径见澄清 #1/#2 答复 2026-10-19：分配绩效、 余额归合同负责营销人员——超 100% 会把负责人份额算成负数，负责人 自分份额则双重计入） """ business = business_repository.get_by_id(session, business_id) if business is None: raise NotFoundError("业务不存在") total = sum((Decimal(item.ratio) for item in items), Decimal("0")) if total > Decimal("100"): raise BadRequestError( f"二次分配比例合计 {total.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)}%" " 超过 100%（余额归合同负责营销人员，比例不可超满额）" ) owner_id = business.owner_personnel_id if owner_id is not None and any(item.personnel_id == owner_id for item in items): raise BadRequestError("分摊参与人不可包含合同负责营销人员（负责人得余额份额）") rows: list[SecondaryAllocation] = [] for item in items: if item.personnel_id in old_ids: person = personnel_repository.get_by_id(session, item.personnel_id) if person is None: raise BadRequestError("分摊参与人员不存在") else: person = _ensure_personnel_active(session, item.personnel_id, "分摊参与") rows.append( SecondaryAllocation( business_id=business_id, personnel_id=item.personnel_id, personnel_name=person.personnel_name, ratio=Decimal(item.ratio), ) ) return rows def _plan_changed(old_rows: list[SecondaryAllocation], new_rows: list[SecondaryAllocation]) -> bool: """方案内容（参与人 + 比例）是否真的变了。 比较口径是**集合**（人员 ID 与其比例的组合），与提交顺序无关——送审 快照同样按库内生效行的顺序生成，若按顺序比较，用户只是把行拖了个 序就会被判成「变了」，进而无谓地新增版本与重提审批单。 """ def _key(row: SecondaryAllocation) -> tuple[int, Decimal]: return (row.personnel_id, Decimal(row.ratio).quantize(Decimal("0.01"))) return {_key(r) for r in old_rows} != {_key(r) for r in new_rows} def _allocation_hint(allocations: list[SecondaryAllocation]) -> str | None: """方案合计/余额提示（不要求合计满额是需求原文；余额归属 = 澄清 #2 答复）。 无方案返回 None；合计恰满额只报合计，避免「余额 0」噪音。 """ if not allocations: return None total = sum((a.ratio for a in allocations), Decimal("0")) total_text = total.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) if total == Decimal("100"): return f"二次分配合计 {total_text}%" rest = (Decimal("100") - total).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) return f"二次分配合计 {total_text}%，余额 {rest}%（归合同负责营销人员）" def create_business( session: Session, payload: BusinessCreateIn, *, operator: User, meta: RequestMeta | None = None, ) -> BusinessOut: """创建业务 + 明细行 + 首条状态留痕，一个事务。 创建时指定的归属不写归属留痕（留痕记「变更」，创建即初始事实）。 :raises NotFoundError: 客户不存在 :raises BadRequestError: 归属人员不存在或已离职 / 明细校验不通过 """ with audit_service.audit( session, operator=operator, record=AuditRecord( module=MODULE_BUSINESS, action=ACTION_CREATE, summary=f"创建业务（客户ID={payload.customer_id}，明细 {len(payload.items)} 行）", request_data=payload.model_dump(exclude_none=True), ), meta=meta, ): _ensure_customer(session, payload.customer_id) if payload.owner_personnel_id is not None: _ensure_personnel_active(session, payload.owner_personnel_id, "负责") if payload.assistant_personnel_id is not None: _ensure_personnel_active(session, payload.assistant_personnel_id, "协助") # 编号唯一键冲突的重试整体收在审计上下文内：rollback 后同一事务重来， # 成功/最终失败各只记一条审计（吞异常 return 会被记成假成功，重入 # 整个 create_* 则违反「不支持嵌套」约定）；上限 _NO_RETRY 防无限重试 # 掩盖死锁类缺陷。审计摘要预取的是首次编号——重试成功时入库编号见 # 应用日志（AuditRecord frozen，不回填）。 for attempt in range(_NO_RETRY + 1): try: business = Business( business_no=_generate_no(session), customer_id=payload.customer_id, business_date=payload.business_date, status=BUSINESS_STATUS_ACTIVE, owner_personnel_id=payload.owner_personnel_id, assistant_personnel_id=payload.assistant_personnel_id, is_special=payload.is_special, special_note=payload.special_note, remark=payload.remark, ) business_repository.create(session, business) business_repository.replace_items( session, business.id, _build_item_rows(business.id, payload.items) ) _recalc_totals(session, business) business_repository.add_status_log( session, BusinessStatusLog( business_id=business.id, old_status=BUSINESS_STATUS_ACTIVE, new_status=BUSINESS_STATUS_ACTIVE, operator_id=operator.id, operator_name=operator.username, ), ) # 二次分配方案与业务同事务（需求：只出现在合同刚建立时）； # 放在重试循环内——rollback 后方案行随业务一并重来 business_repository.replace_allocations( session, business.id, _build_allocation_rows( session, business.id, payload.secondary_allocations, old_ids=set() ), ) # 方案非空才提单：开关关闭=自动通过留痕，开启=待财务复核。 # AP 编号撞键时 IntegrityError 穿透 → 外层编号重试循环把 # 业务/方案/审批单整体重来 if payload.secondary_allocations: approval_service.submit_order( session, order_type=APPROVAL_TYPE_SECONDARY_ALLOCATION, ref_id=business.id, refer_no=business.business_no, submitter=operator, snapshot=approval_service.allocation_content(session, business) or None, ) customer = customer_repository.get_by_id(session, business.customer_id) out = _to_out((business, customer.company_name if customer else None, None, None)) new_id, new_no = business.id, business.business_no # 固化：commit 后不碰 ORM session.commit() break except IntegrityError as exc: session.rollback() if _is_duplicate_entry(exc) and attempt < _NO_RETRY: continue raise logger.info("创建业务 | business_id=%s | no=%s", new_id, new_no) return out def update_business( session: Session, business_id: int, payload: BusinessUpdateIn, *, operator: User, meta: RequestMeta | None = None, ) -> BusinessOut: """编辑业务：全量替换明细 + 重算汇总；归属不在此端点（防旁路）。 客户可改，但有发货或回款记录后禁改 —— 单据 FK 挂在业务上，换客户会让 发货/回款语义断裂（明细同理随业务走）。 :raises NotFoundError: 业务不存在 :raises BadRequestError: 业务已完成或已终止 / 归属人员不合法 / 有发货或回款时换客户 """ row = business_repository.get_with_names(session, business_id, for_update=True) business = row[0] if row else None with audit_service.audit( session, operator=operator, record=AuditRecord( module=MODULE_BUSINESS, action=ACTION_UPDATE, summary=f"修改业务 {business.business_no if business else f'#{business_id}'}", request_data=payload.model_dump(exclude_none=True), ), meta=meta, ): if row is None: raise NotFoundError("业务不存在") business, _, owner_name, assistant_name = row # 终态门槛与发货/回款写准入同口径：明细与汇总是 M3 核算基线， # 已完成/已终止的业务禁改（发货/回款服务以同一理由禁写终态业务） if business.status != BUSINESS_STATUS_ACTIVE: raise BadRequestError("业务已完成或已终止，禁止编辑业务") if payload.customer_id != business.customer_id and business_repository.has_docs( session, business_id ): raise BadRequestError("业务已有发货或回款记录，不能更换客户") customer = _ensure_customer(session, payload.customer_id) if payload.owner_personnel_id is not None: _ensure_personnel_active(session, payload.owner_personnel_id, "负责") if payload.assistant_personnel_id is not None: _ensure_personnel_active(session, payload.assistant_personnel_id, "协助") business.customer_id = payload.customer_id business.business_date = payload.business_date business.is_special = payload.is_special business.special_note = payload.special_note business.remark = payload.remark # 归属刻意不赋值：留痕只由 /ownership 端点产生，编辑路径改归属构成 # 「改归属不留痕」的旁路（与 customers 编辑不触归属同一口径） business_repository.replace_items( session, business_id, _build_item_rows(business_id, payload.items) ) # 方案：内容真的变了才动版本行（内容未变不新增版本，避免「只改备注」 # 也堆出一个方案版本）；空数组=清空=把生效行置历史 old_allocations = business_repository.list_allocations(session, business_id) new_rows = _build_allocation_rows( session, business_id, payload.secondary_allocations, old_ids={a.personnel_id for a in old_allocations}, ) if not payload.secondary_allocations: business_repository.replace_allocations(session, business_id, []) elif _plan_changed(old_allocations, new_rows): business_repository.replace_allocations(session, business_id, new_rows) # 汇总先重算：送审快照里的销售额/成本额必须反映本次编辑后的值， # 放到提单之后会固化成上一版本的金额 _recalc_totals(session, business) # 提单/撤单与方案同步生命周期： # - 方案为空 → 仅撤回在途单（旧在途单挂着已不存在的方案继续走审批） # - 送审内容与在途单一致 → 保留原单，不撤回不重建（避免反复保存 # 刷出「撤回+新建」的审批留痕噪音） # - 其余 → 提新单（内含撤旧建新） if payload.secondary_allocations: snapshot = approval_service.allocation_content(session, business) or None if not approval_service.pending_matches_snapshot( session, order_type=APPROVAL_TYPE_SECONDARY_ALLOCATION, ref_id=business_id, snapshot=snapshot, ): try: approval_service.submit_order( session, order_type=APPROVAL_TYPE_SECONDARY_ALLOCATION, ref_id=business_id, refer_no=business.business_no, submitter=operator, snapshot=snapshot, ) except IntegrityError as exc: # version_no 唯一键的并发冲突（取号是锁读后此项理论不可达， # 留作 DB 兜底）：审计上下文会统一回滚，这里只把 1062 # 翻译成可重试的业务提示，其余完整性错误原样上抛 if _is_duplicate_entry(exc): raise BadRequestError("提交冲突（版本号生成重复），请重试") from exc raise else: approval_service.withdraw_pending_orders( session, order_type=APPROVAL_TYPE_SECONDARY_ALLOCATION, ref_id=business_id, operator=operator, ) # 锁定行上已赋值，同一身份映射实例的 _to_out 即反映新值；名称字段 # 从首查 row 取（换客户时 _ensure_customer 已给出新公司名） out = _to_out((business, customer.company_name, owner_name, assistant_name)) session.commit() logger.info("修改业务 | business_id=%s", business_id) return out def delete_business( session: Session, business_id: int, *, operator: User, meta: RequestMeta | None = None, ) -> None: """删除业务（物理删除）：有发货/回款 → 400 引导走「已终止」。 无单据引用的业务（建错的空单）方可删除；删除时连带清理明细、状态留痕 与业务维度的归属留痕（无外键固化表，须手动清）。 :raises NotFoundError: 业务不存在 :raises BadRequestError: 存在发货或回款记录 / 存在在途审批单 """ business = business_repository.get_by_id(session, business_id, for_update=True) with audit_service.audit( session, operator=operator, record=AuditRecord( module=MODULE_BUSINESS, action=ACTION_DELETE, summary=f"删除业务 {business.business_no if business else f'#{business_id}'}", ), meta=meta, ): if business is None: raise NotFoundError("业务不存在") if business_repository.has_docs(session, business_id): raise BadRequestError("业务已有发货或回款记录，请改为已终止而非删除") # 在途审批单（ref_id 固化无外键，业务删了单还挂着且可被继续审批）： # 禁删引导先完成/撤回审批，而不是静默留下指向已删业务的活单 if _count_pending_approvals(session, business_id): raise BadRequestError("该业务有在途审批单，请先完成或撤回审批后再删除") business_no = business.business_no _hard_delete(session, business_id) session.commit() logger.info("删除业务 | business_id=%s | no=%s", business_id, business_no) def _count_pending_approvals(session: Session, business_id: int) -> int: """该业务的在途审批单数（status 1/2；终态单随业务删除一并清理）。""" from sqlalchemy import func, select from app.models import ApprovalOrder from app.models.approval import ( APPROVAL_STATUS_PENDING_FINANCE, APPROVAL_STATUS_PENDING_LEADER, ) return int( session.execute( select(func.count()) .select_from(ApprovalOrder) .where( ApprovalOrder.order_type == APPROVAL_TYPE_SECONDARY_ALLOCATION, ApprovalOrder.ref_id == business_id, ApprovalOrder.status.in_( (APPROVAL_STATUS_PENDING_FINANCE, APPROVAL_STATUS_PENDING_LEADER) ), ) ).scalar_one() ) def _hard_delete(session: Session, business_id: int) -> None: """按依赖序删除：明细 → 二次分配方案 → 审批单（steps 随 CASCADE）→ 状态留痕 → 业务维度归属留痕 → 主表。""" from sqlalchemy import delete session.execute(delete(BusinessItem).where(BusinessItem.business_id == business_id)) session.execute( delete(SecondaryAllocation).where(SecondaryAllocation.business_id == business_id) ) session.execute( delete(ApprovalOrder).where( ApprovalOrder.order_type == APPROVAL_TYPE_SECONDARY_ALLOCATION, ApprovalOrder.ref_id == business_id, ) ) session.execute(delete(BusinessStatusLog).where(BusinessStatusLog.business_id == business_id)) session.execute( delete(OwnershipChangeLog).where( OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS, OwnershipChangeLog.business_id == business_id, ) ) session.execute(delete(Business).where(Business.id == business_id)) def change_status( session: Session, business_id: int, payload: BusinessStatusIn, *, operator: User, meta: RequestMeta | None = None, ) -> BusinessOut: """状态流转：1→2 / 1→3，终态不可再变更；每次流转写留痕。 :raises NotFoundError: 业务不存在 :raises BadRequestError: 非执行中状态流转 / 目标与当前相同 """ row = business_repository.get_with_names(session, business_id, for_update=True) with audit_service.audit( session, operator=operator, record=AuditRecord( module=MODULE_BUSINESS, action=ACTION_UPDATE, summary=( f"业务状态变更 {row[0].business_no if row else f'#{business_id}'} → " f"{'已完成' if payload.status == BUSINESS_STATUS_DONE else '已终止'}" ), request_data=payload.model_dump(exclude_none=True), ), meta=meta, ): if row is None: raise NotFoundError("业务不存在") business, _, _, _ = row if business.status != BUSINESS_STATUS_ACTIVE: raise BadRequestError("已完成或已终止的业务不可再变更状态") if business.status == payload.status: raise BadRequestError("状态未变化") # 变更前状态须在赋值前固化：ORM 属性赋值后 old_status 就取不到旧值 # （留痕会记成 old==new 失真，时间线无法回溯） old_status = business.status business.status = payload.status business_repository.add_status_log( session, BusinessStatusLog( business_id=business_id, old_status=old_status, new_status=payload.status, change_reason=payload.reason, operator_id=operator.id, operator_name=operator.username, ), ) # 锁定行上已赋值，同一身份映射实例的 _to_out 即反映新状态 out = _to_out(row) session.commit() logger.info( "业务状态变更 | business_id=%s | %s -> %s", business_id, old_status, payload.status, ) return out def change_ownership( session: Session, business_id: int, payload: BusinessOwnershipIn, *, operator: User, meta: RequestMeta | None = None, ) -> BusinessOut: """归属变更（负责/协助）：同事务「更新主表 + 插留痕」，一次 commit。 负责与协助之一发生变化即记一条留痕（前后值都含两个字段）；两者均未变化 → 400「归属未变化」。 :raises NotFoundError: 业务不存在 :raises BadRequestError: 归属人员不存在或已离职 / 归属未变化 """ row = business_repository.get_with_names(session, business_id, for_update=True) with audit_service.audit( session, operator=operator, record=AuditRecord( module=MODULE_BUSINESS, action=ACTION_UPDATE, summary=f"调整业务归属 {row[0].business_no if row else f'#{business_id}'}", request_data=payload.model_dump(exclude_none=True), ), meta=meta, ): if row is None: raise NotFoundError("业务不存在") business, _, old_owner_name, old_assistant_name = row # 归属变更的出参姓名来自校验函数的返回行（锁定读）：首查姓名是变更前 # 固化标量，新归属人名必须按新 ID 重取 new_owner = ( _ensure_personnel_active(session, payload.new_owner_personnel_id, "负责") if payload.new_owner_personnel_id is not None else None ) new_assistant = ( _ensure_personnel_active(session, payload.new_assistant_personnel_id, "协助") if payload.new_assistant_personnel_id is not None else None ) owner_changed = business.owner_personnel_id != payload.new_owner_personnel_id assistant_changed = business.assistant_personnel_id != payload.new_assistant_personnel_id if not owner_changed and not assistant_changed: raise BadRequestError("归属未变化") old_owner_id = business.owner_personnel_id old_assistant_id = business.assistant_personnel_id business.owner_personnel_id = payload.new_owner_personnel_id business.assistant_personnel_id = payload.new_assistant_personnel_id business_repository.add_ownership_log( session, OwnershipChangeLog( owner_type=OWNER_TYPE_BUSINESS, business_id=business_id, old_owner_personnel_id=old_owner_id, new_owner_personnel_id=payload.new_owner_personnel_id, old_assistant_personnel_id=old_assistant_id, new_assistant_personnel_id=payload.new_assistant_personnel_id, change_reason=payload.reason, operator_id=operator.id, operator_name=operator.username, ), ) # 出参组装：ID 在锁定行上已赋值（_to_out 读内存值即反映新值），姓名 # 一侧「变更项用新 ID 的人名、未变项沿用首查名」——首查名是变更前快照 out = _to_out( ( business, row[1], new_owner.personnel_name if new_owner is not None else old_owner_name, new_assistant.personnel_name if new_assistant is not None else old_assistant_name, ) ) session.commit() logger.info( "调整业务归属 | business_id=%s | owner %s->%s | assistant %s->%s", business_id, old_owner_id, payload.new_owner_personnel_id, old_assistant_id, payload.new_assistant_personnel_id, ) return out def list_ownership_logs( session: Session, business_id: int, query: PageQuery ) -> PageResult[OwnershipChangeLogOut]: """某业务的归属变更留痕（分页）。 :raises NotFoundError: 业务不存在 """ if business_repository.get_by_id(session, business_id) is None: raise NotFoundError("业务不存在") rows = business_repository.list_ownership_logs( session, business_id, offset=query.offset, limit=query.page_size ) total = business_repository.count_ownership_logs(session, business_id=business_id) return PageResult[OwnershipChangeLogOut]( records=[_log_to_out(row) for row in rows], total=total, page_num=query.page_num, page_size=query.page_size, ) def _log_to_out(row: Row) -> OwnershipChangeLogOut: log, old_owner_name, new_owner_name = row return OwnershipChangeLogOut( id=log.id, business_id=log.business_id, old_owner_personnel_id=log.old_owner_personnel_id, new_owner_personnel_id=log.new_owner_personnel_id, old_assistant_personnel_id=log.old_assistant_personnel_id, new_assistant_personnel_id=log.new_assistant_personnel_id, old_owner_name=old_owner_name, new_owner_name=new_owner_name, change_reason=log.change_reason, operator_id=log.operator_id, operator_name=log.operator_name, create_time=log.create_time, ) def list_status_logs( session: Session, business_id: int, query: PageQuery ) -> PageResult[BusinessStatusLogOut]: """某业务的状态流转时间线（分页，时间正序）。 :raises NotFoundError: 业务不存在 """ if business_repository.get_by_id(session, business_id) is None: raise NotFoundError("业务不存在") rows = business_repository.list_status_logs( session, business_id, offset=query.offset, limit=query.page_size ) total = business_repository.count_status_logs(session, business_id=business_id) return PageResult[BusinessStatusLogOut]( records=[BusinessStatusLogOut.model_validate(row) for row in rows], total=total, page_num=query.page_num, page_size=query.page_size, ) def gross_profit(session: Session, business_id: int, as_of: date | None) -> GrossProfitOut: """毛利润数据视图（需求 2.2.4）：行级派生值 + 类别汇总，实时计算不落库。 阈值按 asOf（默认今日）取 M1 参数快照（product.threshold，按行类别维度， 存储口径为 % 数如 30，直接进 converted_profit）；毛利润率按 M1 预览同一 出参口径输出 Fraction 分数字符串（如 "97/10000"）；成本缺失的行派生列为 None 并标注原因；汇总只聚合可计算行。 :raises NotFoundError: 业务不存在 """ if business_repository.get_by_id(session, business_id) is None: raise NotFoundError("业务不存在") as_of = as_of or date.today() snapshot = param_service.resolve_snapshot(session, as_of) def threshold_of(category: str) -> Fraction: return snapshot.fraction("product.threshold", product_category=category) rows_out: list[GrossProfitRowOut] = [] missing_count = 0 # 类别聚合容器：{类别: [sale, cost]}，成本缺失行不计入该类别聚合 agg: dict[str, list[Decimal]] = {} for item in business_repository.list_items(session, business_id): threshold = threshold_of(item.product_category) row_kwargs: dict = { "item_id": item.id, "product_category": item.product_category, "product_name": item.product_name, "spec_model": item.spec_model, "quantity": str(item.quantity), "sale_amount": str(item.sale_amount), "threshold": str(threshold), } if item.cost_amount is None: missing_count += 1 row_kwargs.update( cost_amount=None, profit_amount=None, profit_rate=None, converted_profit=None, missing_reason="成本缺失，无法计算毛利润", ) else: profit = item.sale_amount - item.cost_amount rate = profit_rate(profit, item.sale_amount) conv = converted_profit(item.sale_amount, rate, threshold) row_kwargs.update( cost_amount=str(item.cost_amount), profit_amount=str(profit), profit_rate=str(rate), converted_profit=str(conv), missing_reason=None, ) bucket = agg.setdefault(item.product_category, [Decimal("0"), Decimal("0")]) bucket[0] += item.sale_amount bucket[1] += item.cost_amount rows_out.append(GrossProfitRowOut(**row_kwargs)) categories = [ GrossProfitCategoryOut( product_category=category, sale_amount=str(sale), cost_amount=str(cost), profit_amount=str(sale - cost), profit_rate=str(profit_rate(sale - cost, sale)), threshold=str(threshold_of(category)), converted_profit=str( converted_profit(sale, profit_rate(sale - cost, sale), threshold_of(category)) ), ) for category, [sale, cost] in agg.items() ] return GrossProfitOut( business_id=business_id, as_of=as_of, rows=rows_out, categories=categories, missing_count=missing_count, )