102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
"""查询冲突 order 的详细信息"""
|
||
import os
|
||
from pathlib import Path
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv(Path(__file__).resolve().parents[2] / ".env")
|
||
|
||
PG_DSN = os.environ.get("PG_DSN")
|
||
if not PG_DSN:
|
||
raise RuntimeError("PG_DSN 未设置")
|
||
|
||
import psycopg2
|
||
import psycopg2.extras
|
||
|
||
conn = psycopg2.connect(PG_DSN)
|
||
conn.autocommit = True
|
||
|
||
def q(sql, params=None):
|
||
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
||
cur.execute(sql, params or ())
|
||
return cur.fetchall()
|
||
|
||
ORDER_ID = 2987675294190277
|
||
MEMBER_ID = 2976361970370373
|
||
SITE_ID = 2790685415443269
|
||
|
||
print(f"冲突 order 详情: order={ORDER_ID}")
|
||
print("=" * 60)
|
||
|
||
# 1. 在 dws_member_visit_detail 中查找
|
||
rows = q("""
|
||
SELECT id, site_id, member_id, order_settle_id, visit_date, visit_time, created_at
|
||
FROM dws.dws_member_visit_detail
|
||
WHERE order_settle_id = %s AND member_id = %s
|
||
""", (ORDER_ID, MEMBER_ID))
|
||
print(f"\ndws_member_visit_detail 中的记录: {len(rows)}")
|
||
for r in rows:
|
||
print(f" id={r['id']}, visit_date={r['visit_date']}, visit_time={r['visit_time']}")
|
||
print(f" created_at={r['created_at']}")
|
||
|
||
# 2. 在 dwd_settlement_head 中查找
|
||
rows2 = q("""
|
||
SELECT order_settle_id, member_id, pay_time, create_time,
|
||
scd2_valid_from, scd2_valid_to, scd2_is_current
|
||
FROM dwd.dwd_settlement_head
|
||
WHERE order_settle_id = %s
|
||
""", (ORDER_ID,))
|
||
print(f"\ndwd_settlement_head 中的记录: {len(rows2)}")
|
||
for r in rows2:
|
||
print(f" member={r['member_id']}, pay_time={r['pay_time']}")
|
||
print(f" scd2_current={r['scd2_is_current']}, from={r['scd2_valid_from']}, to={r['scd2_valid_to']}")
|
||
|
||
# 3. 检查 biz_date 计算
|
||
rows3 = q("""
|
||
SELECT order_settle_id, pay_time,
|
||
EXTRACT(HOUR FROM pay_time) AS pay_hour,
|
||
(CASE WHEN EXTRACT(HOUR FROM pay_time) < 8
|
||
THEN (pay_time - INTERVAL '1 day')::date
|
||
ELSE pay_time::date END) AS biz_date
|
||
FROM dwd.dwd_settlement_head
|
||
WHERE order_settle_id = %s
|
||
""", (ORDER_ID,))
|
||
print(f"\nbiz_date 计算:")
|
||
for r in rows3:
|
||
print(f" pay_time={r['pay_time']}, hour={r['pay_hour']}, biz_date={r['biz_date']}")
|
||
|
||
# 4. 检查窗口边界问题
|
||
# 窗口 1/4: 2025-10-31 ~ 2025-11-30
|
||
# delete 会删除 visit_date >= 2025-10-31 AND visit_date <= 2025-11-30
|
||
# 但如果这个 order 的 visit_date 在 2025-10-31 之前（比如 2025-10-30），
|
||
# 它不会被 delete 清理，但 insert 时会冲突
|
||
|
||
# 检查这个 order 在旧数据中的 visit_date
|
||
rows4 = q("""
|
||
SELECT visit_date, COUNT(*) AS cnt
|
||
FROM dws.dws_member_visit_detail
|
||
WHERE order_settle_id = %s AND site_id = %s
|
||
GROUP BY visit_date
|
||
""", (ORDER_ID, SITE_ID))
|
||
print(f"\n该 order 在 dws 中的 visit_date 分布:")
|
||
for r in rows4:
|
||
print(f" visit_date={r['visit_date']}, cnt={r['cnt']}")
|
||
|
||
# 5. 检查 dwd 中是否有多条 SCD2 版本
|
||
rows5 = q("""
|
||
SELECT COUNT(*) AS cnt
|
||
FROM dwd.dwd_settlement_head
|
||
WHERE order_settle_id = %s
|
||
""", (ORDER_ID,))
|
||
print(f"\ndwd_settlement_head 中该 order 的记录数: {rows5[0]['cnt']}")
|
||
|
||
# 6. 检查是否有 member_id 不同但 order_settle_id 相同的情况
|
||
rows6 = q("""
|
||
SELECT DISTINCT member_id
|
||
FROM dwd.dwd_settlement_head
|
||
WHERE order_settle_id = %s
|
||
""", (ORDER_ID,))
|
||
print(f"该 order 对应的 member_id: {[r['member_id'] for r in rows6]}")
|
||
|
||
conn.close()
|
||
print("\n诊断完成。")
|