fix(business): 删除保护补齐与单价小数位口径修复(e2e 发现)

- 客户删除保护:被业务引用 → 400(T4 占位注释在业务表就位后漏补,FK 500 暴露)
- 人员删除保护:业务归属(负责/协助)引用纳入检查(同样漏检 500)
- schema 单价类列(DECIMAL 14,4)小数位放行至 4 位:回显 '5000.0000'
  原样提交被两位校验永久拒收(编辑任何带精度单价的历史行都失败)
- 补 2 个集成用例锁定;文案同步「客户归属或业务归属引用」

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
liuwenxinandClaude Code committed 2026-10-13 17:06:46 +08:00
1 parent 299b72483c
commit fba8cfc215
8 files changed
+167 -23

No files matched your search

@@ -23,6 +23,19 @@ def get_by_id(session: Session, customer_id: int) -> Customer | None:
return session.get(Customer, customer_id)
def count_business_refs(session: Session, customer_id: int) -> int:
"""被业务单据引用的次数(删除保护;业务表就位后补齐的检查)。"""
from app.models import Business
return int(
session.execute(
select(func.count())
.select_from(Business)
.where(Business.customer_id == customer_id)
).scalar_one()
)
def get_with_owner_name(session: Session, customer_id: int) -> Row | None:
"""查单个客户并带出归属人员姓名(详情、编辑回显与归属变更共用)。"""
stmt = _base_select().where(Customer.id == customer_id)
@@ -94,6 +94,33 @@ def count_customer_ownerships(session: Session, personnel_ids: list[int]) -> dic
return {int(owner_id): int(count) for owner_id, count in session.execute(stmt).all()}
def count_business_roles_refs(session: Session, personnel_ids: list[int]) -> dict[int, int]:
"""按人员批量统计业务单据归属引用(负责/协助合计;业务表就位后补齐的检查)。
businesses.owner/assistant_personnel_id 均为 RESTRICT 外键——漏检会让
「删除被业务引用的人员」以 500 暴露而非可读的 400。
"""
from app.models import Business
owner = (
select(Business.owner_personnel_id, func.count(Business.id))
.where(Business.owner_personnel_id.in_(personnel_ids))
.group_by(Business.owner_personnel_id)
)
assistant = (
select(Business.assistant_personnel_id, func.count(Business.id))
.where(Business.assistant_personnel_id.in_(personnel_ids))
.group_by(Business.assistant_personnel_id)
)
refs: dict[int, int] = {}
for stmt in (owner, assistant):
for owner_id, count in session.execute(stmt).all():
if owner_id is not None:
refs[int(owner_id)] = refs.get(int(owner_id), 0) + int(count)
return refs
def list_regions(session: Session, personnel_id: int) -> list[PersonnelRegion]:
"""查人员的全部辐射区域,按插入顺序返回。"""
stmt = (