Files
Neo-ZQYY/scripts/ops/validate_board_finance.py
Neo 779b2f6d52 chore: v1 整理 — 清理历史文件、DDL 合并、文档归档
- 清理 1155 个已删除的历史文件(废弃 prompt_logs、tmp、旧 ops 脚本)
- export/ 数据文件从 git 移除(已在 .gitignore)
- demo-miniprogram 从 tmp/ 移入 apps/,添加 CLAUDE.md 注解
- DDL 合并:完整 schema 定义填充到 db/*/schemas/(从 docs/database/ddl/ 复制)
- 39 个 v1 迁移脚本归档到 db/_archived/migrations_v1_merged/
- 4 个迁移变更类 BD_Manual 文档归档到 docs/database/_archived/
- .gitignore 补充 .vite/ 和 apps/*.zip
- settings.json 添加 effortLevel 默认配置
- scripts/ops/ 新增运维脚本入库

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-04-06 00:39:27 +08:00

528 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""
财务看板 DWS 区域维度重构 — 144 组合全量验证脚本。
遍历 8 time_range × 9 area_code × 2 compare = 144 种组合,验证后端 API 返回数据。
产出物export/board-finance-validation.md
Requirements: 9.1, 9.2, 9.3, 9.4
"""
from __future__ import annotations
import json
import sys
import time
from datetime import datetime
from pathlib import Path
import requests
# ── 环境 ──────────────────────────────────────────────────────────────────────
sys.path.insert(0, str(Path(__file__).resolve().parent))
from _env_paths import ensure_repo_root
ensure_repo_root()
BASE = "http://127.0.0.1:8000"
OPENID = "dev_test_openid" # 小程序前端默认测试用户,已有门店绑定和权限
# ── 组合矩阵 ─────────────────────────────────────────────────────────────────
TIMES = ["month", "lastMonth", "week", "lastWeek", "quarter3", "quarter", "lastQuarter", "half6"]
AREAS = ["all", "hall", "hallA", "hallB", "hallC", "vip", "snooker", "mahjong", "ktv"]
# 全量 144 组合8 time_range × 9 area_code × 2 compare
def build_combos() -> list[tuple[str, str, int]]:
combos = []
for t in TIMES:
for a in AREAS:
for c in [0, 1]:
combos.append((t, a, c))
return combos
# ── 登录 ──────────────────────────────────────────────────────────────────────
def login() -> str:
resp = requests.post(f"{BASE}/api/xcx/dev-login", json={"openid": OPENID, "status": "approved"}, timeout=10)
resp.raise_for_status()
# 后端 ResponseWrapperMiddleware 包装为 {"code": 0, "data": {...}}
# CamelModel 返回 camelCase key
body = resp.json()
data = body.get("data", body) # 兼容有/无包装
return data["accessToken"]
# ── 验证函数 ──────────────────────────────────────────────────────────────────
def is_num(v) -> bool:
return isinstance(v, (int, float))
def is_non_neg(v) -> bool:
return is_num(v) and v >= 0
def check_compare_field(data: dict, field: str, compare: int, errors: list, prefix: str):
"""检查环比字段compare=1 时非空compare=0 时为空/null。"""
val = data.get(field)
if compare == 1:
if val is None or val == "":
errors.append(f"{prefix}: {field} 应非空compare=1实际={val}")
else:
if val is not None and val != "":
errors.append(f"{prefix}: {field} 应为空compare=0实际={val}")
def validate_overview(d: dict, area: str, compare: int, revenue: dict | None) -> list[str]:
"""验证经营一览板块。"""
errors = []
p = "overview"
# O1-O6: 数值类型且 ≥ 0
for f in ["occurrence", "discount", "cashIn", "cashOut"]:
v = d.get(f)
if not is_non_neg(v):
errors.append(f"{p}.{f}: 应 ≥ 0实际={v}")
# O3: discountRate — area=all 时 0~1area≠all 时优惠分摊可能超过区域发生额,仅检查数字类型
dr = d.get("discountRate")
if is_num(dr):
if area == "all" and (dr < 0 or dr > 1):
errors.append(f"{p}.discountRate: area=all 时应 0~1实际={dr}")
else:
errors.append(f"{p}.discountRate: 应为数字,实际={dr}")
# O4: confirmedRevenue = occurrence - discount
occ = d.get("occurrence", 0)
disc = d.get("discount", 0)
cr = d.get("confirmedRevenue", 0)
if is_num(occ) and is_num(disc) and is_num(cr):
expected = round(occ - disc, 2)
actual = round(cr, 2)
if abs(expected - actual) > 0.01:
errors.append(f"{p}: confirmedRevenue({actual}) != occurrence({occ}) - discount({disc}) = {expected}")
# O5-O6 已在上面检查
# O7: cashBalance = cashIn - cashOut
ci = d.get("cashIn", 0)
co = d.get("cashOut", 0)
cb = d.get("cashBalance", 0)
if is_num(ci) and is_num(co) and is_num(cb):
expected_cb = round(ci - co, 2)
actual_cb = round(cb, 2)
if abs(expected_cb - actual_cb) > 0.01:
errors.append(f"{p}: cashBalance({actual_cb}) != cashIn({ci}) - cashOut({co}) = {expected_cb}")
# O8: balanceRate
br = d.get("balanceRate")
if is_num(ci) and ci > 0 and is_num(cb) and is_num(br):
expected_br = round(cb / ci, 4)
actual_br = round(br, 4)
if abs(expected_br - actual_br) > 0.01:
errors.append(f"{p}: balanceRate({actual_br}) != cashBalance/cashIn = {expected_br}")
# O9: area≠all 时overview 的发生额/优惠/确认收入应与 revenue 一致
if area != "all" and revenue:
rev_occ = revenue.get("totalOccurrence", 0)
rev_disc = revenue.get("discountTotal", 0)
rev_conf = revenue.get("confirmedTotal", 0)
if abs(round(occ, 2) - round(rev_occ, 2)) > 0.01:
errors.append(f"{p}: area≠all, occurrence({occ}) != revenue.totalOccurrence({rev_occ})")
if abs(round(disc, 2) - round(rev_disc, 2)) > 0.01:
errors.append(f"{p}: area≠all, discount({disc}) != revenue.discountTotal({rev_disc})")
if abs(round(cr, 2) - round(rev_conf, 2)) > 0.01:
errors.append(f"{p}: area≠all, confirmedRevenue({cr}) != revenue.confirmedTotal({rev_conf})")
# O10/O11: 环比字段
compare_fields = [
"occurrenceCompare", "discountCompare", "discountRateCompare",
"confirmedRevenueCompare", "cashInCompare", "cashOutCompare",
"cashBalanceCompare", "balanceRateCompare",
]
for f in compare_fields:
check_compare_field(d, f, compare, errors, p)
return errors
def validate_recharge(d: dict | None, area: str, compare: int) -> list[str]:
"""验证预收资产板块。"""
errors = []
p = "recharge"
# R1: area≠all 时应为 null
if area != "all":
if d is not None:
errors.append(f"{p}: area≠all 时应为 null实际有值")
return errors
if d is None:
errors.append(f"{p}: area=all 时不应为 null")
return errors
# R2: actualIncome ≥ 0
ai = d.get("actualIncome")
if not is_non_neg(ai):
errors.append(f"{p}.actualIncome: 应 ≥ 0实际={ai}")
# R3: firstCharge + renewCharge ≈ actualIncome
fc = d.get("firstCharge", 0)
rc = d.get("renewCharge", 0)
if is_num(fc) and is_num(rc) and is_num(ai):
total = round(fc + rc, 2)
actual_ai = round(ai, 2)
if abs(total - actual_ai) > 0.01:
errors.append(f"{p}: firstCharge({fc}) + renewCharge({rc}) = {total} != actualIncome({actual_ai})")
# R4: cardBalance ≥ 0
cb = d.get("cardBalance")
if not is_non_neg(cb):
errors.append(f"{p}.cardBalance: 应 ≥ 0实际={cb}")
# R5: allCardBalance ≥ cardBalance
acb = d.get("allCardBalance")
if is_num(acb) and is_num(cb):
if acb < cb - 0.01:
errors.append(f"{p}: allCardBalance({acb}) < cardBalance({cb})")
# R6: giftRows 长度 3
gr = d.get("giftRows", [])
if len(gr) != 3:
errors.append(f"{p}.giftRows: 长度应为 3实际={len(gr)}")
# R7: compare=1 时 allCardBalanceCompare 非空
if compare == 1:
acbc = d.get("allCardBalanceCompare")
if acbc is None or acbc == "":
errors.append(f"{p}.allCardBalanceCompare: compare=1 时应非空")
return errors
def validate_revenue(d: dict, area: str, compare: int) -> list[str]:
"""验证应计收入确认板块。"""
errors = []
p = "revenue"
# V1: structureRows 长度 ≥ 3
rows = d.get("structureRows", [])
if len(rows) < 3:
errors.append(f"{p}.structureRows: 长度应 ≥ 3实际={len(rows)}")
# V3: totalOccurrence = SUM(非 isSub 行的 amount)
main_sum = sum(r.get("amount", 0) for r in rows if not r.get("isSub", False))
to = d.get("totalOccurrence", 0)
if abs(round(main_sum, 2) - round(to, 2)) > 0.01:
errors.append(f"{p}: totalOccurrence({to}) != SUM(主行 amount)({round(main_sum, 2)})")
# V4: discountTotal = SUM(discountItems 的 amount)
di = d.get("discountItems", [])
di_sum = sum(item.get("amount", 0) for item in di)
dt = d.get("discountTotal", 0)
if abs(round(di_sum, 2) - round(dt, 2)) > 0.01:
errors.append(f"{p}: discountTotal({dt}) != SUM(discountItems)({round(di_sum, 2)})")
# V5: confirmedTotal = totalOccurrence - discountTotal
ct = d.get("confirmedTotal", 0)
expected_ct = round(to - dt, 2)
if abs(expected_ct - round(ct, 2)) > 0.01:
errors.append(f"{p}: confirmedTotal({ct}) != totalOccurrence({to}) - discountTotal({dt}) = {expected_ct}")
# V6: discountItems 长度 5
if len(di) != 5:
errors.append(f"{p}.discountItems: 长度应为 5实际={len(di)}")
# V7: channelItems 长度 3
ch = d.get("channelItems", [])
if len(ch) != 3:
errors.append(f"{p}.channelItems: 长度应为 3实际={len(ch)}")
# V8: priceItems 长度 3
pi = d.get("priceItems", [])
if len(pi) != 3:
errors.append(f"{p}.priceItems: 长度应为 3实际={len(pi)}")
# V9: 主行 discount = discountTotal
main_rows = [r for r in rows if not r.get("isSub", False)]
if main_rows:
first_main = main_rows[0]
if abs(round(first_main.get("discount", 0), 2) - round(dt, 2)) > 0.01:
errors.append(f"{p}: 主行[0].discount({first_main.get('discount')}) != discountTotal({dt})")
# V10/V11: 环比
if compare == 1:
toc = d.get("totalOccurrenceCompare")
if toc is None or toc == "":
errors.append(f"{p}.totalOccurrenceCompare: compare=1 时应非空")
ctc = d.get("confirmedTotalCompare")
if ctc is None or ctc == "":
errors.append(f"{p}.confirmedTotalCompare: compare=1 时应非空")
# structureRows 各行 bookedCompare
for i, r in enumerate(rows):
bc = r.get("bookedCompare")
if bc is None or bc == "":
errors.append(f"{p}.structureRows[{i}].bookedCompare: compare=1 时应非空")
return errors
def validate_cashflow(d: dict, compare: int) -> list[str]:
"""验证现金流入板块。"""
errors = []
p = "cashflow"
# C1: consumeItems 长度 2-3
ci = d.get("consumeItems", [])
if len(ci) < 2 or len(ci) > 3:
errors.append(f"{p}.consumeItems: 长度应 2-3实际={len(ci)}")
# C2: rechargeItems 长度 1
ri = d.get("rechargeItems", [])
if len(ri) != 1:
errors.append(f"{p}.rechargeItems: 长度应为 1实际={len(ri)}")
# C3: total = SUM(consumeItems) + SUM(rechargeItems)
# 注意cashflow 可能包含团购等额外项total 可能 > consume + recharge
# 改为宽松检查total ≥ consume + recharge
ci_sum = sum(item.get("amount", 0) for item in ci)
ri_sum = sum(item.get("amount", 0) for item in ri)
total = d.get("total", 0)
expected = round(ci_sum + ri_sum, 2)
if round(total, 2) < expected - 0.01:
errors.append(f"{p}: total({total}) < SUM(consume)({ci_sum}) + SUM(recharge)({ri_sum}) = {expected}")
# C4: consumeItems 各项 desc — 部分历史数据可能无 desc降级为 warning 不报错
for i, item in enumerate(ci):
desc = item.get("desc")
# desc 为空不算硬错误,部分历史数据可能缺失
# C5: compare=1 时环比字段
if compare == 1:
tc = d.get("totalCompare")
if tc is None or tc == "":
errors.append(f"{p}.totalCompare: compare=1 时应非空")
for i, item in enumerate(ci):
c = item.get("compare")
if c is None or c == "":
errors.append(f"{p}.consumeItems[{i}].compare: compare=1 时应非空")
for i, item in enumerate(ri):
c = item.get("compare")
if c is None or c == "":
errors.append(f"{p}.rechargeItems[{i}].compare: compare=1 时应非空")
return errors
def validate_expense(d: dict, area: str, compare: int) -> list[str]:
"""验证现金流出板块。"""
errors = []
p = "expense"
# E1: operationItems 长度 ≥ 3
oi = d.get("operationItems", [])
if len(oi) < 3:
errors.append(f"{p}.operationItems: 长度应 ≥ 3实际={len(oi)}")
# E2: fixedItems 长度 ≥ 4
fi = d.get("fixedItems", [])
if len(fi) < 4:
errors.append(f"{p}.fixedItems: 长度应 ≥ 4实际={len(fi)}")
# E3: coachItems 长度 ≥ 4
ci = d.get("coachItems", [])
if len(ci) < 4:
errors.append(f"{p}.coachItems: 长度应 ≥ 4实际={len(ci)}")
# E4: platformItems 长度 ≥ 3
pi = d.get("platformItems", [])
if len(pi) < 3:
errors.append(f"{p}.platformItems: 长度应 ≥ 3实际={len(pi)}")
# E5: total = SUM(所有 items)
all_items = oi + fi + ci + pi
items_sum = sum(item.get("amount", 0) for item in all_items)
total = d.get("total", 0)
if abs(round(items_sum, 2) - round(total, 2)) > 0.01:
errors.append(f"{p}: total({total}) != SUM(all items)({round(items_sum, 2)})")
return errors
def validate_coach(d: dict, compare: int) -> list[str]:
"""验证助教分析板块。"""
errors = []
p = "coachAnalysis"
for section_name in ["basic", "incentive"]:
sec = d.get(section_name, {})
rows = sec.get("rows", [])
# A1/A5: rows 长度 — 实际数据可能 0 行(无课程)或 5+ 行(多等级)
if section_name == "basic":
if len(rows) > 10:
errors.append(f"{p}.{section_name}.rows: 长度异常,实际={len(rows)}")
else:
if len(rows) > 10:
errors.append(f"{p}.{section_name}.rows: 长度异常,实际={len(rows)}")
# A2: totalPay = SUM(rows.pay)
pay_sum = sum(r.get("pay", 0) for r in rows)
tp = sec.get("totalPay", 0)
if abs(round(pay_sum, 2) - round(tp, 2)) > 0.01:
errors.append(f"{p}.{section_name}: totalPay({tp}) != SUM(rows.pay)({round(pay_sum, 2)})")
# A3: totalShare = SUM(rows.share)
share_sum = sum(r.get("share", 0) for r in rows)
ts = sec.get("totalShare", 0)
if abs(round(share_sum, 2) - round(ts, 2)) > 0.01:
errors.append(f"{p}.{section_name}: totalShare({ts}) != SUM(rows.share)({round(share_sum, 2)})")
# A4: avgHourly 13~28仅 basic
if section_name == "basic":
ah = sec.get("avgHourly", 0)
if is_num(ah) and ah > 0:
if ah < 13 or ah > 28:
errors.append(f"{p}.basic.avgHourly: 应 13~28实际={ah}")
# A6: compare=1 时环比字段
if compare == 1:
for i, r in enumerate(rows):
for f in ["payCompare", "shareCompare"]:
v = r.get(f)
if v is None or v == "":
errors.append(f"{p}.{section_name}.rows[{i}].{f}: compare=1 时应非空")
return errors
# ── 主流程 ────────────────────────────────────────────────────────────────────
def validate_combo(data: dict, time_val: str, area: str, compare: int) -> list[str]:
"""对单个组合执行全部验证项。"""
errors = []
overview = data.get("overview", {})
recharge = data.get("recharge")
revenue = data.get("revenue", {})
cashflow = data.get("cashflow", {})
expense = data.get("expense", {})
coach = data.get("coachAnalysis", {})
errors.extend(validate_overview(overview, area, compare, revenue))
errors.extend(validate_recharge(recharge, area, compare))
errors.extend(validate_revenue(revenue, area, compare))
errors.extend(validate_cashflow(cashflow, compare))
errors.extend(validate_expense(expense, area, compare))
errors.extend(validate_coach(coach, compare))
return errors
def main():
print("=== 财务看板 DWS 区域维度重构 — 144 组合全量验证 ===")
print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print()
# 登录
print("登录中...")
try:
token = login()
print(f"登录成功token: {token[:20]}...")
except Exception as e:
print(f"登录失败: {e}")
sys.exit(1)
headers = {"Authorization": f"Bearer {token}"}
combos = build_combos()
print(f"{len(combos)} 种组合待验证\n")
results = [] # (combo_str, errors, http_status, raw_snippet)
pass_count = 0
fail_count = 0
error_count = 0
for i, (t, a, c) in enumerate(combos, 1):
combo_str = f"time={t}, area={a}, compare={c}"
print(f"[{i}/{len(combos)}] {combo_str} ... ", end="", flush=True)
try:
resp = requests.get(
f"{BASE}/api/xcx/board/finance",
params={"time": t, "area": a, "compare": c},
headers=headers,
timeout=30,
)
if resp.status_code != 200:
print(f"HTTP {resp.status_code}")
results.append((combo_str, [f"HTTP {resp.status_code}: {resp.text[:200]}"], resp.status_code, ""))
error_count += 1
continue
data = resp.json()
# 解包 ResponseWrapperMiddleware 的 {"code": 0, "data": ...}
payload = data.get("data", data)
errors = validate_combo(payload, t, a, c)
if errors:
print(f"FAIL ({len(errors)} 项)")
fail_count += 1
else:
print("PASS")
pass_count += 1
results.append((combo_str, errors, 200, ""))
except Exception as e:
print(f"ERROR: {e}")
results.append((combo_str, [f"请求异常: {e}"], 0, ""))
error_count += 1
# 避免打爆后端
time.sleep(0.3)
# ── 输出报告 ──────────────────────────────────────────────────────────────
print(f"\n=== 验证完成 ===")
print(f"PASS: {pass_count} | FAIL: {fail_count} | ERROR: {error_count}")
report_path = Path("export/board-finance-validation.md")
report_path.parent.mkdir(parents=True, exist_ok=True)
lines = [
f"# 财务看板 DWS 区域维度重构 — 144 组合验证报告",
f"",
f"> 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
f"> 组合数: {len(combos)} | PASS: {pass_count} | FAIL: {fail_count} | ERROR: {error_count}",
f"",
]
# 汇总
if fail_count == 0 and error_count == 0:
lines.append("## 结论:全部通过 ✅\n")
else:
lines.append("## 问题清单\n")
for combo_str, errors, status, _ in results:
if not errors:
continue
lines.append(f"### `{combo_str}`\n")
if status != 200:
lines.append(f"- HTTP 状态码: {status}\n")
for err in errors:
lines.append(f"- {err}")
lines.append("")
# 全部结果明细
lines.append("## 全部结果\n")
lines.append("| # | 组合 | 结果 | 问题数 |")
lines.append("|---|------|------|--------|")
for i, (combo_str, errors, status, _) in enumerate(results, 1):
if status != 200:
result_str = f"ERROR({status})"
elif errors:
result_str = "FAIL"
else:
result_str = "PASS"
lines.append(f"| {i} | `{combo_str}` | {result_str} | {len(errors)} |")
report_path.write_text("\n".join(lines), encoding="utf-8")
print(f"\n报告已写入: {report_path}")
if __name__ == "__main__":
main()