PayrollManagementSystem/backend/app/services/tplus/calculation_preview.py
liuwenxin 1461a8d113 feat(后端): 添加授权体系、系统监控与试用期计提
按 2026-11-02 需求补齐计划交付后端能力:
- 角色授权:权限目录化,接口统一准入校验,未登记默认拒绝;
  管理员判断改为按数据库角色重新判定,防止权限码持有者绕过
- 系统监控:单进程采集资源、请求与异常指标,分钟级聚合,
  阈值配置化;敏感配置增加 repr 保护,Redis 密码 URL 转义
- 试用期计提:按回款日判定并写入快照,年度结清与调整单联动;
  人员模型补试用期、离职日期与工资版本历史字段
- 计提模板:公式骨架受控,页面维护名称与参数绑定
- T+ 采集:自动化配置表与进度查询

内置角色禁停用保护对齐模块注释;核算规则版本提为常量并
补充批量核算失败堆栈日志。

影响范围:后端全部模块、数据库迁移 V22-V28
验证:uv run ruff check 通过;922 个单元测试全部通过;
集成测试因本机 MySQL 数据卷被其他容器锁定暂未运行
2026-11-03 11:20:55 +08:00

556 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""固定真实源快照和人工事实,组装隔离的 FIFO 核算预演。"""
from calendar import monthrange
from collections import Counter
from datetime import date
from decimal import InvalidOperation
from sqlalchemy import select
from app.core.exceptions import BadRequestError
from app.models import Customer, Product, TPlusSourceLink, TPlusSyncRecord, TPlusWorkLine
from app.services import param_service
from app.services.rate_rules import personal_values
from app.services.tplus.fifo import START, amount_text, money, preview_group
from app.services.tplus.snapshots import source_records, source_statement
from app.services.tplus.workbench import RULE_VERSION, WorkbenchError, digest, get_run, latest_facts
def _parameters(session, days):
values = {}
for day in sorted(days):
snap = param_service.resolve_snapshot(session, date.fromisoformat(day))
values[day] = {
"parameter_snapshot_hash": digest([
(code, dimension, value, str(snap._effective.get((code, dimension))))
for code, dimension, value in sorted(snap.iter_items())
]),
"start": snap.get("salesperson.rate.start", role_code="salesperson"),
"slope": snap.get("salesperson.rate.slope", role_code="salesperson"),
"floor": snap.get("salesperson.rate.floor", role_code="salesperson"),
"boundary": snap.integer("salesperson.days.boundary", role_code="salesperson"),
"enterprise": snap.get("enterprise.rate"),
"enterprise_by_category": {
category: snap.get("enterprise.rate", product_category=category)
for category in ("municipal_water", "rail_transit")
},
"info": snap.get("info.rate", role_code="info_officer"),
"personal": {
category: personal_values(snap, category)
for category in ("municipal_water", "rail_transit")
},
"thresholds": {
category: snap.get("product.threshold", product_category=category)
for category in ("municipal_water", "rail_transit")
},
}
return values
def _group_records(session, source, entity, ids):
rows = source_records(session, source, entity, ids)
if {r.source_id for r in rows} != set(ids):
raise WorkbenchError("GROUP_SOURCE_MISSING")
return rows
def _guidance(facts, product_id, line_id, month):
candidates = [
f.data_json
for (kind, _), f in facts.items()
if kind == "GUIDANCE"
and f.data_json["product_id"] == product_id
and f.data_json["month"] == month
and (f.data_json["basis"] == "UNIT" or f.data_json.get("shipment_line_id") == line_id)
]
return candidates[0] if len(candidates) == 1 else None
def _shipments(session, run, key, group, facts):
headers = _group_records(session, run.source_database, "shipments", group["shipment_ids"])
details = source_records(
session, run.source_database, "shipment_items", parent_ids=group["shipment_ids"]
)
heads = {r.source_id: r for r in headers}
product_ids = {str(r.payload_json.get("idinventory")) for r in details}
product_targets = dict(
session.execute(
select(TPlusSourceLink.source_id, Product)
.join(Product, Product.id == TPlusSourceLink.target_id)
.where(
TPlusSourceLink.source_database == run.source_database,
TPlusSourceLink.entity_type == "products",
TPlusSourceLink.source_id.in_(product_ids),
)
).all()
)
customer_target = session.scalar(
select(Customer)
.join(TPlusSourceLink, Customer.id == TPlusSourceLink.target_id)
.where(
TPlusSourceLink.source_database == run.source_database,
TPlusSourceLink.entity_type == "customers",
TPlusSourceLink.source_id == group["customer_id"],
)
)
rows, snapshots = [], headers + details
for record in details:
raw = record.payload_json
head_record = heads.get(str(raw.get("source_parent_id")))
if head_record is None:
continue
head = head_record.payload_json
if str(head.get("idcustomer")) != group["customer_id"]:
raise WorkbenchError("GROUP_CUSTOMER_MISMATCH")
supplement = facts.get(("SHIPMENT", record.source_id))
supplement = supplement.data_json if supplement else {}
correction_fact = facts.get(("CORRECTION", record.source_id)) or facts.get(
("CORRECTION", head_record.source_id)
)
correction = (
correction_fact.data_json
if correction_fact and correction_fact.data_json.get("entity") == "SHIPMENT"
else {}
)
product_id = str(raw.get("idinventory"))
product = facts.get(("PRODUCT", product_id))
customer = facts.get(("CUSTOMER", group["customer_id"]))
pending = []
if (
not supplement.get("sales_amount_excludes_freight")
and supplement.get("freight_amount") is None
):
pending.append("FREIGHT_PENDING")
if not supplement.get("business_key"):
pending.append("SHIPPING_OWNERSHIP_PENDING")
# SA_SaleDelivery.isCancel 是 VoucherSettleState(如428未结清、180已结清),
# 并非作废标志;不能将非零编码当作红冲或作废。
if head.get("voucherState") != 189:
pending.append("SHIPMENT_STATE_PENDING")
if head.get("isCounteractDelivery") and not correction.get("exclude_performance"):
pending.append("NEGATIVE_EVENT_REVIEW_REQUIRED")
day = supplement.get("date_override") or str(head["voucherdate"])[:10]
managers = [
f.data_json
for (kind, _), f in facts.items()
if kind == "MANAGER"
and f.data_json["business_key"] == supplement.get("business_key")
and f.data_json["effective_date"] <= day
]
managers.sort(key=lambda v: v["effective_date"])
for (kind, _), fact in facts.items():
if (
kind == "HANDOVER"
and fact.data_json["business_key"] == supplement.get("business_key")
and fact.data_json["mode"] == "ACTIVE"
and not fact.data_json["shipping_ownership_confirmed"]
):
pending.append("HANDOVER_SHIPMENTS_PENDING")
if raw.get("taxAmount") is None:
pending.append("TAX_AMOUNT_PENDING")
gross_amount = money(raw.get("taxAmount") or "0")
freight = (
money(supplement["freight_amount"])
if supplement.get("freight_amount") is not None
else money("0")
)
target_product = product_targets.get(product_id)
rows.append(
{
"source_id": head_record.source_id,
"line_id": record.source_id,
"source_no": head_record.source_no,
"date": day,
"group_key": key,
"business_key": supplement.get("business_key"),
"personnel_id": supplement.get("personnel_id"),
"amount": str(gross_amount - freight),
"gross_tax_amount": str(gross_amount),
"freight_amount": str(freight)
if supplement.get("freight_amount") is not None
else None,
"product_id": product_id,
"quantity": str(raw.get("quantity") or "0"),
"category": target_product.product_category
if target_product
else product.data_json["category"]
if product
else None,
"unit": target_product.unit
if target_product
else product.data_json["unit"]
if product
else None,
"customer_type": customer_target.customer_type
if customer_target
else customer.data_json["customer_type"]
if customer
else None,
"classification_evidence": {
"product_target_id": target_product.id if target_product else None,
"customer_target_id": customer_target.id if customer_target else None,
},
"manager_id": managers[-1]["manager_id"] if managers else None,
"performance_excluded": bool(correction.get("exclude_performance")),
"correction_pair_key": correction.get("pair_key"),
"exclusion_reason": "CORRECTION_EXCLUDED" if correction else None,
"pending": pending,
}
)
covered = {r["source_id"] for r in rows}
if covered != set(group["shipment_ids"]):
raise WorkbenchError("SHIPMENT_DETAILS_MISSING")
return rows, snapshots
def _receipts(session, run, key, group, facts):
records = _group_records(session, run.source_database, "receipts", group["receipt_ids"])
rows = []
for record in records:
raw = record.payload_json
correction_fact = facts.get(("CORRECTION", record.source_id))
correction = (
correction_fact.data_json
if correction_fact and correction_fact.data_json.get("entity") == "RECEIPT"
else {}
)
redirect_target = correction.get("redirect_business_key")
if str(raw.get("idpartner")) != group["customer_id"] and redirect_target != key:
raise WorkbenchError("GROUP_CUSTOMER_MISMATCH")
fact = facts.get(("RECEIPT", record.source_id))
confirmed, voucher_evidence, voucher_pending = _voucher_evidence(raw, fact)
pending = []
pending.extend(voucher_pending)
if raw.get("isReceiveFlag") != 1:
pending.append("NOT_RECEIPT")
# 凭证关系与人工核实冲突时不能强行按人工勾选放行。
if pending:
confirmed = False
if record.source_id in group.get("excluded_receipt_ids", []):
continue
if raw.get("voucherstate") != 189:
confirmed = False
exclusion = None
if raw.get("isReceiveFlag") == 0:
exclusion = "NOT_RECEIPT"
elif (
voucher_evidence.get("basis") == "GL_SrcVoucherInfo/GL_Doc"
and voucher_evidence.get("count") == 0
and not voucher_pending
):
exclusion = "NOT_GENERATED_VOUCHER"
if redirect_target and redirect_target != key:
exclusion = "REDIRECTED_RECEIPT"
rows.append(
{
"source_id": record.source_id,
"source_no": record.source_no,
"date": str(raw["voucherDate"])[:10],
"group_key": key,
"amount": str(raw["amount"]),
"voucher_confirmed": confirmed,
"exclusion": exclusion,
"voucher_evidence": voucher_evidence,
"accounting_voucher_id": raw.get("accounting_voucher_id"),
"accounting_voucher_code": raw.get("accounting_voucher_code"),
"pending": pending,
"redirect_business_key": redirect_target,
"redirected_from_customer_id": str(raw.get("idpartner"))
if redirect_target == key and str(raw.get("idpartner")) != key
else None,
}
)
return rows, records
def _voucher_evidence(raw, fact):
"""以真实总账关系核实生成资格;旧快照只接受有依据的人工核实。"""
count, valid = raw.get("accounting_voucher_count"), raw.get("valid_accounting_voucher_count")
if count is None or valid is None:
return (
bool(fact and fact.data_json["voucher_confirmed"]),
{
"basis": "MANUAL" if fact else "UNKNOWN",
"reference": fact.data_json["voucher_reference"] if fact else None,
},
[],
)
count, valid = int(count), int(valid)
if count < 0 or valid < 0 or valid > count:
raise ValueError("凭证关联数量异常")
generated = count > 0 and valid == count and raw.get("accounting_voucher_id") is not None
pending = ["ACCOUNTING_VOUCHER_INVALID"] if count > 0 and not generated else []
if fact and fact.data_json["voucher_confirmed"] != generated:
pending.append("VOUCHER_EVIDENCE_CONFLICT")
return (
generated and not pending,
{
"basis": "GL_SrcVoucherInfo/GL_Doc",
"count": count,
"valid_count": valid,
"voucher_id": raw.get("accounting_voucher_id"),
"voucher_code": raw.get("accounting_voucher_code"),
"reference": fact.data_json["voucher_reference"] if fact else None,
},
pending,
)
def _costed_shipments(shipments, receipts, facts, group):
"""回款月定价时成本按每笔回款取值,交给 FIFO 逐段计算。"""
basis = "SHIPMENT"
for ship in shipments:
if basis is None:
ship["pending"].append("GUIDANCE_MONTH_BASIS_PENDING")
continue
months = {r["date"][:7] for r in receipts} if basis == "RECEIPT" else {ship["date"][:7]}
costs = {}
for month in months:
guide = _guidance(facts, ship["product_id"], ship["line_id"], month)
if guide is None or guide["unit"] != ship["unit"]:
continue
ship["guidance_finalized"] = bool(guide.get("is_finalized"))
cost = money(guide["amount"])
if guide["basis"] == "UNIT":
if money(ship["quantity"]) <= 0:
ship["pending"].append("QUANTITY_PENDING")
continue
cost *= money(ship["quantity"])
costs[month] = amount_text(cost)
ship["cost_by_month"] = costs
ship["cost_month_basis"] = basis
ship["cost_total"] = costs.get(ship["date"][:7]) if basis == "SHIPMENT" else None
def execute_preview(session, run, *, save_lines=True, customer_scope=False):
"""同一次事务固化所有事实、来源指纹、参数和结果,不写正式台账。"""
if run.input_json.get("replay_no"):
return _replay(session, run)
facts = latest_facts(session, run.source_database)
month = run.input_json["month"]
year, mon = map(int, month.split("-"))
until = date(year, mon, monthrange(year, mon)[1])
selected = []
overlapping = set()
for key in run.input_json["group_keys"]:
fact = facts.get(("GROUP", key))
if fact is None:
selected.append((key, {}))
continue
group = dict(fact.data_json)
if customer_scope:
peers = [
f.data_json
for (kind, _), f in facts.items()
if kind == "GROUP" and f.data_json["customer_id"] == group["customer_id"]
]
def policy(g):
return (
money(g["opening_balance"]) if g.get("opening_balance") is not None else None,
tuple(sorted(g.get("excluded_receipt_ids", []))),
bool(g.get("opening_confirmed")),
bool(g.get("historical_settlement_confirmed")),
)
if len({policy(g) for g in peers}) != 1:
selected.append((key, {"customer_scope_conflict": True}))
continue
# 2026-10-30 E7:同一发货客户名头统一平账,不再按合同/人工业务键拆组。
group["scope_confirmed"] = True
for entity, field, day_field, target_field in (
("shipments", "idcustomer", "voucherdate", "shipment_ids"),
("receipts", "idpartner", "voucherDate", "receipt_ids"),
):
statement = source_statement(session, run.source_database, entity).where(
TPlusSyncRecord.payload_json[field].as_integer() == int(group["customer_id"])
)
source_ids = [
r.source_id
for r in session.scalars(statement)
if START.isoformat()
<= str(r.payload_json.get(day_field, ""))[:10]
<= until.isoformat()
]
if entity == "receipts":
source_ids.extend(
f.fact_key
for (kind, _), f in facts.items()
if kind == "CORRECTION"
and f.data_json.get("entity") == "RECEIPT"
and f.data_json.get("redirect_business_key") == key
and f.fact_key not in source_ids
)
group[target_field] = sorted(source_ids, key=lambda value: int(value))
# 也检查未选组,避免同一单据在另一组被解释为独立资金。
for (kind, other_key), other in facts.items():
if (
kind == "GROUP"
and other_key != key
and not customer_scope
and (
set(group["shipment_ids"]) & set(other.data_json["shipment_ids"])
or set(group["receipt_ids"]) & set(other.data_json["receipt_ids"])
)
):
overlapping.add(key)
selected.append((key, group))
frozen = {
"rule_version": RULE_VERSION,
"month": month,
"source": [],
"facts": [
{
"id": f.id,
"kind": f.kind,
"key": f.fact_key,
"version": f.version,
"data": f.data_json,
"evidence": f.evidence,
}
for f in facts.values()
],
"parameters": {},
"calculation_inputs": {},
}
count, summaries = 0, {}
person_facts = {key: f.data_json for (kind, key), f in facts.items() if kind == "PERSONNEL"}
handovers = [f.data_json for (kind, _), f in facts.items() if kind == "HANDOVER"]
for key, data in selected:
group = {**data, "key": key}
if data.get("customer_scope_conflict"):
result = {"status": "PENDING", "reasons": ["CUSTOMER_SCOPE_CONFLICT"], "lines": []}
elif key in overlapping:
result = {"status": "PENDING", "reasons": ["OVERLAPPING_GROUPS"], "lines": []}
elif not data:
result = {"status": "PENDING", "reasons": ["GROUP_FACT_REQUIRED"], "lines": []}
elif (
not group.get("scope_confirmed")
or not group.get("opening_confirmed")
or not group.get("historical_settlement_confirmed")
):
result = preview_group(group, [], [], {}, {}, until)
else:
try:
ships, ship_refs = _shipments(session, run, key, group, facts)
cash, cash_refs = _receipts(session, run, key, group, facts)
frozen["source"].extend(
(r.entity_type, r.source_id, r.id, r.row_hash) for r in ship_refs + cash_refs
)
_costed_shipments(ships, cash, facts, group)
# max(发货日,回款日) 必为其中一个日期,无需构造单据笛卡尔积。
days = {
row["date"]
for row in [*ships, *cash]
if START.isoformat() <= row["date"] <= until.isoformat()
}
params = _parameters(session, days)
# 已确认的原修改方案和到账日口径不再依赖重复人工勾选。
params["august_confirmed"] = True
params["prepayment_parameter_basis"] = "RECEIPT"
frozen["parameters"][key] = params
frozen["calculation_inputs"][key] = {
"group": group,
"shipments": ships,
"receipts": cash,
"facts": {"personnel": person_facts, "handovers": handovers},
"params": params,
}
result = preview_group(
group,
ships,
cash,
{"personnel": person_facts, "handovers": handovers},
params,
until,
)
except WorkbenchError as exc:
frozen["calculation_inputs"].pop(key, None)
result = {"status": "PENDING", "reasons": [str(exc)], "lines": []}
except BadRequestError:
frozen["calculation_inputs"].pop(key, None)
result = {"status": "PENDING", "reasons": ["PARAMETERS_PENDING"], "lines": []}
except (ValueError, KeyError, InvalidOperation, TypeError):
frozen["calculation_inputs"].pop(key, None)
result = {"status": "PENDING", "reasons": ["DATA_VALUE_INVALID"], "lines": []}
if key not in frozen["calculation_inputs"]:
frozen["calculation_inputs"][key] = {"pending": result}
summaries[key] = {k: v for k, v in result.items() if k != "lines"}
if save_lines:
count = _save_group(session, run, key, result, count)
run.snapshot_json, run.snapshot_hash = frozen, digest(frozen)
run.summary_json = {
"groups": summaries,
"total_lines": count,
"rule_version": RULE_VERSION,
"month": month,
"formal_payroll_written": False,
}
states = Counter(v["status"] for v in summaries.values())
run.status = "PARTIAL" if states["PENDING"] or states["PARTIAL"] else "SUCCEEDED"
def _save_group(session, run, key, result, count):
lines = result["lines"] or (
[{"status": "PENDING", "reasons": result["reasons"], "performance": None}]
if result["status"] == "PENDING"
else []
)
for line in lines:
count += 1
session.add(
TPlusWorkLine(
run_no=run.run_no,
line_no=count,
entity_type="group",
source_id=key,
action=line["status"],
reason=",".join(line.get("reasons", []))[:128] or None,
data_json={**line, "group_key": key, "formal_payroll_written": False},
)
)
return count
def _replay(session, run):
"""只重算原运行冻结输入;不读取当前事实、指导价、参数或源数据。"""
import copy
previous = get_run(session, run.input_json["replay_no"], run.source_database)
frozen = copy.deepcopy(previous.snapshot_json or {})
if (
previous.kind != "CALC_PREVIEW"
or frozen.get("rule_version") not in ("2026-10-30.1", RULE_VERSION)
or not frozen.get("calculation_inputs")
):
raise WorkbenchError("REPLAY_UNAVAILABLE")
year, month = map(int, frozen["month"].split("-"))
until = date(year, month, monthrange(year, month)[1])
summaries, count = {}, 0
for key, values in frozen["calculation_inputs"].items():
result = values.get("pending")
if result is None:
result = preview_group(
values["group"],
values["shipments"],
values["receipts"],
values["facts"],
values["params"],
until,
)
summaries[key] = {k: v for k, v in result.items() if k != "lines"}
count = _save_group(session, run, key, result, count)
run.snapshot_json, run.snapshot_hash = frozen, digest(frozen)
run.summary_json = {
"groups": summaries,
"total_lines": count,
"rule_version": RULE_VERSION,
"month": frozen["month"],
"formal_payroll_written": False,
}
run.status = (
"PARTIAL"
if any(v["status"] in ("PENDING", "PARTIAL") for v in summaries.values())
else "SUCCEEDED"
)