确认 1

etl_billiards/scripts/export_index_tables.py

@@ -0,0 +1,77 @@
+# -*- coding: utf-8 -*-
+"""导出指数表数据到Markdown"""
+import sys
+sys.path.insert(0, '.')
+from config.settings import AppConfig
+from database.connection import DatabaseConnection
+from database.operations import DatabaseOperations
+
+config = AppConfig.load()
+db_conn = DatabaseConnection(config.config['db']['dsn'])
+db = DatabaseOperations(db_conn)
+
+output_lines = []
+
+# 1. 客户召回表
+output_lines.append("## 1. 客户召回表\n")
+output_lines.append("| 客户姓名 | 召回指数 |")
+output_lines.append("|----------|----------|")
+
+sql_recall = """
+    SELECT 
+        COALESCE(m.nickname, CONCAT('会员', r.member_id)) AS member_name,
+        r.display_score AS recall_score
+    FROM billiards_dws.dws_member_recall_index r
+    LEFT JOIN billiards_dwd.dim_member m 
+        ON r.member_id = m.member_id AND m.scd2_is_current = 1
+    ORDER BY r.display_score DESC
+"""
+
+rows = db.query(sql_recall)
+for row in rows:
+    r = dict(row)
+    name = r['member_name'] or '未知'
+    score = r['recall_score']
+    output_lines.append(f"| {name} | {score:.2f} |")
+
+output_lines.append(f"\n共 {len(rows)} 条记录\n")
+
+# 2. 助教客户关系表
+output_lines.append("## 2. 助教客户关系表\n")
+output_lines.append("| 助教花名 | 客户姓名 | 关系指数 |")
+output_lines.append("|----------|----------|----------|")
+
+sql_intimacy = """
+    SELECT 
+        a.nickname AS assistant_name,
+        i.assistant_id AS assistant_no,
+        COALESCE(m.nickname, CONCAT('会员', i.member_id)) AS member_name,
+        i.display_score AS intimacy_score,
+        i.session_count,
+        i.attributed_recharge_amount
+    FROM billiards_dws.dws_member_assistant_intimacy i
+    LEFT JOIN billiards_dwd.dim_member m 
+        ON i.member_id = m.member_id AND m.scd2_is_current = 1
+    LEFT JOIN billiards_dwd.dim_assistant a
+        ON i.assistant_id::text = a.assistant_no AND a.scd2_is_current = 1
+    ORDER BY i.display_score DESC, i.session_count DESC
+"""
+
+rows2 = db.query(sql_intimacy)
+for row in rows2:
+    r = dict(row)
+    assistant = r['assistant_name'] or f"工号{r.get('assistant_no', '?')}"
+    member = r['member_name'] or '未知'
+    score = r['intimacy_score']
+    output_lines.append(f"| {assistant} | {member} | {score:.2f} |")
+
+output_lines.append(f"\n共 {len(rows2)} 条记录")
+
+db_conn.close()
+
+# 写入文件（UTF-8带BOM）
+output_path = 'docs/index_tables.md'
+with open(output_path, 'w', encoding='utf-8-sig') as f:
+    f.write('\n'.join(output_lines))
+
+print(f"已导出到 {output_path}")