618 lines
26 KiB
Python
618 lines
26 KiB
Python
"""
|
||
口径验证脚本：从真实订单案例出发，交叉验证 DWD 层金额/时长/绩效字段。
|
||
|
||
策略：
|
||
1. 自动挑选 5 类有代表性的订单样本
|
||
2. 对每个订单，拉取所有关联子表的实际数值
|
||
3. 验证金额等式是否成立，标注差异
|
||
4. 输出 Markdown 报告到 SYSTEM_LOG_ROOT
|
||
"""
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
from pathlib import Path
|
||
|
||
import psycopg2
|
||
import psycopg2.extras
|
||
from dotenv import load_dotenv
|
||
|
||
# ── 环境 ──
|
||
load_dotenv(Path(__file__).resolve().parents[2] / ".env")
|
||
DSN = os.environ.get("TEST_DB_DSN") or os.environ.get("PG_DSN")
|
||
if not DSN:
|
||
raise RuntimeError("TEST_DB_DSN / PG_DSN 未配置")
|
||
|
||
OUTPUT_DIR = Path(os.environ.get("SYSTEM_LOG_ROOT", ""))
|
||
if not OUTPUT_DIR.is_dir():
|
||
raise RuntimeError(f"SYSTEM_LOG_ROOT 不存在: {OUTPUT_DIR}")
|
||
|
||
D2 = lambda v: Decimal(str(v or 0)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
||
NOW_STR = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
TODAY = datetime.now().strftime("%Y-%m-%d")
|
||
|
||
|
||
def q(cur, sql, params=None):
|
||
"""执行查询，返回 dict 列表"""
|
||
cur.execute(sql, params)
|
||
if cur.description is None:
|
||
return []
|
||
cols = [d[0] for d in cur.description]
|
||
return [dict(zip(cols, row)) for row in cur.fetchall()]
|
||
|
||
|
||
def fmt(v):
|
||
"""格式化值用于 Markdown 输出"""
|
||
if v is None:
|
||
return "NULL"
|
||
if isinstance(v, Decimal):
|
||
return f"{v:,.2f}"
|
||
if isinstance(v, (int,)):
|
||
return f"{v:,}"
|
||
return str(v)
|
||
|
||
|
||
def secs_to_hm(s):
|
||
"""秒数转 'Xh Ym' 格式"""
|
||
if s is None:
|
||
return "NULL"
|
||
h, m = divmod(int(s), 3600)
|
||
m = m // 60
|
||
return f"{h}h {m}m" if h else f"{m}m"
|
||
|
||
|
||
# ── 第一步：挑选样本订单 ──
|
||
SAMPLE_QUERIES = {
|
||
"A_纯台费": """
|
||
SELECT sh.order_settle_id
|
||
FROM dwd.dwd_settlement_head sh
|
||
WHERE sh.table_charge_money > 0
|
||
AND sh.goods_money = 0
|
||
AND sh.assistant_pd_money = 0
|
||
AND sh.assistant_cx_money = 0
|
||
AND sh.settle_type = 1
|
||
ORDER BY sh.pay_time DESC NULLS LAST
|
||
LIMIT 1
|
||
""",
|
||
"B_台费加商品": """
|
||
SELECT sh.order_settle_id
|
||
FROM dwd.dwd_settlement_head sh
|
||
WHERE sh.table_charge_money > 0
|
||
AND sh.goods_money > 0
|
||
AND sh.assistant_pd_money = 0
|
||
AND sh.settle_type = 1
|
||
ORDER BY sh.pay_time DESC NULLS LAST
|
||
LIMIT 1
|
||
""",
|
||
"C_台费加助教": """
|
||
SELECT sh.order_settle_id
|
||
FROM dwd.dwd_settlement_head sh
|
||
WHERE sh.table_charge_money > 0
|
||
AND sh.assistant_pd_money > 0
|
||
AND sh.goods_money = 0
|
||
AND sh.settle_type = 1
|
||
ORDER BY sh.pay_time DESC NULLS LAST
|
||
LIMIT 1
|
||
""",
|
||
"D_全要素": """
|
||
SELECT sh.order_settle_id
|
||
FROM dwd.dwd_settlement_head sh
|
||
WHERE sh.table_charge_money > 0
|
||
AND sh.goods_money > 0
|
||
AND sh.assistant_pd_money > 0
|
||
AND sh.settle_type = 1
|
||
ORDER BY sh.pay_time DESC NULLS LAST
|
||
LIMIT 1
|
||
""",
|
||
"E_含退款": """
|
||
SELECT DISTINCT r.relate_id AS order_settle_id
|
||
FROM dwd.dwd_refund r
|
||
ORDER BY r.relate_id DESC
|
||
LIMIT 1
|
||
""",
|
||
}
|
||
|
||
|
||
# ── 第二步：拉取各子表数据 ──
|
||
SUB_TABLE_QUERIES = {
|
||
"settlement_head": """
|
||
SELECT * FROM dwd.dwd_settlement_head WHERE order_settle_id = %s
|
||
""",
|
||
"settlement_head_ex": """
|
||
SELECT * FROM dwd.dwd_settlement_head_ex WHERE order_settle_id = %s
|
||
""",
|
||
"table_fee_log": """
|
||
SELECT * FROM dwd.dwd_table_fee_log WHERE order_settle_id = %s ORDER BY table_fee_log_id
|
||
""",
|
||
"table_fee_log_ex": """
|
||
SELECT ex.* FROM dwd.dwd_table_fee_log_ex ex
|
||
JOIN dwd.dwd_table_fee_log t ON t.table_fee_log_id = ex.table_fee_log_id
|
||
WHERE t.order_settle_id = %s ORDER BY ex.table_fee_log_id
|
||
""",
|
||
"assistant_service_log": """
|
||
SELECT * FROM dwd.dwd_assistant_service_log WHERE order_settle_id = %s ORDER BY assistant_service_id
|
||
""",
|
||
"assistant_service_log_ex": """
|
||
SELECT ex.* FROM dwd.dwd_assistant_service_log_ex ex
|
||
JOIN dwd.dwd_assistant_service_log a ON a.assistant_service_id = ex.assistant_service_id
|
||
WHERE a.order_settle_id = %s ORDER BY ex.assistant_service_id
|
||
""",
|
||
"store_goods_sale": """
|
||
SELECT * FROM dwd.dwd_store_goods_sale WHERE order_settle_id = %s ORDER BY store_goods_sale_id
|
||
""",
|
||
"store_goods_sale_ex": """
|
||
SELECT ex.* FROM dwd.dwd_store_goods_sale_ex ex
|
||
JOIN dwd.dwd_store_goods_sale g ON g.store_goods_sale_id = ex.store_goods_sale_id
|
||
WHERE g.order_settle_id = %s ORDER BY ex.store_goods_sale_id
|
||
""",
|
||
"payment": """
|
||
SELECT * FROM dwd.dwd_payment WHERE relate_id = %s ORDER BY payment_id
|
||
""",
|
||
"refund": """
|
||
SELECT * FROM dwd.dwd_refund WHERE relate_id = %s ORDER BY refund_id
|
||
""",
|
||
"refund_ex": """
|
||
SELECT ex.* FROM dwd.dwd_refund_ex ex
|
||
JOIN dwd.dwd_refund r ON r.refund_id = ex.refund_id
|
||
WHERE r.relate_id = %s ORDER BY ex.refund_id
|
||
""",
|
||
"table_fee_adjust": """
|
||
SELECT * FROM dwd.dwd_table_fee_adjust WHERE order_settle_id = %s ORDER BY table_fee_adjust_id
|
||
""",
|
||
"table_fee_adjust_ex": """
|
||
SELECT ex.* FROM dwd.dwd_table_fee_adjust_ex ex
|
||
JOIN dwd.dwd_table_fee_adjust a ON a.table_fee_adjust_id = ex.table_fee_adjust_id
|
||
WHERE a.order_settle_id = %s ORDER BY ex.table_fee_adjust_id
|
||
""",
|
||
"groupbuy_redemption": """
|
||
SELECT * FROM dwd.dwd_groupbuy_redemption WHERE order_settle_id = %s ORDER BY redemption_id
|
||
""",
|
||
"member_balance_change": """
|
||
SELECT * FROM dwd.dwd_member_balance_change
|
||
WHERE balance_change_id IN (
|
||
SELECT p.payment_id FROM dwd.dwd_payment p WHERE p.relate_id = %s
|
||
)
|
||
ORDER BY balance_change_id
|
||
""",
|
||
}
|
||
|
||
|
||
# ── 第三步：等式验证逻辑 ──
|
||
def verify_settlement_equation(sh, sh_ex, tfls, asls, gss, pays, refunds, adjusts):
|
||
"""验证结算单级金额等式，返回 (检查项列表, 通过数, 失败数)"""
|
||
checks = []
|
||
|
||
# 等式1: consume_money = table_charge_money + goods_money + assistant_pd_money + assistant_cx_money
|
||
lhs = D2(sh.get("consume_money"))
|
||
rhs = D2(sh.get("table_charge_money")) + D2(sh.get("goods_money")) + \
|
||
D2(sh.get("assistant_pd_money")) + D2(sh.get("assistant_cx_money"))
|
||
diff = lhs - rhs
|
||
checks.append({
|
||
"name": "消费总额 = 台费 + 商品 + 助教陪打 + 助教促销",
|
||
"lhs": f"consume_money = {fmt(lhs)}",
|
||
"rhs": f"table_charge + goods + asst_pd + asst_cx = {fmt(D2(sh.get('table_charge_money')))} + {fmt(D2(sh.get('goods_money')))} + {fmt(D2(sh.get('assistant_pd_money')))} + {fmt(D2(sh.get('assistant_cx_money')))} = {fmt(rhs)}",
|
||
"diff": diff,
|
||
"pass": diff == 0,
|
||
})
|
||
|
||
# 等式2: 台费明细合计 vs settlement_head.table_charge_money
|
||
tfl_sum = sum(D2(t.get("ledger_amount")) for t in tfls)
|
||
sh_table = D2(sh.get("table_charge_money"))
|
||
diff2 = sh_table - tfl_sum
|
||
checks.append({
|
||
"name": "结算单台费 vs 台费明细合计(ledger_amount)",
|
||
"lhs": f"sh.table_charge_money = {fmt(sh_table)}",
|
||
"rhs": f"SUM(tfl.ledger_amount) = {fmt(tfl_sum)} ({len(tfls)} 条)",
|
||
"diff": diff2,
|
||
"pass": diff2 == 0,
|
||
})
|
||
|
||
# 等式3: 助教明细合计 vs settlement_head.assistant_pd_money + assistant_cx_money
|
||
asl_pd = sum(D2(a.get("ledger_amount")) for a in asls if a.get("order_assistant_type") == 1)
|
||
asl_cx = sum(D2(a.get("ledger_amount")) for a in asls if a.get("order_assistant_type") == 2)
|
||
asl_null = sum(D2(a.get("ledger_amount")) for a in asls if a.get("order_assistant_type") is None)
|
||
sh_pd = D2(sh.get("assistant_pd_money"))
|
||
sh_cx = D2(sh.get("assistant_cx_money"))
|
||
diff3_pd = sh_pd - asl_pd
|
||
diff3_cx = sh_cx - asl_cx
|
||
checks.append({
|
||
"name": "结算单助教陪打 vs 助教明细(type=1)合计",
|
||
"lhs": f"sh.assistant_pd_money = {fmt(sh_pd)}",
|
||
"rhs": f"SUM(asl.ledger_amount WHERE type=1) = {fmt(asl_pd)} ({sum(1 for a in asls if a.get('order_assistant_type')==1)} 条)" + (f", type=NULL: {fmt(asl_null)} ({sum(1 for a in asls if a.get('order_assistant_type') is None)} 条)" if asl_null else ""),
|
||
"diff": diff3_pd,
|
||
"pass": diff3_pd == 0,
|
||
})
|
||
if sh_cx > 0 or asl_cx > 0:
|
||
checks.append({
|
||
"name": "结算单助教促销 vs 助教明细(type=2)合计",
|
||
"lhs": f"sh.assistant_cx_money = {fmt(sh_cx)}",
|
||
"rhs": f"SUM(asl.ledger_amount WHERE type=2) = {fmt(asl_cx)}",
|
||
"diff": diff3_cx,
|
||
"pass": diff3_cx == 0,
|
||
})
|
||
|
||
# 等式4: 商品明细合计 vs settlement_head.goods_money
|
||
gs_sum = sum(D2(g.get("ledger_amount")) for g in gss)
|
||
sh_goods = D2(sh.get("goods_money"))
|
||
diff4 = sh_goods - gs_sum
|
||
checks.append({
|
||
"name": "结算单商品 vs 商品明细合计(ledger_amount)",
|
||
"lhs": f"sh.goods_money = {fmt(sh_goods)}",
|
||
"rhs": f"SUM(gs.ledger_amount) = {fmt(gs_sum)} ({len(gss)} 条)",
|
||
"diff": diff4,
|
||
"pass": diff4 == 0,
|
||
})
|
||
|
||
# 等式5: 支付合计 vs settlement_head.pay_amount + balance_amount + gift_card_amount
|
||
pay_sum = sum(D2(p.get("pay_amount")) for p in pays)
|
||
sh_pay_total = D2(sh.get("pay_amount")) + D2(sh.get("balance_amount")) + D2(sh.get("gift_card_amount"))
|
||
diff5 = sh_pay_total - pay_sum
|
||
checks.append({
|
||
"name": "结算单(pay+balance+gift) vs 支付流水合计",
|
||
"lhs": f"sh.(pay+balance+gift) = {fmt(D2(sh.get('pay_amount')))} + {fmt(D2(sh.get('balance_amount')))} + {fmt(D2(sh.get('gift_card_amount')))} = {fmt(sh_pay_total)}",
|
||
"rhs": f"SUM(payment.pay_amount) = {fmt(pay_sum)} ({len(pays)} 笔)",
|
||
"diff": diff5,
|
||
"pass": diff5 == 0,
|
||
})
|
||
|
||
# 等式6: 台费调整合计 vs settlement_head.adjust_amount
|
||
adj_sum = sum(D2(a.get("ledger_amount")) for a in adjusts)
|
||
sh_adj = D2(sh.get("adjust_amount"))
|
||
diff6 = sh_adj - adj_sum
|
||
checks.append({
|
||
"name": "结算单调整额 vs 台费调整明细合计",
|
||
"lhs": f"sh.adjust_amount = {fmt(sh_adj)}",
|
||
"rhs": f"SUM(adj.ledger_amount) = {fmt(adj_sum)} ({len(adjusts)} 条)",
|
||
"diff": diff6,
|
||
"pass": diff6 == 0,
|
||
})
|
||
|
||
passed = sum(1 for c in checks if c["pass"])
|
||
failed = len(checks) - passed
|
||
return checks, passed, failed
|
||
|
||
|
||
def verify_table_fee_equations(tfls):
|
||
"""验证台费明细级等式，返回检查列表"""
|
||
checks = []
|
||
for i, t in enumerate(tfls):
|
||
# ledger_amount ≈ ledger_unit_price × real_table_use_seconds / 3600
|
||
price = D2(t.get("ledger_unit_price"))
|
||
secs = int(t.get("real_table_use_seconds") or 0)
|
||
calc = D2(price * secs / 3600)
|
||
actual = D2(t.get("ledger_amount"))
|
||
diff = actual - calc
|
||
checks.append({
|
||
"name": f"台费#{i+1} ledger_amount vs 单价×时长/3600",
|
||
"lhs": f"ledger_amount = {fmt(actual)}",
|
||
"rhs": f"{fmt(price)} × {secs}s / 3600 = {fmt(calc)}",
|
||
"diff": diff,
|
||
"pass": abs(diff) <= Decimal("0.02"), # 允许 ±0.02 四舍五入误差
|
||
})
|
||
# real_table_charge_money = ledger_amount - adjust_amount - ...
|
||
real = D2(t.get("real_table_charge_money"))
|
||
adj = D2(t.get("adjust_amount"))
|
||
coup = D2(t.get("coupon_promotion_amount"))
|
||
mem = D2(t.get("member_discount_amount"))
|
||
act_disc = D2(t.get("activity_discount_amount"))
|
||
calc_real = actual - adj - coup - mem - act_disc
|
||
diff_r = real - calc_real
|
||
checks.append({
|
||
"name": f"台费#{i+1} real_charge vs ledger-adj-coup-mem-act",
|
||
"lhs": f"real_table_charge_money = {fmt(real)}",
|
||
"rhs": f"{fmt(actual)} - {fmt(adj)} - {fmt(coup)} - {fmt(mem)} - {fmt(act_disc)} = {fmt(calc_real)}",
|
||
"diff": diff_r,
|
||
"pass": abs(diff_r) <= Decimal("0.02"),
|
||
})
|
||
# ledger_count vs real_table_use_seconds
|
||
lc = int(t.get("ledger_count") or 0)
|
||
rts = int(t.get("real_table_use_seconds") or 0)
|
||
checks.append({
|
||
"name": f"台费#{i+1} ledger_count vs real_table_use_seconds",
|
||
"lhs": f"ledger_count = {lc} ({secs_to_hm(lc)})",
|
||
"rhs": f"real_table_use_seconds = {rts} ({secs_to_hm(rts)})",
|
||
"diff": lc - rts,
|
||
"pass": True, # 仅记录，不判定
|
||
"info_only": True,
|
||
})
|
||
return checks
|
||
|
||
|
||
def verify_assistant_equations(asls):
|
||
"""验证助教服务级等式"""
|
||
checks = []
|
||
for i, a in enumerate(asls):
|
||
price = D2(a.get("ledger_unit_price"))
|
||
inc_secs = int(a.get("income_seconds") or 0)
|
||
calc = D2(price * inc_secs / 3600)
|
||
actual = D2(a.get("ledger_amount"))
|
||
diff = actual - calc
|
||
checks.append({
|
||
"name": f"助教#{i+1} ledger_amount vs 单价×计费秒/3600",
|
||
"lhs": f"ledger_amount = {fmt(actual)}",
|
||
"rhs": f"{fmt(price)} × {inc_secs}s / 3600 = {fmt(calc)}",
|
||
"diff": diff,
|
||
"pass": abs(diff) <= Decimal("0.02"),
|
||
})
|
||
# income_seconds vs real_use_seconds
|
||
real_secs = int(a.get("real_use_seconds") or 0)
|
||
checks.append({
|
||
"name": f"助教#{i+1} income_seconds vs real_use_seconds",
|
||
"lhs": f"income_seconds = {inc_secs} ({secs_to_hm(inc_secs)})",
|
||
"rhs": f"real_use_seconds = {real_secs} ({secs_to_hm(real_secs)})",
|
||
"diff": inc_secs - real_secs,
|
||
"pass": True,
|
||
"info_only": True,
|
||
})
|
||
# projected_income / ledger_amount = 分成比例
|
||
proj = D2(a.get("projected_income"))
|
||
if actual > 0:
|
||
ratio = proj / actual
|
||
checks.append({
|
||
"name": f"助教#{i+1} 分成比例 projected_income/ledger_amount",
|
||
"lhs": f"projected_income = {fmt(proj)}",
|
||
"rhs": f"ledger_amount = {fmt(actual)}, 比例 = {ratio:.4f}",
|
||
"diff": Decimal("0"),
|
||
"pass": True,
|
||
"info_only": True,
|
||
})
|
||
return checks
|
||
|
||
|
||
def verify_goods_equations(gss):
|
||
"""验证商品销售级等式"""
|
||
checks = []
|
||
for i, g in enumerate(gss):
|
||
price = D2(g.get("ledger_unit_price"))
|
||
count = int(g.get("ledger_count") or 0)
|
||
calc = D2(price * count)
|
||
actual = D2(g.get("ledger_amount"))
|
||
diff = actual - calc
|
||
checks.append({
|
||
"name": f"商品#{i+1}({g.get('ledger_name','?')}) ledger_amount vs 单价×数量",
|
||
"lhs": f"ledger_amount = {fmt(actual)}",
|
||
"rhs": f"{fmt(price)} × {count} = {fmt(calc)}",
|
||
"diff": diff,
|
||
"pass": abs(diff) <= Decimal("0.02"),
|
||
})
|
||
real = D2(g.get("real_goods_money"))
|
||
disc = D2(g.get("discount_money"))
|
||
coup = D2(g.get("coupon_share_money"))
|
||
calc_real = actual - disc - coup
|
||
diff_r = real - calc_real
|
||
checks.append({
|
||
"name": f"商品#{i+1} real_goods_money vs ledger-disc-coup",
|
||
"lhs": f"real_goods_money = {fmt(real)}",
|
||
"rhs": f"{fmt(actual)} - {fmt(disc)} - {fmt(coup)} = {fmt(calc_real)}",
|
||
"diff": diff_r,
|
||
"pass": abs(diff_r) <= Decimal("0.02"),
|
||
})
|
||
return checks
|
||
|
||
|
||
# ── 第四步：数据新鲜度查询 ──
|
||
FRESHNESS_SQL = """
|
||
SELECT 'dwd_settlement_head' AS tbl, MAX(create_time) AS max_ct, MAX(pay_time) AS max_biz, COUNT(*) AS cnt FROM dwd.dwd_settlement_head
|
||
UNION ALL SELECT 'dwd_table_fee_log', MAX(create_time), MAX(start_use_time), COUNT(*) FROM dwd.dwd_table_fee_log
|
||
UNION ALL SELECT 'dwd_assistant_service_log', MAX(create_time), MAX(start_use_time), COUNT(*) FROM dwd.dwd_assistant_service_log
|
||
UNION ALL SELECT 'dwd_store_goods_sale', MAX(create_time), NULL, COUNT(*) FROM dwd.dwd_store_goods_sale
|
||
UNION ALL SELECT 'dwd_payment', MAX(create_time), MAX(pay_time), COUNT(*) FROM dwd.dwd_payment
|
||
UNION ALL SELECT 'dwd_refund', MAX(create_time), MAX(pay_time), COUNT(*) FROM dwd.dwd_refund
|
||
UNION ALL SELECT 'dwd_groupbuy_redemption', MAX(create_time), NULL, COUNT(*) FROM dwd.dwd_groupbuy_redemption
|
||
UNION ALL SELECT 'dwd_table_fee_adjust', MAX(adjust_time), NULL, COUNT(*) FROM dwd.dwd_table_fee_adjust
|
||
UNION ALL SELECT 'dwd_recharge_order', MAX(create_time), MAX(pay_time), COUNT(*) FROM dwd.dwd_recharge_order
|
||
UNION ALL SELECT 'dwd_member_balance_change', MAX(change_time), NULL, COUNT(*) FROM dwd.dwd_member_balance_change
|
||
ORDER BY tbl
|
||
"""
|
||
|
||
|
||
# ── 第五步：生成报告 ──
|
||
def render_checks_table(checks):
|
||
"""渲染检查结果为 Markdown 表格"""
|
||
lines = []
|
||
lines.append("| # | 检查项 | 左侧 | 右侧 | 差异 | 结果 |")
|
||
lines.append("|---|--------|------|------|------|------|")
|
||
for i, c in enumerate(checks, 1):
|
||
info = c.get("info_only", False)
|
||
if info:
|
||
result = "ℹ️ 信息"
|
||
elif c["pass"]:
|
||
result = "✅ 通过"
|
||
else:
|
||
result = "❌ 不等"
|
||
diff_str = fmt(c["diff"]) if not info else str(c["diff"])
|
||
lines.append(f"| {i} | {c['name']} | {c['lhs']} | {c['rhs']} | {diff_str} | {result} |")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def render_row_detail(label, rows, key_fields):
|
||
"""渲染子表数据明细"""
|
||
if not rows:
|
||
return f"\n**{label}**：无数据\n"
|
||
lines = [f"\n**{label}**（{len(rows)} 条）\n"]
|
||
# 只输出 key_fields 中存在的字段
|
||
cols = [f for f in key_fields if f in rows[0]]
|
||
lines.append("| " + " | ".join(cols) + " |")
|
||
lines.append("| " + " | ".join(["---"] * len(cols)) + " |")
|
||
for r in rows:
|
||
vals = [fmt(r.get(c)) for c in cols]
|
||
lines.append("| " + " | ".join(vals) + " |")
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ── 主流程 ──
|
||
def main():
|
||
conn = psycopg2.connect(DSN)
|
||
conn.set_session(readonly=True)
|
||
cur = conn.cursor()
|
||
|
||
report = []
|
||
report.append(f"# DWD 口径验证报告（订单案例分析）\n")
|
||
report.append(f"> 生成时间：{NOW_STR}")
|
||
report.append(f"> 数据来源：`test_etl_feiqiu` 测试库 `dwd` schema")
|
||
report.append(f"> 方法：从真实订单出发，拉取各子表实际数值，交叉验证金额等式\n")
|
||
|
||
# ── 数据新鲜度 ──
|
||
report.append("---\n\n## 零、数据新鲜度\n")
|
||
freshness = q(cur, FRESHNESS_SQL)
|
||
report.append("| 表 | 记录数 | 最晚 create_time | 最晚业务时间 |")
|
||
report.append("|---|--------|-----------------|-------------|")
|
||
for f in freshness:
|
||
ct = str(f["max_ct"])[:19] if f["max_ct"] else "—"
|
||
bt = str(f["max_biz"])[:19] if f["max_biz"] else "—"
|
||
report.append(f"| `{f['tbl']}` | {f['cnt']:,} | {ct} | {bt} |")
|
||
|
||
# ── 挑选样本 ──
|
||
samples = {}
|
||
for label, sql in SAMPLE_QUERIES.items():
|
||
rows = q(cur, sql)
|
||
if rows:
|
||
samples[label] = rows[0]["order_settle_id"]
|
||
print(f" 样本 {label}: order_settle_id = {rows[0]['order_settle_id']}")
|
||
else:
|
||
print(f" 样本 {label}: 未找到匹配订单")
|
||
|
||
total_pass = 0
|
||
total_fail = 0
|
||
|
||
for label, settle_id in samples.items():
|
||
report.append(f"\n---\n\n## 样本 {label}（order_settle_id = `{settle_id}`）\n")
|
||
|
||
# 拉取所有子表
|
||
data = {}
|
||
for tbl, sql in SUB_TABLE_QUERIES.items():
|
||
data[tbl] = q(cur, sql, (settle_id,))
|
||
|
||
sh = data["settlement_head"][0] if data["settlement_head"] else {}
|
||
sh_ex = data["settlement_head_ex"][0] if data["settlement_head_ex"] else {}
|
||
|
||
if not sh:
|
||
report.append("> ⚠️ 结算单主表无数据，跳过\n")
|
||
continue
|
||
|
||
# 概览
|
||
report.append(f"- 门店：`{sh.get('site_name', '?')}`")
|
||
report.append(f"- 支付时间：{str(sh.get('pay_time', '?'))[:19]}")
|
||
report.append(f"- 结算类型：{sh.get('settle_type')}")
|
||
report.append(f"- 会员：{sh.get('member_name', '无')} ({sh.get('member_phone', '')})")
|
||
report.append(f"- 消费总额：{fmt(sh.get('consume_money'))}")
|
||
report.append(f"- 台费明细：{len(data['table_fee_log'])} 条")
|
||
report.append(f"- 助教服务：{len(data['assistant_service_log'])} 条")
|
||
report.append(f"- 商品销售：{len(data['store_goods_sale'])} 条")
|
||
report.append(f"- 支付流水：{len(data['payment'])} 笔")
|
||
report.append(f"- 退款流水：{len(data['refund'])} 笔")
|
||
report.append(f"- 台费调整：{len(data['table_fee_adjust'])} 条")
|
||
report.append(f"- 团购核销：{len(data['groupbuy_redemption'])} 条\n")
|
||
|
||
# 结算单级金额全貌
|
||
report.append("### 结算单金额全貌\n")
|
||
amt_fields = [
|
||
("consume_money", "消费总额"), ("table_charge_money", "台费"),
|
||
("goods_money", "商品原价"), ("real_goods_money", "商品实收"),
|
||
("assistant_pd_money", "助教陪打"), ("assistant_cx_money", "助教促销"),
|
||
("adjust_amount", "调整额"), ("pay_amount", "实付"),
|
||
("balance_amount", "余额支付"), ("gift_card_amount", "赠送卡"),
|
||
("recharge_card_amount", "充值卡"), ("coupon_amount", "优惠券"),
|
||
("point_amount", "积分"), ("member_discount_amount", "会员折扣"),
|
||
("rounding_amount", "抹零"),
|
||
]
|
||
report.append("| 字段 | 含义 | 值 |")
|
||
report.append("|------|------|-----|")
|
||
for field, desc in amt_fields:
|
||
report.append(f"| `{field}` | {desc} | {fmt(sh.get(field))} |")
|
||
|
||
if sh_ex:
|
||
ex_fields = [
|
||
("service_money", "服务费"), ("cash_amount", "现金"),
|
||
("card_amount", "银行卡"), ("online_amount", "线上"),
|
||
("refund_amount", "退款"), ("prepay_money", "预付"),
|
||
]
|
||
report.append("\n**扩展表（_ex）**\n")
|
||
report.append("| 字段 | 含义 | 值 |")
|
||
report.append("|------|------|-----|")
|
||
for field, desc in ex_fields:
|
||
report.append(f"| `{field}` | {desc} | {fmt(sh_ex.get(field))} |")
|
||
|
||
# 子表明细
|
||
tfl_keys = ["table_fee_log_id", "ledger_name", "ledger_unit_price", "ledger_count",
|
||
"real_table_use_seconds", "ledger_amount", "real_table_charge_money",
|
||
"adjust_amount", "start_use_time", "ledger_end_time", "add_clock_seconds"]
|
||
report.append(render_row_detail("台费明细", data["table_fee_log"], tfl_keys))
|
||
|
||
asl_keys = ["assistant_service_id", "nickname", "skill_name", "order_assistant_type",
|
||
"level_name", "ledger_unit_price", "income_seconds", "real_use_seconds",
|
||
"ledger_amount", "projected_income", "add_clock", "start_use_time", "last_use_time"]
|
||
report.append(render_row_detail("助教服务明细", data["assistant_service_log"], asl_keys))
|
||
|
||
gs_keys = ["store_goods_sale_id", "ledger_name", "ledger_unit_price", "ledger_count",
|
||
"ledger_amount", "discount_money", "real_goods_money", "coupon_share_money", "cost_money"]
|
||
report.append(render_row_detail("商品销售明细", data["store_goods_sale"], gs_keys))
|
||
|
||
pay_keys = ["payment_id", "pay_amount", "payment_method", "pay_time"]
|
||
report.append(render_row_detail("支付流水", data["payment"], pay_keys))
|
||
|
||
ref_keys = ["refund_id", "pay_amount", "payment_method", "pay_time", "channel_fee"]
|
||
report.append(render_row_detail("退款流水", data["refund"], ref_keys))
|
||
|
||
adj_keys = ["table_fee_adjust_id", "ledger_amount", "table_name"]
|
||
report.append(render_row_detail("台费调整", data["table_fee_adjust"], adj_keys))
|
||
|
||
# 等式验证
|
||
report.append("\n### 等式验证\n")
|
||
report.append("#### 结算单级\n")
|
||
checks, p, f = verify_settlement_equation(
|
||
sh, sh_ex, data["table_fee_log"], data["assistant_service_log"],
|
||
data["store_goods_sale"], data["payment"], data["refund"], data["table_fee_adjust"]
|
||
)
|
||
report.append(render_checks_table(checks))
|
||
total_pass += p
|
||
total_fail += f
|
||
|
||
if data["table_fee_log"]:
|
||
report.append("\n#### 台费明细级\n")
|
||
tc = verify_table_fee_equations(data["table_fee_log"])
|
||
report.append(render_checks_table(tc))
|
||
total_pass += sum(1 for c in tc if c["pass"] and not c.get("info_only"))
|
||
total_fail += sum(1 for c in tc if not c["pass"] and not c.get("info_only"))
|
||
|
||
if data["assistant_service_log"]:
|
||
report.append("\n#### 助教服务级\n")
|
||
ac = verify_assistant_equations(data["assistant_service_log"])
|
||
report.append(render_checks_table(ac))
|
||
total_pass += sum(1 for c in ac if c["pass"] and not c.get("info_only"))
|
||
total_fail += sum(1 for c in ac if not c["pass"] and not c.get("info_only"))
|
||
|
||
if data["store_goods_sale"]:
|
||
report.append("\n#### 商品销售级\n")
|
||
gc = verify_goods_equations(data["store_goods_sale"])
|
||
report.append(render_checks_table(gc))
|
||
total_pass += sum(1 for c in gc if c["pass"] and not c.get("info_only"))
|
||
total_fail += sum(1 for c in gc if not c["pass"] and not c.get("info_only"))
|
||
|
||
# 汇总
|
||
report.append(f"\n---\n\n## 汇总\n")
|
||
report.append(f"- 样本数：{len(samples)}")
|
||
report.append(f"- 等式检查通过：{total_pass}")
|
||
report.append(f"- 等式检查失败：{total_fail}")
|
||
if total_fail > 0:
|
||
report.append(f"\n> ⚠️ 存在 {total_fail} 项等式不成立，需进一步排查口径定义。")
|
||
else:
|
||
report.append(f"\n> ✅ 所有等式验证通过。")
|
||
|
||
cur.close()
|
||
conn.close()
|
||
|
||
# 写入文件
|
||
out_path = OUTPUT_DIR / f"{TODAY}__calibration_order_analysis.md"
|
||
out_path.write_text("\n".join(report), encoding="utf-8")
|
||
print(f"\n报告已写入: {out_path}")
|
||
|
||
# 同时写入 docs/reports/ 作为文档归档
|
||
doc_path = Path(__file__).resolve().parents[2] / "docs" / "reports" / "dwd-calibration-order-analysis.md"
|
||
doc_path.write_text("\n".join(report), encoding="utf-8")
|
||
print(f"文档归档: {doc_path}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|