PayrollManagementSystem/backend/app/repositories/performance_ledger_repository.py

"""绩效核算台账仓储：只读写数据并 flush；commit/rollback 在服务层。"""

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models import PerformanceLedger
from app.models.performance_ledger import CALC_TYPE_MANAGER_AGGREGATE
from app.schemas.calc import LedgerQuery


def get_by_id(session: Session, ledger_id: int) -> PerformanceLedger | None:
    """按主键取台账行。"""
    return session.get(PerformanceLedger, ledger_id)


def get_active_by_receipt(
    session: Session, receipt_id: int, *, calc_type: int = 1
) -> PerformanceLedger | None:
    """取某回款的当前生效核算行（is_active=1；无则 None）。

    二次分摊拆行后同回款可有多行（负责人+参与人）——重算作废用
    ``list_active_by_receipt``；本函数仅限确知单行的调用方（重算前的
    存在性判断按业务维度走 has_active_for_business）。
    """
    return session.execute(
        select(PerformanceLedger).where(
            PerformanceLedger.receipt_id == receipt_id,
            PerformanceLedger.calc_type == calc_type,
            PerformanceLedger.is_active == 1,
        )
    ).scalar_one_or_none()


def list_active_by_receipt(
    session: Session, receipt_id: int, *, calc_type: int = 1
) -> list[PerformanceLedger]:
    """取某回款当前生效的**全部**核算行（分摊拆行后负责人+参与人各一行）。"""
    return list(
        session.execute(
            select(PerformanceLedger).where(
                PerformanceLedger.receipt_id == receipt_id,
                PerformanceLedger.calc_type == calc_type,
                PerformanceLedger.is_active == 1,
            )
        ).scalars()
    )


def has_active_for_business(session: Session, business_id: int, *, calc_type: int = 1) -> bool:
    """该业务是否已有生效核算行（核算粒度 #25：业务级一次性计提的判定）。"""
    return (
        session.execute(
            select(func.count())
            .select_from(PerformanceLedger)
            .where(
                PerformanceLedger.business_id == business_id,
                PerformanceLedger.calc_type == calc_type,
                PerformanceLedger.is_active == 1,
            )
        ).scalar_one()
        > 0
    )


def get_active_by_receipt_of_business(
    session: Session, business_id: int, *, calc_type: int = 1
) -> PerformanceLedger:
    """取该业务当前生效的核算行（重复回款登记进度时回显既有台账）。"""
    return (
        session.execute(
            select(PerformanceLedger)
            .where(
                PerformanceLedger.business_id == business_id,
                PerformanceLedger.calc_type == calc_type,
                PerformanceLedger.is_active == 1,
            )
            .order_by(PerformanceLedger.id.desc())
        )
        .scalars()
        .first()
    )


def add(session: Session, row: PerformanceLedger) -> PerformanceLedger:
    """落台账行（flush 取 id；commit 在服务层）。"""
    session.add(row)
    session.flush()
    return row


def mark_inactive(session: Session, row: PerformanceLedger) -> None:
    """重算时把旧版本行翻为已替代（全表唯一的受控 UPDATE 例外；不 commit）。"""
    row.is_active = 0
    session.flush()


def deactivate_month_aggregates(session: Session, month: str) -> None:
    """月结重生成时作废该月全部经理聚合行（calc_type=2）。

    聚合行每次生成重建，旧行不翻 0 会与新行重复计入（按人聚合按 is_active
    过滤）。与 mark_inactive 同属 is_active 翻 0 的受控 UPDATE 例外；set 批量
    版本而非逐行——聚合行一个月至多几行，且无逐行回读必要。
    """
    from sqlalchemy import update

    session.execute(
        update(PerformanceLedger)
        .where(
            PerformanceLedger.settlement_month == month,
            PerformanceLedger.calc_type == CALC_TYPE_MANAGER_AGGREGATE,
            PerformanceLedger.is_active == 1,
        )
        .values(is_active=0)
    )
    session.flush()


def next_version(session: Session, receipt_id: int) -> int:
    """某回款的下一个核算版本号（max+1；无行时 1）。"""
    current = session.execute(
        select(func.max(PerformanceLedger.calc_version)).where(
            PerformanceLedger.receipt_id == receipt_id
        )
    ).scalar_one()
    return int(current or 0) + 1


def list_ledger(session: Session, query: LedgerQuery) -> list[PerformanceLedger]:
    """分页查询台账（筛选：月/人员/业务/生效/类型）。"""
    stmt = select(PerformanceLedger).where(*_filters(query))
    stmt = stmt.order_by(PerformanceLedger.id.desc()).offset(query.offset).limit(query.page_size)
    return list(session.execute(stmt).scalars())


def count_ledger(session: Session, query: LedgerQuery) -> int:
    """台账计数（与 list_ledger 同筛选）。"""
    return int(
        session.execute(
            select(func.count()).select_from(PerformanceLedger).where(*_filters(query))
        ).scalar_one()
    )


def _filters(query: LedgerQuery) -> tuple:
    filters = []
    if query.settlement_month:
        filters.append(PerformanceLedger.settlement_month == query.settlement_month)
    if query.personnel_id is not None:
        filters.append(PerformanceLedger.personnel_id == query.personnel_id)
    if query.business_id is not None:
        filters.append(PerformanceLedger.business_id == query.business_id)
    if query.is_active is not None:
        filters.append(PerformanceLedger.is_active == query.is_active)
    if query.calc_type is not None:
        filters.append(PerformanceLedger.calc_type == query.calc_type)
    return tuple(filters)