"""
会员消费报表对比：数据库 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()
