feat(base): 客户管理接口与归属变更留痕（不回溯历史）

- 归属变更同事务更新主表+插留痕；归属调整唯一入口防旁路

Co-Authored-By: Claude Code <noreply@anthropic.com>

backend/app/api/v1/base/customers.py

@@ -0,0 +1,142 @@
+"""客户管理接口（需求 2.1.3）。
+
+归属调整有专属端点 ``PUT /{id}/ownership``：它是唯一写归属留痕的入口，
+编辑端点的 schema 不含归属字段 —— 防止出现「改了归属却无留痕」的旁路
+（原因详见 `schemas.customer.CustomerUpdateIn`）。
+
+与 `api/v1/system` 的权限分工：基础数据是**业务操作**，登录即可访问
+（不要求管理员）；写操作同样经 `RequestMetaDep` 带入请求元信息，
+由 service 层写入操作日志。
+"""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Path, Query
+
+from app.api.deps import DbSession, RequestMetaDep, get_current_user
+from app.core.response import ApiResponse, PageQuery, PageResult, ok
+from app.models import User
+from app.schemas.customer import (
+    CustomerCreateIn,
+    CustomerOut,
+    CustomerOwnershipIn,
+    CustomerQuery,
+    CustomerUpdateIn,
+    OwnershipChangeLogOut,
+)
+from app.services import customer_service
+
+router = APIRouter(prefix="/customers", tags=["客户管理"])
+
+# 基础数据不要求管理员（见模块 docstring），登录用户即可操作
+CurrentUser = Annotated[User, Depends(get_current_user)]
+
+
+@router.get("", response_model=ApiResponse[PageResult[CustomerOut]], summary="客户列表")
+def list_customers(
+    session: DbSession,
+    _: CurrentUser,
+    query: Annotated[CustomerQuery, Query()],
+) -> ApiResponse[PageResult[CustomerOut]]:
+    """分页查询客户（keyword 对公司名称/客户负责人模糊匹配，附归属人员姓名）。"""
+    return ok(customer_service.list_customers(session, query))
+
+
+@router.post("", response_model=ApiResponse[CustomerOut], summary="新增客户")
+def create_customer(
+    session: DbSession,
+    current_user: CurrentUser,
+    payload: CustomerCreateIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[CustomerOut]:
+    """新增客户。客户分类须在注册表内；「公司名称+负责人」组合唯一。"""
+    return ok(
+        customer_service.create_customer(session, payload, operator=current_user, meta=meta),
+        message="客户已创建",
+    )
+
+
+@router.get("/{customer_id}", response_model=ApiResponse[CustomerOut], summary="客户详情")
+def get_customer(
+    session: DbSession,
+    _: CurrentUser,
+    customer_id: Annotated[int, Path(ge=1)],
+) -> ApiResponse[CustomerOut]:
+    """按主键查询客户详情。"""
+    return ok(customer_service.get_customer(session, customer_id))
+
+
+@router.put("/{customer_id}", response_model=ApiResponse[CustomerOut], summary="编辑客户")
+def update_customer(
+    session: DbSession,
+    current_user: CurrentUser,
+    customer_id: Annotated[int, Path(ge=1)],
+    payload: CustomerUpdateIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[CustomerOut]:
+    """编辑客户档案。
+
+    **不含归属** —— 归属调整统一走 ``PUT /{id}/ownership``；编辑表单即使
+    提交 ownerPersonnelId 也会被忽略，归属以归属端点为准。
+    """
+    return ok(
+        customer_service.update_customer(
+            session, customer_id, payload, operator=current_user, meta=meta
+        ),
+        message="已保存",
+    )
+
+
+@router.delete("/{customer_id}", response_model=ApiResponse[None], summary="删除客户")
+def delete_customer(
+    session: DbSession,
+    current_user: CurrentUser,
+    customer_id: Annotated[int, Path(ge=1)],
+    meta: RequestMetaDep,
+) -> ApiResponse[None]:
+    """删除客户（物理删除）。
+
+    被业务数据引用的保护检查待业务数据表（V5）就位后补充。
+    """
+    customer_service.delete_customer(session, customer_id, operator=current_user, meta=meta)
+    return ok(message="客户已删除")
+
+
+@router.put(
+    "/{customer_id}/ownership",
+    response_model=ApiResponse[CustomerOut],
+    summary="归属变更",
+)
+def change_ownership(
+    session: DbSession,
+    current_user: CurrentUser,
+    customer_id: Annotated[int, Path(ge=1)],
+    payload: CustomerOwnershipIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[CustomerOut]:
+    """调整客户归属（唯一写归属留痕的入口）。
+
+    只影响变更后核算，不回溯历史（对照表 #19）；新旧归属人员ID任一为 null
+    表示清空归属。新旧归属相同报「归属未变化」。
+    """
+    return ok(
+        customer_service.change_ownership(
+            session, customer_id, payload, operator=current_user, meta=meta
+        ),
+        message="归属已调整",
+    )
+
+
+@router.get(
+    "/{customer_id}/ownership-logs",
+    response_model=ApiResponse[PageResult[OwnershipChangeLogOut]],
+    summary="归属变更留痕",
+)
+def list_ownership_logs(
+    session: DbSession,
+    _: CurrentUser,
+    customer_id: Annotated[int, Path(ge=1)],
+    query: Annotated[PageQuery, Query()],
+) -> ApiResponse[PageResult[OwnershipChangeLogOut]]:
+    """查某客户的归属变更留痕（分页，按变更时间倒序，附新旧归属人员姓名）。"""
+    return ok(customer_service.list_ownership_logs(session, customer_id, query))

backend/app/repositories/customer_repository.py

@@ -0,0 +1,181 @@
+"""客户数据访问（含归属变更留痕）。
+
+本层只 flush 不 commit —— 事务边界由 services 层控制。
+"""
+
+from sqlalchemy import ColumnElement, Row, Select, func, or_, select
+from sqlalchemy.orm import Session, aliased
+
+from app.models import Customer, OwnershipChangeLog, Personnel
+from app.models.customer import OWNER_TYPE_CUSTOMER
+from app.repositories.query_utils import fuzzy
+
+# 归属留痕表对人员 ID 无外键固化：新旧人员姓名靠对 personnel 的两次外联带出，
+# 人员被删除后姓名为 null，行仍完整可读
+_old_owner = aliased(Personnel)
+_new_owner = aliased(Personnel)
+
+# ==================== 读取 ====================
+
+
+def get_by_id(session: Session, customer_id: int) -> Customer | None:
+    """按主键查询客户。"""
+    return session.get(Customer, customer_id)
+
+
+def get_with_owner_name(session: Session, customer_id: int) -> Row | None:
+    """查单个客户并带出归属人员姓名（详情、编辑回显与归属变更共用）。"""
+    stmt = _base_select().where(Customer.id == customer_id)
+    return session.execute(stmt).first()
+
+
+def get_by_company_owner(session: Session, company_name: str, owner_name: str) -> Customer | None:
+    """按「公司名称 + 负责人」查客户（uk_customers_company_owner 的先查后写兜底）。"""
+    stmt = select(Customer).where(
+        Customer.company_name == company_name,
+        Customer.owner_name == owner_name,
+    )
+    return session.execute(stmt).scalar_one_or_none()
+
+
+def list_customers(
+    session: Session,
+    *,
+    offset: int,
+    limit: int,
+    keyword: str | None = None,
+    customer_type: str | None = None,
+    owner_personnel_id: int | None = None,
+) -> list[Row]:
+    """分页查询客户，行内容为 (客户, 归属人员姓名)。
+
+    归属人员外键非空但统一用 outerjoin：人员行不可考（被物理删除的脏数据）
+    不应让客户行整个查不出来（同 personnel_repository 的韧性口径）。
+    排序按 id 倒序：客户无排序字段，新增的靠前符合列表页习惯。
+    """
+    stmt = _apply_filters(
+        _base_select(),
+        keyword=keyword,
+        customer_type=customer_type,
+        owner_personnel_id=owner_personnel_id,
+    )
+    stmt = stmt.order_by(Customer.id.desc()).offset(offset).limit(limit)
+    return list(session.execute(stmt).all())
+
+
+def count_customers(
+    session: Session,
+    *,
+    keyword: str | None = None,
+    customer_type: str | None = None,
+    owner_personnel_id: int | None = None,
+) -> int:
+    """统计符合条件的客户总数，供分页使用。
+
+    筛选条件全部落在主表上，计数无需 join（列表与计数的筛选口径由
+    ``_apply_filters`` 单点保证）。
+    """
+    stmt = _apply_filters(
+        select(func.count(Customer.id)),
+        keyword=keyword,
+        customer_type=customer_type,
+        owner_personnel_id=owner_personnel_id,
+    )
+    return int(session.execute(stmt).scalar_one())
+
+
+def list_ownership_logs(
+    session: Session,
+    *,
+    customer_id: int,
+    offset: int,
+    limit: int,
+) -> list[Row]:
+    """查某客户的归属变更留痕，行内容为 (留痕, 旧归属人名, 新归属人名)。
+
+    按 (create_time, id) 双键倒序：create_time 是秒级精度，同一秒内多条
+    记录排序不确定，id 兜底保证翻页稳定（与操作日志同一理由）。
+    """
+    stmt = (
+        select(OwnershipChangeLog, _old_owner.personnel_name, _new_owner.personnel_name)
+        .outerjoin(_old_owner, OwnershipChangeLog.old_owner_personnel_id == _old_owner.id)
+        .outerjoin(_new_owner, OwnershipChangeLog.new_owner_personnel_id == _new_owner.id)
+        .where(
+            OwnershipChangeLog.owner_type == OWNER_TYPE_CUSTOMER,
+            OwnershipChangeLog.customer_id == customer_id,
+        )
+        .order_by(OwnershipChangeLog.create_time.desc(), OwnershipChangeLog.id.desc())
+        .offset(offset)
+        .limit(limit)
+    )
+    return list(session.execute(stmt).all())
+
+
+def count_ownership_logs(session: Session, *, customer_id: int) -> int:
+    """统计某客户的归属留痕总数，供分页使用。"""
+    stmt = select(func.count(OwnershipChangeLog.id)).where(
+        OwnershipChangeLog.owner_type == OWNER_TYPE_CUSTOMER,
+        OwnershipChangeLog.customer_id == customer_id,
+    )
+    return int(session.execute(stmt).scalar_one())
+
+
+def _base_select() -> Select:
+    """客户主表外联归属人员姓名。"""
+    return select(Customer, Personnel.personnel_name).outerjoin(
+        Personnel, Customer.owner_personnel_id == Personnel.id
+    )
+
+
+def _apply_filters(
+    stmt: Select,
+    *,
+    keyword: str | None,
+    customer_type: str | None,
+    owner_personnel_id: int | None,
+) -> Select:
+    """应用通用筛选条件。列表与计数共用，避免两处口径不一致。"""
+    if keyword:
+        stmt = stmt.where(_keyword_clause(keyword))
+    if customer_type:
+        stmt = stmt.where(Customer.customer_type == customer_type)
+    if owner_personnel_id is not None:
+        stmt = stmt.where(Customer.owner_personnel_id == owner_personnel_id)
+    return stmt
+
+
+def _keyword_clause(keyword: str) -> ColumnElement[bool]:
+    """关键词条件：公司名称或客户负责人两列 OR 模糊。
+
+    借 ``query_utils.fuzzy`` 在空 select 上生成单列条件后取 whereclause 做
+    OR 合并 —— LIKE 通配符转义口径单点维护；两处手写 ``contains`` 迟早有
+    一处漏掉 ``autoescape``（人名含下划线是常态，不转义会当单字符通配执行）。
+    """
+    probe = select(Customer.id)
+    return or_(
+        fuzzy(probe, Customer.company_name, keyword).whereclause,
+        fuzzy(probe, Customer.owner_name, keyword).whereclause,
+    )
+
+
+# ==================== 写入 ====================
+
+
+def create(session: Session, customer: Customer) -> Customer:
+    """新增客户。只 flush 以取得自增 id。"""
+    session.add(customer)
+    session.flush()
+    return customer
+
+
+def delete(session: Session, customer: Customer) -> None:
+    """删除客户。调用方须先完成引用保护校验。"""
+    session.delete(customer)
+    session.flush()
+
+
+def add_ownership_log(session: Session, log: OwnershipChangeLog) -> OwnershipChangeLog:
+    """写入一条归属变更留痕。只 flush 不 commit，与主表归属更新同事务提交。"""
+    session.add(log)
+    session.flush()
+    return log

backend/app/schemas/customer.py

@@ -0,0 +1,147 @@
+"""客户管理的请求/响应模型（需求 2.1.3）。"""
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.core.response import PageQuery
+
+
+class CustomerOut(BaseModel):
+    """客户响应（列表、详情与写操作回包共用）。
+
+    owner_personnel_name 由 LEFT JOIN 带出：归属人员不可考（被物理删除的
+    脏数据）时为 null，owner_personnel_id 仍返回原值供追溯 —— 与归属留痕
+    「ID 固化、姓名随联表」同一口径。
+    """
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    company_name: str
+    owner_name: str
+    contact_person: str | None = None
+    contact_phone: str | None = None
+    address: str | None = None
+    industry: str | None = None
+    customer_type: str
+    owner_personnel_id: int | None = None
+    owner_personnel_name: str | None = None
+    remark: str | None = None
+
+
+class CustomerCreateIn(BaseModel):
+    """新增客户。
+
+    owner_personnel_id 为**初始归属**，可空（客户先建档、后派归属也合法）；
+    归属的后续调整不走本模型，统一走 ``PUT /{id}/ownership``（见
+    ``CustomerUpdateIn`` 的说明）。camelCase 别名与蛇形字段名均可绑定，
+    兼容前端两种拼写。
+    """
+
+    model_config = ConfigDict(str_strip_whitespace=True, populate_by_name=True)
+
+    company_name: str = Field(min_length=1, max_length=128, description="公司名称")
+    owner_name: str = Field(
+        min_length=1,
+        max_length=64,
+        description="客户负责人姓名（客户公司的负责人人名，非系统内营销人员）",
+    )
+    contact_person: str | None = Field(default=None, max_length=64, description="联系人")
+    contact_phone: str | None = Field(default=None, max_length=20, description="联系电话")
+    address: str | None = Field(default=None, max_length=255, description="地址")
+    industry: str | None = Field(default=None, max_length=64, description="所属行业")
+    customer_type: str = Field(
+        min_length=1,
+        max_length=32,
+        description="客户分类：personal个人户 manager_personal经理个人户 enterprise企业户",
+    )
+    owner_personnel_id: int | None = Field(
+        default=None,
+        ge=1,
+        alias="ownerPersonnelId",
+        description="归属负责营销人员ID（初始归属，可空）",
+    )
+    remark: str | None = Field(default=None, max_length=255, description="备注")
+
+
+class CustomerUpdateIn(BaseModel):
+    """编辑客户（全量替换，前端表单全量提交）。
+
+    **不含 owner_personnel_id** —— 归属调整统一走 ``PUT /{id}/ownership``，
+    只有那条路径才写归属变更留痕。若在本端点放开该字段，会出现「改了归属
+    却无留痕」的旁路，破坏留痕的 append-only 完整性；编辑表单即使提交
+    ownerPersonnelId，pydantic 默认忽略未声明字段，归属以归属端点为准。
+    """
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    company_name: str = Field(min_length=1, max_length=128, description="公司名称")
+    owner_name: str = Field(min_length=1, max_length=64, description="客户负责人姓名")
+    contact_person: str | None = Field(default=None, max_length=64, description="联系人")
+    contact_phone: str | None = Field(default=None, max_length=20, description="联系电话")
+    address: str | None = Field(default=None, max_length=255, description="地址")
+    industry: str | None = Field(default=None, max_length=64, description="所属行业")
+    customer_type: str = Field(
+        min_length=1,
+        max_length=32,
+        description="客户分类：personal个人户 manager_personal经理个人户 enterprise企业户",
+    )
+    remark: str | None = Field(default=None, max_length=255, description="备注")
+
+
+class CustomerQuery(PageQuery):
+    """客户列表查询条件。
+
+    customer_type / owner_personnel_id 同时接受蛇形（项目既有契约，见前端
+    `types.ts` 的说明）与驼峰别名，两种拼写不并存即可。
+    """
+
+    keyword: str | None = Field(
+        default=None, max_length=64, description="公司名称或客户负责人，模糊匹配"
+    )
+    customer_type: str | None = Field(
+        default=None, max_length=32, alias="customerType", description="客户分类筛选"
+    )
+    owner_personnel_id: int | None = Field(
+        default=None, ge=1, alias="ownerPersonnelId", description="归属人员ID筛选"
+    )
+
+
+class CustomerOwnershipIn(BaseModel):
+    """归属变更入参。
+
+    new_owner_personnel_id 可空：null 表示清空归属（客户暂无负责人也合法）。
+    """
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    new_owner_personnel_id: int | None = Field(
+        default=None,
+        ge=1,
+        alias="newOwnerPersonnelId",
+        description="变更后归属人员ID，null=清空归属",
+    )
+    reason: str | None = Field(default=None, max_length=255, description="变更原因")
+
+
+class OwnershipChangeLogOut(BaseModel):
+    """归属变更留痕响应。
+
+    新旧归属人员姓名由 LEFT JOIN 带出，人员被删除后为 null；固化的
+    ``*_personnel_id`` 始终原样返回 —— 留痕表对人员 ID 无外键固化
+    （见 `models/customer.py` 的论证），姓名可随人员消失，ID 不可。
+    """
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    customer_id: int
+    old_owner_personnel_id: int | None = None
+    new_owner_personnel_id: int | None = None
+    old_owner_name: str | None = None
+    new_owner_name: str | None = None
+    change_reason: str | None = None
+    operator_id: int | None = None
+    operator_name: str
+    create_time: datetime

backend/app/services/customer_service.py

@@ -0,0 +1,425 @@
+"""客户管理业务编排（需求 2.1.3）。
+
+**事务边界在本层**：repository 只 flush，commit/rollback 只写在这里。
+
+- 「公司名称 + 负责人」组合唯一（uk_customers_company_owner）：先查后写不是
+  原子操作，并发提交撞唯一键时捕 IntegrityError 判 1062 翻译为 400
+  （同 M1 的并发兜底模式）
+- **归属变更的裁决语义是「只影响变更后核算，不回溯历史」（对照表 #19）**：
+  实现为同一事务内「更新 customers.owner_personnel_id + 插入一条
+  ownership_change_logs」，一次 commit —— 两步拆开提交时，中途失败会出现
+  「主表已换人、留痕缺失」的静默不一致，事后追溯无据可查
+- 归属调整只有 ``change_ownership`` 一条路径；编辑端点不触归属，
+  防止出现「改归属不留痕」的旁路
+"""
+
+import logging
+
+from sqlalchemy import Row
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.core.constants import (
+    ACTION_CREATE,
+    ACTION_DELETE,
+    ACTION_UPDATE,
+    CUSTOMER_TYPES,
+    MODULE_CUSTOMER,
+)
+from app.core.exceptions import BadRequestError, NotFoundError
+from app.core.response import PageQuery, PageResult
+from app.models import Customer, OwnershipChangeLog, Personnel, User
+from app.models.customer import OWNER_TYPE_CUSTOMER
+from app.repositories import customer_repository, personnel_repository
+from app.schemas.customer import (
+    CustomerCreateIn,
+    CustomerOut,
+    CustomerOwnershipIn,
+    CustomerQuery,
+    CustomerUpdateIn,
+    OwnershipChangeLogOut,
+)
+from app.services import audit_service
+from app.services.audit_service import AuditRecord, RequestMeta
+
+logger = logging.getLogger(__name__)
+
+# personnel.status 的在职值（列注释口径：1在职 0离职）。离职人员不可再被
+# 客户归属引用；人员侧抽出共享常量后此处替换
+_PERSONNEL_STATUS_ACTIVE = 1
+
+
+def _to_out(customer: Customer, owner_name: str | None) -> CustomerOut:
+    """ORM 对象转响应模型。禁止把 ORM 对象直接当 response_model。"""
+    return CustomerOut(
+        id=customer.id,
+        company_name=customer.company_name,
+        owner_name=customer.owner_name,
+        contact_person=customer.contact_person,
+        contact_phone=customer.contact_phone,
+        address=customer.address,
+        industry=customer.industry,
+        customer_type=customer.customer_type,
+        owner_personnel_id=customer.owner_personnel_id,
+        owner_personnel_name=owner_name,
+        remark=customer.remark,
+    )
+
+
+def _to_log_out(row: Row) -> OwnershipChangeLogOut:
+    """留痕行 (留痕, 旧归属人名, 新归属人名) 转响应模型。"""
+    log, old_name, new_name = row
+    return OwnershipChangeLogOut(
+        id=log.id,
+        customer_id=log.customer_id,
+        old_owner_personnel_id=log.old_owner_personnel_id,
+        new_owner_personnel_id=log.new_owner_personnel_id,
+        old_owner_name=old_name,
+        new_owner_name=new_name,
+        change_reason=log.change_reason,
+        operator_id=log.operator_id,
+        operator_name=log.operator_name,
+        create_time=log.create_time,
+    )
+
+
+def _ensure_customer_type(customer_type: str) -> None:
+    """客户分类必须是注册表内的三个值之一（前端为受控下拉，后端仍须校验）。
+
+    customer_type 是 M3 计提规则匹配的锚点，脏值会让该客户在核算时匹配不到
+    任何计提规则而静默漏算。
+    """
+    if customer_type not in CUSTOMER_TYPES:
+        raise BadRequestError("客户分类不合法")
+
+
+def _ensure_owner_personnel(session: Session, personnel_id: int) -> Personnel:
+    """归属人员必须存在且在职，返回人员行。
+
+    「不存在」与「已离职」合并为同一文案：人员 ID 是否有效不构成需要
+    向调用方区分的信息（同登录防枚举的思路）。
+
+    :raises BadRequestError: 归属人员不存在或已离职
+    """
+    person = personnel_repository.get_by_id(session, personnel_id)
+    if person is None or person.status != _PERSONNEL_STATUS_ACTIVE:
+        raise BadRequestError("归属人员不存在或已离职")
+    return person
+
+
+def _is_duplicate_entry(exc: IntegrityError) -> bool:
+    """判断 IntegrityError 是否为 MySQL 唯一键冲突（错误码 1062）。
+
+    customers 上还有归属人员外键（RESTRICT），并发窗口内删除人员等场景
+    会抛 1452 —— 不能一并翻译成「已存在」，否则用户会被错误提示带偏；
+    非 1062 原样上抛，由全局处理器按 500 收敛。
+    """
+    args = getattr(exc.orig, "args", ())
+    return bool(args) and args[0] == 1062
+
+
+def list_customers(session: Session, query: CustomerQuery) -> PageResult[CustomerOut]:
+    """分页查询客户（附归属人员姓名）。"""
+    rows = customer_repository.list_customers(
+        session,
+        offset=query.offset,
+        limit=query.page_size,
+        keyword=query.keyword,
+        customer_type=query.customer_type,
+        owner_personnel_id=query.owner_personnel_id,
+    )
+    total = customer_repository.count_customers(
+        session,
+        keyword=query.keyword,
+        customer_type=query.customer_type,
+        owner_personnel_id=query.owner_personnel_id,
+    )
+    return PageResult[CustomerOut](
+        records=[_to_out(customer, owner_name) for customer, owner_name in rows],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )
+
+
+def get_customer(session: Session, customer_id: int) -> CustomerOut:
+    """客户详情。
+
+    :raises NotFoundError: 客户不存在
+    """
+    row = customer_repository.get_with_owner_name(session, customer_id)
+    if row is None:
+        raise NotFoundError("客户不存在")
+    customer, owner_name = row
+    return _to_out(customer, owner_name)
+
+
+def create_customer(
+    session: Session,
+    payload: CustomerCreateIn,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> CustomerOut:
+    """新增客户。
+
+    :raises BadRequestError: 客户分类不合法 / 归属人员不存在或已离职 /
+                            「公司名称+负责人」已存在（含并发兜底）
+    """
+    _ensure_customer_type(payload.customer_type)
+    owner_name: str | None = None
+    if payload.owner_personnel_id is not None:
+        owner_name = _ensure_owner_personnel(session, payload.owner_personnel_id).personnel_name
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_CUSTOMER,
+            action=ACTION_CREATE,
+            summary=f"新增客户 {payload.company_name}({payload.owner_name})",
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        if (
+            customer_repository.get_by_company_owner(
+                session, payload.company_name, payload.owner_name
+            )
+            is not None
+        ):
+            raise BadRequestError("该公司名称与负责人的客户已存在")
+
+        customer = Customer(
+            company_name=payload.company_name,
+            owner_name=payload.owner_name,
+            contact_person=payload.contact_person,
+            contact_phone=payload.contact_phone,
+            address=payload.address,
+            industry=payload.industry,
+            customer_type=payload.customer_type,
+            owner_personnel_id=payload.owner_personnel_id,
+            remark=payload.remark,
+        )
+        try:
+            customer_repository.create(session, customer)
+            # 响应在 commit 之前构造（同 user_service 顶部说明）
+            out = _to_out(customer, owner_name)
+            session.commit()
+        except IntegrityError as exc:
+            # 先查后写不是原子操作：并发提交同一「公司名称+负责人」会撞
+            # uk_customers_company_owner，回滚并翻译为可读的 400，而非 500
+            session.rollback()
+            if _is_duplicate_entry(exc):
+                raise BadRequestError("该公司名称与负责人的客户已存在") from exc
+            raise
+
+        logger.info("新增客户 | customer_id=%s | company=%s", customer.id, payload.company_name)
+        return out
+
+
+def update_customer(
+    session: Session,
+    customer_id: int,
+    payload: CustomerUpdateIn,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> CustomerOut:
+    """编辑客户档案（全量替换）。
+
+    **不触归属**：owner_personnel_id 保持原值，调整统一走
+    ``PUT /{id}/ownership``（唯一写留痕的路径），见 ``CustomerUpdateIn`` 说明。
+
+    :raises NotFoundError: 客户不存在
+    :raises BadRequestError: 客户分类不合法 / 「公司名称+负责人」已存在（含并发兜底）
+    """
+    _ensure_customer_type(payload.customer_type)
+    # 只读预取在审计上下文之前：目标不存在（NotFound 失败）时也能留下可读记录，
+    # 摘要与回包都要用客户行；归属姓名本端点不变更，一并带出
+    row = customer_repository.get_with_owner_name(session, customer_id)
+    customer = row[0] if row else None
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_CUSTOMER,
+            action=ACTION_UPDATE,
+            summary=f"修改客户 {customer.company_name if customer else f'#{customer_id}'}",
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        if row is None:
+            raise NotFoundError("客户不存在")
+        customer, owner_name = row
+
+        duplicate = customer_repository.get_by_company_owner(
+            session, payload.company_name, payload.owner_name
+        )
+        if duplicate is not None and duplicate.id != customer_id:
+            raise BadRequestError("该公司名称与负责人的客户已存在")
+
+        customer.company_name = payload.company_name
+        customer.owner_name = payload.owner_name
+        customer.contact_person = payload.contact_person
+        customer.contact_phone = payload.contact_phone
+        customer.address = payload.address
+        customer.industry = payload.industry
+        customer.customer_type = payload.customer_type
+        # owner_personnel_id 刻意不赋值 —— 归属调整只走 /ownership
+
+        try:
+            # 响应在 commit 之前构造（同 user_service 顶部说明）
+            out = _to_out(customer, owner_name)
+            session.commit()
+        except IntegrityError as exc:
+            # 把自身以外的组合改成已存在的「公司名称+负责人」时，先查后写
+            # 的并发窗口会撞唯一键，兜底同 create_customer
+            session.rollback()
+            if _is_duplicate_entry(exc):
+                raise BadRequestError("该公司名称与负责人的客户已存在") from exc
+            raise
+
+        logger.info("修改客户 | customer_id=%s | company=%s", customer_id, payload.company_name)
+        return out
+
+
+def delete_customer(
+    session: Session,
+    customer_id: int,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> None:
+    """删除客户（物理删除）。
+
+    **被业务数据引用的保护检查待业务数据表（V5）就位后补充**：customers
+    当前仅被 ownership_change_logs 以无外键的固化 ID 引用，而留痕本就要求
+    「被引用方删除后记录仍完整可读」（见 models/customer.py 的固化论证），
+    故现在删除不会产生悬挂外键。
+
+    :raises NotFoundError: 客户不存在
+    """
+    customer = customer_repository.get_by_id(session, customer_id)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_CUSTOMER,
+            action=ACTION_DELETE,
+            summary=f"删除客户 {customer.company_name if customer else f'#{customer_id}'}",
+        ),
+        meta=meta,
+    ):
+        if customer is None:
+            raise NotFoundError("客户不存在")
+
+        company_name = customer.company_name  # 固化：对象随删除失效
+        customer_repository.delete(session, customer)
+        session.commit()
+        logger.info("删除客户 | customer_id=%s | company=%s", customer_id, company_name)
+
+
+def change_ownership(
+    session: Session,
+    customer_id: int,
+    payload: CustomerOwnershipIn,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> CustomerOut:
+    """归属变更：同一事务内「改主表归属 + 插留痕」，一次 commit。
+
+    裁决口径为「只影响变更后核算，不回溯历史」（对照表 #19）：历史核算结果
+    已由 M3 台账固化，留痕仅承担追溯职责，故无 effective_date 时点化设计。
+
+    :raises NotFoundError: 客户不存在
+    :raises BadRequestError: 归属人员不存在或已离职 / 归属未变化
+    """
+    # 只读预取在审计上下文之前：摘要需要客户名与新旧归属人名；校验仍在
+    # 上下文内进行，失败同样要留下可追溯的记录
+    row = customer_repository.get_with_owner_name(session, customer_id)
+    old_person = (
+        personnel_repository.get_by_id(session, row[0].owner_personnel_id)
+        if row is not None and row[0].owner_personnel_id is not None
+        else None
+    )
+    new_person = (
+        personnel_repository.get_by_id(session, payload.new_owner_personnel_id)
+        if payload.new_owner_personnel_id is not None
+        else None
+    )
+    old_label = old_person.personnel_name if old_person else "无"
+    new_label = new_person.personnel_name if new_person else "无"
+    customer_name = row[0].company_name if row else f"#{customer_id}"
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_CUSTOMER,
+            action=ACTION_UPDATE,
+            summary=f"调整客户归属 {customer_name}：{old_label} → {new_label}",
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        if row is None:
+            raise NotFoundError("客户不存在")
+        customer, _ = row
+
+        if payload.new_owner_personnel_id is not None:
+            _ensure_owner_personnel(session, payload.new_owner_personnel_id)
+        old_owner_id = customer.owner_personnel_id
+        if old_owner_id == payload.new_owner_personnel_id:
+            raise BadRequestError("归属未变化")
+
+        # 主表更新与留痕插入同事务：只在本方法内一次 commit（见模块 docstring）
+        customer.owner_personnel_id = payload.new_owner_personnel_id
+        customer_repository.add_ownership_log(
+            session,
+            OwnershipChangeLog(
+                owner_type=OWNER_TYPE_CUSTOMER,
+                customer_id=customer.id,
+                old_owner_personnel_id=old_owner_id,
+                new_owner_personnel_id=payload.new_owner_personnel_id,
+                change_reason=payload.reason,
+                operator_id=operator.id,
+                operator_name=operator.username,
+            ),
+        )
+        # 响应在 commit 之前构造（同 user_service 顶部说明）；归属人名以新值为准
+        out = _to_out(customer, new_person.personnel_name if new_person else None)
+        session.commit()
+        logger.info(
+            "调整客户归属 | customer_id=%s | old=%s | new=%s",
+            customer_id,
+            old_owner_id,
+            payload.new_owner_personnel_id,
+        )
+        return out
+
+
+def list_ownership_logs(
+    session: Session,
+    customer_id: int,
+    query: PageQuery,
+) -> PageResult[OwnershipChangeLogOut]:
+    """某客户的归属变更留痕（分页，按变更时间倒序）。
+
+    :raises NotFoundError: 客户不存在
+    """
+    if customer_repository.get_by_id(session, customer_id) is None:
+        raise NotFoundError("客户不存在")
+    rows = customer_repository.list_ownership_logs(
+        session,
+        customer_id=customer_id,
+        offset=query.offset,
+        limit=query.page_size,
+    )
+    total = customer_repository.count_ownership_logs(session, customer_id=customer_id)
+    return PageResult[OwnershipChangeLogOut](
+        records=[_to_log_out(row) for row in rows],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )