a5c86640e2
- V7 四表：performance_ledgers（append-only 版本行，uk(receipt_id,calc_version) 库级防重兜底）/monthly_settlements（月份 uk，通过即锁定终态）/ settlement_items（按人固化可负工资，uk 三列）/adjustment_orders（TZ 单号） - registry +1：approval.adjustment.enabled（19 项，BOOL 首组扩至三开关） - formula 补核算纯函数：clamp_weight_r（从预览内联提取共用）/monthly_salary 三口径/probation_annual_note/manager_team_share（#12 暂口径：基数不含 经理个人户业绩，聚合集中一处答复后只改） - 迁移 downgrade 只保留 drop_table+alter 回滚+种子行反向 DELETE； performance_ledgers 表名复数（test_models 红线断言） - 断言同步：EXPECTED_TABLES 四新表/registry 19/BOOL 三开关/conftest 清理四新表与 ParamValue 边界 22 - schema_sync+全量单测 460 passed Co-Authored-By: Claude Code <noreply@anthropic.com>
689 lines
27 KiB
Python
689 lines
27 KiB
Python
"""计提参数服务：时点化取值、版本管理（变更/撤回/历史/对比）、计算预览与导出。
|
||
|
||
ParamSnapshot 是 M3 核算与计算预览共用的输入契约：按业务日期（回款到账日）
|
||
冻结当时口径，禁读「当前值」计算历史数据（算法红线）。
|
||
变更批次 = 版本（一次提交一批次）；条目逐项留痕（旧值/新值），append-only。
|
||
"""
|
||
|
||
import logging
|
||
from collections.abc import Iterable
|
||
from dataclasses import dataclass
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
from fractions import Fraction
|
||
from io import BytesIO
|
||
|
||
from sqlalchemy.exc import IntegrityError
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.constants import (
|
||
ACTION_DELETE,
|
||
ACTION_UPDATE,
|
||
MODULE_PARAM,
|
||
NO_PREFIX_PARAM,
|
||
PRODUCT_CATEGORIES,
|
||
)
|
||
from app.core.exceptions import BadRequestError, NotFoundError
|
||
from app.core.response import PageResult
|
||
from app.models import ParamChangeBatch, ParamDefinition, ParamValue, User
|
||
from app.repositories import business_role_repository, param_repository
|
||
from app.schemas.param import (
|
||
ParamBatchDetailOut,
|
||
ParamBatchItemOut,
|
||
ParamBatchOut,
|
||
ParamBatchQuery,
|
||
ParamCatalogItemOut,
|
||
ParamChangeBatchOut,
|
||
ParamChangeIn,
|
||
ParamChangeItemIn,
|
||
ParamDiffOut,
|
||
ParamDimensionValueOut,
|
||
ParamHistoryOut,
|
||
ParamHistoryQuery,
|
||
ParamPreviewIn,
|
||
ParamPreviewOut,
|
||
ParamUsedOut,
|
||
)
|
||
from app.services import audit_service, formula
|
||
from app.services.audit_service import AuditRecord, RequestMeta
|
||
from app.services.formula import to_decimal
|
||
from app.services.no_generator import next_no
|
||
from app.services.param_registry import get_spec, parse_fraction, validate_and_normalize
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ==================== 快照 ====================
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class ParamSnapshot:
|
||
"""某业务日期下的参数快照（冻结不可变，核算期间保持一致）。
|
||
|
||
覆盖链：角色维度 > 产品类别维度 > 默认行（''）；均无匹配时回退注册表默认值。
|
||
"""
|
||
|
||
biz_date: date
|
||
_values: dict[str, dict[str, str]]
|
||
_effective: dict[tuple[str, str], date]
|
||
|
||
@classmethod
|
||
def from_rows(
|
||
cls, biz_date: date, rows: Iterable[tuple[str, str, str, date]]
|
||
) -> "ParamSnapshot":
|
||
"""由 (param_code, dimension_value, value_text, effective_date) 行构造。
|
||
|
||
rows 须按仓储输出顺序（同一 (code, dim) 的最新生效值在前）——以先出现者为准。
|
||
"""
|
||
values: dict[str, dict[str, str]] = {}
|
||
effective: dict[tuple[str, str], date] = {}
|
||
for code, dimension, value_text, effective_date in rows:
|
||
if (code, dimension) in effective:
|
||
continue
|
||
effective[(code, dimension)] = effective_date
|
||
values.setdefault(code, {}).setdefault(dimension, value_text)
|
||
return cls(biz_date=biz_date, _values=values, _effective=effective)
|
||
|
||
def get(
|
||
self, code: str, *, role_code: str | None = None, product_category: str | None = None
|
||
) -> str:
|
||
"""取参数原始值（覆盖链解析）。无值且无注册表默认值时抛 400。"""
|
||
candidates = self._values.get(code)
|
||
if candidates:
|
||
for dimension in (role_code, product_category, ""):
|
||
if dimension is not None and dimension in candidates:
|
||
return candidates[dimension]
|
||
spec = get_spec(code) # 未注册编码 → ValueError（代码缺陷）
|
||
if spec.default_value is None:
|
||
raise BadRequestError(f"参数 {code} 无生效值且无默认值，请先在系统设置中维护")
|
||
return spec.default_value
|
||
|
||
def effective_date_of(
|
||
self, code: str, *, role_code: str | None = None, product_category: str | None = None
|
||
) -> date | None:
|
||
"""取参数当前生效值的生效日期（回退注册表默认值时为 None）。"""
|
||
candidates = self._values.get(code)
|
||
if candidates:
|
||
for dimension in (role_code, product_category, ""):
|
||
if dimension is not None and (code, dimension) in self._effective:
|
||
return self._effective[(code, dimension)]
|
||
return None
|
||
|
||
def iter_items(self) -> Iterable[tuple[str, str, str]]:
|
||
"""遍历快照全部 (param_code, dimension_value, value_text)，供对比。"""
|
||
for code, dims in self._values.items():
|
||
for dimension, value_text in dims.items():
|
||
yield code, dimension, value_text
|
||
|
||
def fraction(
|
||
self, code: str, *, role_code: str | None = None, product_category: str | None = None
|
||
) -> Fraction:
|
||
"""取参数值并精确解析为 Fraction（"4/360" → 1/90，不经 float）。"""
|
||
return parse_fraction(
|
||
self.get(code, role_code=role_code, product_category=product_category)
|
||
)
|
||
|
||
def integer(
|
||
self, code: str, *, role_code: str | None = None, product_category: str | None = None
|
||
) -> int:
|
||
"""取参数值为整数。配置成非整数属数据缺陷（写入经校验，不应发生）。"""
|
||
value = self.fraction(code, role_code=role_code, product_category=product_category)
|
||
if value.denominator != 1:
|
||
raise ValueError(f"参数 {code} 配置为非整数：{value}")
|
||
return int(value)
|
||
|
||
def boolean(
|
||
self, code: str, *, role_code: str | None = None, product_category: str | None = None
|
||
) -> bool:
|
||
"""取布尔参数（0/1）。"""
|
||
return self.get(code, role_code=role_code, product_category=product_category) == "1"
|
||
|
||
def text(
|
||
self, code: str, *, role_code: str | None = None, product_category: str | None = None
|
||
) -> str:
|
||
"""取参数原文（枚举码/文本）。"""
|
||
return self.get(code, role_code=role_code, product_category=product_category)
|
||
|
||
|
||
def resolve_snapshot(session: Session, biz_date: date) -> ParamSnapshot:
|
||
"""按业务日期解析全量参数快照（M3 核算与计算预览共用入口）。"""
|
||
return ParamSnapshot.from_rows(biz_date, param_repository.resolve_values(session, biz_date))
|
||
|
||
|
||
# ==================== 目录 ====================
|
||
|
||
|
||
def get_catalog(session: Session, *, as_of: date | None = None) -> list[ParamCatalogItemOut]:
|
||
"""参数目录：全部启用定义 + 各维度截至 asOf（默认今日）的生效值。"""
|
||
biz_date = as_of or date.today()
|
||
resolved = param_repository.resolve_values(session, biz_date)
|
||
by_code: dict[str, list[ParamDimensionValueOut]] = {}
|
||
for code, dimension, value_text, effective_date in resolved:
|
||
by_code.setdefault(code, []).append(
|
||
ParamDimensionValueOut(
|
||
dimension_value=dimension, value_text=value_text, effective_date=effective_date
|
||
)
|
||
)
|
||
items: list[ParamCatalogItemOut] = []
|
||
for definition in param_repository.list_definitions(session, status=1):
|
||
spec = get_spec(definition.param_code)
|
||
items.append(
|
||
ParamCatalogItemOut(
|
||
param_code=definition.param_code,
|
||
param_name=definition.param_name,
|
||
param_group=definition.param_group,
|
||
value_type=definition.value_type,
|
||
unit=definition.unit,
|
||
dimension_type=definition.dimension_type,
|
||
options=list(spec.options or ()),
|
||
default_value=definition.default_value,
|
||
value_min=definition.value_min,
|
||
value_max=definition.value_max,
|
||
description=definition.description,
|
||
values=by_code.get(definition.param_code, []),
|
||
)
|
||
)
|
||
return items
|
||
|
||
|
||
# ==================== 变更与版本管理 ====================
|
||
|
||
|
||
def _normalize_items(
|
||
session: Session, items: list[ParamChangeItemIn]
|
||
) -> list[tuple[ParamDefinition, str, str]]:
|
||
"""全量校验条目并归一为 (定义, 规范维度值, 规范值)，任何一项不合法整批拒绝。
|
||
|
||
定义与岗位各一条 IN 批量查询预载（逐条查询在批次扩大时是写侧 N+1）；
|
||
岗位预载收集全部非空维度值 —— 集合可能略大于实际所需，但省去按维度
|
||
类型先分类的第二轮遍历，无害多取由校验逻辑自行过滤。
|
||
|
||
:raises BadRequestError: 参数未定义 / 维度不合法 / 值校验失败 / 同批次重复条目
|
||
"""
|
||
codes = list({item.param_code for item in items})
|
||
definitions = param_repository.list_definitions_by_codes(session, codes)
|
||
dimension_values = list({item.dimension_value for item in items if item.dimension_value})
|
||
# 预载字典键归一为 casefold：单条版 get_by_code 走 SQL ci 校对（大小写
|
||
# 变形可命中），批量版改字典查找后须自行等价归一，否则变形大小写会
|
||
# 被误判「不是已存在的岗位」；岗位编码均为 ascii 小写蛇形，casefold 足够
|
||
roles = {
|
||
code.casefold(): role
|
||
for code, role in business_role_repository.list_by_codes(session, dimension_values).items()
|
||
}
|
||
normalized: list[tuple[ParamDefinition, str, str]] = []
|
||
seen: set[tuple[str, str]] = set()
|
||
for item in items:
|
||
definition = definitions.get(item.param_code)
|
||
if definition is None:
|
||
# 用户输入先行查库给 400；未注册编码出现在库里属数据与注册表漂移
|
||
raise BadRequestError(f"参数 {item.param_code} 未在系统中定义")
|
||
# ci 校对规则下大小写变形也能命中库，此后一律用库里的规范值：
|
||
# 注册表 get_spec 与快照字典均按精确键匹配，原始大小写入库会导致
|
||
# 取值链静默不命中（按错误参数算钱且无报错）
|
||
spec = get_spec(definition.param_code)
|
||
dimension = item.dimension_value
|
||
if spec.dimension_type == "NONE" and dimension != "":
|
||
raise BadRequestError(f"参数 {item.param_code} 不带维度，维度值须为空")
|
||
if spec.dimension_type == "BUSINESS_ROLE" and dimension:
|
||
role = roles.get(dimension.casefold())
|
||
if role is None:
|
||
raise BadRequestError(
|
||
f"参数 {item.param_code} 的维度值 {dimension!r} 不是已存在的岗位"
|
||
)
|
||
dimension = role.role_code
|
||
if (
|
||
spec.dimension_type == "PRODUCT_CATEGORY"
|
||
and dimension
|
||
and dimension not in PRODUCT_CATEGORIES
|
||
):
|
||
raise BadRequestError(
|
||
f"参数 {item.param_code} 的维度值 {dimension!r} 不是合法的产品类别"
|
||
)
|
||
value = validate_and_normalize(spec, item.value)
|
||
key = (item.param_code, dimension)
|
||
if key in seen:
|
||
raise BadRequestError(
|
||
f"同一批次内重复提交参数 {item.param_code}（{dimension or '默认'}）"
|
||
)
|
||
seen.add(key)
|
||
normalized.append((definition, dimension, value))
|
||
return normalized
|
||
|
||
|
||
def submit_change(
|
||
session: Session,
|
||
payload: ParamChangeIn,
|
||
*,
|
||
operator: User,
|
||
meta: RequestMeta | None = None,
|
||
) -> ParamChangeBatchOut:
|
||
"""提交变更批次：全量校验 → 建批次 → 逐项写条目（旧值取提交时生效值）。
|
||
|
||
:raises BadRequestError: 参数未定义 / 维度不合法 / 值校验失败 / 生效日期早于今日 /
|
||
同批次重复条目
|
||
"""
|
||
effective = payload.effective_date if payload.change_type == 2 else date.today()
|
||
with audit_service.audit(
|
||
session,
|
||
operator=operator,
|
||
record=AuditRecord(
|
||
module=MODULE_PARAM,
|
||
action=ACTION_UPDATE,
|
||
summary=(
|
||
f"提交计提参数变更 {len(payload.items)} 项，"
|
||
f"{'立即生效' if payload.change_type == 1 else f'{effective} 起生效'}"
|
||
),
|
||
request_data=payload.model_dump(exclude_none=True),
|
||
),
|
||
meta=meta,
|
||
):
|
||
if payload.change_type == 2 and effective < date.today():
|
||
raise BadRequestError("指定生效日期不得早于今日")
|
||
|
||
# 先全量校验、再写入：任何一项不合法整批拒绝（不留半批脏数据）
|
||
normalized = _normalize_items(session, payload.items)
|
||
|
||
try:
|
||
batch = param_repository.create_batch(
|
||
session,
|
||
ParamChangeBatch(
|
||
# 版本号与三单据同口径：MAX FOR UPDATE 锁当日序号区间；
|
||
# 残余并发冲突由下方 IntegrityError 分支翻译为可重试 400
|
||
version_no=next_no(
|
||
session,
|
||
model=ParamChangeBatch,
|
||
column_name="version_no",
|
||
prefix=NO_PREFIX_PARAM,
|
||
seq_width=2,
|
||
),
|
||
effective_date=effective,
|
||
change_type=payload.change_type,
|
||
remark=payload.remark,
|
||
operator_id=operator.id,
|
||
operator_name=operator.username,
|
||
item_count=len(normalized),
|
||
),
|
||
)
|
||
# 旧值留痕取「本条目生效日时点」的值：存在多个未来生效批次时，
|
||
# 每条记录的是它实际顶替的那一版（而非提交时刻的当前值）
|
||
for definition, dimension, value in normalized:
|
||
old = param_repository.latest_value_before(
|
||
session, definition.id, dimension, effective
|
||
)
|
||
param_repository.create_value(
|
||
session,
|
||
ParamValue(
|
||
param_id=definition.id,
|
||
dimension_value=dimension,
|
||
effective_date=effective,
|
||
value_text=value,
|
||
old_value_text=old.value_text if old else None,
|
||
batch_id=batch.id,
|
||
),
|
||
)
|
||
except IntegrityError as exc:
|
||
# version_no 唯一键的并发冲突：回滚并翻译为可重试的 400
|
||
session.rollback()
|
||
raise BadRequestError("提交冲突（版本号生成重复），请重试") from exc
|
||
session.commit()
|
||
logger.info("提交参数变更 | version_no=%s | items=%s", batch.version_no, batch.item_count)
|
||
return ParamChangeBatchOut(
|
||
id=batch.id,
|
||
version_no=batch.version_no,
|
||
effective_date=batch.effective_date,
|
||
change_type=batch.change_type,
|
||
remark=batch.remark,
|
||
operator_name=batch.operator_name,
|
||
item_count=batch.item_count,
|
||
create_time=batch.create_time,
|
||
)
|
||
|
||
|
||
def withdraw_batch(
|
||
session: Session, batch_id: int, *, operator: User, meta: RequestMeta | None = None
|
||
) -> None:
|
||
"""撤回未生效批次。已生效禁撤；条目删除后解析自然回退到更早版本。
|
||
|
||
:raises NotFoundError: 批次不存在
|
||
:raises BadRequestError: 批次已生效
|
||
"""
|
||
batch = param_repository.get_batch_by_id(session, batch_id, for_update=True)
|
||
with audit_service.audit(
|
||
session,
|
||
operator=operator,
|
||
record=AuditRecord(
|
||
module=MODULE_PARAM,
|
||
action=ACTION_DELETE,
|
||
summary=f"撤回参数变更批次 {batch.version_no if batch else f'#{batch_id}'}",
|
||
),
|
||
meta=meta,
|
||
):
|
||
if batch is None:
|
||
raise NotFoundError("变更批次不存在")
|
||
if batch.effective_date <= date.today():
|
||
raise BadRequestError("批次已生效，不可撤回")
|
||
|
||
param_repository.delete_values_by_batch(session, batch.id)
|
||
param_repository.delete_batch(session, batch)
|
||
session.commit()
|
||
logger.info("撤回参数变更批次 | version_no=%s", batch.version_no)
|
||
|
||
|
||
def get_history(
|
||
session: Session, param_code: str, query: ParamHistoryQuery
|
||
) -> PageResult[ParamHistoryOut]:
|
||
"""单参数版本历史（分页，最新在前）。
|
||
|
||
:raises NotFoundError: 参数不存在
|
||
"""
|
||
definition = param_repository.get_definition_by_code(session, param_code)
|
||
if definition is None:
|
||
raise NotFoundError("参数不存在")
|
||
rows = param_repository.list_history(
|
||
session,
|
||
definition.id,
|
||
offset=query.offset,
|
||
limit=query.page_size,
|
||
dimension_value=query.dimension_value,
|
||
)
|
||
total = param_repository.count_history(
|
||
session, definition.id, dimension_value=query.dimension_value
|
||
)
|
||
return PageResult[ParamHistoryOut](
|
||
records=[
|
||
ParamHistoryOut(
|
||
version_no=batch.version_no,
|
||
dimension_value=value.dimension_value,
|
||
old_value_text=value.old_value_text,
|
||
value_text=value.value_text,
|
||
effective_date=value.effective_date,
|
||
operator_name=batch.operator_name,
|
||
create_time=value.create_time,
|
||
)
|
||
for value, batch in rows
|
||
],
|
||
total=total,
|
||
page_num=query.page_num,
|
||
page_size=query.page_size,
|
||
)
|
||
|
||
|
||
def list_batches(session: Session, query: ParamBatchQuery) -> PageResult[ParamBatchOut]:
|
||
"""变更批次分页（按创建时间倒序）。"""
|
||
batches = param_repository.list_batches(
|
||
session,
|
||
offset=query.offset,
|
||
limit=query.page_size,
|
||
start_date=query.start_date,
|
||
end_date=query.end_date,
|
||
)
|
||
total = param_repository.count_batches(
|
||
session, start_date=query.start_date, end_date=query.end_date
|
||
)
|
||
return PageResult[ParamBatchOut](
|
||
records=[
|
||
ParamBatchOut(
|
||
id=batch.id,
|
||
version_no=batch.version_no,
|
||
effective_date=batch.effective_date,
|
||
change_type=batch.change_type,
|
||
remark=batch.remark,
|
||
operator_name=batch.operator_name,
|
||
item_count=batch.item_count,
|
||
create_time=batch.create_time,
|
||
)
|
||
for batch in batches
|
||
],
|
||
total=total,
|
||
page_num=query.page_num,
|
||
page_size=query.page_size,
|
||
)
|
||
|
||
|
||
def get_batch_detail(session: Session, batch_id: int) -> ParamBatchDetailOut:
|
||
"""批次明细：批次头 + 逐项旧值 → 新值。
|
||
|
||
:raises NotFoundError: 批次不存在
|
||
"""
|
||
batch = param_repository.get_batch_by_id(session, batch_id)
|
||
if batch is None:
|
||
raise NotFoundError("变更批次不存在")
|
||
items = [
|
||
ParamBatchItemOut(
|
||
param_code=definition.param_code,
|
||
param_name=definition.param_name,
|
||
dimension_value=value.dimension_value,
|
||
old_value_text=value.old_value_text,
|
||
value_text=value.value_text,
|
||
effective_date=value.effective_date,
|
||
)
|
||
for value, definition in param_repository.list_values_by_batch(session, batch_id)
|
||
]
|
||
return ParamBatchDetailOut(
|
||
id=batch.id,
|
||
version_no=batch.version_no,
|
||
effective_date=batch.effective_date,
|
||
change_type=batch.change_type,
|
||
remark=batch.remark,
|
||
operator_name=batch.operator_name,
|
||
item_count=batch.item_count,
|
||
create_time=batch.create_time,
|
||
items=items,
|
||
)
|
||
|
||
|
||
# ==================== 对比与预览 ====================
|
||
|
||
|
||
def compare(session: Session, *, from_date: date, to_date: date) -> list[ParamDiffOut]:
|
||
"""两日期快照逐项对比（全量参数，changed 标记变化）。
|
||
|
||
:raises BadRequestError: 起始日期晚于截止日期
|
||
"""
|
||
if from_date > to_date:
|
||
raise BadRequestError("起始日期不得晚于截止日期")
|
||
old_items = {
|
||
(code, dimension): value
|
||
for code, dimension, value in resolve_snapshot(session, from_date).iter_items()
|
||
}
|
||
new_items = {
|
||
(code, dimension): value
|
||
for code, dimension, value in resolve_snapshot(session, to_date).iter_items()
|
||
}
|
||
out: list[ParamDiffOut] = []
|
||
for code, dimension in sorted(set(old_items) | set(new_items)):
|
||
spec = get_spec(code)
|
||
old_value = old_items.get((code, dimension))
|
||
new_value = new_items.get((code, dimension))
|
||
out.append(
|
||
ParamDiffOut(
|
||
param_code=code,
|
||
param_name=spec.name,
|
||
param_group=spec.group,
|
||
dimension_value=dimension,
|
||
old_value=old_value,
|
||
new_value=new_value,
|
||
changed=old_value != new_value,
|
||
)
|
||
)
|
||
return out
|
||
|
||
|
||
def _preview_rate(
|
||
session: Session,
|
||
payload: ParamPreviewIn,
|
||
snapshot: ParamSnapshot,
|
||
days: int,
|
||
take,
|
||
) -> tuple[Fraction, str, Decimal | None]:
|
||
"""按岗位模板分派计提比例，返回（纯比例, 人读说明, 绩效基数覆盖值）。
|
||
|
||
基数覆盖值非 None 表示该模板的绩效基数不是折算毛利润额（如信息员按
|
||
「本人销售额」计提），预览层直接采用并跳过折算与阈值。
|
||
聚合模板（营销经理 12.5%+5%）依赖月度团队数据，不支持单笔预览。
|
||
"""
|
||
if payload.customer_type == "enterprise":
|
||
# 存储口径为 ‰ 数（如 "3"），除以 1000 转纯比例 —— performance_amount
|
||
# 按纯比例相乘，漏换算会把绩效金额放大 1000 倍（与业务员分支同防线）
|
||
rate = take("enterprise.rate") / 1000
|
||
return rate, "企业户固定计提比例", None
|
||
|
||
role_code = payload.business_role
|
||
template = "TPL_SALESPERSON_DECAY"
|
||
if role_code:
|
||
role = business_role_repository.get_by_code(session, role_code)
|
||
if role is None:
|
||
raise BadRequestError(f"岗位 {role_code!r} 不存在")
|
||
template = role.calc_template
|
||
|
||
if template == "TPL_MANAGER_SHARE":
|
||
raise BadRequestError(
|
||
"营销经理绩效为团队聚合口径（本团队 12.5% + 全公司 5%），不支持单笔预览"
|
||
)
|
||
if template == "TPL_INFO_SALES_FIXED":
|
||
# ‰ 存储口径 → 纯比例，同企业户分支（不换算则绩效放大 1000 倍）
|
||
rate = take("info.rate") / 1000
|
||
# 信息员的基数是本人销售额（无折算与阈值），覆盖默认基数
|
||
return rate, "信息员：本人销售额 × 计提比例", Decimal(payload.revenue)
|
||
if template == "TPL_SALESPERSON_DECAY":
|
||
start = take("salesperson.rate.start")
|
||
slope = take("salesperson.rate.slope")
|
||
floor = take("salesperson.rate.floor")
|
||
boundary = int(take("salesperson.days.boundary"))
|
||
rate = formula.salesperson_rate(
|
||
days, start=start, slope=slope, floor=floor, boundary_days=boundary
|
||
)
|
||
return (
|
||
rate,
|
||
f"业务员递减：{to_decimal(start)}‰ 起，{days} 天（分界 {boundary} 天，超界取 {to_decimal(floor)}‰）",
|
||
None,
|
||
)
|
||
raise BadRequestError(f"公式模板 {template} 暂不支持预览")
|
||
|
||
|
||
def preview(session: Session, payload: ParamPreviewIn) -> ParamPreviewOut:
|
||
"""计算预览：入参（发货/回款/金额/岗位）→ 公式纯函数 → 出参（含所用参数）。"""
|
||
# 参数按回款到账日取值（对照表 #11）
|
||
snapshot = resolve_snapshot(session, payload.receipt_date)
|
||
revenue = Decimal(payload.revenue)
|
||
profit = Decimal(payload.profit)
|
||
received = Decimal(payload.received_amount)
|
||
used: list[ParamUsedOut] = []
|
||
|
||
def take(code: str) -> Fraction:
|
||
raw = snapshot.get(
|
||
code,
|
||
role_code=payload.business_role,
|
||
product_category=payload.product_category,
|
||
)
|
||
used.append(
|
||
ParamUsedOut(
|
||
param_code=code,
|
||
value_text=raw,
|
||
effective_date=snapshot.effective_date_of(
|
||
code,
|
||
role_code=payload.business_role,
|
||
product_category=payload.product_category,
|
||
),
|
||
)
|
||
)
|
||
return parse_fraction(raw)
|
||
|
||
if payload.product_category and payload.product_category not in PRODUCT_CATEGORIES:
|
||
raise BadRequestError(
|
||
f"产品类别 {payload.product_category!r} 不存在，可选：municipal_water / rail_transit"
|
||
)
|
||
|
||
days = formula.payment_days(payload.ship_date, payload.receipt_date)
|
||
rate, rate_text, base_override = _preview_rate(session, payload, snapshot, days, take)
|
||
|
||
profit_rate = formula.profit_rate(profit, revenue)
|
||
if base_override is not None:
|
||
# 基数覆盖（信息员按本人销售额计提）：不走折算与阈值
|
||
converted = base_override
|
||
else:
|
||
threshold = take("product.threshold")
|
||
converted = formula.converted_profit(revenue, profit_rate, threshold)
|
||
performance = formula.performance_amount(converted, rate)
|
||
|
||
barter_applied = payload.barter
|
||
after_barter: Decimal | None = None
|
||
if barter_applied:
|
||
discount = take("barter.discount")
|
||
after_barter = formula.apply_barter(performance, discount)
|
||
|
||
start_ratio = take("payment.start.ratio")
|
||
start_ok = formula.start_payment_ok(received, revenue, start_ratio)
|
||
# R 钳制已提取为纯函数 clamp_weight_r（④ 核算与预览共用同一实现）
|
||
weight_r = formula.clamp_weight_r(received, revenue)
|
||
weight = formula.payment_weight(weight_r)
|
||
base_amount = after_barter if barter_applied else performance
|
||
payable = to_decimal(Fraction(base_amount) * weight)
|
||
|
||
probation_note: str | None = None
|
||
if payload.probation:
|
||
annual = take("probation.annual.threshold")
|
||
probation_note = f"试用期绩效照算不发放；年累计超 {to_decimal(annual)} 元部分年底一次性结算"
|
||
|
||
return ParamPreviewOut(
|
||
payment_days=days,
|
||
rate=str(rate),
|
||
rate_text=rate_text,
|
||
profit_rate=str(profit_rate),
|
||
converted_profit=str(converted),
|
||
performance_amount=str(performance),
|
||
barter_applied=barter_applied,
|
||
performance_after_barter=str(after_barter) if after_barter is not None else None,
|
||
start_payment_ok=start_ok,
|
||
payment_weight_r=weight_r,
|
||
payment_weight=str(weight),
|
||
payable_amount=str(payable),
|
||
probation_note=probation_note,
|
||
params_used=used,
|
||
)
|
||
|
||
|
||
# ==================== 导出 ====================
|
||
|
||
|
||
def export_changes(
|
||
session: Session, *, start_date: date | None = None, end_date: date | None = None
|
||
) -> bytes:
|
||
"""变更记录导出 Excel（openpyxl）。列：版本号/变更人/变更时间/生效日期/参数/维度/旧值/新值。"""
|
||
from openpyxl import Workbook
|
||
|
||
rows = param_repository.list_export_rows(session, start_date=start_date, end_date=end_date)
|
||
workbook = Workbook()
|
||
sheet = workbook.active
|
||
sheet.title = "参数变更记录"
|
||
sheet.append(
|
||
[
|
||
"版本号",
|
||
"变更人",
|
||
"变更时间",
|
||
"生效日期",
|
||
"参数编码",
|
||
"参数名称",
|
||
"维度值",
|
||
"旧值",
|
||
"新值",
|
||
]
|
||
)
|
||
for batch, value, definition in rows:
|
||
sheet.append(
|
||
[
|
||
batch.version_no,
|
||
batch.operator_name,
|
||
batch.create_time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
value.effective_date.strftime("%Y-%m-%d"),
|
||
definition.param_code,
|
||
definition.param_name,
|
||
value.dimension_value,
|
||
value.old_value_text,
|
||
value.value_text,
|
||
]
|
||
)
|
||
buffer = BytesIO()
|
||
workbook.save(buffer)
|
||
return buffer.getvalue()
|