5ece9f6dab
仓储层的 list_*/count_* 此前以 keyword/status/offset/limit 四五个散参数 传参，调用点重复铺开、易漏传且不可扩展；按项目「参数 >4 封装 DTO」的 约定统一收敛为各模块的 Query 对象（如 RoleQuery、BusinessQuery）， 服务层与测试调用同步调整，行为不变。 - repositories 层查询方法签名改为 (session, query)，分页与筛选条件 由 Query 承载，_apply_filters 同步收口 - schemas 层补齐各模块 Query 模型（approval/business_role/report 等） - services 层调用点同步改为传 Query - tests 随之调整；新增用例：待办角色流转（财务通过后领导出现待办）、 参数预览顺序稳定 验证：全量 pytest 900 passed（11:08）；单元 526 passed；ruff check 绿。 Co-Authored-By: Claude Code <noreply@anthropic.com>
1036 lines
42 KiB
Python
1036 lines
42 KiB
Python
"""业务管理业务编排（需求 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 collections.abc import Callable
|
||
from dataclasses import dataclass
|
||
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, query)
|
||
total = business_repository.count_businesses(session, query)
|
||
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
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class _BusinessOwnershipResolution:
|
||
"""业务归属变更的解析结果（新归属人行 + 变更前固化 ID）。"""
|
||
|
||
owner: Personnel | None
|
||
assistant: Personnel | None
|
||
old_owner_id: int | None
|
||
old_assistant_id: int | None
|
||
|
||
|
||
def _resolve_business_ownership(
|
||
session: Session, business: Business, payload: BusinessOwnershipIn
|
||
) -> _BusinessOwnershipResolution:
|
||
"""校验新归属人（须在职）并判定归属是否真的变化。
|
||
|
||
校验在审计上下文内执行 —— 失败也留痕。两者均未变化 → 400「归属未变化」。
|
||
"""
|
||
# 归属变更的出参姓名来自校验函数的返回行（锁定读）：首查姓名是变更前
|
||
# 固化标量，新归属人名必须按新 ID 重取
|
||
owner = (
|
||
_ensure_personnel_active(session, payload.new_owner_personnel_id, "负责")
|
||
if payload.new_owner_personnel_id is not None
|
||
else None
|
||
)
|
||
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("归属未变化")
|
||
return _BusinessOwnershipResolution(
|
||
owner=owner,
|
||
assistant=assistant,
|
||
old_owner_id=business.owner_personnel_id,
|
||
old_assistant_id=business.assistant_personnel_id,
|
||
)
|
||
|
||
|
||
def _business_ownership_log(
|
||
business: Business,
|
||
payload: BusinessOwnershipIn,
|
||
resolution: _BusinessOwnershipResolution,
|
||
operator: User,
|
||
) -> OwnershipChangeLog:
|
||
"""归属变更留痕行（负责/协助两个字段的前后值都记）。"""
|
||
return OwnershipChangeLog(
|
||
owner_type=OWNER_TYPE_BUSINESS,
|
||
business_id=business.id,
|
||
old_owner_personnel_id=resolution.old_owner_id,
|
||
new_owner_personnel_id=payload.new_owner_personnel_id,
|
||
old_assistant_personnel_id=resolution.old_assistant_id,
|
||
new_assistant_personnel_id=payload.new_assistant_personnel_id,
|
||
change_reason=payload.reason,
|
||
operator_id=operator.id,
|
||
operator_name=operator.username,
|
||
)
|
||
|
||
|
||
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
|
||
resolution = _resolve_business_ownership(session, business, payload)
|
||
|
||
business.owner_personnel_id = payload.new_owner_personnel_id
|
||
business.assistant_personnel_id = payload.new_assistant_personnel_id
|
||
business_repository.add_ownership_log(
|
||
session, _business_ownership_log(business, payload, resolution, operator)
|
||
)
|
||
# 出参组装：ID 在锁定行上已赋值（_to_out 读内存值即反映新值），姓名
|
||
# 一侧「变更项用新 ID 的人名、未变项沿用首查名」——首查名是变更前快照
|
||
out = _to_out(
|
||
(
|
||
business,
|
||
row[1],
|
||
resolution.owner.personnel_name if resolution.owner is not None else old_owner_name,
|
||
resolution.assistant.personnel_name
|
||
if resolution.assistant is not None
|
||
else old_assistant_name,
|
||
)
|
||
)
|
||
session.commit()
|
||
logger.info(
|
||
"调整业务归属 | business_id=%s | owner %s->%s | assistant %s->%s",
|
||
business_id,
|
||
resolution.old_owner_id,
|
||
payload.new_owner_personnel_id,
|
||
resolution.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,
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class _GrossProfitScan:
|
||
"""毛利润逐行扫描结果（行 DTO、类别聚合桶、成本缺失计数）。"""
|
||
|
||
rows: list[GrossProfitRowOut]
|
||
category_agg: dict[str, list[Decimal]]
|
||
missing_count: int
|
||
|
||
|
||
def _gross_profit_row(
|
||
item: BusinessItem, threshold: Fraction
|
||
) -> tuple[GrossProfitRowOut, tuple[Decimal, Decimal] | None]:
|
||
"""单行毛利润派生值（行 DTO + 类别聚合桶）。
|
||
|
||
聚合桶为 None 表示成本缺失行 —— 这类行不计入类别汇总，但仍要输出
|
||
派生列为 None 的行与缺失原因。阈值对两类行都要取（成本缺失也展示阈值）。
|
||
"""
|
||
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:
|
||
row_kwargs.update(
|
||
cost_amount=None,
|
||
profit_amount=None,
|
||
profit_rate=None,
|
||
converted_profit=None,
|
||
missing_reason="成本缺失，无法计算毛利润",
|
||
)
|
||
return GrossProfitRowOut(**row_kwargs), None
|
||
|
||
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,
|
||
)
|
||
return GrossProfitRowOut(**row_kwargs), (item.sale_amount, item.cost_amount)
|
||
|
||
|
||
def _scan_gross_profit_rows(
|
||
session: Session, business_id: int, threshold_of: Callable[[str], Fraction]
|
||
) -> _GrossProfitScan:
|
||
"""逐行扫描业务明细：出行 DTO、类别聚合桶与成本缺失计数。
|
||
|
||
类别顺序沿用明细行的首次出现顺序（不排序）—— 前端汇总展示依赖该顺序。
|
||
"""
|
||
rows: list[GrossProfitRowOut] = []
|
||
agg: dict[str, list[Decimal]] = {}
|
||
missing_count = 0
|
||
for item in business_repository.list_items(session, business_id):
|
||
row, bucket = _gross_profit_row(item, threshold_of(item.product_category))
|
||
rows.append(row)
|
||
if bucket is None:
|
||
missing_count += 1
|
||
continue
|
||
current = agg.setdefault(item.product_category, [Decimal("0"), Decimal("0")])
|
||
current[0] += bucket[0]
|
||
current[1] += bucket[1]
|
||
return _GrossProfitScan(rows=rows, category_agg=agg, missing_count=missing_count)
|
||
|
||
|
||
def _gross_profit_category(
|
||
category: str, sale: Decimal, cost: Decimal, threshold: Fraction
|
||
) -> GrossProfitCategoryOut:
|
||
"""类别汇总行（派生口径与行级一致：利润率/折算复用 formula 纯函数）。"""
|
||
profit = sale - cost
|
||
rate = profit_rate(profit, sale)
|
||
return GrossProfitCategoryOut(
|
||
product_category=category,
|
||
sale_amount=str(sale),
|
||
cost_amount=str(cost),
|
||
profit_amount=str(profit),
|
||
profit_rate=str(rate),
|
||
threshold=str(threshold),
|
||
converted_profit=str(converted_profit(sale, rate, threshold)),
|
||
)
|
||
|
||
|
||
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)
|
||
|
||
scan = _scan_gross_profit_rows(session, business_id, threshold_of)
|
||
categories = [
|
||
_gross_profit_category(category, sale, cost, threshold_of(category))
|
||
for category, [sale, cost] in scan.category_agg.items()
|
||
]
|
||
return GrossProfitOut(
|
||
business_id=business_id,
|
||
as_of=as_of,
|
||
rows=scan.rows,
|
||
categories=categories,
|
||
missing_count=scan.missing_count,
|
||
)
|