diff --git a/backend/scripts/_archive/verify_bep_threshold_fix.snapshot-20260908.py b/backend/scripts/_archive/verify_bep_threshold_fix.snapshot-20260908.py new file mode 100644 index 00000000..0a94600c --- /dev/null +++ b/backend/scripts/_archive/verify_bep_threshold_fix.snapshot-20260908.py @@ -0,0 +1,130 @@ +"""BEP保本KPI阈值+现金口径修复验收脚本 — fin-fix-bep-threshold-20260908 + +验收: +1. kpi_definitions 457/458 阈值与财务口径一致(COVER: 100/85/85;GAP: 0/15/15),456 保持 NULL +2. 现金口径2条定义 F_BEP_REVENUE_CASH/F_BEP_COVER_CASH 存在(无阈值/100/85/85) +3. 2026-07 派生5条落库:账面 169.9/66.9/33.1 + 现金 121.0/93.9(±1%财务权威 121.4/93.5) +4. 告警闭环:575/576 = red pending(66.9<85, 33.1>15);577 = yellow pending(93.9>=85) + 旧黄色 573/574 已 resolved +5. bep-card 端点:book.cover_level=red, cash.cover_level=yellow, derived 5 条 + +用法: venv/bin/python3 scripts/verify_bep_threshold_fix.py (需后端8010在线) + +⚠️ 历史快照守卫(2026-09-12 加固,任务单 concurrent-duplicate-session-and-stale-verify-script-20260912): + 本文件**不是活检查**,是 fin-fix-bep-threshold-20260908 的验收快照,硬编码 + period='2026-07' 期望值 169.9/66.9/33.1 与告警自增 ID 573~577。期值随派生器滚动 + (2026-07 现为 170.2/66.7/33.3,最新期已滚动到 2026-08)、告警 ID 已变为 596~598, + 且脚本所用的 cma_user 账号已按凭据治理退役 → 直接复跑必然报错/FAIL,属"快照过期", + **不代表本轮回归**。现行活检查 = ../verify_bep_threshold_fix.py(按最新派生期动态断言 + 只读凭据)。 + 原版逐字留存于 git(HEAD:backend/scripts/verify_bep_threshold_fix.py)。 + 确需复跑历史行为:FORCE_SNAPSHOT=1 python3 scripts/_archive/verify_bep_threshold_fix.snapshot-20260908.py +""" +import json +import os +import sys +import urllib.request + +import pymysql + +if os.environ.get("FORCE_SNAPSHOT") != "1": + print("⛔ 这是 20260908 的历史验收快照,不是活检查;复跑结果不代表当前回归。") + print(" 现行活检查:python3 backend/scripts/verify_bep_threshold_fix.py") + print(" 历史行为复跑:FORCE_SNAPSHOT=1 python3 scripts/_archive/verify_bep_threshold_fix.snapshot-20260908.py") + sys.exit(2) + +BASE = "http://127.0.0.1:8010" + +passed = [] + + +def check(name, cond, detail=""): + if cond: + passed.append(name) + print(f" PASS {name}") + else: + print(f" FAIL {name} {detail}") + sys.exit(1) + + +def _db_cfg(): + """从 backend/.env 读生产 DB 凭据(root@localhost 走 auth_socket,pymysql 直连须用应用账号 cma_user)""" + env = {} + env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".env") + if os.path.exists(env_path): + for line in open(env_path): + line = line.strip() + if line and "=" in line and not line.startswith("#"): + k, v = line.split("=", 1) + env[k] = v.strip().strip('"').strip("'") + return dict( + host=env.get("CMA_DB_HOST", "127.0.0.1"), + port=int(env.get("CMA_DB_PORT", "3306")), + user=env.get("CMA_DB_USER", "cma_user"), + password=env.get("CMA_DB_PASS", ""), + database=env.get("CMA_DB_NAME", "cma"), + charset="utf8mb4", + ) + + +def db_q(sql): + conn = pymysql.connect(**_db_cfg()) + try: + with conn.cursor() as cur: + cur.execute(sql) + return cur.fetchall() + finally: + conn.close() + + +def http_json(path, method="GET", payload=None, token=None): + headers = {} + if token: + headers["Authorization"] = "Bearer " + token + data = json.dumps(payload).encode() if payload is not None else None + if data: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(BASE + path, data=data, headers=headers, method=method) + with urllib.request.urlopen(req) as r: + return json.loads(r.read().decode()) + + +print("== 1. 库阈值 ==") +rows = db_q("SELECT id, kpi_code, threshold_green, threshold_yellow, threshold_red FROM kpi_definitions WHERE id IN (456,457,458)") +m = {r[0]: r for r in rows} +check("457 COVER 100/85/85", m[457][2:5] == (">=100", ">=85", "<85"), str(m[457])) +check("458 GAP 0/15/15", m[458][2:5] == ("<=0", "<=15", ">15"), str(m[458])) +check("456 REVENUE 无阈值", m[456][2] is None and m[456][3] is None and m[456][4] is None, str(m[456])) + +print("== 2. 现金口径定义 ==") +rows = db_q("SELECT kpi_code, unit, threshold_green, threshold_yellow, threshold_red FROM kpi_definitions WHERE kpi_code IN ('F_BEP_REVENUE_CASH','F_BEP_COVER_CASH')") +cash = {r[0]: r for r in rows} +check("REVENUE_CASH 存在且无阈值", "F_BEP_REVENUE_CASH" in cash and cash["F_BEP_REVENUE_CASH"][1] == "万元" and cash["F_BEP_REVENUE_CASH"][2] is None, str(cash.get("F_BEP_REVENUE_CASH"))) +check("COVER_CASH 存在 100/85/85", "F_BEP_COVER_CASH" in cash and cash["F_BEP_COVER_CASH"][1] == "%" and cash["F_BEP_COVER_CASH"][2:5] == (">=100", ">=85", "<85"), str(cash.get("F_BEP_COVER_CASH"))) + +print("== 3. 2026-07 派生5条 ==") +rows = db_q("SELECT k.kpi_code, v.actual_value, v.source_type FROM kpi_values v JOIN kpi_definitions k ON v.kpi_id=k.id WHERE v.period='2026-07' AND k.kpi_code LIKE 'F_BEP%' AND v.source_type='derived'") +vals = {r[0]: r[1] for r in rows} +exp_book = {"F_BEP_REVENUE": 169.9, "F_BEP_COVER": 66.9, "F_BEP_GAP": 33.1} +for code, ev in exp_book.items(): + check(f"{code} = {ev}", vals.get(code) == ev, f"got {vals.get(code)}") +check("REVENUE_CASH ~121.4±1%", abs(vals.get("F_BEP_REVENUE_CASH", 0) - 121.4) / 121.4 < 0.01, f"got {vals.get('F_BEP_REVENUE_CASH')}") +check("COVER_CASH ~93.5±1%", abs(vals.get("F_BEP_COVER_CASH", 0) - 93.5) / 93.5 < 0.01, f"got {vals.get('F_BEP_COVER_CASH')}") + +print("== 4. 告警闭环 ==") +rows = db_q("SELECT id, alert_level, status FROM kpi_alerts WHERE kpi_id IN (457,458,463) ORDER BY id") +alerts = {r[0]: (r[1], r[2]) for r in rows} +check("573 旧黄已resolved", alerts.get(573) == ("yellow", "resolved"), str(alerts.get(573))) +check("574 旧黄已resolved", alerts.get(574) == ("yellow", "resolved"), str(alerts.get(574))) +check("575 COVER red pending", alerts.get(575) == ("red", "pending"), str(alerts.get(575))) +check("576 GAP red pending", alerts.get(576) == ("red", "pending"), str(alerts.get(576))) +check("577 COVER_CASH yellow pending", alerts.get(577) == ("yellow", "pending"), str(alerts.get(577))) + +print("== 5. bep-card 端点 ==") +login = http_json("/api/cma/auth/login", "POST", {"username": "admin", "password": "admin123", "entity_id": 1}) +card = http_json("/api/cma/dashboard/bep-card", token=login["token"]) +check("available", card.get("available") is True) +check("book red", card["book"]["cover_level"] == "red", str(card["book"])) +check("cash yellow", card["cash"]["cover_level"] == "yellow", str(card["cash"])) +check("derived 5条", len(card.get("derived", {})) == 5, str(list(card.get("derived", {}).keys()))) + +print(f"\nALL {len(passed)} CHECKS PASSED") diff --git a/backend/scripts/verify_bep_threshold_fix.py b/backend/scripts/verify_bep_threshold_fix.py index 54c3c380..47429a1e 100644 --- a/backend/scripts/verify_bep_threshold_fix.py +++ b/backend/scripts/verify_bep_threshold_fix.py @@ -1,25 +1,41 @@ -"""BEP保本KPI阈值+现金口径修复验收脚本 — fin-fix-bep-threshold-20260908 +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""BEP 保本 KPI 口径一致性验证(**动态期**)— 原 fin-fix-bep-threshold-20260908 快照改造版 -验收: -1. kpi_definitions 457/458 阈值与财务口径一致(COVER: 100/85/85;GAP: 0/15/15),456 保持 NULL -2. 现金口径2条定义 F_BEP_REVENUE_CASH/F_BEP_COVER_CASH 存在(无阈值/100/85/85) -3. 2026-07 派生5条落库:账面 169.9/66.9/33.1 + 现金 121.0/93.9(±1%财务权威 121.4/93.5) -4. 告警闭环:575/576 = red pending(66.9<85, 33.1>15);577 = yellow pending(93.9>=85) - 旧黄色 573/574 已 resolved -5. bep-card 端点:book.cover_level=red, cash.cover_level=yellow, derived 5 条 +改造说明(2026-09-12,任务单 concurrent-duplicate-session-and-stale-verify-script-20260912): + 原脚本是 fin-fix-bep-threshold-20260908 的**验收快照**:硬编码 period='2026-07'、 + 期望值 169.9/66.9/33.1、告警自增 ID 573~577。期值随派生器持续更新 + (2026-07 现为 170.2/66.7/33.3,最新派生期已滚动到 2026-08)→ 复跑必然 + `FAIL F_BEP_REVENUE = 169.9 got 170.2`,后人会把"快照过期"误读成"本轮回归"。 + 处置:改为**按最新派生期动态断言**(不是删除——阈值口径与派生一致性本身就是活的检查项): + · 结构契约:阈值口径按 kpi_code 查,不写死 ID / 不写死期 + · 数值检查:改判**定义级一致性**(GAP = 100 − COVER、现金收入 ≤ 账面收入、值 ≥ 0) + · 告警检查:改为**按阈值复算级别**(不再比对自增 ID),另对"历史期仍 pending"给软告警 + · 端点检查:凭据读 .secrets/cma-admin.env(不再硬编码账号口令),期望值与 card 复算比对 + 历史快照留档:backend/scripts/_archive/verify_bep_threshold_fix.snapshot-20260908.py -用法: venv/bin/python3 scripts/verify_bep_threshold_fix.py (需后端8010在线) +口径与访问纪律: + · DB 只读 = cma_reader(.secrets/cma-reader.env,可 CMA_READER_ENV 覆盖;仅 SELECT 权限) + · 登录口令 = .secrets/cma-admin.env(600,禁止把口令写进代码/文档/日志) +用法: python3 backend/scripts/verify_bep_threshold_fix.py (需后端 8010 在线) +退出码: 0 = 全部结构/一致性断言通过;1 = 有 FAIL(=真漂移,不是期回滚) """ import json import os +import re import sys import urllib.request import pymysql -BASE = "http://127.0.0.1:8010" +BASE = os.environ.get("CMA_API_BASE", "http://127.0.0.1:8010") +PROFILE_HOME = "/root/.hermes/profiles/wecom-project" +READER_ENV = os.environ.get("CMA_READER_ENV", os.path.join(PROFILE_HOME, ".secrets/cma-reader.env")) +ADMIN_ENV = os.environ.get("CMA_ADMIN_ENV", os.path.join(PROFILE_HOME, ".secrets/cma-admin.env")) +BEP_CODES = ("F_BEP_REVENUE", "F_BEP_COVER", "F_BEP_GAP", + "F_BEP_REVENUE_CASH", "F_BEP_COVER_CASH") -passed = [] +passed, warned = [], [] def check(name, cond, detail=""): @@ -31,36 +47,56 @@ def check(name, cond, detail=""): sys.exit(1) -def _db_cfg(): - """从 backend/.env 读生产 DB 凭据(root@localhost 走 auth_socket,pymysql 直连须用应用账号 cma_user)""" +def warn(name, detail=""): + warned.append(name) + print(f" WARN {name} {detail}") + + +def _dotenv(path): env = {} - env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".env") - if os.path.exists(env_path): - for line in open(env_path): + try: + for line in open(path, encoding="utf-8"): line = line.strip() if line and "=" in line and not line.startswith("#"): k, v = line.split("=", 1) env[k] = v.strip().strip('"').strip("'") - return dict( - host=env.get("CMA_DB_HOST", "127.0.0.1"), - port=int(env.get("CMA_DB_PORT", "3306")), - user=env.get("CMA_DB_USER", "cma_user"), - password=env.get("CMA_DB_PASS", ""), - database=env.get("CMA_DB_NAME", "cma"), - charset="utf8mb4", - ) + except Exception as e: + print(f" FAIL 读取凭据文件失败 {path}: {str(e)[:80]}") + sys.exit(1) + return env -def db_q(sql): - conn = pymysql.connect(**_db_cfg()) +def db_q(sql, args=None): + e = _dotenv(READER_ENV) + conn = pymysql.connect(host=e["CMA_READER_DB_HOST"], port=int(e["CMA_READER_DB_PORT"]), + user=e["CMA_READER_DB_USER"], password=e["CMA_READER_DB_PASS"], + database=e["CMA_READER_DB_NAME"], charset="utf8mb4") try: with conn.cursor() as cur: - cur.execute(sql) + cur.execute(sql, args or ()) return cur.fetchall() finally: conn.close() +def level_of(value, thr_yellow, thr_red): + """阈值口径 → 级别:red 命中优先,其次 yellow,否则 green""" + def hit(expr): + if not expr or value is None: + return False + m = re.match(r"^\s*(>=|<=|>|<|=)\s*(-?[\d.]+)\s*$", str(expr)) + if not m: + return False + op, num = m.group(1), float(m.group(2)) + return {">=": value >= num, "<=": value <= num, ">": value > num, + "<": value < num, "=": value == num}[op] + if hit(thr_red): + return "red" + if hit(thr_yellow): + return "yellow" + return "green" + + def http_json(path, method="GET", payload=None, token=None): headers = {} if token: @@ -69,47 +105,104 @@ def http_json(path, method="GET", payload=None, token=None): if data: headers["Content-Type"] = "application/json" req = urllib.request.Request(BASE + path, data=data, headers=headers, method=method) - with urllib.request.urlopen(req) as r: + with urllib.request.urlopen(req, timeout=15) as r: return json.loads(r.read().decode()) -print("== 1. 库阈值 ==") -rows = db_q("SELECT id, kpi_code, threshold_green, threshold_yellow, threshold_red FROM kpi_definitions WHERE id IN (456,457,458)") -m = {r[0]: r for r in rows} -check("457 COVER 100/85/85", m[457][2:5] == (">=100", ">=85", "<85"), str(m[457])) -check("458 GAP 0/15/15", m[458][2:5] == ("<=0", "<=15", ">15"), str(m[458])) -check("456 REVENUE 无阈值", m[456][2] is None and m[456][3] is None and m[456][4] is None, str(m[456])) +# ── 1. 阈值口径(结构契约:不写死 ID、不写死期)─────────────────────────────── +print("== 1. 阈值口径(按 kpi_code,结构契约)==") +rows = db_q("SELECT kpi_code, threshold_green, threshold_yellow, threshold_red FROM kpi_definitions " + "WHERE kpi_code IN %s", (BEP_CODES,)) +thr = {r[0]: r[1:4] for r in rows} +check("5 个 BEP 口径均已定义", all(c in thr for c in BEP_CODES), + f"missing={[c for c in BEP_CODES if c not in thr]}") +check("COVER 口径 100/85/85", thr.get("F_BEP_COVER") == (">=100", ">=85", "<85"), str(thr.get("F_BEP_COVER"))) +check("GAP 口径 0/15/15", thr.get("F_BEP_GAP") == ("<=0", "<=15", ">15"), str(thr.get("F_BEP_GAP"))) +check("REVENUE 无阈值(派生值不判级)", thr.get("F_BEP_REVENUE") == (None, None, None), str(thr.get("F_BEP_REVENUE"))) +check("COVER_CASH 口径 100/85/85", thr.get("F_BEP_COVER_CASH") == (">=100", ">=85", "<85"), + str(thr.get("F_BEP_COVER_CASH"))) +check("REVENUE_CASH 无阈值", thr.get("F_BEP_REVENUE_CASH") == (None, None, None), str(thr.get("F_BEP_REVENUE_CASH"))) -print("== 2. 现金口径定义 ==") -rows = db_q("SELECT kpi_code, unit, threshold_green, threshold_yellow, threshold_red FROM kpi_definitions WHERE kpi_code IN ('F_BEP_REVENUE_CASH','F_BEP_COVER_CASH')") -cash = {r[0]: r for r in rows} -check("REVENUE_CASH 存在且无阈值", "F_BEP_REVENUE_CASH" in cash and cash["F_BEP_REVENUE_CASH"][1] == "万元" and cash["F_BEP_REVENUE_CASH"][2] is None, str(cash.get("F_BEP_REVENUE_CASH"))) -check("COVER_CASH 存在 100/85/85", "F_BEP_COVER_CASH" in cash and cash["F_BEP_COVER_CASH"][1] == "%" and cash["F_BEP_COVER_CASH"][2:5] == (">=100", ">=85", "<85"), str(cash.get("F_BEP_COVER_CASH"))) +# ── 2. 最新派生期 + 5 口径齐全(实体动态:BEP 数据挂在哪个实体就验哪个实体)──── +print("== 2. 最新派生期与落库完整性 ==") +rows = db_q("SELECT v.entity_id, v.period, k.kpi_code, v.actual_value FROM kpi_values v " + "JOIN kpi_definitions k ON v.kpi_id=k.id " + "WHERE k.kpi_code LIKE 'F_BEP%%' AND v.source_type='derived'") +per_entity = {} +for eid, period, code, val in rows: + per_entity.setdefault(int(eid), {}).setdefault(period, {})[code] = \ + float(val) if val is not None else None +check("存在派生期数据", bool(per_entity), "entities=%s" % sorted(per_entity)) +ENTITY = max(per_entity, key=lambda e: max(per_entity[e])) +LATEST = max(per_entity[ENTITY]) +vals = per_entity[ENTITY][LATEST] +print(f" 实测实体 entity_id={ENTITY},最新派生期 P = {LATEST}(当前值: " + + ", ".join(f"{c}={vals.get(c)}" for c in BEP_CODES) + ")") +check(f"最新派生期 {LATEST} 五口径齐全且非空", + all(vals.get(c) is not None for c in BEP_CODES), + f"missing={[c for c in BEP_CODES if vals.get(c) is None]}") +check("派生值均 ≥ 0", all(vals[c] >= 0 for c in BEP_CODES if vals.get(c) is not None)) -print("== 3. 2026-07 派生5条 ==") -rows = db_q("SELECT k.kpi_code, v.actual_value, v.source_type FROM kpi_values v JOIN kpi_definitions k ON v.kpi_id=k.id WHERE v.period='2026-07' AND k.kpi_code LIKE 'F_BEP%' AND v.source_type='derived'") -vals = {r[0]: r[1] for r in rows} -exp_book = {"F_BEP_REVENUE": 169.9, "F_BEP_COVER": 66.9, "F_BEP_GAP": 33.1} -for code, ev in exp_book.items(): - check(f"{code} = {ev}", vals.get(code) == ev, f"got {vals.get(code)}") -check("REVENUE_CASH ~121.4±1%", abs(vals.get("F_BEP_REVENUE_CASH", 0) - 121.4) / 121.4 < 0.01, f"got {vals.get('F_BEP_REVENUE_CASH')}") -check("COVER_CASH ~93.5±1%", abs(vals.get("F_BEP_COVER_CASH", 0) - 93.5) / 93.5 < 0.01, f"got {vals.get('F_BEP_COVER_CASH')}") +# ── 3. 定义级一致性(替代硬编码数值)───────────────────────────────────────── +print("== 3. 定义级一致性(GAP = 100 − COVER;现金口径 ≤ 账面口径)==") +check("GAP ≈ 100 − COVER(±0.2)", abs(vals["F_BEP_GAP"] - (100 - vals["F_BEP_COVER"])) <= 0.2, + f"gap={vals['F_BEP_GAP']} cover={vals['F_BEP_COVER']}") +check("REVENUE_CASH ≤ REVENUE(现金口径不高于账面口径)", + vals["F_BEP_REVENUE_CASH"] <= vals["F_BEP_REVENUE"] + 0.05, + f"cash={vals['F_BEP_REVENUE_CASH']} book={vals['F_BEP_REVENUE']}") -print("== 4. 告警闭环 ==") -rows = db_q("SELECT id, alert_level, status FROM kpi_alerts WHERE kpi_id IN (457,458,463) ORDER BY id") -alerts = {r[0]: (r[1], r[2]) for r in rows} -check("573 旧黄已resolved", alerts.get(573) == ("yellow", "resolved"), str(alerts.get(573))) -check("574 旧黄已resolved", alerts.get(574) == ("yellow", "resolved"), str(alerts.get(574))) -check("575 COVER red pending", alerts.get(575) == ("red", "pending"), str(alerts.get(575))) -check("576 GAP red pending", alerts.get(576) == ("red", "pending"), str(alerts.get(576))) -check("577 COVER_CASH yellow pending", alerts.get(577) == ("yellow", "pending"), str(alerts.get(577))) +# ── 4. 告警与阈值一致性(按阈值复算,不比对自增 ID)────────────────────────── +print("== 4. 告警级别与阈值复算一致 ==") +rows = db_q("SELECT a.id, a.alert_level, a.status, v.period, k.kpi_code, v.actual_value " + "FROM kpi_alerts a JOIN kpi_definitions k ON a.kpi_id=k.id " + "JOIN kpi_values v ON a.kpi_value_id=v.id WHERE k.kpi_code LIKE 'F_BEP%%'") +latest_alerts = [r for r in rows if r[3] == LATEST] +for _id, lvl, st, period, code, val in latest_alerts: + exp = level_of(val, thr[code][1], thr[code][2]) + if exp == "green": + check(f"告警 {_id}({code} {period})不应存在(实测值 {val} 未越阈)", False, + f"level={lvl} status={st}") + else: + check(f"告警 {_id} 级别与阈值一致({code} {period} 实测 {val} → {exp})", lvl == exp, + f"expected={exp} got={lvl}") +pending_latest = [r for r in latest_alerts if r[2] == "pending"] +check(f"最新派生期 {LATEST} 的红/黄项均有 pending 告警", + all(any(r[4] == c and r[2] == "pending" and level_of(r[5], thr[c][1], thr[c][2]) + for r in latest_alerts) + for c in ("F_BEP_COVER", "F_BEP_GAP")), + "期值越阈但无 pending 告警") +hist_pending = [r for r in rows if r[3] != LATEST and r[2] == "pending"] +if hist_pending: + warn(f"历史期仍有 {len(hist_pending)} 条 pending 告警(新期派生未自动解除旧期)" + f":{[ (r[0], r[4], r[3]) for r in hist_pending ]}") -print("== 5. bep-card 端点 ==") -login = http_json("/api/cma/auth/login", "POST", {"username": "admin", "password": "admin123", "entity_id": 1}) +# ── 5. bep-card 端点(动态期,凭据来自 .secrets)──────────────────────────── +print("== 5. bep-card 端点(period 必须等于最新派生期,级别必须与阈值复算一致)==") +adm = _dotenv(ADMIN_ENV) +# 实体必须用第 2 段实测出来的 ENTITY(数据在哪就验哪),**不要信凭据文件里的 CMA_ADMIN_ENTITY_ID**: +# 该字段是"默认登录实体",与"目标数据归属实体"无关(实测凭据=2/博海,而 BEP 派生数据只在 1/酣客, +# 照凭据登录会拿到 available=false,把"验证脚本选错实体"误判成"端点回归")。 +cred_entity = int(adm.get("CMA_ADMIN_ENTITY_ID", "1") or 1) +if cred_entity != ENTITY: + warn(f"凭据默认实体 {cred_entity} ≠ BEP 数据归属实体 {ENTITY}(按实测实体登录,不按凭据字段)") +login = http_json("/api/cma/auth/login", "POST", + {"username": adm.get("CMA_ADMIN_USER"), "password": adm.get("CMA_ADMIN_PASS"), + "entity_id": ENTITY}) card = http_json("/api/cma/dashboard/bep-card", token=login["token"]) -check("available", card.get("available") is True) -check("book red", card["book"]["cover_level"] == "red", str(card["book"])) -check("cash yellow", card["cash"]["cover_level"] == "yellow", str(card["cash"])) -check("derived 5条", len(card.get("derived", {})) == 5, str(list(card.get("derived", {}).keys()))) +check(f"available(entity={ENTITY})", card.get("available") is True, + f"message={card.get('message')} missing={card.get('missing')} found={card.get('found')}") +if card.get("entity_id") is not None: + check(f"card.entity_id == 数据归属实体 {ENTITY}", int(card["entity_id"]) == ENTITY, str(card.get("entity_id"))) +check(f"card.period == 最新派生期 {LATEST}", card.get("period") == LATEST, str(card.get("period"))) +check("derived 5 条", len(card.get("derived", {})) == 5, str(list(card.get("derived", {}).keys()))) +for leg in ("book", "cash"): + code = "F_BEP_COVER" if leg == "book" else "F_BEP_COVER_CASH" + d = card.get(leg, {}) + exp = level_of(d.get("cover"), thr[code][1], thr[code][2]) + check(f"card.{leg}.cover_level 与阈值复算一致({d.get('cover')} → {exp})", + d.get("cover_level") == exp, f"got={d.get('cover_level')}") + check(f"card.{leg}.cover 与库内派生值一致", + d.get("cover") is not None and abs(float(d["cover"]) - vals[code]) <= 0.05, + f"card={d.get('cover')} db={vals[code]}") -print(f"\nALL {len(passed)} CHECKS PASSED") +print(f"\nALL {len(passed)} CHECKS PASSED" + (f"(另有 {len(warned)} 条 WARN)" if warned else ""))