liuwenxin和Claude Code
ccfd20651b
- _personal_rate 的 TPL_MANAGER_SHARE 分支读 manager.personal.included：1=按业务员递减公式计提，0=单笔落 0（现状） - _load_allocations 装载生效方案并防御拦截（合计超 100%、参与人含负责人） - _persist_ledger 拆行落库：参与人=份额×比例(q2)、负责人=q2 全额−已分累计（行合计守恒防分漂）；快照带 alloc_plan/alloc_role/alloc_ratio - _prepare_mode 重算改 list_active_by_receipt 作废全部拆分行；uk 三列版支持同回款按人多行 - business_service 保存侧同口径校验（超 100% 400、负责人自分 400、hint 文案定稿） - 集成用例 +5（拆行守恒/叠加进度登记/重算全作废/双拦截）+保存侧 3 条 Co-Authored-By: Claude Code <noreply@anthropic.com>
157 行
5.6 KiB
Python
157 行
5.6 KiB
Python
"""绩效核算台账仓储：只读写数据并 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)
|