208 lines
7.3 KiB
Python
208 lines
7.3 KiB
Python
"""客户数据访问（含归属变更留痕）。
|
||
|
||
本层只 flush 不 commit —— 事务边界由 services 层控制。
|
||
"""
|
||
|
||
from sqlalchemy import ColumnElement, Row, Select, func, or_, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import Customer, CustomerRating, OwnershipChangeLog, Personnel
|
||
from app.models.customer import OWNER_TYPE_CUSTOMER
|
||
from app.repositories import ownership_log_repository
|
||
from app.repositories.query_utils import fuzzy
|
||
|
||
# ==================== 读取 ====================
|
||
|
||
|
||
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)
|
||
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,
|
||
order_field: str | None = None,
|
||
order_type: str | None = None,
|
||
) -> list[Row]:
|
||
"""分页查询客户，行内容为 (客户, 归属人员姓名, 评级总分, 评级等级)。
|
||
|
||
归属人员/评级行均用 outerjoin：关联行不可考（人员被物理删除、客户
|
||
未评级）不应让客户行整个查不出来（同 personnel_repository 韧性口径）。
|
||
默认排序按 id 倒序：新增的靠前符合列表页习惯；评级排序见 ``_apply_order``。
|
||
"""
|
||
stmt = _apply_filters(
|
||
_base_select(),
|
||
keyword=keyword,
|
||
customer_type=customer_type,
|
||
owner_personnel_id=owner_personnel_id,
|
||
)
|
||
stmt = _apply_order(stmt, order_field=order_field, order_type=order_type)
|
||
stmt = stmt.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]:
|
||
"""查某客户的归属变更留痕（共享实现，owner_type/customer_id 参数化）。"""
|
||
return ownership_log_repository.list_by_ref(
|
||
session,
|
||
owner_type=OWNER_TYPE_CUSTOMER,
|
||
ref_column=OwnershipChangeLog.customer_id,
|
||
ref_id=customer_id,
|
||
offset=offset,
|
||
limit=limit,
|
||
)
|
||
|
||
|
||
def count_ownership_logs(session: Session, *, customer_id: int) -> int:
|
||
"""统计某客户的归属留痕总数，供分页使用。"""
|
||
return ownership_log_repository.count_by_ref(
|
||
session,
|
||
owner_type=OWNER_TYPE_CUSTOMER,
|
||
ref_column=OwnershipChangeLog.customer_id,
|
||
ref_id=customer_id,
|
||
)
|
||
|
||
|
||
def _base_select() -> Select:
|
||
"""客户主表外联归属人员姓名与评级快照（等级名/总分，列表评级列免 N+1）。"""
|
||
return (
|
||
select(
|
||
Customer,
|
||
Personnel.personnel_name,
|
||
CustomerRating.total_score,
|
||
CustomerRating.grade_name,
|
||
)
|
||
.outerjoin(Personnel, Customer.owner_personnel_id == Personnel.id)
|
||
.outerjoin(CustomerRating, CustomerRating.customer_id == Customer.id)
|
||
)
|
||
|
||
|
||
def _apply_order(stmt: Select, *, order_field: str | None, order_type: str | None) -> Select:
|
||
"""应用列表排序。默认 id 倒序；order_field=rating 按评级总分排序。
|
||
|
||
等级由总分判定、与总分同向，故按总分即等价按等级排序。MySQL 没有
|
||
``NULLS LAST`` 语法，用布尔键 ``total_score IS NULL`` 升序模拟 ——
|
||
未评级行恒在末尾，正序/倒序一致；同分以 id 倒序兜底保证次序稳定。
|
||
非法字段/方向不落到 SQL（schema Literal 已在入口拦 400），此处静默
|
||
回退默认排序作为纵深防御。
|
||
"""
|
||
if order_field == "rating" and order_type in ("asc", "desc"):
|
||
score_order = (
|
||
CustomerRating.total_score.asc()
|
||
if order_type == "asc"
|
||
else CustomerRating.total_score.desc()
|
||
)
|
||
return stmt.order_by(CustomerRating.total_score.is_(None), score_order, Customer.id.desc())
|
||
return stmt.order_by(Customer.id.desc())
|
||
|
||
|
||
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，与主表更新同事务。"""
|
||
return ownership_log_repository.add(session, log)
|