diff --git a/backend/app/api/treasury.py b/backend/app/api/treasury.py new file mode 100644 index 00000000..2944f72e --- /dev/null +++ b/backend/app/api/treasury.py @@ -0,0 +1,1199 @@ +"""司库（资金）体系 · 阶段一 端点层 — /api/cma/treasury/* + +PRD v1.1 附录 A + 阶段一范围（账户池 / 头寸 / 内控闸门 / 授权矩阵 / 两页面数据面）。 +G1 裁定落地： + * 不建物理外键 → 应用层校验覆盖（DDL §F-2 逐表说明：账户同实体校验 / AC-001 账外账户 / 预算同实体 …） + * 金额 FLOAT（万元）：写库前 round(x,4)，比较一律用容差 1e-4（禁用浮点等值比较） + * 分级授权「取最严」：amount 档（rules.params.tiers）× treasury_limits（role/account/tier scopes）× 拆单累计（同一提交人×同一收款方×同一自然日） + * 全量留痕：每条规则每次评估写一行 treasury_control_hits；detail 固定结构化字段 + {rule_code, rule_version, matched_scope, threshold, cumulative, verdict_basis} + * 豁免通道：指定审批人 + 必填事由 + 全量留痕；豁免级别不得低于该笔所需审批级数 +边界：**零改动 cash.py**；只读复用 cash_plans / budget_plans；不动其他模块表结构。 +""" +import logging +from datetime import datetime, date +from typing import Optional, List, Any, Dict + +from fastapi import APIRouter, HTTPException, Depends, Query +from pydantic import BaseModel, Field +from sqlalchemy import func, or_ +from sqlalchemy.orm import Session + +from app.database import get_db +from app.deps import get_entity_id +from app.auth_middleware import require_role, require_auth +from app.models import BudgetPlan, CashPlan, User, OrgNode +from app.models.treasury import ( + TreasuryAccount, TreasuryBalance, TreasuryControlRule, + TreasuryPayment, TreasuryControlHit, TreasuryLimit, +) + +logger = logging.getLogger("cma.treasury") + +router = APIRouter( + prefix="/api/cma/treasury", + tags=["司库（资金）"], + dependencies=[Depends(require_role("ceo", "finance", "business", "it"))], +) + +TOL = 1e-4 # 金额比较容差（G1 §3-3 附条件：禁浮点等值比较） +ACTIVE_STATUSES = ("pending_approval", "approved", "paid") # NP-001 占用口径：已审批未支付 + 已支付 +INTERNAL_TRANSFER = "内部划转" + + +def _r(x) -> float: + """金额写库/计算统一 round(x, 4)（万元，FLOAT）。""" + return round(float(x or 0), 4) + + +def _ge(a, b) -> bool: + return float(a) >= float(b) - TOL + + +def _lt(a, b) -> bool: + return float(a) < float(b) - TOL + + +def _now() -> datetime: + return datetime.now() + + +# ───────────────────────────────────────────────────────────── +# 规则引擎（阶段一：AC-001 / NP-001 / AU-001 / BL-001 / LQ-001） +# ───────────────────────────────────────────────────────────── + +def _load_rules(db: Session, entity_id: int) -> Dict[str, TreasuryControlRule]: + rows = db.query(TreasuryControlRule).filter( + TreasuryControlRule.entity_id == entity_id, + TreasuryControlRule.enabled == 1, + ).all() + return {r.rule_code: r for r in rows} + + +def _hit_detail(rule_code: str, rule_version: str, matched_scope: str, + threshold: Any, cumulative: Any, verdict_basis: str) -> dict: + """[调整2] detail 固定结构化字段（审计追问要能自证）。""" + return { + "rule_code": rule_code, + "rule_version": rule_version, + "matched_scope": matched_scope, + "threshold": threshold, + "cumulative": cumulative, + "verdict_basis": verdict_basis, + } + + +def _record_hits(db: Session, entity_id: int, target_type: str, target_id: int, + results: List[dict]) -> None: + """append-only 留痕：每条规则每次评估一行（pass 也留，审计可复现全链路）。""" + for res in results: + db.add(TreasuryControlHit( + entity_id=entity_id, + target_type=target_type, + target_id=target_id, + rule_code=res["rule_code"], + verdict=res["verdict"], + detail=res["detail"], + )) + + +def _pick_account(db: Session, entity_id: int, account_id: int) -> TreasuryAccount: + acc = db.query(TreasuryAccount).filter( + TreasuryAccount.id == account_id, + TreasuryAccount.entity_id == entity_id, # 防跨实体误挂（G1 附加验收①） + ).first() + if not acc: + raise HTTPException(404, f"账户 {account_id} 不存在或不属于当前账套") + return acc + + +def _resolve_submitter_role(db: Session, submitter: str) -> Optional[str]: + if not submitter: + return None + u = db.query(User).filter(or_(User.name == submitter, User.username == submitter)).first() + return u.role if u else None + + +def _tier_of(tiers: List[dict], amount: float): + """金额段命中（口径2-④ 左闭右开 [from,to)）：返回 (required_approvals, scope_text)。""" + for t in sorted(tiers, key=lambda x: float(x.get("from") or 0)): + lo = float(t.get("from") or 0) + hi = t.get("to") + if _ge(amount, lo) and (hi is None or _lt(amount, float(hi))): + return int(t.get("required_approvals") or 0), f"[{lo},{hi if hi is not None else '∞'})" + return 0, "none" + + +def _eval_ac001(db, entity_id, payment, rule) -> dict: + """AC-001 账外账户：付款账户必须已登记且 active（结构性校验，无金额参数）。""" + acc = db.query(TreasuryAccount).filter( + TreasuryAccount.id == payment.from_account_id, + TreasuryAccount.entity_id == entity_id, + ).first() + if not acc: + verdict, basis = "blocked", "account_not_registered（账外账户：未登记入台账）" + elif acc.status != "active": + verdict, basis = "blocked", f"account_status={acc.status}（非 active 账户不允许发起支付）" + else: + verdict, basis = "pass", f"account#{acc.id} {acc.account_no} status=active" + return {"rule_code": "AC-001", "verdict": verdict, "severity": "block", + "detail": _hit_detail("AC-001", "builtin-v1", "account", None, None, basis)} + + +def _eval_np001(db, entity_id, payment, rule) -> dict: + """NP-001 无预算不开支：占用 = 已支付 + 已审批未支付 + 本单；匹配不到走指定/补录预算流程。""" + params = rule.params or {} + severity = params.get("severity") or rule.severity + budget_value, scope = None, None + if payment.budget_plan_id: + bp = db.query(BudgetPlan).filter( + BudgetPlan.id == payment.budget_plan_id, BudgetPlan.entity_id == entity_id).first() + if bp: + budget_value, scope = bp.budget_value, f"budget_plan:{bp.id}" + if budget_value is None and payment.kpi_id: + period = payment.pay_date.strftime("%Y-%m") if payment.pay_date else datetime.now().strftime("%Y-%m") + bp = db.query(BudgetPlan).filter( + BudgetPlan.entity_id == entity_id, BudgetPlan.kpi_id == payment.kpi_id, + BudgetPlan.period == period).first() + if bp: + budget_value, scope = bp.budget_value, f"kpi:{bp.kpi_id}@{period}" + this_amt = _r(payment.amount) + if budget_value is None: + return {"rule_code": "NP-001", "verdict": severity, "severity": severity, + "detail": _hit_detail("NP-001", rule.rule_version, "none", None, this_amt, + (params.get("unmatched") or "specify_budget_flow") + + "（未匹配到预算：走指定/补录预算流程并留痕，不直接卡死）")} + q = db.query(func.coalesce(func.sum(TreasuryPayment.amount), 0)).filter( + TreasuryPayment.entity_id == entity_id, + TreasuryPayment.status.in_(ACTIVE_STATUSES), + TreasuryPayment.id != payment.id, + ) + if payment.budget_plan_id: + q = q.filter(TreasuryPayment.budget_plan_id == payment.budget_plan_id) + elif payment.kpi_id: + q = q.filter(TreasuryPayment.kpi_id == payment.kpi_id) + prev = float(q.scalar() or 0) + cumulative = _r(prev + this_amt) + ok = cumulative <= float(budget_value) + TOL + verdict = "pass" if ok else severity + basis = ("占用口径 %s；占用=已支付+已审批未支付+本单" % (params.get("occupancy") or "paid+approved+pending_approval+this")) + if not ok: + basis += f"；占用 {cumulative} 超过预算 {_r(budget_value)}" + return {"rule_code": "NP-001", "verdict": verdict, "severity": severity, + "detail": _hit_detail("NP-001", rule.rule_version, scope or "none", + _r(budget_value), cumulative, basis)} + + +def _eval_au001(db, entity_id, payment, rule) -> dict: + """AU-001 超授权拦截（拆单累计穿透 + 日累计升级 + 多档取最严 + 左闭右开）。""" + params = rule.params or {} + tiers = params.get("tiers") or [] + amount = _r(payment.amount) + daily_cum_limit = params.get("daily_cum_limit") + pay_day = payment.pay_date or date.today() + + base_req, base_scope = _tier_of(tiers, amount) + + # 拆单累计穿透：同一提交人 × 同一收款方 × 同一自然日（含本单） + cum_q = db.query(func.coalesce(func.sum(TreasuryPayment.amount), 0)).filter( + TreasuryPayment.entity_id == entity_id, + TreasuryPayment.status.in_(ACTIVE_STATUSES), + TreasuryPayment.id != payment.id, + TreasuryPayment.submitter == payment.submitter, + TreasuryPayment.to_name == payment.to_name, + TreasuryPayment.pay_date == pay_day, + ) + prev_same = float(cum_q.scalar() or 0) + cum_same = _r(prev_same + amount) + cum_req, cum_scope = _tier_of(tiers, cum_same) + + # 日累计升级：同提交人当日全部（含本单）超 daily_cum_limit → 自主档升一级 + day_q = db.query(func.coalesce(func.sum(TreasuryPayment.amount), 0)).filter( + TreasuryPayment.entity_id == entity_id, + TreasuryPayment.status.in_(ACTIVE_STATUSES), + TreasuryPayment.id != payment.id, + TreasuryPayment.submitter == payment.submitter, + TreasuryPayment.pay_date == pay_day, + ) + day_total = _r(float(day_q.scalar() or 0) + amount) + upgraded = False + if daily_cum_limit is not None and base_req == 0 and day_total > float(daily_cum_limit) + TOL: + upgraded = True + + # 多档取最严：叠加 treasury_limits（role/account/tier 维度） + limits_req, limits_scope = _limits_strictest(db, entity_id, payment, amount, cum_same) + + required = max(base_req, cum_req, limits_req, 1 if upgraded else 0) + scopes = [f"amount_tier{base_scope}"] + if cum_req > base_req: + scopes.append(f"cumulative_tier{cum_scope}({cum_same})") + if limits_req: + scopes.append(f"limits:{limits_scope}") + if upgraded: + scopes.append(f"daily_cum_upgrade(total={day_total}>limit={_r(daily_cum_limit)})") + matched_scope = ";".join(scopes) + basis = (f"单笔 {amount} 档位 {base_scope}→{base_req}级；拆单累计（同提交人×同收款方×同自然日）{cum_same}" + f"→{cum_req}级；取最严={required}级；边界左闭右开") + return {"rule_code": "AU-001", "verdict": "pass", "severity": rule.severity, + "required_approvals": required, "day_cum_amount": cum_same, + "detail": _hit_detail("AU-001", rule.rule_version, matched_scope, + _r(daily_cum_limit) if daily_cum_limit is not None else None, + cum_same, basis)} + + +def _limits_strictest(db, entity_id, payment, amount, cum_amount): + """treasury_limits 中 role/account/tier 维度的最严档（不硬编码，走表）。""" + rows = db.query(TreasuryLimit).filter( + TreasuryLimit.entity_id == entity_id, + TreasuryLimit.status == "active", + ).all() + role = _resolve_submitter_role(db, payment.submitter) + acc = db.query(TreasuryAccount).filter( + TreasuryAccount.id == payment.from_account_id, + TreasuryAccount.entity_id == entity_id).first() + best, scope = 0, None + for r in rows: + if r.scope_type == "role" and (not role or r.scope_value != role): + continue + if r.scope_type == "account" and str(r.scope_value) != str(payment.from_account_id): + continue + if r.scope_type == "tier" and (not acc or r.scope_value != acc.tier): + continue + for amt in (amount, cum_amount): + if _ge(amt, r.amount_from) and (r.amount_to is None or _lt(amt, r.amount_to)): + if int(r.required_approvals or 0) > best: + best = int(r.required_approvals or 0) + scope = (f"{r.scope_type}={r.scope_value}@[{_r(r.amount_from)}," + f"{_r(r.amount_to) if r.amount_to is not None else '∞'})") + return best, scope + + +def _eval_bl001_requirement(db, entity_id, payment, rule) -> dict: + """BL-001 预检阶段：登记"本单需双签"要求（≥min_amount）；实质约束在 /approve 执行。""" + params = rule.params or {} + min_amount = float(params.get("min_amount") or 0) + if payment.pay_type == INTERNAL_TRANSFER: + return {"rule_code": "BL-001", "verdict": "pass", "severity": rule.severity, + "detail": _hit_detail("BL-001", rule.rule_version, "pay_type=内部划转", + _r(min_amount), _r(payment.amount), + "同一法人内部账户划转/归集不适用 BL-001（走单签+全量留痕）")} + if _ge(payment.amount, min_amount): + return {"rule_code": "BL-001", "verdict": "warn", "severity": rule.severity, + "detail": _hit_detail("BL-001", rule.rule_version, + f"amount>={_r(min_amount)}", _r(min_amount), _r(payment.amount), + "awaiting_dual_sign：需 2 个不同账号且不同自然人的审批（审批人" + "不得为提交人本人）；未满足前不得置 approved")} + return {"rule_code": "BL-001", "verdict": "pass", "severity": rule.severity, + "detail": _hit_detail("BL-001", rule.rule_version, f"amount<{_r(min_amount)}", + _r(min_amount), _r(payment.amount), "未达双签门槛")} + + +def _eval_lq001(db, entity_id, payment, rule) -> dict: + """LQ-001 流动性下限（首版 warn-only；estimated 头寸禁 block）。""" + params = rule.params or {} + target = float(params.get("target") or 0) + warn = float(params.get("warn") or 0) + block = float(params.get("block") or 0) + block_enabled = bool(params.get("block_enabled")) + bal = db.query(TreasuryBalance).filter( + TreasuryBalance.entity_id == entity_id, + TreasuryBalance.account_id == payment.from_account_id, + ).order_by(TreasuryBalance.period.desc()).first() + if not bal: + return {"rule_code": "LQ-001", "verdict": "pass", "severity": "warn", + "detail": _hit_detail("LQ-001", rule.rule_version, "none", _r(warn), None, + "无头寸数据（no_position_data）：不判定，不阻断")} + if bal.data_status == "estimated": + return {"rule_code": "LQ-001", "verdict": "pass", "severity": "warn", + "detail": _hit_detail("LQ-001", rule.rule_version, + f"account:{bal.account_id}@{bal.period}", _r(warn), + _r(bal.balance), + "skip_when=data_status=estimated（期初未落定，estimated 禁 block）")} + after = _r(float(bal.balance) - float(payment.amount)) + if block_enabled and _lt(after, block): + verdict, basis = "blocked", f"支付后头寸 {after} < 阻断线 {_r(block)}" + elif _lt(after, warn): + verdict, basis = "warn", f"支付后头寸 {after} < 警戒线 {_r(warn)}" + elif _lt(after, target): + verdict, basis = "warn", f"支付后头寸 {after} < 目标线 {_r(target)}" + else: + verdict, basis = "pass", f"支付后头寸 {after} >= 目标线 {_r(target)}" + return {"rule_code": "LQ-001", "verdict": verdict, "severity": "warn", + "detail": _hit_detail("LQ-001", rule.rule_version, + f"account:{bal.account_id}@{bal.period}", + {"target": _r(target), "warn": _r(warn), "block": _r(block)}, + after, basis)} + + +def eval_payment(db: Session, entity_id: int, payment: TreasuryPayment, + persist: bool = True) -> dict: + """阶段一规则引擎主入口（预检）：AC-001 / NP-001 / AU-001 / BL-001 / LQ-001。""" + rules = _load_rules(db, entity_id) + results: List[dict] = [] + + r_ac = rules.get("AC-001") + results.append(_eval_ac001(db, entity_id, payment, r_ac)) + + if r_ac and results[-1]["verdict"] == "blocked": + severity_all = "blocked" + detail = {"results": [{"rule_code": r["rule_code"], "verdict": r["verdict"], + "matched_scope": r["detail"]["matched_scope"]} for r in results], + "reason": "AC-001 账外账户拦截（后续规则不再评估）"} + if persist: + _record_hits(db, entity_id, "payment", payment.id, results) + return {"check_result": severity_all, "required_approvals": None, + "day_cum_amount": None, "check_detail": detail, + "dual_sign_required": False} + + r_np = rules.get("NP-001") + if r_np: + results.append(_eval_np001(db, entity_id, payment, r_np)) + r_au = rules.get("AU-001") + req = 0 + day_cum = None + if r_au: + au = _eval_au001(db, entity_id, payment, r_au) + req = au["required_approvals"] + day_cum = au["day_cum_amount"] + results.append({k: v for k, v in au.items() if k not in ("required_approvals", "day_cum_amount")}) + r_bl = rules.get("BL-001") + dual_required = False + if r_bl: + bl = _eval_bl001_requirement(db, entity_id, payment, r_bl) + dual_required = bl["verdict"] == "warn" + results.append(bl) + r_lq = rules.get("LQ-001") + if r_lq: + results.append(_eval_lq001(db, entity_id, payment, r_lq)) + + if any(x["verdict"] == "blocked" for x in results): + check_result = "blocked" + elif any(x["verdict"] in ("warn", "blocked") for x in results): + check_result = "warn" + else: + check_result = "pass" + + detail = {"results": [{"rule_code": r["rule_code"], "verdict": r["verdict"], + "matched_scope": r["detail"]["matched_scope"], + "threshold": r["detail"]["threshold"], + "cumulative": r["detail"]["cumulative"], + "verdict_basis": r["detail"]["verdict_basis"], + "rule_version": r["detail"]["rule_version"]} for r in results], + "required_approvals": req, + "dual_sign_required": dual_required, + "evaluated_at": _now().isoformat(timespec="seconds")} + if persist: + _record_hits(db, entity_id, "payment", payment.id, results) + return {"check_result": check_result, "required_approvals": req, + "day_cum_amount": day_cum, "check_detail": detail, + "dual_sign_required": dual_required} + + +def eval_dual_sign_at_approve(db: Session, entity_id: int, payment: TreasuryPayment) -> dict: + """BL-001 实质约束（审批时执行）：两账号须不同自然人，否则形式双签（不放行，审计标红）。""" + rule = _load_rules(db, entity_id).get("BL-001") + if not rule: + return {"verdict": "pass", "basis": "BL-001 未启用"} + params = rule.params or {} + if payment.pay_type == INTERNAL_TRANSFER: + return {"verdict": "pass", "basis": "内部划转不适用 BL-001"} + min_amount = float(params.get("min_amount") or 0) + if _lt(payment.amount, min_amount): + return {"verdict": "pass", "basis": f"未达双签门槛 {_r(min_amount)} 万元"} + entries = payment.approvals or [] + persons = {str(e.get("person_name") or e.get("account") or "") + for e in entries if e.get("account") != payment.submitter} + persons.discard("") + if len(persons) >= 2: + return {"verdict": "pass", "basis": f"双签有效：{len(persons)} 个不同自然人 {sorted(persons)}"} + return {"verdict": "blocked", "basis": + f"形式双签拦截：审批数 {len(entries)}，不同自然人 {len(persons)} < 2" + f"（dual_sign_effective=0，审计标红，不作为有效双签放行）"} + + +def _dual_sign_effective(payment: TreasuryPayment) -> Optional[int]: + """两账号是否属不同自然人：0=形式双签（放行无效，审计标红）。""" + entries = payment.approvals or [] + persons = {str(e.get("person_name") or e.get("account") or "") + for e in entries if e.get("account") != payment.submitter} + persons.discard("") + return 1 if len(persons) >= 2 else 0 + + +# ───────────────────────────────────────────────────────────── +# Pydantic 输入 +# ───────────────────────────────────────────────────────────── + +class AccountIn(BaseModel): + account_no: str + account_name: str + account_type: str + tier: str + bank_name: Optional[str] = None + subject_code: Optional[str] = None + currency: str = "CNY" + owner_node_id: Optional[int] = None + open_balance: Optional[float] = None + open_period: Optional[str] = None + is_pool_member: int = 0 + status: str = "active" + remark: Optional[str] = None + + +class BalanceIn(BaseModel): + account_id: Optional[int] = None + account_no: Optional[str] = None + period: str + balance: float + available: Optional[float] = None + frozen: Optional[float] = None + source: str = "import" + data_status: str = "estimated" # fail-safe：verified 必须显式传 + batch: Optional[str] = None + + +class RuleIn(BaseModel): + rule_code: str + rule_name: str + rule_type: str + severity: str + params: Optional[dict] = None + rule_version: str = "v1" + enabled: int = 1 + remark: Optional[str] = None + + +class LimitIn(BaseModel): + scope_type: str + scope_value: str + amount_from: float + amount_to: Optional[float] = None + required_approvals: int = 1 + approver_roles: List[str] = [] + require_budget: int = 1 + daily_cum_limit: Optional[float] = None + status: str = "active" + + +class PaymentIn(BaseModel): + pay_no: Optional[str] = None + from_account_id: int + to_name: str + to_account_no: Optional[str] = None + amount: float + currency: str = "CNY" + purpose: str + pay_type: Optional[str] = None + pay_date: Optional[str] = None + budget_plan_id: Optional[int] = None + kpi_id: Optional[int] = None + related_plan_id: Optional[int] = None + submitter: Optional[str] = None + + +class ApproveIn(BaseModel): + approve: bool = True + comment: Optional[str] = None + + +class WaiveIn(BaseModel): + waived_by: str = Field(..., description="豁免审批人（姓名/账号）") + waived_reason: str = Field(..., min_length=4, description="豁免事由（必填）") + + +# ───────────────────────────────────────────────────────────── +# A. 账户主数据 +# ───────────────────────────────────────────────────────────── + +def _account_dict(a: TreasuryAccount) -> dict: + return { + "id": a.id, "entity_id": a.entity_id, "account_no": a.account_no, + "account_name": a.account_name, "account_type": a.account_type, + "bank_name": a.bank_name, "subject_code": a.subject_code, "currency": a.currency, + "tier": a.tier, "tier_basis": a.tier_basis, "owner_node_id": a.owner_node_id, + "open_balance": a.open_balance, "open_period": a.open_period, + "is_pool_member": a.is_pool_member, "status": a.status, "remark": a.remark, + } + + +@router.get("/accounts") +def list_accounts(status: Optional[str] = None, tier: Optional[str] = None, + account_type: Optional[str] = None, keyword: Optional[str] = None, + db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + q = db.query(TreasuryAccount).filter(TreasuryAccount.entity_id == entity_id) + if status: + q = q.filter(TreasuryAccount.status == status) + if tier: + q = q.filter(TreasuryAccount.tier == tier) + if account_type: + q = q.filter(TreasuryAccount.account_type == account_type) + if keyword: + like = f"%{keyword}%" + q = q.filter(or_(TreasuryAccount.account_name.like(like), + TreasuryAccount.account_no.like(like), + TreasuryAccount.subject_code.like(like))) + rows = q.order_by(TreasuryAccount.account_no).all() + return {"total": len(rows), "items": [_account_dict(a) for a in rows]} + + +@router.post("/accounts") +def create_account(payload: AccountIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_auth)): + if payload.tier not in ("hq", "platform", "grassroots"): + raise HTTPException(400, "tier 取值须为 hq/platform/grassroots（按账户性质判定，禁按 level 派生）") + if payload.owner_node_id: + node = db.query(OrgNode).filter(OrgNode.id == payload.owner_node_id, + OrgNode.entity_id == entity_id).first() + if not node: + raise HTTPException(400, "owner_node_id 不存在于当前账套（禁造虚层级）") + exists = db.query(TreasuryAccount).filter( + TreasuryAccount.entity_id == entity_id, + TreasuryAccount.account_no == payload.account_no).first() + if exists: + raise HTTPException(409, f"账号 {payload.account_no} 已在该账套登记（唯一键 uq_tacct_entity_no）") + acc = TreasuryAccount(entity_id=entity_id, **payload.model_dump()) + acc.open_balance = None if payload.open_balance is None else _r(payload.open_balance) + db.add(acc) + db.commit() + db.refresh(acc) + return _account_dict(acc) + + +@router.put("/accounts/{acc_id}") +def update_account(acc_id: int, payload: AccountIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_auth)): + acc = _pick_account(db, entity_id, acc_id) + for k, v in payload.model_dump().items(): + setattr(acc, k, _r(v) if k == "open_balance" and v is not None else v) + db.commit() + db.refresh(acc) + return _account_dict(acc) + + +@router.delete("/accounts/{acc_id}") +def delete_account(acc_id: int, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_auth)): + acc = _pick_account(db, entity_id, acc_id) + used = db.query(TreasuryPayment).filter( + TreasuryPayment.entity_id == entity_id, + TreasuryPayment.from_account_id == acc_id).count() + bal = db.query(TreasuryBalance).filter( + TreasuryBalance.entity_id == entity_id, + TreasuryBalance.account_id == acc_id).count() + if used or bal: + raise HTTPException(409, f"账户已被引用（支付单 {used} 笔 / 头寸 {bal} 行），请置 inactive 而非删除") + db.delete(acc) + db.commit() + return {"deleted": acc_id} + + +@router.post("/accounts/import") +def import_accounts(rows: List[AccountIn], db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_auth)): + """批量导入（幂等：按 (entity_id, account_no) upsert）。F1 账户主数据清单即走此通道。""" + created = updated = 0 + for row in rows: + acc = db.query(TreasuryAccount).filter( + TreasuryAccount.entity_id == entity_id, + TreasuryAccount.account_no == row.account_no).first() + data = row.model_dump() + if data.get("open_balance") is not None: + data["open_balance"] = _r(data["open_balance"]) + if acc: + for k, v in data.items(): + setattr(acc, k, v) + updated += 1 + else: + db.add(TreasuryAccount(entity_id=entity_id, **data)) + created += 1 + db.commit() + return {"entity_id": entity_id, "created": created, "updated": updated, "total": len(rows)} + + +# ───────────────────────────────────────────────────────────── +# B. 头寸 +# ───────────────────────────────────────────────────────────── + +def _balance_dict(b: TreasuryBalance) -> dict: + return {"id": b.id, "entity_id": b.entity_id, "account_id": b.account_id, + "period": b.period, "balance": b.balance, "available": b.available, + "frozen": b.frozen, "source": b.source, "data_status": b.data_status, + "batch": b.batch, "created_at": b.created_at} + + +def _upsert_balance(db, entity_id, row: BalanceIn) -> str: + """按 (account_id, period) 幂等 upsert；写入前校验账户同实体（G1 附加验收①）。""" + acc = None + if row.account_id: + acc = db.query(TreasuryAccount).filter( + TreasuryAccount.id == row.account_id, + TreasuryAccount.entity_id == entity_id).first() + elif row.account_no: + acc = db.query(TreasuryAccount).filter( + TreasuryAccount.entity_id == entity_id, + TreasuryAccount.account_no == row.account_no).first() + if not acc: + raise HTTPException(400, f"账户未登记或不属于当前账套（account_id={row.account_id} " + f"account_no={row.account_no}）—— 头寸禁止跨实体误挂") + if row.data_status not in ("verified", "estimated"): + raise HTTPException(400, "data_status 须为 verified/estimated") + data = {"balance": _r(row.balance), + "available": None if row.available is None else _r(row.available), + "frozen": _r(row.frozen or 0), "source": row.source, + "data_status": row.data_status, "batch": row.batch} + bal = db.query(TreasuryBalance).filter( + TreasuryBalance.entity_id == entity_id, + TreasuryBalance.account_id == acc.id, + TreasuryBalance.period == row.period).first() + if bal: + for k, v in data.items(): + setattr(bal, k, v) + return "updated" + db.add(TreasuryBalance(entity_id=entity_id, account_id=acc.id, period=row.period, **data)) + return "created" + + +@router.get("/balances") +def list_balances(account_id: Optional[int] = None, period_from: Optional[str] = None, + period_to: Optional[str] = None, data_status: Optional[str] = None, + db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + q = db.query(TreasuryBalance).filter(TreasuryBalance.entity_id == entity_id) + if account_id: + q = q.filter(TreasuryBalance.account_id == account_id) + if period_from: + q = q.filter(TreasuryBalance.period >= period_from) + if period_to: + q = q.filter(TreasuryBalance.period <= period_to) + if data_status: + q = q.filter(TreasuryBalance.data_status == data_status) + rows = q.order_by(TreasuryBalance.period.desc(), TreasuryBalance.account_id).all() + return {"total": len(rows), "items": [_balance_dict(b) for b in rows]} + + +@router.post("/balances") +def upsert_balance(row: BalanceIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_auth)): + action = _upsert_balance(db, entity_id, row) + db.commit() + return {"action": action, "account": row.account_id or row.account_no, "period": row.period} + + +@router.post("/balances/import") +def import_balances(rows: List[BalanceIn], db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_auth)): + """批量导入（幂等：同一 period 重复导入结果不变）。""" + created = updated = 0 + for row in rows: + action = _upsert_balance(db, entity_id, row) + created += action == "created" + updated += action == "updated" + db.commit() + total = db.query(TreasuryBalance).filter(TreasuryBalance.entity_id == entity_id).count() + return {"entity_id": entity_id, "created": created, "updated": updated, + "rows_in": len(rows), "balances_total": total} + + +@router.get("/positions/summary") +def positions_summary(period: Optional[str] = None, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id)): + """实体头寸汇总：按 account_type / tier / data_status 分解（账户级 → 实体级）。""" + accounts = db.query(TreasuryAccount).filter(TreasuryAccount.entity_id == entity_id).all() + q = db.query(TreasuryBalance).filter(TreasuryBalance.entity_id == entity_id) + if period: + q = q.filter(TreasuryBalance.period == period) + balances = q.all() + latest: Dict[int, TreasuryBalance] = {} + for b in balances: + cur = latest.get(b.account_id) + if cur is None or b.period > cur.period: + latest[b.account_id] = b + by_type: Dict[str, float] = {} + by_tier: Dict[str, float] = {} + by_status: Dict[str, int] = {} + total = 0.0 + for a in accounts: + b = latest.get(a.id) + amt = float(b.balance) if b else float(a.open_balance or 0) + src = (b.data_status if b else ("estimated" if a.open_balance is None else "verified")) + total += amt + by_type[a.account_type] = _r(by_type.get(a.account_type, 0) + amt) + by_tier[a.tier] = _r(by_tier.get(a.tier, 0) + amt) + by_status[src] = by_status.get(src, 0) + 1 + return {"entity_id": entity_id, "period": period, "accounts": len(accounts), + "total_balance": _r(total), "by_account_type": by_type, "by_tier": by_tier, + "by_data_status": by_status, + "unit": "万元", "accounts_without_position": len(accounts) - len(latest)} + + +# ───────────────────────────────────────────────────────────── +# C. 规则库 / 授权矩阵 +# ───────────────────────────────────────────────────────────── + +def _rule_dict(r: TreasuryControlRule) -> dict: + return {"id": r.id, "entity_id": r.entity_id, "rule_code": r.rule_code, + "rule_name": r.rule_name, "rule_type": r.rule_type, "severity": r.severity, + "params": r.params, "rule_version": r.rule_version, "enabled": r.enabled, + "remark": r.remark} + + +def _limit_dict(r: TreasuryLimit) -> dict: + return {"id": r.id, "entity_id": r.entity_id, "scope_type": r.scope_type, + "scope_value": r.scope_value, "amount_from": r.amount_from, + "amount_to": r.amount_to, "required_approvals": r.required_approvals, + "approver_roles": r.approver_roles, "require_budget": r.require_budget, + "daily_cum_limit": r.daily_cum_limit, "status": r.status} + + +@router.get("/control-rules") +def list_rules(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + rows = db.query(TreasuryControlRule).filter( + TreasuryControlRule.entity_id == entity_id).order_by(TreasuryControlRule.rule_code).all() + return {"total": len(rows), "items": [_rule_dict(r) for r in rows]} + + +@router.post("/control-rules") +def create_rule(payload: RuleIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "finance", "it"))): + exists = db.query(TreasuryControlRule).filter( + TreasuryControlRule.entity_id == entity_id, + TreasuryControlRule.rule_code == payload.rule_code).first() + if exists: + raise HTTPException(409, f"规则 {payload.rule_code} 已存在（唯一键 uq_tcr_entity_code）") + r = TreasuryControlRule(entity_id=entity_id, **payload.model_dump()) + db.add(r) + db.commit() + db.refresh(r) + return _rule_dict(r) + + +@router.put("/control-rules/{rule_id}") +def update_rule(rule_id: int, payload: RuleIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "finance", "it"))): + r = db.query(TreasuryControlRule).filter( + TreasuryControlRule.id == rule_id, + TreasuryControlRule.entity_id == entity_id).first() + if not r: + raise HTTPException(404, "规则不存在或不属于当前账套") + for k, v in payload.model_dump().items(): + setattr(r, k, v) + db.commit() + db.refresh(r) + return _rule_dict(r) + + +@router.delete("/control-rules/{rule_id}") +def delete_rule(rule_id: int, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "it"))): + r = db.query(TreasuryControlRule).filter( + TreasuryControlRule.id == rule_id, + TreasuryControlRule.entity_id == entity_id).first() + if not r: + raise HTTPException(404, "规则不存在或不属于当前账套") + db.delete(r) + db.commit() + return {"deleted": rule_id} + + +@router.get("/limits") +def list_limits(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + rows = db.query(TreasuryLimit).filter(TreasuryLimit.entity_id == entity_id).order_by( + TreasuryLimit.scope_type, TreasuryLimit.scope_value, TreasuryLimit.amount_from).all() + return {"total": len(rows), "items": [_limit_dict(r) for r in rows]} + + +@router.post("/limits") +def create_limit(payload: LimitIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "finance", "it"))): + data = payload.model_dump() + data["amount_from"] = _r(data["amount_from"]) + data["amount_to"] = None if data["amount_to"] is None else _r(data["amount_to"]) + data["daily_cum_limit"] = None if data["daily_cum_limit"] is None else _r(data["daily_cum_limit"]) + l = TreasuryLimit(entity_id=entity_id, **data) + db.add(l) + db.commit() + db.refresh(l) + return _limit_dict(l) + + +@router.put("/limits/{limit_id}") +def update_limit(limit_id: int, payload: LimitIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "finance", "it"))): + l = db.query(TreasuryLimit).filter(TreasuryLimit.id == limit_id, + TreasuryLimit.entity_id == entity_id).first() + if not l: + raise HTTPException(404, "授权矩阵行不存在或不属于当前账套") + for k, v in payload.model_dump().items(): + setattr(l, k, _r(v) if k in ("amount_from", "amount_to", "daily_cum_limit") and v is not None else v) + db.commit() + db.refresh(l) + return _limit_dict(l) + + +@router.delete("/limits/{limit_id}") +def delete_limit(limit_id: int, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "it"))): + l = db.query(TreasuryLimit).filter(TreasuryLimit.id == limit_id, + TreasuryLimit.entity_id == entity_id).first() + if not l: + raise HTTPException(404, "授权矩阵行不存在或不属于当前账套") + db.delete(l) + db.commit() + return {"deleted": limit_id} + + +# ───────────────────────────────────────────────────────────── +# D. 支付单 + 内控闸门 +# ───────────────────────────────────────────────────────────── + +def _payment_dict(p: TreasuryPayment) -> dict: + return {"id": p.id, "entity_id": p.entity_id, "pay_no": p.pay_no, + "from_account_id": p.from_account_id, "to_name": p.to_name, + "to_account_no": p.to_account_no, "amount": p.amount, "currency": p.currency, + "purpose": p.purpose, "pay_type": p.pay_type, + "pay_date": p.pay_date.isoformat() if p.pay_date else None, + "budget_plan_id": p.budget_plan_id, "kpi_id": p.kpi_id, + "related_plan_id": p.related_plan_id, "status": p.status, + "check_result": p.check_result, "check_detail": p.check_detail, + "required_approvals": p.required_approvals, "day_cum_amount": p.day_cum_amount, + "approvals": p.approvals or [], "dual_sign_effective": p.dual_sign_effective, + "is_waived": p.is_waived, "submitter": p.submitter, "created_by": p.created_by, + "submitted_at": p.submitted_at, "approved_at": p.approved_at, + "paid_at": p.paid_at, "created_at": p.created_at} + + +def _gen_pay_no(db, entity_id) -> str: + today = datetime.now().strftime("%Y%m%d") + n = db.query(func.count(TreasuryPayment.id)).filter( + TreasuryPayment.entity_id == entity_id, + TreasuryPayment.pay_no.like(f"TP{today}%")).scalar() or 0 + return f"TP{today}{int(n) + 1:04d}" + + +def _get_payment(db, entity_id, pid) -> TreasuryPayment: + p = db.query(TreasuryPayment).filter(TreasuryPayment.id == pid, + TreasuryPayment.entity_id == entity_id).first() + if not p: + raise HTTPException(404, f"支付单 {pid} 不存在或不属于当前账套") + return p + + +@router.get("/payments") +def list_payments(status: Optional[str] = None, check_result: Optional[str] = None, + db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + q = db.query(TreasuryPayment).filter(TreasuryPayment.entity_id == entity_id) + if status: + q = q.filter(TreasuryPayment.status == status) + if check_result: + q = q.filter(TreasuryPayment.check_result == check_result) + rows = q.order_by(TreasuryPayment.id.desc()).limit(500).all() + return {"total": len(rows), "items": [_payment_dict(p) for p in rows]} + + +@router.post("/payments") +def create_payment(payload: PaymentIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_auth)): + acc = _pick_account(db, entity_id, payload.from_account_id) # 同实体校验 + submitter = payload.submitter or (current_user.name or current_user.username) + if payload.related_plan_id: + plan = db.query(CashPlan).filter(CashPlan.id == payload.related_plan_id, + CashPlan.entity_id == entity_id).first() + if not plan: + raise HTTPException(400, "related_plan_id 不在当前账套（cash_plans 只读复用，禁跨实体）") + if payload.budget_plan_id: + bp = db.query(BudgetPlan).filter(BudgetPlan.id == payload.budget_plan_id, + BudgetPlan.entity_id == entity_id).first() + if not bp: + raise HTTPException(400, "budget_plan_id 不在当前账套") + p = TreasuryPayment( + entity_id=entity_id, + pay_no=payload.pay_no or _gen_pay_no(db, entity_id), + from_account_id=acc.id, + to_name=payload.to_name, + to_account_no=payload.to_account_no, + amount=_r(payload.amount), + currency=payload.currency, + purpose=payload.purpose, + pay_type=payload.pay_type, + pay_date=date.fromisoformat(payload.pay_date) if payload.pay_date else date.today(), + budget_plan_id=payload.budget_plan_id, + kpi_id=payload.kpi_id, + related_plan_id=payload.related_plan_id, + status="draft", + submitter=submitter, + created_by=current_user.username if hasattr(current_user, "username") else submitter, + ) + db.add(p) + db.commit() + db.refresh(p) + return _payment_dict(p) + + +@router.put("/payments/{pid}") +def update_payment(pid: int, payload: PaymentIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_auth)): + """草稿/预检态支付单修改（已进入审批链后不可改，需作废重建）。""" + p = _get_payment(db, entity_id, pid) + if p.status not in ("draft", "precheck"): + raise HTTPException(409, f"支付单状态 {p.status} 不可修改（审批链已开始）") + acc = _pick_account(db, entity_id, payload.from_account_id) + if payload.budget_plan_id: + bp = db.query(BudgetPlan).filter(BudgetPlan.id == payload.budget_plan_id, + BudgetPlan.entity_id == entity_id).first() + if not bp: + raise HTTPException(400, "budget_plan_id 不在当前账套") + if payload.related_plan_id: + plan = db.query(CashPlan).filter(CashPlan.id == payload.related_plan_id, + CashPlan.entity_id == entity_id).first() + if not plan: + raise HTTPException(400, "related_plan_id 不在当前账套") + p.from_account_id = acc.id + p.to_name = payload.to_name + p.to_account_no = payload.to_account_no + p.amount = _r(payload.amount) + p.currency = payload.currency + p.purpose = payload.purpose + p.pay_type = payload.pay_type + p.pay_date = date.fromisoformat(payload.pay_date) if payload.pay_date else p.pay_date + p.budget_plan_id = payload.budget_plan_id + p.kpi_id = payload.kpi_id + p.related_plan_id = payload.related_plan_id + if payload.submitter: + p.submitter = payload.submitter + db.commit() + db.refresh(p) + return _payment_dict(p) + + +@router.post("/payments/{pid}/precheck") +def precheck_payment(pid: int, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_auth)): + """预检（规则引擎）：命中拦截 → check_result=blocked（可走豁免）；否则转 pending_approval。""" + p = _get_payment(db, entity_id, pid) + if p.status in ("approved", "paid"): + raise HTTPException(409, f"支付单已 {p.status}，不可重复预检") + res = eval_payment(db, entity_id, p, persist=True) + p.check_result = res["check_result"] + p.check_detail = res["check_detail"] + p.required_approvals = res["required_approvals"] + p.day_cum_amount = res["day_cum_amount"] + p.submitted_at = p.submitted_at or _now() + if res["check_result"] == "blocked": + p.status = "precheck" # 被拦：不得进入 pending_approval + else: + p.status = "pending_approval" + db.commit() + db.refresh(p) + return _payment_dict(p) + + +@router.post("/payments/{pid}/approve") +def approve_payment(pid: int, payload: ApproveIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "finance"))): + """审批（双签校验）：同一账号重复审批无效；审批人不得为提交人本人；BL-001 需不同自然人。""" + p = _get_payment(db, entity_id, pid) + if p.check_result == "blocked" and not p.is_waived: + raise HTTPException(409, "该支付单已被 block 规则拦截，请先走豁免通道（/waive）") + if p.status in ("paid", "cancelled", "rejected"): + raise HTTPException(409, f"支付单状态 {p.status} 不可审批") + approver = current_user.name if hasattr(current_user, "name") else current_user.username + account = current_user.username + if account == p.submitter or approver == p.submitter: + raise HTTPException(403, "审批人不得为提交人本人（口径4）") + entries = list(p.approvals or []) + if any(e.get("account") == account for e in entries): + raise HTTPException(400, "同一账号重复审批无效（须不同账号，口径4）") + if not payload.approve: + p.approvals = entries + p.status = "rejected" + db.commit() + db.refresh(p) + return _payment_dict(p) + entries.append({"account": account, "role": current_user.role, "person_name": approver, + "at": _now().isoformat(timespec="seconds"), "comment": payload.comment}) + p.approvals = entries + + need = int(p.required_approvals or 0) + dual = eval_dual_sign_at_approve(db, entity_id, p) + p.dual_sign_effective = _dual_sign_effective(p) + if need >= 2: + persons = {e.get("person_name") for e in entries} + if len(persons) < 2: + p.status = "pending_approval" # 未达双签：不得置 approved + db.add(TreasuryControlHit( + entity_id=entity_id, target_type="payment", target_id=p.id, rule_code="BL-001", + verdict="blocked", + detail=_hit_detail("BL-001", "v1", f"required_approvals={need}", need, + len(persons), + "审批数不足：" + dual["basis"]))) + db.commit() + db.refresh(p) + return {"payment": _payment_dict(p), "approved": False, + "reason": "required_approvals=%d 未满足：需 %d 个不同自然人的审批" % (need, need), + "dual_sign_basis": dual["basis"]} + if dual["verdict"] == "blocked": + db.add(TreasuryControlHit( + entity_id=entity_id, target_type="payment", target_id=p.id, rule_code="BL-001", + verdict="blocked", + detail=_hit_detail("BL-001", "v1", "approve", need, len(entries), dual["basis"]))) + db.commit() + db.refresh(p) + return {"payment": _payment_dict(p), "approved": False, "reason": dual["basis"]} + p.status = "approved" + p.approved_at = _now() + db.add(TreasuryControlHit( + entity_id=entity_id, target_type="payment", target_id=p.id, rule_code="BL-001", + verdict="pass", + detail=_hit_detail("BL-001", "v1", "approve", need, len(entries), dual["basis"]))) + db.commit() + db.refresh(p) + return {"payment": _payment_dict(p), "approved": True, "reason": dual["basis"]} + + +@router.post("/payments/{pid}/pay") +def pay_payment(pid: int, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "finance"))): + p = _get_payment(db, entity_id, pid) + if p.status != "approved": + raise HTTPException(409, f"支付单状态 {p.status}，未放行不得登记支付") + p.status = "paid" + p.paid_at = _now() + db.commit() + db.refresh(p) + return _payment_dict(p) + + +@router.post("/payments/{pid}/waive") +def waive_payment(pid: int, payload: WaiveIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "finance", "it"))): + """豁免通道：必填审批人 + 事由；豁免级别不得低于该笔所需审批级数（≥B 需 ceo）。全量留痕。""" + p = _get_payment(db, entity_id, pid) + if p.check_result != "blocked": + raise HTTPException(409, "仅被 block 规则拦截的支付单可走豁免通道") + need = 1 if int(p.required_approvals or 0) < 2 else 2 + wu = db.query(User).filter(or_(User.name == payload.waived_by, + User.username == payload.waived_by)).first() + if not wu: + raise HTTPException(400, "豁免审批人不存在（须为系统用户姓名或账号）") + if need >= 2 and wu.role != "ceo": + raise HTTPException(403, f"本单所需审批级数 {need}，豁免须 ceo 级审批人（口径4：豁免级别不得低于该笔所需）") + p.is_waived = 1 + entries = list(p.approvals or []) + entries.append({"account": wu.username, "role": wu.role, "person_name": wu.name, + "at": _now().isoformat(timespec="seconds"), + "comment": f"豁免：{payload.waived_reason}"}) + p.approvals = entries + p.dual_sign_effective = _dual_sign_effective(p) + p.status = "approved" + p.approved_at = _now() + db.add(TreasuryControlHit( + entity_id=entity_id, target_type="payment", target_id=p.id, rule_code="WAIVE", + verdict="waived", + detail=_hit_detail("WAIVE", "v1", f"required_approvals={need}", need, None, + f"豁免放行：事由={payload.waived_reason}"), + waived_by=payload.waived_by, waived_reason=payload.waived_reason)) + db.commit() + db.refresh(p) + return _payment_dict(p) + + +# ───────────────────────────────────────────────────────────── +# E. 命中留痕 / 试算 / 总览 +# ───────────────────────────────────────────────────────────── + +@router.get("/control-hits") +def list_hits(rule_code: Optional[str] = None, verdict: Optional[str] = None, + target_id: Optional[int] = None, limit: int = Query(200, le=1000), + db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + q = db.query(TreasuryControlHit).filter(TreasuryControlHit.entity_id == entity_id) + if rule_code: + q = q.filter(TreasuryControlHit.rule_code == rule_code) + if verdict: + q = q.filter(TreasuryControlHit.verdict == verdict) + if target_id: + q = q.filter(TreasuryControlHit.target_id == target_id) + rows = q.order_by(TreasuryControlHit.id.desc()).limit(limit).all() + return {"total": len(rows), "items": [{ + "id": h.id, "entity_id": h.entity_id, "target_type": h.target_type, + "target_id": h.target_id, "rule_code": h.rule_code, "verdict": h.verdict, + "detail": h.detail, "waived_by": h.waived_by, "waived_reason": h.waived_reason, + "created_at": h.created_at} for h in rows]} + + +@router.post("/control-rules/test") +def test_rules(payload: PaymentIn, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "finance", "it"))): + """规则试算（不落库、不写留痕）：给一笔模拟支付单，返回引擎判定。""" + submitter = payload.submitter or (current_user.name or current_user.username) + p = TreasuryPayment( + entity_id=entity_id, pay_no=payload.pay_no or "SIMULATED", + from_account_id=payload.from_account_id, to_name=payload.to_name, + to_account_no=payload.to_account_no, amount=_r(payload.amount), + currency=payload.currency, purpose=payload.purpose, pay_type=payload.pay_type, + pay_date=date.fromisoformat(payload.pay_date) if payload.pay_date else date.today(), + budget_plan_id=payload.budget_plan_id, kpi_id=payload.kpi_id, + related_plan_id=payload.related_plan_id, submitter=submitter, created_by=submitter, + id=0, status="simulated", + ) + res = eval_payment(db, entity_id, p, persist=False) + db.rollback() + return {"simulated": True, **res} + + +@router.get("/dashboard") +def treasury_dashboard(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + """资金总览：账户池 + 头寸 + 规则命中 + 待批支付单。""" + accounts = db.query(TreasuryAccount).filter(TreasuryAccount.entity_id == entity_id).all() + balances = db.query(TreasuryBalance).filter(TreasuryBalance.entity_id == entity_id).all() + latest: Dict[int, TreasuryBalance] = {} + for b in balances: + cur = latest.get(b.account_id) + if cur is None or b.period > cur.period: + latest[b.account_id] = b + tier_stat: Dict[str, int] = {} + type_stat: Dict[str, int] = {} + total = 0.0 + for a in accounts: + tier_stat[a.tier] = tier_stat.get(a.tier, 0) + 1 + type_stat[a.account_type] = type_stat.get(a.account_type, 0) + 1 + b = latest.get(a.id) + total += float(b.balance) if b else float(a.open_balance or 0) + payments = db.query(TreasuryPayment).filter(TreasuryPayment.entity_id == entity_id) + pending = payments.filter(TreasuryPayment.status == "pending_approval").count() + blocked = payments.filter(TreasuryPayment.check_result == "blocked", + TreasuryPayment.is_waived == 0).count() + hits = db.query(TreasuryControlHit).filter(TreasuryControlHit.entity_id == entity_id) + return { + "entity_id": entity_id, + "accounts": {"total": len(accounts), + "by_tier": tier_stat, "by_type": type_stat, + "active": sum(1 for a in accounts if a.status == "active")}, + "balance_total": _r(total), "unit": "万元", + "positions": {"rows": len(balances), + "verified_accounts": sum(1 for b in latest.values() if b.data_status == "verified"), + "estimated_accounts": sum(1 for b in latest.values() if b.data_status == "estimated"), + "accounts_without_position": len(accounts) - len(latest)}, + "payments": {"total": payments.count(), "pending_approval": pending, "blocked": blocked}, + "hits": {"total": hits.count(), + "blocked": hits.filter(TreasuryControlHit.verdict == "blocked").count(), + "warn": hits.filter(TreasuryControlHit.verdict == "warn").count(), + "waived": hits.filter(TreasuryControlHit.verdict == "waived").count()}, + } diff --git a/backend/app/main.py b/backend/app/main.py index 26b36f1f..64f5a75c 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -7,7 +7,7 @@ from fastapi.responses import JSONResponse from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from dotenv import load_dotenv from app.database import init_db, get_db -from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json, ext_input_ledgers +from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json, ext_input_ledgers, treasury from app.utils.cache import clear_all as clear_cache, delete as delete_cache from scripts.erp_sync import run_sync as run_erp_sync from app.auth_middleware import require_auth, require_auth_401 @@ -99,6 +99,7 @@ app.include_router(data_classification.router) app.include_router(tax_compliance.router) app.include_router(verify.router) app.include_router(ext_input_ledgers.router) # 外部输入台账（M2 准入3 派生，2026-09-15） +app.include_router(treasury.router) # 司库（资金）阶段一：账户池/头寸/内控闸门（2026-09-15） @app.exception_handler(Exception) async def global_exception_handler(request: Request, exc: Exception): diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 889dcdf0..fb2aa569 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -8,6 +8,10 @@ from app.models.knowledge import KnowledgeEvent, KnowledgeSummary from app.models.driver_budget import DriverFactorTemplate, DriverFactorBudget from app.models.product_sales import ProductSales from app.models.ext_input_ledger import TaxFilingLog, HrMonthlyLedger, LicenceLedger +from app.models.treasury import ( + TreasuryAccount, TreasuryBalance, TreasuryControlRule, + TreasuryPayment, TreasuryControlHit, TreasuryLimit, +) class Entity(Base): diff --git a/backend/app/models/treasury.py b/backend/app/models/treasury.py new file mode 100644 index 00000000..538b3353 --- /dev/null +++ b/backend/app/models/treasury.py @@ -0,0 +1,178 @@ +"""司库（资金）体系 · 阶段一 ORM 模型 — 管理会计OS + +对齐 DDL: backend/scripts/treasury_ddl.sql（md5 见阶段1回报） +G1 裁定：**不建物理外键**（entity_id 亦不建）——软链 + 应用层校验，覆盖说明见 DDL §F-2。 +金额：FLOAT（万元），写库前 round(x,4)，比较用容差 1e-4（禁用浮点等值比较）。 +多租户：6 张表全部 entity_id NOT NULL（铁律）。 +""" +from sqlalchemy import ( + Column, Integer, String, Float, DateTime, Date, JSON, Text, Index, + UniqueConstraint, func, +) +from app.database import Base + + +class TreasuryAccount(Base): + """司库·资金账户主数据（统一管理：台账唯一入口）""" + __tablename__ = "treasury_accounts" + __table_args__ = ( + UniqueConstraint("entity_id", "account_no", name="uq_tacct_entity_no"), + Index("ix_tacct_entity", "entity_id", "status"), + Index("ix_tacct_subject", "entity_id", "subject_code"), + {"comment": "司库·资金账户主数据"}, + ) + + id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, nullable=False, comment="企业ID（多租户铁律，必带）") + account_no = Column(String(64), nullable=False, comment="账号/虚拟账户号（掩码）") + account_name = Column(String(100), nullable=False, comment="账户名") + account_type = Column(String(20), nullable=False, comment="cash/bank/e_wallet/other") + bank_name = Column(String(100), nullable=True) + subject_code = Column(String(20), nullable=True, comment="对应科目 1001xx/1002xx") + currency = Column(String(10), default="CNY") + tier = Column(String(20), nullable=False, comment="hq/platform/grassroots（按账户性质判定）") + tier_basis = Column(String(20), default="nature", comment="[D-1] tier 判定依据 nature/level/manual") + owner_node_id = Column(Integer, nullable=True, comment="使用/保管部门标注 org_nodes.id（软链，不派生 tier）") + open_balance = Column(Float, nullable=True, comment="期初余额(万元)；未落定留 NULL") + open_period = Column(String(20), nullable=True, comment="期初期间 YYYY-MM") + is_pool_member = Column(Integer, default=0, comment="阶段三资金池用") + status = Column(String(20), default="active", comment="active/inactive/closed") + remark = Column(String(500), nullable=True) + created_at = Column(DateTime, server_default=func.now()) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + + +class TreasuryBalance(Base): + """司库·账户头寸（日/期末）——余额 = 期初 + 流水；禁把流水净额当余额""" + __tablename__ = "treasury_balances" + __table_args__ = ( + UniqueConstraint("account_id", "period", name="uq_tbal_acct_period"), + Index("ix_tbal_entity_period", "entity_id", "period"), + Index("ix_tbal_entity_status", "entity_id", "data_status"), + {"comment": "司库·账户头寸（日/期末）"}, + ) + + id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, nullable=False) + account_id = Column(Integer, nullable=False, comment="treasury_accounts.id（软链）") + period = Column(String(20), nullable=False, comment="YYYY-MM-DD 日 / YYYY-MM 月末") + balance = Column(Float, nullable=False, comment="账面余额(万元)") + available = Column(Float, nullable=True, comment="可用余额(万元)（NULL=未采集）") + frozen = Column(Float, default=0, comment="冻结/在途(万元)") + source = Column(String(30), nullable=False, comment="ledger/import/manual/pool") + # [P2-6] fail-safe：默认 estimated，verified 必须显式写入 + data_status = Column(String(20), default="estimated", + comment="verified/estimated；默认 estimated（fail-safe），estimated 禁 block") + batch = Column(String(100), nullable=True, comment="导入批次指纹（幂等）") + created_at = Column(DateTime, server_default=func.now()) + + +class TreasuryControlRule(Base): + """司库·资金内控规则库（P0 核心）""" + __tablename__ = "treasury_control_rules" + __table_args__ = ( + UniqueConstraint("entity_id", "rule_code", name="uq_tcr_entity_code"), + {"comment": "司库·资金内控规则库"}, + ) + + id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, nullable=False) + rule_code = Column(String(20), nullable=False, comment="NP-001/AU-001/BL-001/LQ-001/...") + rule_name = Column(String(100), nullable=False) + rule_type = Column(String(30), nullable=False, comment="budget_required/authorization/dual_sign/liquidity/...") + severity = Column(String(10), nullable=False, comment="block/warn") + params = Column(JSON, nullable=True, comment="阈值参数（金额段/下限，不硬编码）") + rule_version = Column(String(20), default="v1", comment="[调整2] 规则版本（写入 hits.detail）") + enabled = Column(Integer, default=1) + remark = Column(String(500), nullable=True) + created_at = Column(DateTime, server_default=func.now()) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + + +class TreasuryPayment(Base): + """司库·支付单（资金内控闸门载体）""" + __tablename__ = "treasury_payments" + __table_args__ = ( + UniqueConstraint("entity_id", "pay_no", name="uq_tpay_entity_no"), + Index("ix_tpay_entity_status", "entity_id", "status"), + Index("ix_tpay_daysplit", "entity_id", "pay_date", "to_name", "submitter"), + Index("ix_tpay_acct", "entity_id", "from_account_id", "status"), + {"comment": "司库·支付单（资金内控闸门）"}, + ) + + id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, nullable=False) + pay_no = Column(String(50), nullable=False, comment="支付单号") + from_account_id = Column(Integer, nullable=False, comment="付款账户（必须已登记）") + to_name = Column(String(200), nullable=False, comment="收款方") + to_account_no = Column(String(64), nullable=True) + amount = Column(Float, nullable=False, comment="金额(万元)") + currency = Column(String(10), default="CNY") + purpose = Column(String(500), nullable=False, comment="用途（必填）") + pay_type = Column(String(30), nullable=True, comment="货款/费用/税费/薪酬/融资/内部划转") + pay_date = Column(Date, nullable=True, comment="业务付款日（拆单累计按此自然日归集）") + budget_plan_id = Column(Integer, nullable=True, comment="关联 budget_plans.id（NP-001 首选判据）") + kpi_id = Column(Integer, nullable=True) + related_plan_id = Column(Integer, nullable=True, comment="关联 cash_plans.id（只读复用）") + status = Column(String(20), default="draft", + comment="draft/precheck/pending_approval/approved/rejected/paid/cancelled") + check_result = Column(String(20), nullable=True, comment="pass/blocked/warn") + check_detail = Column(JSON, nullable=True, comment="[{rule_code,verdict,detail,matched_scope}]") + required_approvals = Column(Integer, nullable=True, comment="[D-2] 本单所需审批级数（取最严档）") + day_cum_amount = Column(Float, nullable=True, comment="[D-3] 同提交人×同收款方×同自然日累计(万元，含本单)") + approvals = Column(JSON, nullable=True, comment="[{account,role,person_name,at,comment}]") + dual_sign_effective = Column(Integer, nullable=True, comment="BL-001：0=形式双签（放行无效，标红）") + is_waived = Column(Integer, default=0, comment="[D-4] 是否经豁免通道放行") + submitter = Column(String(100), nullable=False, comment="[D-8] 提交人（拆单累计判据）") + submitted_at = Column(DateTime, nullable=True, comment="[D-5] 提交时间") + approved_at = Column(DateTime, nullable=True, comment="[D-5] 放行时间") + paid_at = Column(DateTime, nullable=True) + created_by = Column(String(100), nullable=False, comment="[D-8] 建单人（首版与 submitter 同值）") + created_at = Column(DateTime, server_default=func.now()) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + + +class TreasuryControlHit(Base): + """司库·内控规则命中留痕（append-only 审计）""" + __tablename__ = "treasury_control_hits" + __table_args__ = ( + Index("ix_tch_target", "target_type", "target_id"), + Index("ix_tch_entity_rule", "entity_id", "rule_code", "created_at"), + Index("ix_tch_waived", "entity_id", "verdict", "created_at"), + {"comment": "司库·内控规则命中留痕（append-only 审计）"}, + ) + + id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, nullable=False) + target_type = Column(String(20), nullable=False, comment="payment/plan/internal_transfer") + target_id = Column(Integer, nullable=False) + rule_code = Column(String(20), nullable=False) + verdict = Column(String(20), nullable=False, comment="pass/blocked/waived/warn") + detail = Column(JSON, nullable=True, + comment="[调整2] {rule_code,rule_version,matched_scope,threshold,cumulative,verdict_basis}") + waived_by = Column(String(100), nullable=True, comment="豁免审批人（级别不得低于本单所需）") + waived_reason = Column(String(500), nullable=True) + created_at = Column(DateTime, server_default=func.now()) + + +class TreasuryLimit(Base): + """司库·分级授权矩阵（角色 × 金额段 × 审批级数；金额段左闭右开）""" + __tablename__ = "treasury_limits" + __table_args__ = ( + UniqueConstraint("entity_id", "scope_type", "scope_value", "amount_from", name="uq_tlim_scope"), + Index("ix_tlim_entity", "entity_id", "status"), + {"comment": "司库·分级授权矩阵"}, + ) + + id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, nullable=False) + scope_type = Column(String(20), nullable=False, comment="role/account/tier") + scope_value = Column(String(50), nullable=False, comment="finance/ceo/.../account_id/tier") + amount_from = Column(Float, nullable=False, comment="段下限(万元,含)") + amount_to = Column(Float, nullable=True, comment="段上限(万元,不含)，NULL=无上限") + required_approvals = Column(Integer, default=1, comment="所需审批级数（≥2=双签）") + approver_roles = Column(JSON, nullable=False, comment='["ceo","finance"]（先复核后批准）') + require_budget = Column(Integer, default=1) + daily_cum_limit = Column(Float, nullable=True, comment="自主档日累计上限（=3×A）") + status = Column(String(20), default="active") + created_at = Column(DateTime, server_default=func.now(), comment="[D-9] 审计必需") diff --git a/backend/scripts/treasury_ddl.sql b/backend/scripts/treasury_ddl.sql new file mode 100644 index 00000000..72457c12 --- /dev/null +++ b/backend/scripts/treasury_ddl.sql @@ -0,0 +1,324 @@ +-- ============================================================================ +-- CMA 司库（资金）体系 · 阶段一 MVP —— DDL（阶段 1 落盘版） +-- ============================================================================ +-- 状态 : 落地版（2026-09-15 阶段 1）。G1 = 放行（有条件），见 +-- wecom-project/pending-tasks/_verification/fullstack-202609151735-cma-treasury-g1-review.md +-- 评审稿出处 : wecom-fullstack/pending-tasks/_reports/treasury-stage1-ddl-draft-20260915.sql +-- （md5 3aa4d944691b49cfda9b1f08f6bea549 —— 归档件，不回改） +-- 本版修正 : P2-6 data_status 默认 fail-safe / D-8 submitter 语义 / 调整2 detail 结构化 +-- / P2-8 金额纪律注释 / 无物理外键的应用层覆盖声明（§F） +-- 依据 : PRD v1.1 §3.2/§3.3（md5 cc9ca839b7fd6f633f4066ba385819be） +-- 财务口径签认 v1（md5 74b43cfdf25abf5b7654eba54e82d099） +-- 财务侧前置交付 F1/F2/F3（md5 a49ae4340c05dfa20aba7daad25cc25f） +-- 目标环境 : MySQL 8.0（InnoDB），charset/collation 以只读探针实测为准 +-- 范围 : 6 张新表 + cash_plans 加 4 列（向后兼容，只加列不改列） +-- 阶段三的 4 张表（pools/pool_members/credits/internal_transfers）不建 +-- 幂等契约 : 同一脚本重复执行结果不变（见 §E） +-- 命名约定 : 约束名与 PRD §3.2 字面一致（唯一键 uq_* / 普通索引 ix_*） +-- 执行方式 : 仅由 backend/scripts/treasury_migrate.py --apply 执行（本文件不含裸 ALTER 执行逻辑） +-- ============================================================================ + + +-- ============================================================================ +-- §A. 建表（6 张，CREATE TABLE IF NOT EXISTS 天然幂等） +-- 应用层软链（不建物理外键，G1 §3-2 裁定）覆盖声明见 §F +-- ============================================================================ + +-- A1. 资金账户主数据（统一管理：台账唯一入口） +-- tier 口径（财务签认 口径1 / PRD M1）：按【账户性质】判定，禁止按 org_nodes.level 派生 +-- hq : 法人银行户 / 收款通道（1002xx、微信、支付宝、友赞商户） +-- grassroots: 现金 / 备用金 / 备用卡 / 中间账户（100101/100104/100105/100106/100107） +-- platform : 分公司独立法人账户（当前两实体均无，禁止造虚层级） +CREATE TABLE IF NOT EXISTS treasury_accounts ( + id INT AUTO_INCREMENT PRIMARY KEY, + entity_id INT NOT NULL COMMENT '企业ID（多租户铁律，必带）', + account_no VARCHAR(64) NOT NULL COMMENT '账号/虚拟账户号（掩码）', + account_name VARCHAR(100) NOT NULL COMMENT '账户名（如 建设银行（货款账户））', + account_type VARCHAR(20) NOT NULL COMMENT 'cash现金/bank银行/e_wallet微信支付宝/other', + bank_name VARCHAR(100) NULL, + subject_code VARCHAR(20) NULL COMMENT '对应科目（1001xx/1002xx）— 流水/头寸归集用', + currency VARCHAR(10) DEFAULT 'CNY', + tier VARCHAR(20) NOT NULL COMMENT 'hq总部/platform平台/grassroots基层（按账户性质判定）', + tier_basis VARCHAR(20) DEFAULT 'nature' COMMENT '[D-1] tier 判定依据：nature账户性质/level派生/manual人工；口径1 要求 accounts 一律 nature', + owner_node_id INT NULL COMMENT '使用/保管部门标注 org_nodes.id（不参与 tier 派生、不作授权依据）— 软链，应用层校验见 §F', + open_balance FLOAT NULL COMMENT '期初余额(万元)；F2 到位前留 NULL。金额纪律见 §F-1', + open_period VARCHAR(20) NULL COMMENT '期初期间 YYYY-MM', + is_pool_member TINYINT DEFAULT 0 COMMENT '阶段三资金池用', + status VARCHAR(20) DEFAULT 'active' COMMENT 'active/inactive/closed', + remark VARCHAR(500) NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uq_tacct_entity_no (entity_id, account_no), + KEY ix_tacct_entity (entity_id, status), + KEY ix_tacct_subject (entity_id, subject_code) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='司库·资金账户主数据'; + + +-- A2. 账户头寸（日/期末） +-- 铁律：余额 = 期初 + 流水（期初缺失时 data_status='estimated'）； +-- 禁止把 voucher_details 的"流水净额累计"当余额（PRD §2.2-1 实证） +CREATE TABLE IF NOT EXISTS treasury_balances ( + id INT AUTO_INCREMENT PRIMARY KEY, + entity_id INT NOT NULL, + account_id INT NOT NULL COMMENT 'treasury_accounts.id（软链，应用层校验见 §F）', + period VARCHAR(20) NOT NULL COMMENT 'YYYY-MM-DD 日头寸 / YYYY-MM 月末头寸', + balance FLOAT NOT NULL COMMENT '账面余额(万元)。金额纪律见 §F-1', + available FLOAT NULL COMMENT '可用余额(万元)（NULL=未采集，勿以 balance 冒充）', + frozen FLOAT DEFAULT 0 COMMENT '冻结/在途(万元)', + source VARCHAR(30) NOT NULL COMMENT 'ledger期初+流水/import导入/manual手工/pool归集', + data_status VARCHAR(20) DEFAULT 'estimated' COMMENT 'verified/estimated —— [P2-6] 默认 estimated（fail-safe）：verified 必须显式写入（仅期初已落定/对账通过）；estimated 头寸禁止 block 规则生效', + batch VARCHAR(100) NULL COMMENT '导入批次指纹（幂等用，对齐 cash_vouchers_json._content_fingerprint 范式）', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + UNIQUE KEY uq_tbal_acct_period (account_id, period), + KEY ix_tbal_entity_period (entity_id, period), + KEY ix_tbal_entity_status (entity_id, data_status) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='司库·账户头寸（日/期末）'; + + +-- A3. 资金内控规则库（P0 核心） +-- params JSON 结构与口径映射： +-- NP-001 budget_required {"judge":"budget_plans","entity1":"block","entity2":"warn", +-- "occupancy":"paid+approved+pending_approval+this", +-- "unmatched":"specify_budget_flow"} +-- AU-001 authorization {"tiers":[{"from":0,"to":3,"required_approvals":0},...], +-- "daily_cum_limit":9,"max_strict":true,"bounds":"[from,to)"} +-- BL-001 dual_sign {"min_amount":30,"distinct_natural_person":true} +-- LQ-001 liquidity {"target":26,"warn":12,"block":6,"block_enabled":false, +-- "skip_when":"data_status=estimated"} +CREATE TABLE IF NOT EXISTS treasury_control_rules ( + id INT AUTO_INCREMENT PRIMARY KEY, + entity_id INT NOT NULL, + rule_code VARCHAR(20) NOT NULL COMMENT 'NP-001/AU-001/BL-001/LQ-001/LQ-002/TF-001/AC-001', + rule_name VARCHAR(100) NOT NULL, + rule_type VARCHAR(30) NOT NULL COMMENT 'budget_required/authorization/dual_sign/liquidity/two_funds/account_valid', + severity VARCHAR(10) NOT NULL COMMENT 'block拦截/warn预警', + params JSON NULL COMMENT '阈值参数（金额段/下限/比例等，不硬编码）', + rule_version VARCHAR(20) DEFAULT 'v1' COMMENT '[调整2] 规则版本，写入 control_hits.detail.rule_version（审计可自证）', + enabled TINYINT DEFAULT 1, + remark VARCHAR(500) NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uq_tcr_entity_code (entity_id, rule_code) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='司库·资金内控规则库'; + + +-- A4. 支付单（内控闸门载体） +-- 拆单累计穿透（口径2/M4）所需索引：ix_tpay_daysplit（同提交人×同收款方×同自然日累计） +-- [D-8] submitter / created_by 语义（G1 复算发现的未披露增补，追认采纳）： +-- submitter = 支付单提交人 —— AU-001 拆单累计穿透的"同一提交人"判据（业务语义字段） +-- created_by = 建单人 —— 历史兼容列（与既有表 created_by 命名一致） +-- 首版两者同值（提交即建单）；保留双列以便后续"审批链与建单分离"时语义不破坏 +CREATE TABLE IF NOT EXISTS treasury_payments ( + id INT AUTO_INCREMENT PRIMARY KEY, + entity_id INT NOT NULL, + pay_no VARCHAR(50) NOT NULL COMMENT '支付单号', + from_account_id INT NOT NULL COMMENT '付款账户（必须已登记，禁未入台账账户）——软链，应用层校验见 §F', + to_name VARCHAR(200) NOT NULL COMMENT '收款方', + to_account_no VARCHAR(64) NULL, + amount FLOAT NOT NULL COMMENT '金额(万元)。金额纪律见 §F-1', + currency VARCHAR(10) DEFAULT 'CNY', + purpose VARCHAR(500) NOT NULL COMMENT '用途（必填，内控留痕）', + pay_type VARCHAR(30) NULL COMMENT '货款/费用/税费/薪酬/融资/内部划转', + pay_date DATE NULL COMMENT '业务付款日（拆单累计穿透按此自然日归集；口径2）', + budget_plan_id INT NULL COMMENT '关联预算计划（NP-001 授权性判据首选）——软链', + kpi_id INT NULL, + related_plan_id INT NULL COMMENT '关联 cash_plans.id（不新建表，复用收付计划）——软链', + status VARCHAR(20) DEFAULT 'draft' COMMENT 'draft/precheck/pending_approval/approved/rejected/paid/cancelled', + check_result VARCHAR(20) NULL COMMENT 'pass/blocked/warn', + check_detail JSON NULL COMMENT '[{rule_code,verdict,detail,matched_scope}] 命中档位写在 matched_scope（口径2-③）', + required_approvals TINYINT NULL COMMENT '[D-2] 本单实际所需审批级数（取最严档结果，落库便于审计复现）', + day_cum_amount FLOAT NULL COMMENT '[D-3] 判定时 同提交人×同收款方×同自然日 累计金额(万元，含本单)', + approvals JSON NULL COMMENT '[{account,role,person_name,at,comment}] — 双签须记账号+角色+自然人姓名（口径4）', + dual_sign_effective TINYINT NULL COMMENT 'BL-001：两账号是否不同自然人；0=形式双签（放行无效，审计标红）', + is_waived TINYINT DEFAULT 0 COMMENT '[D-4] 是否经豁免通道放行（豁免人/事由落 treasury_control_hits）', + submitter VARCHAR(100) NOT NULL COMMENT '[D-8] 提交人（AU-001 拆单累计穿透"同一提交人"判据）', + submitted_at DATETIME NULL COMMENT '[D-5] 提交时间（口径2 日累计/占用口径需要审批链时间锚）', + approved_at DATETIME NULL COMMENT '[D-5] 放行时间', + paid_at DATETIME NULL, + created_by VARCHAR(100) NOT NULL COMMENT '[D-8] 建单人（历史兼容列；首版与 submitter 同值）', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uq_tpay_entity_no (entity_id, pay_no), + KEY ix_tpay_entity_status (entity_id, status), + KEY ix_tpay_daysplit (entity_id, pay_date, to_name, submitter), + KEY ix_tpay_acct (entity_id, from_account_id, status) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='司库·支付单（资金内控闸门）'; + + +-- A5. 规则命中留痕（审计，append-only，不可省） +-- [调整2 / D-6] detail 固定结构化字段（规则引擎按此写入，审计追问要能自证）： +-- {"rule_code":..,"rule_version":..,"matched_scope":..,"threshold":.., +-- "cumulative":..,"verdict_basis":".."} +CREATE TABLE IF NOT EXISTS treasury_control_hits ( + id INT AUTO_INCREMENT PRIMARY KEY, + entity_id INT NOT NULL, + target_type VARCHAR(20) NOT NULL COMMENT 'payment/plan/internal_transfer', + target_id INT NOT NULL, + rule_code VARCHAR(20) NOT NULL, + verdict VARCHAR(20) NOT NULL COMMENT 'pass/blocked/waived/warn', + detail JSON NULL COMMENT '[调整2] 结构化判据快照，固定字段 {rule_code, rule_version, matched_scope, threshold, cumulative, verdict_basis}', + waived_by VARCHAR(100) NULL COMMENT '豁免审批人（豁免级别不得低于该笔所需审批级别，口径4）', + waived_reason VARCHAR(500) NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + KEY ix_tch_target (target_type, target_id), + KEY ix_tch_entity_rule (entity_id, rule_code, created_at), + KEY ix_tch_waived (entity_id, verdict, created_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='司库·内控规则命中留痕（append-only 审计）'; + + +-- A6. 分级授权矩阵（不硬编码；A/B 数值生效前需任总过目） +-- 边界口径（口径2-④ 左闭右开）：amount_from 含、amount_to 不含；amount_to IS NULL = 无上限 +CREATE TABLE IF NOT EXISTS treasury_limits ( + id INT AUTO_INCREMENT PRIMARY KEY, + entity_id INT NOT NULL, + scope_type VARCHAR(20) NOT NULL COMMENT 'role/account/tier', + scope_value VARCHAR(50) NOT NULL COMMENT 'finance/ceo/…/account_id/tier', + amount_from FLOAT NOT NULL COMMENT '段下限(万元,含)。金额纪律见 §F-1', + amount_to FLOAT NULL COMMENT '段上限(万元,不含)，NULL=无上限', + required_approvals TINYINT DEFAULT 1 COMMENT '所需审批级数（≥2=双签）', + approver_roles JSON NOT NULL COMMENT '["ceo","finance"]（顺序=先复核后批准，口径4）', + require_budget TINYINT DEFAULT 1, + daily_cum_limit FLOAT NULL COMMENT '自主档日累计上限（=3×A；超限升一级审批，口径2-②）', + status VARCHAR(20) DEFAULT 'active', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '[D-9] 审计必需（PRD 该表无此列，G1 追认采纳）', + UNIQUE KEY uq_tlim_scope (entity_id, scope_type, scope_value, amount_from), + KEY ix_tlim_entity (entity_id, status) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='司库·分级授权矩阵'; + + +-- ============================================================================ +-- §B. 扩展既有表（只加列、不改列、默认 NULL —— 存量数据零影响） +-- 注意：MySQL 8.0 不支持 ADD COLUMN IF NOT EXISTS， +-- 裸 ALTER 重复执行会报 1060 Duplicate column name。 +-- 故 §B 由 migrate 脚本按 information_schema 预检结果条件执行（见 §E-②）。 +-- ============================================================================ + +-- B0. 幂等预检（migrate 脚本执行前必跑；输出缺失列清单，全 0 即跳过） +-- SELECT COLUMN_NAME FROM information_schema.COLUMNS +-- WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'cash_plans' +-- AND COLUMN_NAME IN ('account_id','tier','pool_id','is_locked'); + +ALTER TABLE cash_plans + ADD COLUMN account_id INT NULL COMMENT '资金来源/去向账户 treasury_accounts.id（软链，无物理外键）', + ADD COLUMN tier VARCHAR(20) NULL COMMENT 'hq/platform/grassroots（三位一体视角；可由 org_nodes.level 派生）', + ADD COLUMN pool_id INT NULL COMMENT '归属资金池（阶段三启用，本阶段恒 NULL）', + ADD COLUMN is_locked TINYINT DEFAULT 0 COMMENT '是否已被预算/内控锁定'; + +-- B1. 可选索引 [D-7]（G1 已随 D-1~D-7 全部采纳；与 §B 同批执行） +ALTER TABLE cash_plans ADD KEY ix_cashplans_entity_acct (entity_id, account_id); + + +-- ============================================================================ +-- §C. 数据种子（首版数值 = 财务建议值；任总拍板后只改数据行、不改代码） +-- 幂等：INSERT ... ON DUPLICATE KEY UPDATE（唯一键 = uq_tcr_entity_code / uq_tlim_scope） +-- 行数契约：规则 8 行（e1 4 + e2 4）+ 授权矩阵 6 行 +-- ============================================================================ + +-- C1. 三条 block 规则 + LQ-001（entity1 / entity2 各一套） +-- NP-001：entity1=block，entity2=warn（分期上线，过渡期 1 个月后切 block） +-- AU-001：tiers 采用签认首版 A/B（酣客 3/30 万；博海 5/50 万）—— 待任总过目 +-- BL-001：distinct_natural_person=true（两账号须属不同自然人） +-- LQ-001：首版一律 warn、block_enabled=false（阶段二接预警） +INSERT INTO treasury_control_rules (entity_id, rule_code, rule_name, rule_type, severity, params, rule_version, enabled) +VALUES + (1,'NP-001','无预算不开支（授权性判据）','budget_required','block', + JSON_OBJECT('judge','budget_plans','severity','block', + 'occupancy','paid+approved+pending_approval+this', + 'unmatched','specify_budget_flow','transition_until','2026-10-15'),'v1',1), + (1,'AU-001','超授权拦截（含拆单累计穿透+日累计升级）','authorization','block', + JSON_OBJECT('tiers',JSON_ARRAY(JSON_OBJECT('from',0,'to',3,'required_approvals',0), + JSON_OBJECT('from',3,'to',30,'required_approvals',1), + JSON_OBJECT('from',30,'to',NULL,'required_approvals',2)), + 'daily_cum_limit',9,'merge_same_payee_same_day',true,'max_strict',true,'bounds','[from,to)'),'v1',1), + (1,'BL-001','大额支付双签（须不同自然人）','dual_sign','block', + JSON_OBJECT('min_amount',30,'required_approvals',2,'distinct_natural_person',true, + 'record_fields',JSON_ARRAY('account','role','person_name'),'order','review_then_approve'),'v1',1), + (1,'LQ-001','流动性下限（首版 warn-only 页面并列对照）','liquidity','warn', + JSON_OBJECT('target',26,'warn',12,'block',6,'block_enabled',false, + 'skip_when','data_status=estimated'),'v1',1), + (2,'NP-001','无预算不开支（博海首版 warn+强制补录）','budget_required','warn', + JSON_OBJECT('judge','budget_plans','severity','warn','occupancy','paid+approved+pending_approval+this', + 'unmatched','specify_budget_flow','transition_until','2026-10-15'),'v1',1), + (2,'AU-001','超授权拦截（含拆单累计穿透+日累计升级）','authorization','block', + JSON_OBJECT('tiers',JSON_ARRAY(JSON_OBJECT('from',0,'to',5,'required_approvals',0), + JSON_OBJECT('from',5,'to',50,'required_approvals',1), + JSON_OBJECT('from',50,'to',NULL,'required_approvals',2)), + 'daily_cum_limit',15,'merge_same_payee_same_day',true,'max_strict',true,'bounds','[from,to)'),'v1',1), + (2,'BL-001','大额支付双签（须不同自然人）','dual_sign','block', + JSON_OBJECT('min_amount',50,'required_approvals',2,'distinct_natural_person',true, + 'record_fields',JSON_ARRAY('account','role','person_name'),'order','review_then_approve'),'v1',1), + (2,'LQ-001','流动性下限（博海首版默认 warn/关闭）','liquidity','warn', + JSON_OBJECT('target',50,'warn',10,'block',5,'block_enabled',false, + 'skip_when','data_status=estimated'),'v1',0) +ON DUPLICATE KEY UPDATE rule_name=VALUES(rule_name), rule_type=VALUES(rule_type), + severity=VALUES(severity), params=VALUES(params), rule_version=VALUES(rule_version); + +-- C2. 授权矩阵（scope_type='role'，按 entity 生效；双签档 approver_roles 顺序=先复核后批准） +INSERT INTO treasury_limits (entity_id, scope_type, scope_value, amount_from, amount_to, + required_approvals, approver_roles, require_budget, daily_cum_limit) +VALUES + (1,'role','finance',0, 3, 0, JSON_ARRAY(), 1, 9), + (1,'role','ceo', 3, 30, 1, JSON_ARRAY('ceo'), 1, NULL), + (1,'role','ceo', 30, NULL,2, JSON_ARRAY('finance','ceo'),1, NULL), + (2,'role','finance',0, 5, 0, JSON_ARRAY(), 1, 15), + (2,'role','ceo', 5, 50, 1, JSON_ARRAY('ceo'), 1, NULL), + (2,'role','ceo', 50, NULL,2, JSON_ARRAY('finance','ceo'),1, NULL) +ON DUPLICATE KEY UPDATE amount_to=VALUES(amount_to), required_approvals=VALUES(required_approvals), + approver_roles=VALUES(approver_roles), daily_cum_limit=VALUES(daily_cum_limit); + + +-- ============================================================================ +-- §D. 回滚 / DOWN（由 treasury_migrate.py --down 执行；顺序与 UP 相反） +-- ============================================================================ +-- D1. 回滚 cash_plans 扩展列：[P1-3] 按列存在性**逐列** DROP（部分态不再 1091） +-- 回滚前 [P1-4] 必先 mysqldump cash_plans（写路径账号 + 临时 defaults-file 注入口令， +-- 不硬编码、不回显；dump 失败 fail-closed —— 先 dump 成功才允许删列） +-- D2. 回滚新表（先删留痕/引用方，再删主数据）：DROP TABLE IF EXISTS 逐张 +-- D3. 回滚安全性：6 张新表首版为空表（DROP 无损）；一旦有业务数据， +-- 回滚须经人工确认（--yes），不提供 --force 免确认开关 +-- D4. 回滚不触碰 cash.py 既有表结构，11 端点契约不受影响 + + +-- ============================================================================ +-- §E. 幂等与验证（由 treasury_migrate.py 逐条实现） +-- ============================================================================ +-- E1. 幂等三件套 + 计数口径（[P0-2] 修正） +-- ① CREATE TABLE IF NOT EXISTS（表级幂等） +-- ② §B 列扩展按 information_schema 预检（列级幂等） +-- ③ §C 种子走 ON DUPLICATE KEY UPDATE（数据级幂等） +-- ⇒ 计数口径 = **新增对象数**（apply 前后 information_schema diff），不是"执行语句数"： +-- 首跑：created_new=6 / added_cols=4 / 种子 8 规则行 + 6 授权行 +-- 二跑：created_new=0 / added_cols=0 / 种子行数不变 +-- ④ up() 结尾断言 check()["missing_tables"] == []，否则非零退出（禁静默成功） +-- ⑤ 指纹行写入 import_logs(batch=treasury-ddl-)，[P2-7] entity_id=1 为约定： +-- 迁移指纹属实体无关元数据，import_logs.entity_id NOT NULL 约束下以 1 占位； +-- 幂等按 (import_type,batch) 先查后写，不依赖表上是否有唯一键 +-- E2. 表级自检（落库后） +-- SHOW CREATE TABLE treasury_accounts; -- 逐表核对 entity_id/唯一键/引擎字符集 +-- SELECT TABLE_NAME FROM information_schema.TABLES +-- WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME LIKE 'treasury%'; -- 期望 6 行 +-- SELECT TABLE_NAME, COLUMN_NAME, IS_NULLABLE FROM information_schema.COLUMNS +-- WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME LIKE 'treasury%' +-- AND COLUMN_NAME='entity_id'; -- 期望 6 行且全 NOT NULL +-- E3. 存量零影响证明 +-- SELECT COUNT(*) FROM cash_plans; -- 迁移前后一致（基线以探针实测为准） + + +-- ============================================================================ +-- §F. 应用层约束声明（G1 裁定"不建物理外键"的附条件覆盖说明 + 金额纪律） +-- ============================================================================ +-- F-1 金额纪律（[P2-8] / G1 §3-3 附条件）： +-- * 全部金额列为 FLOAT（万元）—— 与 cash_plans.amount / budget_plans.budget_value 同口径； +-- FLOAT 非精确货币类型（技术债登记：迁 DECIMAL(18,4) 须与 cash_plans/budget_plans 同批） +-- * 写库前一律 round(x, 4)；比较一律用容差 TOL=1e-4，禁止浮点等值比较（引擎实现见 treasury.py） +-- F-2 软链的应用层校验覆盖（逐表说明，G1 §3-2 附条件）： +-- ① treasury_balances.account_id → 写入/查询前校验 account 存在 且 account.entity_id == 传入 entity_id（防跨实体误挂） +-- ② treasury_payments.from_account_id → AC-001：必须命中 treasury_accounts 且 status='active'（账外账户拦截） +-- ③ treasury_payments.budget_plan_id → NP-001：命中 budget_plans 且 entity_id 一致；未命中走"指定/补录预算"流程 +-- ④ treasury_payments.related_plan_id → 校验 cash_plans 同 entity_id（只读复用，不改 cash.py 契约） +-- ⑤ treasury_payments.kpi_id → 校验 kpis 同 entity_id（NP-001 兜底匹配用） +-- ⑥ treasury_control_hits.target_id → target_type 限定取值域（payment/plan/internal_transfer）， +-- 写入前校验目标单据存在且同 entity_id +-- ⑦ treasury_accounts.owner_node_id → 校验 org_nodes 同 entity_id（不存在则留空，禁造虚层级） +-- * 唯一键 + 应用层校验 + AC-001 覆盖主要一致性风险；跨实体写入在应用层一律 403/400 拒绝 +-- ============================================================================ diff --git a/backend/scripts/treasury_migrate.py b/backend/scripts/treasury_migrate.py new file mode 100644 index 00000000..b9562046 --- /dev/null +++ b/backend/scripts/treasury_migrate.py @@ -0,0 +1,388 @@ +#!/usr/bin/env python3 +"""CMA 司库（资金）阶段一 —— 迁移脚本（阶段 1 落地版） + +落地路径：/root/cma-management/backend/scripts/treasury_migrate.py +评审稿 ：wecom-fullstack/pending-tasks/_reports/treasury-stage1-migration-draft-20260915.py（md5 ceaa3b3bf563b8d664ddd7224f01a5bf，归档不回改） +G1 修正项 ：P0-1 up() 语句过滤器 / P0-2 计数口径 / P1-3 down() 部分态 / P1-4 down() dump 凭据 / + P1-5 种子展开 / P2-6 data_status fail-safe（DDL 侧）/ P2-7 import_logs 指纹行 / P2-8 金额纪律 + +用法: + python3 treasury_migrate.py --parse-only # 零 DB：只解析 DDL，打印命中语句与期望计数（P0-1 复算证据） + python3 treasury_migrate.py --check # 只读预检：缺表/缺列/缺索引/种子行数（零写入，默认动作） + python3 treasury_migrate.py --apply # 建表 + 加列 + 索引 + 种子（幂等） + python3 treasury_migrate.py --down # 回滚（先 dump 后删；有数据时要求 --yes） + +幂等契约（P0-2 计数口径 = 新增对象数，不是执行语句数）: + 首跑 --apply 期望：created_new=6 / added_cols=4 / added_indexes=1 / rules=8 行 / limits=6 行 + 二跑 --apply 期望：created_new=0 / added_cols=0 / added_indexes=0 / 种子行数不变 + 指纹行：import_logs(batch=treasury-ddl-)，按 (import_type,batch) 先查后写， + 不依赖表上是否有唯一键（P2-7：entity_id=1 为约定——迁移指纹属实体无关元数据） +边界：只动 6 张 treasury_* 新表 + cash_plans 加列/加索引；不碰 cash.py / 既有端点契约 / 其他模块表结构。 +""" +import argparse +import hashlib +import json +import os +import subprocess +import sys +import tempfile +from datetime import datetime + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sqlalchemy import text # noqa: E402 +from app.database import get_engine # noqa: E402 + +HERE = os.path.dirname(os.path.abspath(__file__)) +BACKUP_DIR = os.path.join(HERE, "backups") +DDL_FILE = os.path.join(HERE, "treasury_ddl.sql") + +NEW_TABLES = [ + "treasury_accounts", + "treasury_balances", + "treasury_control_rules", + "treasury_payments", + "treasury_control_hits", + "treasury_limits", +] +CASH_PLAN_NEW_COLS = ["account_id", "tier", "pool_id", "is_locked"] +CASH_PLAN_NEW_INDEX = "ix_cashplans_entity_acct" +EXPECTED_RULE_ROWS = 8 # e1 4 + e2 4 +EXPECTED_LIMIT_ROWS = 6 # e1 3 + e2 3 +CASH_PLAN_ALTER = """ALTER TABLE cash_plans + ADD COLUMN account_id INT NULL COMMENT '资金来源/去向账户 treasury_accounts.id（软链，无物理外键）', + ADD COLUMN tier VARCHAR(20) NULL COMMENT 'hq/platform/grassroots（三位一体视角；可由 org_nodes.level 派生）', + ADD COLUMN pool_id INT NULL COMMENT '归属资金池（阶段三启用，本阶段恒 NULL）', + ADD COLUMN is_locked TINYINT DEFAULT 0 COMMENT '是否已被预算/内控锁定'""" + + +# ───────────────────────────────────────────────────────────── +# DDL 解析（P0-1：先剔除注释行再判定语句类型，不再用 startswith 裸判） +# ───────────────────────────────────────────────────────────── + +def _strip_comment_lines(sql_text: str) -> str: + """剔除整行注释（-- 开头）+ 行尾注释，避免注释行带动导致语句被跳过/误判。""" + out = [] + for raw in sql_text.splitlines(): + line = raw + idx = line.find("--") + if idx >= 0: + line = line[:idx] + if not line.strip(): + continue + out.append(line) + return "\n".join(out) + + +def _split_statements(sql_text: str) -> list: + return [s.strip() for s in _strip_comment_lines(sql_text).split(";") if s.strip()] + + +def _stmt_table_name(stmt: str, keyword: str): + """取 `KEYWORD TABLE [IF NOT EXISTS] ` 中的表名。""" + toks = stmt.replace("(", " ").replace("\n", " ").split() + up = [t.upper() for t in toks] + try: + i = up.index(keyword) + except ValueError: + return None + j = i + 1 + while j < len(up) and up[j] in ("IF", "NOT", "EXISTS"): + j += 1 + return toks[j].strip("`") if j < len(toks) else None + + +def parse_ddl(path=DDL_FILE): + """解析 DDL → {creates: [(table, stmt)], seeds: [stmt], alter: str|None, index: str|None}""" + sql = open(path, encoding="utf-8").read() + creates, seeds, alter, index_stmt = [], [], None, None + for stmt in _split_statements(sql): + head = stmt.upper() + if head.startswith("CREATE TABLE"): + creates.append((_stmt_table_name(stmt, "TABLE"), stmt)) + elif head.startswith("INSERT INTO"): + seeds.append(stmt) + elif head.startswith("ALTER TABLE CASH_PLANS ADD KEY"): + index_stmt = stmt + elif head.startswith("ALTER TABLE CASH_PLANS"): + alter = stmt + return {"creates": creates, "seeds": seeds, "alter": alter, "index": index_stmt, + "ddl_text": sql} + + +def assert_ddl_parseable(p): + """P0-1 断言：6 条 CREATE 全命中且表名与 NEW_TABLES 一致；种子 2 条；ALTER/B1 各 1 条。""" + names = [t for t, _ in p["creates"]] + errs = [] + if len(p["creates"]) != len(NEW_TABLES): + errs.append(f"CREATE TABLE 命中数 {len(p['creates'])} != {len(NEW_TABLES)}（{names}）") + if set(names) != set(NEW_TABLES): + errs.append(f"CREATE TABLE 表名集合不一致: {sorted(names)} vs {sorted(NEW_TABLES)}") + if len(p["seeds"]) != 2: + errs.append(f"种子 INSERT 语句数 {len(p['seeds'])} != 2") + if not p["alter"]: + errs.append("DDL §B 未解析到 ALTER TABLE cash_plans 加列语句") + if not p["index"]: + errs.append("DDL §B1 未解析到 ALTER TABLE cash_plans ADD KEY 语句") + if p["alter"] and p["alter"].strip() not in CASH_PLAN_ALTER.strip(): + errs.append("DDL §B ALTER 文本与脚本 CASH_PLAN_ALTER 常量漂移（指纹口径会不一致）") + return errs + + +def _fingerprint(texts): + canon = json.dumps(sorted(texts), ensure_ascii=False, separators=(",", ":")) + return hashlib.sha1(canon.encode("utf-8")).hexdigest()[:10] + + +# ───────────────────────────────────────────────────────────── +# information_schema 内省（只读） +# ───────────────────────────────────────────────────────────── + +def _introspect(conn): + rows = conn.execute(text( + "SELECT TABLE_NAME FROM information_schema.TABLES " + "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME LIKE 'treasury%'" + )).fetchall() + tables = {r[0] for r in rows} + cols = conn.execute(text( + "SELECT COLUMN_NAME FROM information_schema.COLUMNS " + "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='cash_plans'" + )).fetchall() + colset = {r[0] for r in cols} + idx = conn.execute(text( + "SELECT DISTINCT INDEX_NAME FROM information_schema.STATISTICS " + "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='cash_plans'" + )).fetchall() + return tables, colset, {r[0] for r in idx} + + +def _seed_counts(conn): + counts = {} + for tbl, key in (("treasury_control_rules", "rules_rows"), ("treasury_limits", "limits_rows")): + try: + counts[key] = conn.execute(text(f"SELECT COUNT(*) FROM {tbl}")).scalar() + except Exception: + counts[key] = None + return counts + + +def check(conn): + """只读预检 —— 缺表 / 缺列 / 缺索引 / 存量行数 / 种子行数；零写入。""" + tables, cols, idx = _introspect(conn) + cash_rows = conn.execute(text("SELECT COUNT(*) FROM cash_plans")).scalar() + report = { + "missing_tables": [t for t in NEW_TABLES if t not in tables], + "existing_tables": sorted(tables), + "missing_cash_plan_cols": [c for c in CASH_PLAN_NEW_COLS if c not in cols], + "missing_cash_plan_index": [] if CASH_PLAN_NEW_INDEX in idx else [CASH_PLAN_NEW_INDEX], + "cash_plans_rows": cash_rows, + } + report.update(_seed_counts(conn)) + print(json.dumps(report, ensure_ascii=False, indent=2)) + return report + + +# ───────────────────────────────────────────────────────────── +# UP / DOWN +# ───────────────────────────────────────────────────────────── + +def _write_fingerprint(conn, batch, tables_created): + """P2-7：指纹行按 (import_type,batch) 先查后写（不依赖唯一键；entity_id=1 为约定）。""" + cols = {r[0] for r in conn.execute(text( + "SELECT COLUMN_NAME FROM information_schema.COLUMNS " + "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='import_logs'" + )).fetchall()} + need = {"entity_id", "filename", "batch", "total_rows", "success_rows", "import_type"} + if not need <= cols: + raise RuntimeError(f"import_logs 结构不符，缺列 {sorted(need - cols)}（fail-closed，不静默跳过）") + row = conn.execute(text( + "SELECT id FROM import_logs WHERE import_type='treasury-ddl' AND batch=:b ORDER BY id LIMIT 1" + ), {"b": batch}).fetchone() + if row: + conn.execute(text( + "UPDATE import_logs SET total_rows=:t, success_rows=:s WHERE id=:i" + ), {"t": tables_created, "s": tables_created, "i": row[0]}) + return "updated" + conn.execute(text( + "INSERT INTO import_logs (entity_id, filename, batch, total_rows, success_rows, import_type, created_by) " + "VALUES (1, 'treasury_ddl.sql', :b, :t, :s, 'treasury-ddl', 'treasury_migrate')" + ), {"b": batch, "t": tables_created, "s": tables_created}) + return "inserted" + + +def up(conn): + p = parse_ddl() + errs = assert_ddl_parseable(p) + if errs: + for e in errs: + print("❌ DDL 解析失败:", e, file=sys.stderr) + return 3 + + before_tables, before_cols, before_idx = _introspect(conn) + + # ① 建表（6 条 CREATE，IF NOT EXISTS 幂等） + for _table, stmt in p["creates"]: + conn.execute(text(stmt)) + + # ② 条件加列（列级幂等） + missing_cols = [c for c in CASH_PLAN_NEW_COLS if c not in before_cols] + if missing_cols: + conn.execute(text(p["alter"])) + + # ③ 条件加索引（D-7） + if CASH_PLAN_NEW_INDEX not in before_idx: + conn.execute(text(p["index"])) + + # ④ 种子（ON DUPLICATE KEY UPDATE，数据级幂等） + for stmt in p["seeds"]: + conn.execute(text(stmt)) + + conn.commit() + + after_tables, after_cols, after_idx = _introspect(conn) + created_new = len(after_tables - before_tables) + added_cols = len(after_cols - before_cols) + added_indexes = len(after_idx - before_idx) + counts = _seed_counts(conn) + + # ⑤ 自检断言：缺表即非零退出（P0-1，禁静默成功） + post = check(conn) + if post["missing_tables"]: + print(f"❌ up() 后仍有缺表 {post['missing_tables']}（非零退出）", file=sys.stderr) + return 4 + + # ⑥ 指纹行 + batch = "treasury-ddl-" + _fingerprint([p["ddl_text"], CASH_PLAN_ALTER]) + action = _write_fingerprint(conn, batch, created_new or len(NEW_TABLES)) + conn.commit() + + print(json.dumps({ + "phase": "up", + "created_new": created_new, + "added_cols": added_cols, + "added_indexes": added_indexes, + "seeded_statements": len(p["seeds"]), + "rules_rows": counts["rules_rows"], + "limits_rows": counts["limits_rows"], + "import_logs_row": action, + "batch": batch, + "missing_tables": post["missing_tables"], + "cash_plans_rows": post["cash_plans_rows"], + }, ensure_ascii=False)) + return 0 + + +def _dump_cash_plans(out_path): + """P1-4：写路径账号 + 临时 defaults-file 注入口令（不硬编码、不回显）；失败即抛异常（fail-closed）。""" + from app import database as db + + fd, cnf = tempfile.mkstemp(prefix=".cma-dump-", suffix=".cnf") + try: + with os.fdopen(fd, "w") as f: + f.write("[client]\nuser=%s\npassword=%s\nhost=%s\nport=%s\n" + % (db.DB_USER, db.DB_PASS, db.DB_HOST, db.DB_PORT)) + os.chmod(cnf, 0o600) + with open(out_path, "w") as fh: + proc = subprocess.run( + ["mysqldump", f"--defaults-extra-file={cnf}", "--single-transaction", + "--skip-comments", db.DB_NAME, "cash_plans"], + stdout=fh, stderr=subprocess.PIPE, text=True) + if proc.returncode != 0: + raise RuntimeError("mysqldump 失败(rc=%s): %s" % (proc.returncode, (proc.stderr or "").strip()[:300])) + finally: + try: + os.remove(cnf) + except OSError: + pass + + +def down(conn): + """回滚：先 dump（fail-closed）→ 逐列删扩展列（P1-3）→ 删索引 → 删表（IF EXISTS）。""" + os.makedirs(BACKUP_DIR, exist_ok=True) + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + dump = os.path.join(BACKUP_DIR, f"cash_plans-before-rollback-{stamp}.sql") + try: + _dump_cash_plans(dump) + except Exception as exc: # fail-closed：dump 不成功就不允许删列 + print(f"❌ 备份失败，回滚中止（未做任何结构变更）: {exc}", file=sys.stderr) + return 5 + print("backup:", dump, os.path.getsize(dump), "bytes") + + tables, cols, idx = _introspect(conn) + dropped = [] + for col in CASH_PLAN_NEW_COLS: # P1-3：按列存在性逐列 DROP（部分态不再 1091） + if col in cols: + conn.execute(text(f"ALTER TABLE cash_plans DROP COLUMN {col}")) + dropped.append(col) + if CASH_PLAN_NEW_INDEX in idx: + conn.execute(text(f"ALTER TABLE cash_plans DROP INDEX {CASH_PLAN_NEW_INDEX}")) + dropped.append(CASH_PLAN_NEW_INDEX) + for t in reversed(NEW_TABLES): + conn.execute(text(f"DROP TABLE IF EXISTS {t}")) + conn.commit() + print(json.dumps({"phase": "down", "dropped_cash_plan_objects": dropped, + "dropped_tables": list(reversed(NEW_TABLES)), "backup": dump}, + ensure_ascii=False)) + return 0 + + +# ───────────────────────────────────────────────────────────── +# 入口 +# ───────────────────────────────────────────────────────────── + +def parse_only(): + """零 DB 的 DDL 解析复算（P0-1 证据）：命中 6 条 CREATE + 表名清单 + 种子语句。""" + p = parse_ddl() + errs = assert_ddl_parseable(p) + out = { + "ddl_file": DDL_FILE, + "ddl_md5": hashlib.md5(open(DDL_FILE, "rb").read()).hexdigest(), + "statements_total": len(_split_statements(p["ddl_text"])), + "create_table_hits": len(p["creates"]), + "create_table_names": [t for t, _ in p["creates"]], + "seed_statements": len(p["seeds"]), + "has_cash_plan_alter": bool(p["alter"]), + "has_cash_plan_index": bool(p["index"]), + "expected_first_run": {"created_new": 6, "added_cols": 4, "added_indexes": 1, + "rules_rows": EXPECTED_RULE_ROWS, "limits_rows": EXPECTED_LIMIT_ROWS}, + "expected_second_run": {"created_new": 0, "added_cols": 0, "added_indexes": 0, + "rules_rows": EXPECTED_RULE_ROWS, "limits_rows": EXPECTED_LIMIT_ROWS}, + "errors": errs, + } + print(json.dumps(out, ensure_ascii=False, indent=2)) + return 1 if errs else 0 + + +def main(): + ap = argparse.ArgumentParser() + g = ap.add_mutually_exclusive_group() + g.add_argument("--parse-only", action="store_true", help="零 DB：只解析 DDL（P0-1 复算证据）") + g.add_argument("--check", action="store_true", help="只读预检（默认）") + g.add_argument("--apply", action="store_true") + g.add_argument("--down", action="store_true") + ap.add_argument("--yes", action="store_true", help="回滚确认（表内有数据时必需）") + args = ap.parse_args() + + if args.parse_only: + return parse_only() + + engine = get_engine() + with engine.connect() as conn: + if args.apply: + return up(conn) + if args.down: + tables, _c, _i = _introspect(conn) + payload = 0 + if {"treasury_payments", "treasury_balances"} <= tables: + payload = conn.execute(text( + "SELECT (SELECT COUNT(*) FROM treasury_payments) + " + "(SELECT COUNT(*) FROM treasury_balances)")).scalar() + if payload and not args.yes: + print(f"REFUSE: 已有 {payload} 行业务数据，回滚需显式 --yes", file=sys.stderr) + return 2 + return down(conn) + return 0 if check(conn) is not None else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/scripts/treasury_seed_accounts.py b/backend/scripts/treasury_seed_accounts.py new file mode 100644 index 00000000..d6256f56 --- /dev/null +++ b/backend/scripts/treasury_seed_accounts.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +"""CMA 司库 · 阶段一 —— 账户池初始化（F1 两份 CSV 导入） + +数据源（财务侧前置交付 F1，md5 a49ae4340c05dfa20aba7daad25cc25f 的 F1F2F3 交付件附表）： + /root/.hermes/profiles/wecom-finance/reports/CMA_司库P0_F1_账户主数据清单_entity1.csv （12 行，酣客） + /root/.hermes/profiles/wecom-finance/reports/CMA_司库P0_F1_账户主数据清单_entity2.csv （12 行，博海） +表头 14 列与 PRD §3.2 treasury_accounts 逐列一致（tier_basis 不在 CSV → 走 DDL 默认 'nature'）。 + +口径（任务单步骤 6 + G1 §5 + F1 执行建议）： + * e1（entity_id=1 酣客）12 个货币资金科目；e2（entity_id=2 博海）12 个账户 + * `100107 现金备用卡（唐）` → status=inactive + * `owner_node_id` 首版留空（部门写在 remark；e2 无 org_nodes，禁造虚层级） + * **排除** `1002007 榆林酣客工行存款`（属酣客系，不得入 e2；也不并入 e1——科目码体系不同） + * 期初：e2 用 F1/F2 实测值（treasury_balances，data_status='verified'）； + e1 分账户 open_balance 留 NULL + 头寸 estimated（F2 分账户未交付，LQ-001 skip_when=estimated） + * 金额纪律：写库前 round(x,4) + +用法: + python3 treasury_seed_accounts.py --dry-run # 解析 CSV + 打印将写入的行（零 DB） + python3 treasury_seed_accounts.py --apply # 幂等 upsert（按 (entity_id, account_no) / (account_id, period)） + python3 treasury_seed_accounts.py --verify # 只读回查：账户数/排除项/双实体交集 +""" +import argparse +import csv +import hashlib +import json +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sqlalchemy import text # noqa: E402 +from app.database import get_engine # noqa: E402 + +CSV_DIR = "/root/.hermes/profiles/wecom-finance/reports" +CSV_FILES = { + 1: "CMA_司库P0_F1_账户主数据清单_entity1.csv", + 2: "CMA_司库P0_F1_账户主数据清单_entity2.csv", +} +EXCLUDED_ACCOUNT_NO = "1002007" # 榆林酣客工行存款（酣客系，不得入 e2） +INACTIVE_ACCOUNT_NOS = {"100107"} # 现金备用卡（唐）：首版停用 +ACCOUNT_COLS = ["entity_id", "account_no", "account_name", "account_type", "bank_name", + "subject_code", "currency", "tier", "owner_node_id", "open_balance", + "open_period", "is_pool_member", "status", "remark"] + + +def _load_csv(entity_id: int): + path = os.path.join(CSV_DIR, CSV_FILES[entity_id]) + with open(path, encoding="utf-8-sig", newline="") as f: + rows = list(csv.DictReader(f)) + return path, rows + + +def _norm(row: dict, entity_id: int) -> dict: + def s(k): + v = (row.get(k) or "").strip() + return v or None + + out = { + "entity_id": entity_id, + "account_no": s("account_no"), + "account_name": s("account_name"), + "account_type": s("account_type") or "other", + "bank_name": s("bank_name"), + "subject_code": s("subject_code"), + "currency": s("currency") or "CNY", + "tier": s("tier") or "grassroots", + "owner_node_id": None, # 首版留空（部门写在 remark） + "open_balance": None, + "open_period": s("open_period"), + "is_pool_member": int(s("is_pool_member") or 0), + "status": s("status") or "active", + "remark": s("remark"), + } + if entity_id == 1: + out["open_balance"] = None # e1 分账户期初留 NULL（F2 分账户未交付） + else: + raw = s("open_balance") + out["open_balance"] = None if raw is None else round(float(raw), 4) + if out["account_no"] in INACTIVE_ACCOUNT_NOS: + out["status"] = "inactive" + return out + + +def plan(): + """解析两份 CSV → {entity_id: [rows]}，并做清单级断言。""" + data, problems = {}, [] + for eid in (1, 2): + path, rows = _load_csv(eid) + norms = [_norm(r, eid) for r in rows] + nos = [r["account_no"] for r in norms] + if EXCLUDED_ACCOUNT_NO in nos: + problems.append(f"entity{eid} CSV 含被排除账户 {EXCLUDED_ACCOUNT_NO}") + if len(set(nos)) != len(nos): + problems.append(f"entity{eid} CSV 账号重复") + nos_set = set(nos) + data[eid] = {"csv": path, "rows": norms, "account_nos": sorted(nos_set)} + # 双实体账号集合必须零交集（G1 §5） + inter = set(data[1]["account_nos"]) & set(data[2]["account_nos"]) + if inter: + problems.append(f"双实体账号集合交集非空: {sorted(inter)}") + if len(data[1]["rows"]) != 12 or len(data[2]["rows"]) != 12: + problems.append(f"行数异常: e1={len(data[1]['rows'])} e2={len(data[2]['rows'])}（期望 12/12）") + return data, problems + + +def apply(data): + engine = get_engine() + summary = {"accounts": {}, "balances": {}, "batches": {}} + with engine.begin() as conn: + for eid, payload in data.items(): + created = updated = 0 + for row in payload["rows"]: + cols = ", ".join(ACCOUNT_COLS) + ph = ", ".join(":" + c for c in ACCOUNT_COLS) + upd = ", ".join(f"{c}=VALUES({c})" for c in ACCOUNT_COLS + if c not in ("entity_id", "account_no")) + found = conn.execute(text( + "SELECT id FROM treasury_accounts WHERE entity_id=:e AND account_no=:a" + ), {"e": eid, "a": row["account_no"]}).fetchone() + params = dict(row) + conn.execute(text( + f"INSERT INTO treasury_accounts ({cols}) VALUES ({ph}) " + f"ON DUPLICATE KEY UPDATE {upd}"), params) + created += found is None + updated += found is not None + + # e2 期初头寸（F1/F2 实测值 → verified）；e1 无分账户期初 → 不写头寸（estimated） + bal_created = 0 + if eid == 2: + batch = "f1-f2-e2-" + hashlib.md5( + json.dumps([(r["account_no"], r["open_balance"], r["open_period"]) + for r in payload["rows"]], ensure_ascii=False).encode()).hexdigest()[:10] + for row in payload["rows"]: + if row["open_balance"] is None: + continue + acc = conn.execute(text( + "SELECT id FROM treasury_accounts WHERE entity_id=:e AND account_no=:a" + ), {"e": eid, "a": row["account_no"]}).fetchone() + if not acc: + continue + exists = conn.execute(text( + "SELECT id FROM treasury_balances WHERE entity_id=:e AND account_id=:i AND period=:p" + ), {"e": eid, "i": acc[0], "p": row["open_period"]}).fetchone() + conn.execute(text( + "INSERT INTO treasury_balances (entity_id, account_id, period, balance, " + "available, frozen, source, data_status, batch) VALUES (:e,:i,:p,:b,NULL,0," + "'import:F1-F2账户主数据清单','verified',:bt) " + "ON DUPLICATE KEY UPDATE balance=VALUES(balance), source=VALUES(source), " + "data_status=VALUES(data_status), batch=VALUES(batch)"), + {"e": eid, "i": acc[0], "p": row["open_period"], + "b": row["open_balance"], "bt": batch}) + bal_created += exists is None + summary["batches"][f"entity{eid}"] = batch + summary["accounts"][f"entity{eid}"] = {"rows_in": len(payload["rows"]), + "created": created, "updated": updated} + summary["balances"][f"entity{eid}"] = {"created": bal_created} + return summary + + +def verify(): + """只读回查：账户数 / 排除项 / 双实体零交集 / 100107 inactive。""" + engine = get_engine() + with engine.connect() as conn: + out = {} + rows = conn.execute(text( + "SELECT entity_id, COUNT(*) FROM treasury_accounts GROUP BY entity_id ORDER BY entity_id" + )).fetchall() + out["accounts_by_entity"] = {str(r[0]): int(r[1]) for r in rows} + e1 = {r[0] for r in conn.execute(text( + "SELECT account_no FROM treasury_accounts WHERE entity_id=1")).fetchall()} + e2 = {r[0] for r in conn.execute(text( + "SELECT account_no FROM treasury_accounts WHERE entity_id=2")).fetchall()} + out["entity1_accounts"] = sorted(e1) + out["entity2_accounts"] = sorted(e2) + out["intersection"] = sorted(e1 & e2) + out["excluded_present"] = sorted({EXCLUDED_ACCOUNT_NO} & (e1 | e2)) + out["inactive_100107_status"] = conn.execute(text( + "SELECT status FROM treasury_accounts WHERE entity_id=1 AND account_no='100107'" + )).scalar() + out["open_balance_null_e1"] = conn.execute(text( + "SELECT COUNT(*) FROM treasury_accounts WHERE entity_id=1 AND open_balance IS NULL" + )).scalar() + out["balances_e2"] = conn.execute(text( + "SELECT COUNT(*) FROM treasury_balances WHERE entity_id=2")).scalar() + out["balances_e1"] = conn.execute(text( + "SELECT COUNT(*) FROM treasury_balances WHERE entity_id=1")).scalar() + out["balances_e2_bad_status"] = conn.execute(text( + "SELECT COUNT(*) FROM treasury_balances WHERE entity_id=2 AND data_status<>'verified'" + )).scalar() + out["owner_node_id_nonnull"] = conn.execute(text( + "SELECT COUNT(*) FROM treasury_accounts WHERE owner_node_id IS NOT NULL")).scalar() + print(json.dumps(out, ensure_ascii=False, indent=2)) + return out + + +def main(): + ap = argparse.ArgumentParser() + g = ap.add_mutually_exclusive_group() + g.add_argument("--dry-run", action="store_true") + g.add_argument("--apply", action="store_true") + g.add_argument("--verify", action="store_true", help="只读回查（默认）") + args = ap.parse_args() + + if args.verify or not (args.dry_run or args.apply): + verify() + return 0 + + data, problems = plan() + plan_out = {f"entity{e}": {"csv": data[e]["csv"], "rows": len(data[e]["rows"]), + "account_nos": data[e]["account_nos"]} for e in (1, 2)} + if problems: + print(json.dumps({"problems": problems, "plan": plan_out}, ensure_ascii=False, indent=2)) + return 3 + if args.dry_run: + print(json.dumps({"plan": plan_out, + "sample_entity1": data[1]["rows"][:2], + "sample_entity2": data[2]["rows"][:2]}, ensure_ascii=False, indent=2)) + return 0 + summary = apply(data) + print(json.dumps({"applied": summary, "plan": plan_out}, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 87ba881d..53f0b0e0 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -493,4 +493,36 @@ export const aiSuggestionApi = { preview: (id: number) => api.get(`/ai/suggestions/${id}/preview`), } +// ── 资金管理（账户池/头寸/支付内控 2026-09-15）── +export const treasuryApi = { + // 账户池 + listAccounts: (params?: any) => api.get('/treasury/accounts', { params }), + createAccount: (data: any) => api.post('/treasury/accounts', data), + updateAccount: (id: number, data: any) => api.put(`/treasury/accounts/${id}`, data), + deleteAccount: (id: number) => api.delete(`/treasury/accounts/${id}`), + importAccounts: (rows: any[]) => api.post('/treasury/accounts/import', rows), + // 头寸 + listBalances: (params?: any) => api.get('/treasury/balances', { params }), + upsertBalance: (data: any) => api.post('/treasury/balances', data), + importBalances: (rows: any[]) => api.post('/treasury/balances/import', rows), + positionsSummary: (params?: any) => api.get('/treasury/positions/summary', { params }), + dashboard: () => api.get('/treasury/dashboard'), + // 规则库 / 授权矩阵 + listRules: (params?: any) => api.get('/treasury/control-rules', { params }), + updateRule: (id: number, data: any) => api.put(`/treasury/control-rules/${id}`, data), + testRules: (data: any) => api.post('/treasury/control-rules/test', data), + listLimits: (params?: any) => api.get('/treasury/limits', { params }), + updateLimit: (id: number, data: any) => api.put(`/treasury/limits/${id}`, data), + // 支付单 + listPayments: (params?: any) => api.get('/treasury/payments', { params }), + createPayment: (data: any) => api.post('/treasury/payments', data), + updatePayment: (id: number, data: any) => api.put(`/treasury/payments/${id}`, data), + precheckPayment: (id: number) => api.post(`/treasury/payments/${id}/precheck`), + approvePayment: (id: number, data: any) => api.post(`/treasury/payments/${id}/approve`, data), + payPayment: (id: number) => api.post(`/treasury/payments/${id}/pay`), + waivePayment: (id: number, data: any) => api.post(`/treasury/payments/${id}/waive`, data), + // 规则命中留痕 + listHits: (params?: any) => api.get('/treasury/control-hits', { params }), +} + export default api diff --git a/frontend/src/permission.ts b/frontend/src/permission.ts index c540d235..ab235cb7 100644 --- a/frontend/src/permission.ts +++ b/frontend/src/permission.ts @@ -9,10 +9,10 @@ interface MenuItem { // ── 角色路由映射 ── export const ROLE_ROUTES: Record = { - ceo: ['/my-dashboard', '/dashboard', '/ai-suggestions', '/kpis', '/maps', '/maps-review', '/maps/canvas', '/maps/review', '/alerts', '/notifications', '/org', '/users', '/permissions', '/budget', '/deviations', '/cost', '/cost-intelligence', '/predict', '/action-plans', '/knowledge', '/guide', '/customer', '/learning-dashboard', '/reports', '/alignment', '/dupont-analysis', '/analysis-confidence', '/expenses', '/cash-plan', '/receivables', '/growth-quality', '/product-matrix', '/tax-compliance', '/data-classification', '/cashflow-structure'], - finance: ['/my-dashboard', '/dashboard', '/ai-suggestions', '/kpis', '/maps', '/maps-review', '/maps/canvas', '/maps/review', '/alerts', '/data', '/budget', '/deviations', '/cost', '/cost-intelligence', '/predict', '/action-plans', '/knowledge', '/guide', '/customer', '/learning-dashboard', '/reports', '/alignment', '/dupont-analysis', '/analysis-confidence', '/expenses', '/cash-plan', '/receivables', '/growth-quality', '/product-matrix', '/tax-compliance', '/data-classification', '/cashflow-structure'], + ceo: ['/my-dashboard', '/dashboard', '/ai-suggestions', '/kpis', '/maps', '/maps-review', '/maps/canvas', '/maps/review', '/alerts', '/notifications', '/org', '/users', '/permissions', '/budget', '/deviations', '/cost', '/cost-intelligence', '/predict', '/action-plans', '/knowledge', '/guide', '/customer', '/learning-dashboard', '/reports', '/alignment', '/dupont-analysis', '/analysis-confidence', '/expenses', '/cash-plan', '/receivables', '/growth-quality', '/product-matrix', '/tax-compliance', '/data-classification', '/cashflow-structure', '/treasury-dashboard', '/treasury-payment'], + finance: ['/my-dashboard', '/dashboard', '/ai-suggestions', '/kpis', '/maps', '/maps-review', '/maps/canvas', '/maps/review', '/alerts', '/data', '/budget', '/deviations', '/cost', '/cost-intelligence', '/predict', '/action-plans', '/knowledge', '/guide', '/customer', '/learning-dashboard', '/reports', '/alignment', '/dupont-analysis', '/analysis-confidence', '/expenses', '/cash-plan', '/receivables', '/growth-quality', '/product-matrix', '/tax-compliance', '/data-classification', '/cashflow-structure', '/treasury-dashboard', '/treasury-payment'], business: ['/my-dashboard', '/dashboard', '/ai-suggestions', '/kpis', '/alerts', '/budget', '/deviations', '/action-plans', '/knowledge', '/guide', '/customer', '/expenses', '/cash-plan', '/growth-quality', '/product-matrix', '/tax-compliance', '/data-classification'], - it: ['/my-dashboard', '/dashboard', '/ai-suggestions', '/kpis', '/alerts', '/data', '/org', '/users', '/permissions', '/budget', '/deviations', '/cost', '/cost-intelligence', '/predict', '/action-plans', '/knowledge', '/guide', '/customer', '/learning-dashboard', '/analysis-confidence', '/expenses', '/cash-plan', '/product-matrix', '/tax-compliance', '/data-classification'], + it: ['/my-dashboard', '/dashboard', '/ai-suggestions', '/kpis', '/alerts', '/data', '/org', '/users', '/permissions', '/budget', '/deviations', '/cost', '/cost-intelligence', '/predict', '/action-plans', '/knowledge', '/guide', '/customer', '/learning-dashboard', '/analysis-confidence', '/expenses', '/cash-plan', '/product-matrix', '/tax-compliance', '/data-classification', '/treasury-payment'], } export const ROLE_ACTIONS: Record = { @@ -36,6 +36,8 @@ export const MENU_ITEMS: MenuItem[] = [ { path: '/tax-compliance', label: '税务合规', icon: 'DataAnalysis', roles: ['ceo', 'finance', 'business', 'it'], group: '🟢 D 执行与控制' }, { path: '/cash-plan', label: '收付款计划', icon: 'Money', roles: ['ceo', 'finance', 'business', 'it'], group: '🟢 D 执行与控制' }, { path: '/receivables', label: '应收款催收', icon: 'Money', roles: ['ceo', 'finance', 'business'], group: '🟢 D 执行与控制' }, + { path: '/treasury-dashboard', label: '资金总览', icon: 'Money', roles: ['ceo','finance'], group: '🟢 D 执行与控制' }, + { path: '/treasury-payment', label: '资金内控', icon: 'Lock', roles: ['ceo','finance'], group: '🟢 D 执行与控制' }, { path: '/growth-quality', label: '增长质量诊断', icon: 'TrendCharts', roles: ['ceo', 'finance', 'business', 'it'], group: '🟡 C 监控与评价' }, { path: '/product-matrix', label: '波士顿产品矩阵', icon: 'PieChart', roles: ['ceo', 'finance', 'business', 'it'], group: '🟡 C 监控与评价' }, { path: '/deviations', label: '差异分析', icon: 'DataAnalysis', roles: ['ceo', 'finance', 'business', 'it'], group: '🟢 D 执行与控制' }, diff --git a/frontend/src/router/index.ts b/frontend/src/router/index.ts index 6a25a441..43c72675 100644 --- a/frontend/src/router/index.ts +++ b/frontend/src/router/index.ts @@ -56,6 +56,8 @@ const routes = [ { path: 'expenses', name: 'ExpenseManage', component: () => import('@/views/ExpenseManage.vue'), meta: { title: '费用审核', roles: ['ceo', 'finance', 'business', 'it'], zone: 'B', editable: true } }, { path: 'cash-plan', name: 'CashPlan', component: () => import('@/views/CashPlan.vue'), meta: { title: '收付款计划', roles: ['ceo', 'finance', 'business', 'it'], zone: 'B', editable: true } }, { path: 'receivables', name: 'Receivables', component: () => import('@/views/Receivables.vue'), meta: { title: '应收款催收', roles: ['ceo', 'finance', 'business'], zone: 'B' } }, + { path: 'treasury-dashboard', name: 'TreasuryDashboard', component: () => import('@/views/TreasuryDashboard.vue'), meta: { title: '资金总览', roles: ['ceo','finance'], zone: 'B', editable: true } }, + { path: 'treasury-payment', name: 'TreasuryPayment', component: () => import('@/views/TreasuryPayment.vue'), meta: { title: '资金内控', roles: ['ceo','finance'], zone: 'B', editable: true } }, { path: 'growth-quality', name: 'GrowthQuality', component: () => import('@/views/GrowthQuality.vue'), meta: { title: '增长质量诊断', roles: ['ceo', 'finance', 'business', 'it'], zone: 'A' } }, { path: 'product-matrix', name: 'ProductMatrix', component: () => import('@/views/ProductMatrix.vue'), meta: { title: '波士顿产品矩阵', roles: ['ceo', 'finance', 'business', 'it'], zone: 'A' } }, // /tax-compliance: B区（页内职责=税务记录/发票/社保人工录入维护；见盘点表） diff --git a/frontend/src/views/TreasuryDashboard.vue b/frontend/src/views/TreasuryDashboard.vue new file mode 100644 index 00000000..48391c19 --- /dev/null +++ b/frontend/src/views/TreasuryDashboard.vue @@ -0,0 +1,539 @@ + + + + + diff --git a/frontend/src/views/TreasuryPayment.vue b/frontend/src/views/TreasuryPayment.vue new file mode 100644 index 00000000..30b9a856 --- /dev/null +++ b/frontend/src/views/TreasuryPayment.vue @@ -0,0 +1,573 @@ + + + + +