378 lines
14 KiB
Python
378 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
行课查询自动回填 — 从飞书表格读取待处理数据，匹配手机号→ID、查询行课记录并回填
|
||
执行频率：每30分钟 cron 巡检 + 群聊关键词触发
|
||
归属 Agent：小溪 (xiaoxi)
|
||
|
||
流程：
|
||
1. 读取 Sheet 2DOxEI → 找出"待查询ID"行 → 手机号脱敏匹配 account_id → 回填 F/G/H 列
|
||
2. 读取 Sheet 55b0eb → 找出"待查询"行 → 查行课记录(进度/耗时/付费) → 回填 D/E 列
|
||
"""
|
||
import json, requests, os, re, sys, psycopg2
|
||
from datetime import datetime
|
||
from collections import defaultdict
|
||
|
||
SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, SCRIPTS_DIR)
|
||
from phone_encrypt import encrypt_phone
|
||
|
||
# ── 配置 ──
|
||
PG_HOST = "bj-postgres-16pob4sg.sql.tencentcdb.com"
|
||
PG_PORT = 28591
|
||
PG_USER = "ai_member"
|
||
PG_DB = "vala_bi"
|
||
|
||
SPREADSHEET_TOKEN = "RFIJsXT8FhGHhctY4RwczcOfnac"
|
||
CRED_DIR = "/root/.openclaw/credentials/xiaoxi"
|
||
|
||
SHEET_ID_QUERY = "2DOxEI" # 手机号→ID 匹配
|
||
SHEET_COURSE = "55b0eb" # 行课记录查询
|
||
|
||
LOG_FILE = "/var/log/xiaoxi_xingke_query.log"
|
||
|
||
def log(msg):
|
||
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
line = f"[{ts}] {msg}"
|
||
print(line)
|
||
with open(LOG_FILE, "a") as f:
|
||
f.write(line + "\n")
|
||
|
||
def get_pg_password():
|
||
secrets_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "secrets.env")
|
||
with open(secrets_path) as f:
|
||
for line in f:
|
||
if line.startswith("PG_ONLINE_PASSWORD="):
|
||
return line.strip().split("=", 1)[1].strip("'\"")
|
||
|
||
def get_fs_token():
|
||
with open(os.path.join(CRED_DIR, "config.json")) as f:
|
||
cfg = json.load(f)
|
||
resp = requests.post(
|
||
"https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
|
||
json={"app_id": cfg["apps"][0]["appId"], "app_secret": cfg["apps"][0]["appSecret"]},
|
||
timeout=15
|
||
)
|
||
return resp.json()["tenant_access_token"]
|
||
|
||
def read_sheet(token, sheet_id):
|
||
url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{SPREADSHEET_TOKEN}/values/{sheet_id}"
|
||
resp = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=30)
|
||
data = resp.json()
|
||
if data.get("code") != 0:
|
||
raise RuntimeError(f"读取Sheet失败: {data}")
|
||
return data["data"]["valueRange"]["values"]
|
||
|
||
def put_values(token, sheet_id, range_str, values):
|
||
url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{SPREADSHEET_TOKEN}/values"
|
||
body = {"valueRange": {"range": f"{sheet_id}!{range_str}", "values": values}}
|
||
resp = requests.put(url, headers={
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json"
|
||
}, json=body, timeout=30)
|
||
return resp.json()
|
||
|
||
def encrypt_phone_local(phone):
|
||
phone = str(phone).strip()
|
||
if "." in phone:
|
||
parts = phone.split(".")
|
||
if parts[1] in ("0", "00"):
|
||
phone = parts[0]
|
||
if re.match(r"^1\d{10}$", phone):
|
||
return encrypt_phone(phone)
|
||
return None
|
||
|
||
def batch_in(cur, sql_tpl, params, chunk=500):
|
||
results = []
|
||
for i in range(0, len(params), chunk):
|
||
batch = params[i:i+chunk]
|
||
ph = ",".join(["%s"] * len(batch))
|
||
cur.execute(sql_tpl % ph, batch)
|
||
results.extend(cur.fetchall())
|
||
return results
|
||
|
||
# ── Step 1: 手机号→ID 匹配 ──
|
||
def process_id_matching(token, conn):
|
||
log("Step 1: 手机号→ID 匹配")
|
||
rows = read_sheet(token, SHEET_ID_QUERY)
|
||
pending = []
|
||
for idx, row in enumerate(rows[1:], start=1):
|
||
if len(row) > 4 and str(row[4]).strip() == "待查询ID":
|
||
phone = str(row[1]).strip() if len(row) > 1 and row[1] else ""
|
||
pending.append({"row_idx": idx + 1, "phone": phone})
|
||
|
||
log(f" 待查询ID: {len(pending)}")
|
||
if not pending:
|
||
log(" 无待处理, 跳过")
|
||
return {"processed": 0, "matched": 0}
|
||
|
||
phones_raw = list(set(r["phone"] for r in pending if r["phone"]))
|
||
valid_phones = [(p, encrypt_phone_local(p)) for p in phones_raw if encrypt_phone_local(p)]
|
||
enc_list = list(set(m[1] for m in valid_phones))
|
||
|
||
cur = conn.cursor()
|
||
enc_to_aid = {}
|
||
for i in range(0, len(enc_list), 500):
|
||
batch = enc_list[i:i+500]
|
||
ph = ",".join(["%s"] * len(batch))
|
||
cur.execute(
|
||
f"SELECT id, tel_encrypt FROM bi_vala_app_account WHERE tel_encrypt IN ({ph}) AND status=1 AND deleted_at IS NULL",
|
||
batch
|
||
)
|
||
for aid, tel_enc in cur.fetchall():
|
||
if tel_enc not in enc_to_aid:
|
||
enc_to_aid[tel_enc] = aid
|
||
cur.close()
|
||
|
||
phone_to_aid = {}
|
||
for phone, enc in valid_phones:
|
||
if enc in enc_to_aid:
|
||
phone_to_aid[phone] = enc_to_aid[enc]
|
||
|
||
log(f" 匹配成功: {len(phone_to_aid)}, 未匹配: {len(valid_phones)-len(phone_to_aid)}")
|
||
|
||
# 组装回填数据
|
||
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
results = []
|
||
for r in pending:
|
||
phone = r["phone"]
|
||
if phone in phone_to_aid:
|
||
results.append({"row": r["row_idx"], "user_id": str(phone_to_aid[phone]), "status": "已回填"})
|
||
else:
|
||
results.append({"row": r["row_idx"], "user_id": "", "status": "未查到"})
|
||
|
||
results.sort(key=lambda x: x["row"])
|
||
|
||
# 分批写入 (按 contiguous rows 分组)
|
||
groups = []
|
||
cur_grp = []
|
||
for r in results:
|
||
if not cur_grp or r["row"] == cur_grp[-1]["row"] + 1:
|
||
cur_grp.append(r)
|
||
else:
|
||
groups.append(cur_grp)
|
||
cur_grp = [r]
|
||
if cur_grp:
|
||
groups.append(cur_grp)
|
||
|
||
for g in groups:
|
||
sr, er = g[0]["row"], g[-1]["row"]
|
||
f_vals = [[r["user_id"]] for r in g]
|
||
g_vals = [[r["status"]] for r in g]
|
||
h_vals = [[now_str] for r in g]
|
||
put_values(token, SHEET_ID_QUERY, f"F{sr}:F{er}", f_vals)
|
||
put_values(token, SHEET_ID_QUERY, f"G{sr}:G{er}", g_vals)
|
||
put_values(token, SHEET_ID_QUERY, f"H{sr}:H{er}", h_vals)
|
||
# 将"请求状态"列 E 标记为已处理
|
||
e_vals = [["已处理"] for _ in g]
|
||
put_values(token, SHEET_ID_QUERY, f"E{sr}:E{er}", e_vals)
|
||
|
||
log(f" ID回填完成: {len(results)} 行, 匹配{len(phone_to_aid)}")
|
||
return {"processed": len(results), "matched": len(phone_to_aid)}
|
||
|
||
|
||
# ── Step 2: 行课记录查询 ──
|
||
def process_course_records(token, conn):
|
||
log("Step 2: 行课记录查询")
|
||
rows = read_sheet(token, SHEET_COURSE)
|
||
pending = []
|
||
for idx, row in enumerate(rows[1:], start=1):
|
||
if len(row) > 2 and str(row[2]).strip() == "待查询":
|
||
uid = str(row[1]).strip() if len(row) > 1 and row[1] else ""
|
||
sales = str(row[0]).strip() if len(row) > 0 and row[0] else ""
|
||
pending.append({"row_idx": idx + 1, "user_id": uid, "sales": sales})
|
||
|
||
log(f" 待查询: {len(pending)}")
|
||
if not pending:
|
||
log(" 无待处理, 跳过")
|
||
return {"processed": 0, "with_records": 0}
|
||
|
||
# 解析有效 user_id
|
||
valid = []
|
||
for r in pending:
|
||
try:
|
||
aid = int(float(r["user_id"]))
|
||
if aid > 0:
|
||
r["account_id"] = aid
|
||
valid.append(r)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
log(f" 有效用户ID: {len(valid)}")
|
||
if not valid:
|
||
return {"processed": 0, "with_records": 0}
|
||
|
||
uid_set = list(set(r["account_id"] for r in valid))
|
||
|
||
# 获取角色
|
||
cur = conn.cursor()
|
||
account_chars = defaultdict(list)
|
||
char_to_account = {}
|
||
rc = batch_in(cur,
|
||
"SELECT account_id, id, nickname FROM bi_vala_app_character WHERE account_id IN (%s) AND nickname IS NOT NULL AND nickname != '' AND deleted_at IS NULL",
|
||
uid_set
|
||
)
|
||
for aid, cid, nick in rc:
|
||
account_chars[aid].append(cid)
|
||
char_to_account[cid] = aid
|
||
char_ids = list(char_to_account.keys())
|
||
|
||
# 课程映射
|
||
cur.execute("SELECT id, course_level, course_season, course_unit, course_lesson FROM bi_level_unit_lesson")
|
||
chapter_map = {}
|
||
for ch_id, cl, cs, cu, cl2 in cur.fetchall():
|
||
chapter_map[ch_id] = (cl or "", cs or "", cu or "", cl2 or "")
|
||
|
||
# 课时完成记录
|
||
char_plays = defaultdict(lambda: {"latest_time": None, "latest_chapter": None, "total_ms": 0})
|
||
for tbl_idx in range(8):
|
||
table = f"bi_user_chapter_play_record_{tbl_idx}"
|
||
try:
|
||
cur.execute(
|
||
f"SELECT user_id, chapter_id, created_at FROM {table} WHERE play_status=1 AND deleted_at IS NULL AND user_id = ANY(%s)",
|
||
(char_ids,)
|
||
)
|
||
for uid, ch_id, created_at in cur.fetchall():
|
||
ch_data = chapter_map.get(ch_id)
|
||
if ch_data:
|
||
rec = char_plays[uid]
|
||
if rec["latest_time"] is None or created_at > rec["latest_time"]:
|
||
rec["latest_time"] = created_at
|
||
rec["latest_chapter"] = (ch_id, ch_data)
|
||
except Exception as e:
|
||
log(f" 警告 {table}: {e}")
|
||
|
||
# 学习总耗时
|
||
for tbl_idx in range(8):
|
||
table = f"bi_user_component_play_record_{tbl_idx}"
|
||
try:
|
||
cur.execute(
|
||
f"SELECT user_id, SUM(COALESCE(interval_time,0)) FROM {table} WHERE user_id = ANY(%s) AND deleted_at IS NULL GROUP BY user_id",
|
||
(char_ids,)
|
||
)
|
||
for uid, total_ms in cur.fetchall():
|
||
if uid in char_plays:
|
||
char_plays[uid]["total_ms"] += (total_ms or 0)
|
||
except Exception as e:
|
||
log(f" 警告 {table}: {e}")
|
||
|
||
# 付费状态
|
||
ph = ",".join(["%s"] * len(uid_set))
|
||
cur.execute(
|
||
f"SELECT account_id, COUNT(*) FROM bi_vala_order WHERE account_id IN ({ph}) AND pay_success_date IS NOT NULL AND order_status=3 AND deleted_at IS NULL GROUP BY account_id",
|
||
uid_set
|
||
)
|
||
paid = {r[0]: r[1] for r in cur.fetchall()}
|
||
|
||
# 激活状态
|
||
try:
|
||
cur.execute(
|
||
f"SELECT t.account_id, t.season_package_level FROM bi_vala_seasonal_ticket t INNER JOIN bi_vala_app_account a ON t.account_id=a.id AND a.status=1 WHERE t.account_id IN ({ph}) AND t.status=1 AND t.deleted_at IS NULL AND t.season_package_level IN ('A1','A2')",
|
||
uid_set
|
||
)
|
||
activation = {}
|
||
for aid, lvl in cur.fetchall():
|
||
if aid not in activation:
|
||
activation[aid] = lvl
|
||
except:
|
||
activation = {}
|
||
|
||
cur.close()
|
||
|
||
# 组装结果
|
||
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
results = []
|
||
|
||
for r in valid:
|
||
aid = r["account_id"]
|
||
chars = account_chars.get(aid, [])
|
||
|
||
best_time = None
|
||
best_ch = None
|
||
total_ms = 0
|
||
|
||
for cid in chars:
|
||
play = char_plays.get(cid)
|
||
if play and play["latest_chapter"]:
|
||
if best_time is None or play["latest_time"] > best_time:
|
||
best_time = play["latest_time"]
|
||
best_ch = play["latest_chapter"]
|
||
total_ms += play["total_ms"]
|
||
|
||
if not best_ch:
|
||
p = "已付费" if aid in paid else "未付费"
|
||
record = f"销售:{r['sales']} | 用户:{aid} | 当前:无行课记录 | 最近行课:无 | 学习0min | {p}"
|
||
else:
|
||
ch_id, (cl, cs, cu, cl2) = best_ch
|
||
full_label = f"{cl}-{cs}-{cu}-{cl2}"
|
||
act = activation.get(aid, "")
|
||
prefix = f"{act}体验课" if act else f"{cl}体验课"
|
||
current = f"{prefix}-{full_label}"
|
||
recent = best_time.strftime("%Y-%m-%d") if best_time else "无"
|
||
total_min = round(total_ms / 60000, 1)
|
||
if total_min == int(total_min):
|
||
total_min = int(total_min)
|
||
p = "已付费" if aid in paid else "未付费"
|
||
record = f"销售:{r['sales']} | 用户:{aid} | 当前:{current} | 最近行课:{recent} | 学习{total_min}min | {p}"
|
||
|
||
results.append({"row": r["row_idx"], "record": record})
|
||
|
||
# 回填
|
||
results.sort(key=lambda x: x["row"])
|
||
groups = []
|
||
cur_grp = []
|
||
for r in results:
|
||
if not cur_grp or r["row"] == cur_grp[-1]["row"] + 1:
|
||
cur_grp.append(r)
|
||
else:
|
||
groups.append(cur_grp)
|
||
cur_grp = [r]
|
||
if cur_grp:
|
||
groups.append(cur_grp)
|
||
|
||
for g in groups:
|
||
sr, er = g[0]["row"], g[-1]["row"]
|
||
d_vals = [[r["record"]] for r in g]
|
||
e_vals = [[now_str] for r in g]
|
||
put_values(token, SHEET_COURSE, f"D{sr}:D{er}", d_vals)
|
||
put_values(token, SHEET_COURSE, f"E{sr}:E{er}", e_vals)
|
||
|
||
with_records = sum(1 for r in results if "无行课记录" not in r["record"])
|
||
log(f" 行课回填完成: {len(results)} 行, 有记录: {with_records}")
|
||
return {"processed": len(results), "with_records": with_records}
|
||
|
||
|
||
# ── Main ──
|
||
def main():
|
||
log("=" * 50)
|
||
log("行课查询自动回填 启动")
|
||
|
||
try:
|
||
token = get_fs_token()
|
||
conn = psycopg2.connect(
|
||
host=PG_HOST, port=PG_PORT, user=PG_USER,
|
||
password=get_pg_password(), dbname=PG_DB, connect_timeout=30
|
||
)
|
||
|
||
r1 = process_id_matching(token, conn)
|
||
r2 = process_course_records(token, conn)
|
||
|
||
conn.close()
|
||
|
||
total = r1["processed"] + r2["processed"]
|
||
summary = f"ID回填:{r1['processed']}(匹配{r1['matched']}) | 行课:{r2['processed']}(有记录{r2['with_records']})"
|
||
log(f"完成: {summary}")
|
||
|
||
if total == 0:
|
||
log("无待处理任务, 静默退出")
|
||
return 0
|
||
return 0
|
||
except Exception as e:
|
||
log(f"ERROR: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|