PayrollManagementSystem/backend/app/repositories/customer_repository.py

"""客户数据访问（含归属变更留痕）。

本层只 flush 不 commit —— 事务边界由 services 层控制。
"""

from sqlalchemy import ColumnElement, Row, Select, func, or_, select
from sqlalchemy.orm import Session

from app.models import Customer, 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,
) -> 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]:
    """查某客户的归属变更留痕（共享实现，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:
    """客户主表外联归属人员姓名。"""
    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，与主表更新同事务。"""
    return ownership_log_repository.add(session, log)