63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
验证加微判断逻辑：
|
||
bi_vala_app_account.tel_encrypt ↔ stride_contact_bindings.tel_encrypt
|
||
匹配上 = 已加班主任微信，匹配不上 = 未加
|
||
"""
|
||
import psycopg2
|
||
import os
|
||
|
||
PG_HOST = "bj-postgres-16pob4sg.sql.tencentcdb.com"
|
||
PG_USER = "ai_member"
|
||
PG_PASS = os.environ.get("PG_ONLINE_PASSWORD", "")
|
||
|
||
def get_conn(db):
|
||
return psycopg2.connect(host=PG_HOST, port=28591, user=PG_USER, password=PG_PASS, dbname=db)
|
||
|
||
# 1. 从 vala_bi 取所有非测试账号的 tel_encrypt
|
||
conn_bi = get_conn("vala_bi")
|
||
cur = conn_bi.cursor()
|
||
cur.execute("""
|
||
SELECT id, tel_encrypt
|
||
FROM bi_vala_app_account
|
||
WHERE status = 1 AND deleted_at IS NULL AND tel_encrypt IS NOT NULL AND tel_encrypt != ''
|
||
""")
|
||
accounts = {row[1]: row[0] for row in cur.fetchall()} # tel_encrypt -> account_id (取第一个匹配)
|
||
cur.close()
|
||
conn_bi.close()
|
||
|
||
# 2. 从 vala_class 取所有 stride_contact_bindings 的 tel_encrypt
|
||
conn_class = get_conn("vala_class")
|
||
cur = conn_class.cursor()
|
||
cur.execute("""
|
||
SELECT DISTINCT tel_encrypt
|
||
FROM stride_contact_bindings
|
||
WHERE tel_encrypt IS NOT NULL AND tel_encrypt != ''
|
||
""")
|
||
bound_tels = {row[0] for row in cur.fetchall()}
|
||
cur.close()
|
||
conn_class.close()
|
||
|
||
# 3. 匹配
|
||
matched = accounts.keys() & bound_tels
|
||
unmatched = accounts.keys() - bound_tels
|
||
|
||
print(f"=== 加微判断逻辑验证 ===")
|
||
print(f"非测试账号总数（有tel_encrypt）: {len(accounts):,}")
|
||
print(f"stride_contact_bindings 去重 tel_encrypt: {len(bound_tels):,}")
|
||
print(f"")
|
||
print(f"已加微（匹配上）: {len(matched):,} ({len(matched)/len(accounts)*100:.2f}%)")
|
||
print(f"未加微（未匹配）: {len(unmatched):,} ({len(unmatched)/len(accounts)*100:.2f}%)")
|
||
|
||
# 4. 查看几个匹配上的样例
|
||
conn_bi = get_conn("vala_bi")
|
||
cur = conn_bi.cursor()
|
||
sample_tels = list(matched)[:3]
|
||
for tel in sample_tels:
|
||
cur.execute("SELECT id FROM bi_vala_app_account WHERE tel_encrypt = %s LIMIT 1", (tel,))
|
||
row = cur.fetchone()
|
||
if row:
|
||
print(f" 示例: account_id={row[0]}, tel_encrypt={tel[:6]}...{tel[-4:]}")
|
||
cur.close()
|
||
conn_bi.close()
|