PayrollManagementSystem/backend/app/repositories/business_repository.py

"""业务数据仓储：业务单、明细、状态留痕与归属留痕。

只读写数据并 flush；commit/rollback 与业务异常一律在 service 层。
归属留痕复用 ownership_change_logs（owner_type=2 为业务维度，常量在
models/customer.py 定义）。
"""

from datetime import date

from sqlalchemy import Row, Select, delete, exists, func, or_, select
from sqlalchemy.orm import Session, aliased

from app.models import (
    Business,
    BusinessItem,
    BusinessStatusLog,
    Customer,
    OwnershipChangeLog,
    Personnel,
    Receipt,
    Shipment,
)
from app.models.customer import OWNER_TYPE_BUSINESS


def get_by_id(session: Session, business_id: int, *, for_update: bool = False) -> Business | None:
    """按主键取业务行；for_update 供删除保护、编辑门槛与状态机使用。"""
    stmt = select(Business).where(Business.id == business_id)
    if for_update:
        stmt = stmt.with_for_update()
    return session.execute(stmt).scalars().first()


def has_docs(session: Session, business_id: int) -> bool:
    """业务名下是否已有发货单或回款（换客户/删除保护的引用检查）。

    单条 OR exists 一次往返；新增挂靠业务的单据类型时在此补充条件，
    编辑与删除两处共用同一判定，防「delete 补了检查、update 漏补」的分叉。
    """
    stmt = select(
        or_(
            exists().where(Shipment.business_id == business_id),
            exists().where(Receipt.business_id == business_id),
        )
    )
    return bool(session.execute(stmt).scalar())


def create(session: Session, business: Business) -> None:
    """新增业务行（不 commit，flush 让自增 id 就位）。"""
    session.add(business)
    session.flush()


def get_with_names(
    session: Session, business_id: int, *, for_update: bool = False
) -> tuple[Business, str, str | None, str | None] | None:
    """业务行 + 客户名 + 负责人名 + 协助人名（DTO 与审计摘要共用）。

    for_update 供编辑/状态流转/归属变更使用：与发货/回款写路径的
    get_by_id(for_update=True) 同一锁口径，编辑事务与单据写事务互斥，
    「有发货禁换客户」的快照读校验不再被并发新增旁路。
    """
    owner = aliased(Personnel)
    assistant = aliased(Personnel)
    stmt = (
        select(Business, Customer.company_name, owner.personnel_name, assistant.personnel_name)
        .join(Customer, Customer.id == Business.customer_id)
        .outerjoin(owner, owner.id == Business.owner_personnel_id)
        .outerjoin(assistant, assistant.id == Business.assistant_personnel_id)
        .where(Business.id == business_id)
    )
    if for_update:
        stmt = stmt.with_for_update()
    return session.execute(stmt).first()


def _apply_filters(
    stmt: Select,
    *,
    keyword: str | None,
    status: int | None,
    customer_id: int | None,
    owner_personnel_id: int | None,
    date_from: date | None,
    date_to: date | None,
) -> Select:
    """业务列表的筛选条件（调用方须已 join Customer）。"""
    if keyword:
        # 业务编号/公司名称双列 OR 模糊；contains(autoescape) 同 query_utils.fuzzy
        # 的转义口径（fuzzy 为单列 where 语义，双列 OR 在此直接组合）
        stmt = stmt.where(
            or_(
                Business.business_no.contains(keyword, autoescape=True),
                Customer.company_name.contains(keyword, autoescape=True),
            )
        )
    if status is not None:
        stmt = stmt.where(Business.status == status)
    if customer_id is not None:
        stmt = stmt.where(Business.customer_id == customer_id)
    if owner_personnel_id is not None:
        stmt = stmt.where(Business.owner_personnel_id == owner_personnel_id)
    if date_from is not None:
        stmt = stmt.where(Business.business_date >= date_from)
    if date_to is not None:
        stmt = stmt.where(Business.business_date <= date_to)
    return stmt


def list_businesses(
    session: Session,
    *,
    offset: int,
    limit: int,
    keyword: str | None,
    status: int | None,
    customer_id: int | None,
    owner_personnel_id: int | None,
    date_from: date | None,
    date_to: date | None,
) -> list[tuple[Business, str, str | None, str | None]]:
    """分页查询业务（附客户名/负责人名/协助人名），按 id 倒序。"""
    owner = aliased(Personnel)
    assistant = aliased(Personnel)
    stmt = select(
        Business, Customer.company_name, owner.personnel_name, assistant.personnel_name
    ).join(Customer, Customer.id == Business.customer_id)
    stmt = _apply_filters(
        stmt,
        keyword=keyword,
        status=status,
        customer_id=customer_id,
        owner_personnel_id=owner_personnel_id,
        date_from=date_from,
        date_to=date_to,
    )
    stmt = (
        stmt.outerjoin(owner, owner.id == Business.owner_personnel_id)
        .outerjoin(assistant, assistant.id == Business.assistant_personnel_id)
        .order_by(Business.id.desc())
        .offset(offset)
        .limit(limit)
    )
    return list(session.execute(stmt).all())


def count_businesses(
    session: Session,
    *,
    keyword: str | None,
    status: int | None,
    customer_id: int | None,
    owner_personnel_id: int | None,
    date_from: date | None,
    date_to: date | None,
) -> int:
    stmt = (
        select(func.count())
        .select_from(Business)
        .join(Customer, Customer.id == Business.customer_id)
    )
    stmt = _apply_filters(
        stmt,
        keyword=keyword,
        status=status,
        customer_id=customer_id,
        owner_personnel_id=owner_personnel_id,
        date_from=date_from,
        date_to=date_to,
    )
    return int(session.execute(stmt).scalar_one())


def list_items(session: Session, business_id: int) -> list[BusinessItem]:
    """业务明细行（编辑回显与汇总重算共用）。"""
    return list(
        session.execute(
            select(BusinessItem)
            .where(BusinessItem.business_id == business_id)
            .order_by(BusinessItem.id)
        ).scalars()
    )


def replace_items(session: Session, business_id: int, rows: list[BusinessItem]) -> None:
    """全量替换明细：先删、立即 flush、再插入（replace_roles 先例的防御写法）。

    同一 flush 中 SQLAlchemy 先处理 INSERT 后处理 DELETE，先删后插不落库
    会在有唯一键的表上撞键；business_items 虽无唯一键，统一按此写法防御。
    """
    session.execute(delete(BusinessItem).where(BusinessItem.business_id == business_id))
    session.flush()
    for row in rows:
        row.business_id = business_id
        session.add(row)
    session.flush()


def count_items(session: Session, business_ids: list[int]) -> dict[int, int]:
    """业务明细行数（删除保护备用）。"""
    if not business_ids:
        return {}
    rows = session.execute(
        select(BusinessItem.business_id, func.count())
        .where(BusinessItem.business_id.in_(business_ids))
        .group_by(BusinessItem.business_id)
    ).all()
    return dict(rows)


def add_status_log(session: Session, log: BusinessStatusLog) -> None:
    """追加状态留痕（不 commit）。"""
    session.add(log)
    session.flush()


def list_status_logs(
    session: Session, business_id: int, *, offset: int, limit: int
) -> list[BusinessStatusLog]:
    """状态流转时间线（详情页），按时间正序（时间线从创建起展示）。"""
    return list(
        session.execute(
            select(BusinessStatusLog)
            .where(BusinessStatusLog.business_id == business_id)
            .order_by(BusinessStatusLog.create_time, BusinessStatusLog.id)
            .offset(offset)
            .limit(limit)
        ).scalars()
    )


def count_status_logs(session: Session, business_id: int) -> int:
    return int(
        session.execute(
            select(func.count())
            .select_from(BusinessStatusLog)
            .where(BusinessStatusLog.business_id == business_id)
        ).scalar_one()
    )


def add_ownership_log(session: Session, log: OwnershipChangeLog) -> None:
    """追加归属变更留痕（不 commit）。"""
    session.add(log)
    session.flush()


def list_ownership_logs(
    session: Session, business_id: int, *, offset: int, limit: int
) -> list[Row]:
    """某业务的归属变更留痕（join 人员出姓名；人员删除后为 None，留痕固化 ID 兜底）。"""
    old_p = aliased(Personnel)
    new_p = aliased(Personnel)
    return list(
        session.execute(
            select(OwnershipChangeLog, old_p.personnel_name, new_p.personnel_name)
            .outerjoin(old_p, old_p.id == OwnershipChangeLog.old_owner_personnel_id)
            .outerjoin(new_p, new_p.id == OwnershipChangeLog.new_owner_personnel_id)
            .where(
                OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS,
                OwnershipChangeLog.business_id == business_id,
            )
            .order_by(OwnershipChangeLog.create_time.desc(), OwnershipChangeLog.id.desc())
            .offset(offset)
            .limit(limit)
        ).all()
    )


def count_ownership_logs(session: Session, business_id: int) -> int:
    return int(
        session.execute(
            select(func.count())
            .select_from(OwnershipChangeLog)
            .where(
                OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS,
                OwnershipChangeLog.business_id == business_id,
            )
        ).scalar_one()
    )