378 lines
16 KiB
Python
378 lines
16 KiB
Python
"""
|
||
会员消费报表对比：数据库 vs 飞球导出 CSV
|
||
对比范围：2025-12-09 ~ 2026-03-10
|
||
飞球 CSV：tmp/结账记录_朗朗桌球_20251209_20260310.csv
|
||
通过小票号（order_settle_id）反查数据库获取会员信息
|
||
"""
|
||
import os
|
||
import csv
|
||
from decimal import Decimal
|
||
from collections import defaultdict
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv()
|
||
|
||
PG_DSN = os.environ.get("TEST_DB_DSN") or os.environ.get("PG_DSN")
|
||
if not PG_DSN:
|
||
raise RuntimeError("TEST_DB_DSN / PG_DSN 未配置")
|
||
|
||
SYSTEM_ANALYZE_ROOT = os.environ.get("SYSTEM_ANALYZE_ROOT")
|
||
if not SYSTEM_ANALYZE_ROOT:
|
||
raise RuntimeError("SYSTEM_ANALYZE_ROOT 未配置")
|
||
|
||
import psycopg2
|
||
|
||
FEIQIU_CSV = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
|
||
"tmp", "结账记录_朗朗桌球_20251209_20260310.csv"
|
||
)
|
||
|
||
# ── SQL: 按小票号批量查数据库记录 + 会员信息 ──
|
||
SQL_BY_TICKET = r"""
|
||
SELECT
|
||
sh.order_settle_id,
|
||
sh.settle_type,
|
||
sh.pay_time,
|
||
COALESCE(sh.table_charge_money, 0) AS table_charge_money,
|
||
COALESCE(sh.goods_money, 0) AS goods_money,
|
||
COALESCE(sh.assistant_pd_money, 0) AS assistant_pd_money,
|
||
COALESCE(sh.assistant_cx_money, 0) AS assistant_cx_money,
|
||
COALESCE(sh.electricity_money, 0) AS electricity_money,
|
||
COALESCE(sh.table_charge_money, 0)
|
||
+ COALESCE(sh.goods_money, 0)
|
||
+ COALESCE(sh.assistant_pd_money, 0)
|
||
+ COALESCE(sh.assistant_cx_money, 0)
|
||
+ COALESCE(sh.electricity_money, 0) AS items_sum,
|
||
sh.consume_money,
|
||
sh.member_id,
|
||
COALESCE(NULLIF(dm.mobile, ''), NULLIF(sh.member_phone, '')) AS phone,
|
||
COALESCE(NULLIF(dm.nickname, ''), NULLIF(sh.member_name, '')) AS member_name
|
||
FROM dwd.dwd_settlement_head sh
|
||
LEFT JOIN dwd.dim_member dm
|
||
ON dm.member_id = sh.member_id AND dm.scd2_is_current = 1
|
||
WHERE sh.order_settle_id = ANY(%s)
|
||
"""
|
||
|
||
# ── SQL: 数据库全量汇总（同期） ──
|
||
SQL_DB_TOTAL = r"""
|
||
SELECT
|
||
COUNT(*) AS total_orders,
|
||
ROUND(SUM(COALESCE(table_charge_money,0) + COALESCE(goods_money,0)
|
||
+ COALESCE(assistant_pd_money,0) + COALESCE(assistant_cx_money,0)
|
||
+ COALESCE(electricity_money,0)), 2) AS total_items_sum,
|
||
ROUND(SUM(COALESCE(table_charge_money,0)), 2) AS total_table_charge,
|
||
ROUND(SUM(COALESCE(goods_money,0)), 2) AS total_goods,
|
||
ROUND(SUM(COALESCE(assistant_pd_money,0) + COALESCE(assistant_cx_money,0)), 2) AS total_assistant,
|
||
ROUND(SUM(COALESCE(consume_money,0)), 2) AS total_consume_money,
|
||
COUNT(CASE WHEN member_id IS NOT NULL AND member_id != 0 THEN 1 END) AS member_orders,
|
||
COUNT(CASE WHEN member_id IS NULL OR member_id = 0 THEN 1 END) AS non_member_orders
|
||
FROM dwd.dwd_settlement_head
|
||
WHERE settle_type IN (1, 3)
|
||
AND pay_time >= '2025-12-09'
|
||
AND pay_time < '2026-03-11'
|
||
"""
|
||
|
||
|
||
def d(val: str) -> Decimal:
|
||
"""安全转 Decimal"""
|
||
return Decimal(val.strip() if val and val.strip() else "0")
|
||
|
||
|
||
def parse_feiqiu_csv(path: str) -> dict:
|
||
"""解析飞球 CSV，返回 {小票号: row_dict} + 汇总"""
|
||
tickets = {}
|
||
totals = {
|
||
"orders": 0, "consume": Decimal("0"), "table_fee": Decimal("0"),
|
||
"goods": Decimal("0"), "service": Decimal("0"), "course": Decimal("0"),
|
||
"incentive": Decimal("0"), "bill_types": defaultdict(int),
|
||
}
|
||
|
||
with open(path, "r", encoding="utf-8-sig") as f:
|
||
reader = csv.DictReader(f)
|
||
for row in reader:
|
||
ticket = row.get("小票号", "").strip()
|
||
if not ticket or ticket == "-":
|
||
continue
|
||
totals["orders"] += 1
|
||
consume = d(row.get("消费金额", "0"))
|
||
table_fee = d(row.get("台费", "0"))
|
||
goods = d(row.get("商品费", "0"))
|
||
service = d(row.get("服务加收", "0"))
|
||
course = d(row.get("课程费", "0"))
|
||
incentive = d(row.get("激励费", "0"))
|
||
|
||
totals["consume"] += consume
|
||
totals["table_fee"] += table_fee
|
||
totals["goods"] += goods
|
||
totals["service"] += service
|
||
totals["course"] += course
|
||
totals["incentive"] += incentive
|
||
bt = row.get("账单类型", "未知")
|
||
totals["bill_types"][bt] += 1
|
||
|
||
tickets[ticket] = {
|
||
"consume": consume, "table_fee": table_fee,
|
||
"goods": goods, "service": service, "course": course,
|
||
"incentive": incentive, "bill_type": bt,
|
||
"table_no": row.get("台桌号", ""),
|
||
"pay_time": row.get("结账时间", ""),
|
||
}
|
||
|
||
return tickets, totals
|
||
|
||
|
||
def run_compare():
|
||
print("=" * 90, flush=True)
|
||
print("会员消费报表对比：数据库 vs 飞球导出", flush=True)
|
||
print("对比范围：2025-12-09 ~ 2026-03-10", flush=True)
|
||
print("=" * 90, flush=True)
|
||
|
||
# ── 1. 解析飞球 CSV ──
|
||
print("\n📄 解析飞球 CSV ...", flush=True)
|
||
if not os.path.exists(FEIQIU_CSV):
|
||
print(f" ❌ 文件不存在: {FEIQIU_CSV}")
|
||
return
|
||
fq_tickets, fq_totals = parse_feiqiu_csv(FEIQIU_CSV)
|
||
print(f" 订单数: {fq_totals['orders']}", flush=True)
|
||
print(f" 消费金额合计: {fq_totals['consume']}", flush=True)
|
||
print(f" 台费: {fq_totals['table_fee']}", flush=True)
|
||
print(f" 商品费: {fq_totals['goods']}", flush=True)
|
||
print(f" 服务加收(陪打): {fq_totals['service']}", flush=True)
|
||
print(f" 课程费(超休): {fq_totals['course']}", flush=True)
|
||
print(f" 激励费: {fq_totals['incentive']}", flush=True)
|
||
print(f" 助教合计: {fq_totals['service'] + fq_totals['course']}", flush=True)
|
||
print(f" 账单类型: {dict(fq_totals['bill_types'])}", flush=True)
|
||
|
||
# ── 2. 用小票号反查数据库 ──
|
||
print("\n🗄️ 用小票号反查数据库 ...", flush=True)
|
||
ticket_ids = [int(t) for t in fq_tickets.keys() if t.isdigit()]
|
||
print(f" 待查小票数: {len(ticket_ids)}", flush=True)
|
||
|
||
db_records = {}
|
||
with psycopg2.connect(PG_DSN, connect_timeout=15,
|
||
options="-c statement_timeout=300000") as conn:
|
||
with conn.cursor() as cur:
|
||
# 分批查（每批 500）
|
||
for i in range(0, len(ticket_ids), 500):
|
||
batch = ticket_ids[i:i+500]
|
||
cur.execute(SQL_BY_TICKET, (batch,))
|
||
for row in cur.fetchall():
|
||
db_records[str(row[0])] = {
|
||
"order_settle_id": row[0],
|
||
"settle_type": row[1],
|
||
"pay_time": row[2],
|
||
"table_charge": float(row[3]),
|
||
"goods": float(row[4]),
|
||
"assistant_pd": float(row[5]),
|
||
"assistant_cx": float(row[6]),
|
||
"electricity": float(row[7]),
|
||
"items_sum": float(row[8]),
|
||
"consume_money": float(row[9]) if row[9] else 0,
|
||
"member_id": row[10],
|
||
"phone": row[11],
|
||
"member_name": row[12],
|
||
}
|
||
|
||
# 全量汇总
|
||
cur.execute(SQL_DB_TOTAL)
|
||
db_total = cur.fetchone()
|
||
|
||
print(f" 数据库匹配到: {len(db_records)} 条", flush=True)
|
||
print(f" 飞球有但数据库无: {len(fq_tickets) - len(db_records)} 条", flush=True)
|
||
|
||
# ── 3. 数据库全量 ──
|
||
print(f"\n🗄️ 数据库全量（同期 settle_type IN 1,3）:", flush=True)
|
||
print(f" 订单数: {db_total[0]} (会员: {db_total[6]}, 非会员: {db_total[7]})", flush=True)
|
||
print(f" items_sum: {db_total[1]}", flush=True)
|
||
print(f" 台费: {db_total[2]}", flush=True)
|
||
print(f" 商品费: {db_total[3]}", flush=True)
|
||
print(f" 助教费: {db_total[4]}", flush=True)
|
||
print(f" consume_money: {db_total[5]}", flush=True)
|
||
|
||
# ── 4. 总量对比 ──
|
||
print("\n" + "=" * 90, flush=True)
|
||
print("📊 总量对比（飞球 CSV vs 数据库全量）", flush=True)
|
||
print("=" * 90, flush=True)
|
||
|
||
fq_assistant = float(fq_totals["service"] + fq_totals["course"])
|
||
comparisons = [
|
||
("订单数", float(fq_totals["orders"]), float(db_total[0])),
|
||
("消费金额 vs items_sum", float(fq_totals["consume"]), float(db_total[1])),
|
||
("消费金额 vs consume_money", float(fq_totals["consume"]), float(db_total[5])),
|
||
("台费", float(fq_totals["table_fee"]), float(db_total[2])),
|
||
("商品费", float(fq_totals["goods"]), float(db_total[3])),
|
||
("助教费(服务+课程)", fq_assistant, float(db_total[4])),
|
||
]
|
||
|
||
print(f"{'指标':<28} {'飞球CSV':>14} {'数据库':>14} {'差额':>14} {'差异%':>9}", flush=True)
|
||
print("-" * 90, flush=True)
|
||
for label, fq_val, db_val in comparisons:
|
||
diff = db_val - fq_val
|
||
pct = (diff / fq_val * 100) if fq_val else 0
|
||
print(f"{label:<28} {fq_val:>14,.2f} {db_val:>14,.2f} {diff:>+14,.2f} {pct:>+8.2f}%", flush=True)
|
||
|
||
|
||
# ── 5. 逐单对比：找差异订单 ──
|
||
print("\n" + "=" * 90, flush=True)
|
||
print("🔍 逐单对比（飞球消费金额 vs 数据库 items_sum）", flush=True)
|
||
print("=" * 90, flush=True)
|
||
|
||
# 提前定义输出目录
|
||
report_dir = os.path.join(os.path.dirname(SYSTEM_ANALYZE_ROOT), "member_reports")
|
||
os.makedirs(report_dir, exist_ok=True)
|
||
|
||
diffs = []
|
||
only_fq = []
|
||
only_db_settle_types = defaultdict(int) # 飞球有但数据库 settle_type 不同
|
||
|
||
for ticket, fq_row in fq_tickets.items():
|
||
db_row = db_records.get(ticket)
|
||
if not db_row:
|
||
only_fq.append(ticket)
|
||
continue
|
||
|
||
fq_consume = float(fq_row["consume"])
|
||
db_items = db_row["items_sum"]
|
||
diff = abs(db_items - fq_consume)
|
||
if diff > 0.02: # 容差 2 分钱
|
||
diffs.append({
|
||
"ticket": ticket,
|
||
"fq_pay_time": fq_row["pay_time"],
|
||
"db_pay_time": str(db_row["pay_time"]) if db_row["pay_time"] else "",
|
||
"fq_consume": fq_consume,
|
||
"db_items_sum": db_items,
|
||
"db_consume_money": db_row["consume_money"],
|
||
"diff": db_items - fq_consume,
|
||
"fq_table": float(fq_row["table_fee"]),
|
||
"db_table": db_row["table_charge"],
|
||
"fq_goods": float(fq_row["goods"]),
|
||
"db_goods": db_row["goods"],
|
||
"fq_assistant": float(fq_row["service"] + fq_row["course"]),
|
||
"db_assistant": db_row["assistant_pd"] + db_row["assistant_cx"],
|
||
"member_name": db_row["member_name"] or "",
|
||
"phone": db_row["phone"] or "",
|
||
"bill_type": fq_row["bill_type"],
|
||
"table_no": fq_row["table_no"],
|
||
"settle_type": db_row["settle_type"],
|
||
})
|
||
|
||
print(f" 总对比: {len(fq_tickets)} 单", flush=True)
|
||
print(f" 完全匹配（差额≤0.02）: {len(fq_tickets) - len(only_fq) - len(diffs)} 单", flush=True)
|
||
print(f" 有差异: {len(diffs)} 单", flush=True)
|
||
print(f" 飞球有但数据库无: {len(only_fq)} 单", flush=True)
|
||
|
||
if only_fq:
|
||
print(f"\n 飞球有但数据库无的小票号（前20）:", flush=True)
|
||
for t in only_fq[:20]:
|
||
fq_row = fq_tickets[t]
|
||
print(f" {t} 消费={fq_row['consume']} 类型={fq_row['bill_type']} 台桌={fq_row['table_no']}", flush=True)
|
||
|
||
if diffs:
|
||
# 导出完整差异明细 CSV
|
||
diff_csv_path = os.path.join(report_dir, "order_diff_detail_20251209.csv")
|
||
sorted_diffs = sorted(diffs, key=lambda x: abs(x["diff"]), reverse=True)
|
||
with open(diff_csv_path, "w", newline="", encoding="utf-8-sig") as f:
|
||
writer = csv.writer(f)
|
||
writer.writerow(["小票号", "飞球结账时间", "DB结账时间", "账单类型", "台桌",
|
||
"settle_type", "飞球消费", "DB_items_sum", "DB_consume_money",
|
||
"差额", "飞球台费", "DB台费", "飞球商品", "DB商品",
|
||
"飞球助教", "DB助教", "会员昵称", "手机号"])
|
||
for d_row in sorted_diffs:
|
||
writer.writerow([
|
||
d_row["ticket"], d_row["fq_pay_time"], d_row["db_pay_time"],
|
||
d_row["bill_type"], d_row["table_no"], d_row["settle_type"],
|
||
f"{d_row['fq_consume']:.2f}", f"{d_row['db_items_sum']:.2f}",
|
||
f"{d_row['db_consume_money']:.2f}", f"{d_row['diff']:+.2f}",
|
||
f"{d_row['fq_table']:.2f}", f"{d_row['db_table']:.2f}",
|
||
f"{d_row['fq_goods']:.2f}", f"{d_row['db_goods']:.2f}",
|
||
f"{d_row['fq_assistant']:.2f}", f"{d_row['db_assistant']:.2f}",
|
||
d_row["member_name"], d_row["phone"],
|
||
])
|
||
print(f"\n ✅ 差异明细已导出: {diff_csv_path}", flush=True)
|
||
|
||
print(f"\n 差异订单（前30，按差额绝对值排序）:", flush=True)
|
||
print(f" {'小票号':<20} {'结账时间':<20} {'飞球消费':>10} {'DB items':>10} {'差额':>10} {'台桌':<10} {'会员':>8}", flush=True)
|
||
print(f" {'-'*100}", flush=True)
|
||
for d_row in sorted_diffs[:30]:
|
||
print(f" {d_row['ticket']:<20} {d_row['fq_pay_time']:<20} {d_row['fq_consume']:>10,.2f} {d_row['db_items_sum']:>10,.2f} "
|
||
f"{d_row['diff']:>+10,.2f} {d_row['table_no']:<10} {d_row['member_name']:>8}", flush=True)
|
||
|
||
# ── 6. 按会员归总（飞球 CSV 通过小票号反查会员） ──
|
||
print("\n" + "=" * 90, flush=True)
|
||
print("👤 按会员归总（飞球订单通过小票号反查数据库会员信息）", flush=True)
|
||
print("=" * 90, flush=True)
|
||
|
||
member_agg = defaultdict(lambda: {
|
||
"name": "", "fq_consume": Decimal("0"), "db_items": 0.0,
|
||
"fq_table": Decimal("0"), "fq_goods": Decimal("0"),
|
||
"fq_assistant": Decimal("0"), "orders": 0,
|
||
})
|
||
no_member_count = 0
|
||
no_db_match = 0
|
||
|
||
for ticket, fq_row in fq_tickets.items():
|
||
db_row = db_records.get(ticket)
|
||
if not db_row:
|
||
no_db_match += 1
|
||
continue
|
||
phone = db_row.get("phone") or ""
|
||
if not phone:
|
||
# 无会员
|
||
if db_row.get("member_id") and db_row["member_id"] != 0:
|
||
phone = f"member_{db_row['member_id']}"
|
||
else:
|
||
no_member_count += 1
|
||
continue
|
||
|
||
agg = member_agg[phone]
|
||
agg["name"] = db_row.get("member_name") or agg["name"]
|
||
agg["fq_consume"] += fq_row["consume"]
|
||
agg["db_items"] += db_row["items_sum"]
|
||
agg["fq_table"] += fq_row["table_fee"]
|
||
agg["fq_goods"] += fq_row["goods"]
|
||
agg["fq_assistant"] += fq_row["service"] + fq_row["course"]
|
||
agg["orders"] += 1
|
||
|
||
print(f" 有会员的订单: {sum(a['orders'] for a in member_agg.values())} 单", flush=True)
|
||
print(f" 非会员订单: {no_member_count} 单", flush=True)
|
||
print(f" 数据库无匹配: {no_db_match} 单", flush=True)
|
||
print(f" 会员数: {len(member_agg)} 位", flush=True)
|
||
|
||
# 导出 CSV
|
||
csv_path = os.path.join(report_dir, "member_consumption_20251209_compare.csv")
|
||
|
||
sorted_members = sorted(member_agg.items(), key=lambda x: float(x[1]["fq_consume"]), reverse=True)
|
||
|
||
with open(csv_path, "w", newline="", encoding="utf-8-sig") as f:
|
||
writer = csv.writer(f)
|
||
writer.writerow(["会员昵称", "手机号", "飞球消费合计", "DB items_sum合计",
|
||
"差额", "飞球台费", "飞球商品费", "飞球助教费", "订单数"])
|
||
for phone, agg in sorted_members:
|
||
diff = agg["db_items"] - float(agg["fq_consume"])
|
||
writer.writerow([
|
||
agg["name"], phone,
|
||
f"{float(agg['fq_consume']):.2f}",
|
||
f"{agg['db_items']:.2f}",
|
||
f"{diff:+.2f}",
|
||
f"{float(agg['fq_table']):.2f}",
|
||
f"{float(agg['fq_goods']):.2f}",
|
||
f"{float(agg['fq_assistant']):.2f}",
|
||
agg["orders"],
|
||
])
|
||
|
||
print(f"\n ✅ 会员对比报表已导出: {csv_path}", flush=True)
|
||
|
||
# 终端显示前 20
|
||
print(f"\n {'会员昵称':<12} {'手机号':<14} {'飞球消费':>10} {'DB items':>10} {'差额':>10} {'单数':>5}", flush=True)
|
||
print(f" {'-'*70}", flush=True)
|
||
for phone, agg in sorted_members[:20]:
|
||
diff = agg["db_items"] - float(agg["fq_consume"])
|
||
print(f" {agg['name']:<12} {phone:<14} {float(agg['fq_consume']):>10,.2f} "
|
||
f"{agg['db_items']:>10,.2f} {diff:>+10,.2f} {agg['orders']:>5}", flush=True)
|
||
if len(sorted_members) > 20:
|
||
print(f" ... 还有 {len(sorted_members) - 20} 位（见 CSV）", flush=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run_compare()
|