46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""实测 sender_id：get_messages 与 Listener 回调两条路径的真实值"""
|
||
import sys, os, tempfile, time, threading
|
||
BASE = r"C:\Users\Administrator\wechatauto-replica"
|
||
os.chdir(BASE)
|
||
sys.path.insert(0, BASE)
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
except AttributeError:
|
||
pass
|
||
|
||
from wechatauto.db import WeChatDB, Listener
|
||
|
||
wd = tempfile.mkdtemp(prefix="diag_sid_")
|
||
db = WeChatDB(workdir=wd)
|
||
info = db.get_self_info()
|
||
self_wxid = info.get("username")
|
||
print("SELF wxid:", self_wxid, "| nick:", info.get("nick_name"))
|
||
|
||
# 路径1：get_messages 的 sender_id
|
||
who = "wxid_yxnyrj5sul1g22"
|
||
try:
|
||
msgs = db.get_messages(who, limit=12)
|
||
print("\n[get_messages] 好友", who, "最近", len(msgs), "条:")
|
||
for m in reversed(msgs):
|
||
sid = str(m.get("sender_id"))
|
||
me = "★自己" if (sid == "1" or sid == self_wxid) else "对方"
|
||
print(f" sender_id={sid!r} {me} | {m.get('type')} | {str(m.get('content'))[:28]}")
|
||
except Exception as e:
|
||
print("get_messages ERR:", repr(e))
|
||
|
||
# 路径2：Listener 回调的 sender_id（真实监听一条，看回调字段）
|
||
print("\n[Listener] 用 add_listener 监听该好友，等 3s 观察 watermark 初始化…")
|
||
seen = {}
|
||
lst = Listener(db, interval=1.0)
|
||
|
||
def on_msg(m, l):
|
||
sid = str(m.get("sender_id"))
|
||
print(" Listener 回调 sender_id=%r | content=%r" % (sid, str(m.get("content"))[:30]))
|
||
|
||
lst.add_listener(who, on_msg)
|
||
lst.start()
|
||
time.sleep(3)
|
||
lst.stop()
|
||
print("[Listener] 结束（若无新消息则无回调输出，属正常）")
|