UIA 读微信群消息没有 id,怎么判断哪条是新的:后缀对齐在翻历史和撤回时整屏重报
我那个微信群机器人不走 hook,也不走协议,就是每 1.5 秒用 Windows UI Automation 读一遍当前会话屏上能看到的消息,再和上一轮比,算出哪些是新的。这件事听起来简单,真写起来最难的一块其实是「怎么判断新消息」。
UIA 读出来的消息节点是叶子:只有文字内容,没有消息 id、没有时间戳,昵称也拿不到。能用的只有 (kind, content)。这篇记录我重新审这段对齐逻辑时,在 Linux 上把 uiautomation 换成空壳、喂假数据跑出来的几个问题。
现在的做法:后缀对前缀
适配层给每条消息分一个稳定 id,上层 MessageStream 按 id 去重。分 id 的方式是把上一屏的末尾和这一屏的开头对齐:
for k in range(min(len(prev_keys), len(cur_keys)), 0, -1):
if prev_keys[-k:] == cur_keys[:k]:
overlap = k
break对上的部分沿用老 id,后面多出来的发新 id。
上一屏:A B C D
这一屏: C D E F -> 沿用 C D 的 id,E F 是新的这个设计有两个地方我觉得是对的:
- 比较键只有
kind + content,不含 is_self。is_self 是靠截图判气泡颜色得来的,消息滚到视口边缘被裁掉一截时会判不出来。代码注释里记着,之前把它放进键里,一条消息从 self 变成 unknown,对齐直接断掉,12 条消息被重报了一遍。 - 不用集合去重,所以同一个人连发两条「在吗」,会拿到两个不同 id。项目里本来还有一层内容指纹去重(
DEDUP_WINDOW),默认就是 0,因为它会把这种合法的重复发言吞掉。
正常追加我跑出来没问题:
来了一条 D 判为新消息=['D']
同一人连发 D 判为新消息=['D']我还做了个小模拟:窗口固定显示 8 条,200 轮里每轮随机来 0~3 条,应收 230 条,实收 230 条,顺序一致。
问题:对不上的时候,整屏都成了新消息
overlap 找不到时是 0,代码会继续往下走:0 条沿用老 id,整屏全部发新 id。上层只认 id,于是整屏都被当成新消息送进回复链。
适配层里其实另有一个 diff_new(),对不上时返回 None,让上层重建基线,思路是对的。但机器人实际走的是兼容层 GetAllMessage() → read_all_with_ids() 这条路,这里没有这个分支。
我用真实的 bot.py、compat.py、adapter.py,只把读屏函数换成返回假数据,跑了三个场景。
场景一:有人往上翻聊天记录。 机器人跑在一台常开的电脑上,谁路过往上滚一下很正常:
首轮(建基线) 屏上=['A', 'B', 'C', 'D', 'E', 'F']
人往上翻了 3 条 屏上=['h1', 'h2', 'h3', 'A', 'B', 'C']
判为新消息=['h1', 'h2', 'h3', 'A', 'B', 'C']
又滚回底部 屏上=['A', 'B', 'C', 'D', 'E', 'F']
判为新消息=['D', 'E', 'F']往上翻时,上一屏的后缀 D E F 和这一屏的前缀 h1 h2 h3 对不上,6 条全成了新的。更难发现的是滚回底部这一步:此时「上一屏」已经是翻上去那屏,A B C 对上了,后面的 D E F 又被当成新消息报了一次。也就是说 D E F 这三条老消息,一共被处理了两次。
场景二:最后一条被撤回。 微信里撤回后,原消息位置会变成一行灰色提示。我按「原地换成一条系统消息」来模拟:
首轮(建基线) 屏上=['A', 'B', 'C', 'D']
D 被撤回成系统提示 屏上=['A', 'B', 'C', '"张三" 撤回了一条消息']
判为新消息=['A', 'B', 'C', '"张三" 撤回了一条消息']只改了最后一条,后缀就整个对不上了,A B C 三条重新进回复链(系统提示会被 should_handle 挡掉)。
场景三:撤回的同时来了新消息。 这时候我们希望只报 E:
屏上=['B', 'C', '"张三" 撤回了一条消息', 'E']
判为新消息=['B', 'C', '"张三" 撤回了一条消息', 'E']好在后面还有限流兜底:同一个人 30 秒内不重复回、群里两条回复至少隔 5 秒、每小时最多 40 条,所以不至于真的刷屏。但这几条历史消息确实被当成新消息走完了整条决策链,被回复的那条完全看运气。
我的改法:对齐「流水」,不只是上一屏
改动有两点。
一是对不上时不再当新消息。 给消息加一个 replay 标记,表示「新分了 id,但不是新消息」。上层看到 replay 就只记成已读,同时打一条 warning,写清楚吞了几条,不再悄悄跳过:
if getattr(m, "replay", False):
replayed += 1
continue二是对齐对象从上一屏换成最近 300 条的流水。 在流水里找和这一屏最长的连续公共段,长度相同时取流水里更靠后的那段:
- 公共段上方的消息:翻上去看到的更早历史,标 replay;
- 公共段之后、流水里也有的位置,内容对得上就沿用 id,对不上(被撤回之类)就标 replay;
- 超出流水末尾的,才是真正的新消息。
流水里保留着翻上去之前的 D E F,所以滚回底部时它们能找回原来的 id。补丁版跑同样的场景:
人往上翻了 3 条 判为新消息=[] (warning:3 条对不上流水,按已读处理)
又滚回底部 判为新消息=[]
D 被撤回成系统提示 判为新消息=[] (warning:1 条)
D 撤回 + 同时来了 E 判为新消息=['E']正常追加和 200 轮随机模拟的结果和改之前一样,230 条一条不差。最长公共段用的是 O(屏 × 流水) 的 DP,我本机上 30 × 300 每轮大约 0.3 ms,相对 1.5 秒的轮询间隔可以忽略。
还没解决的:内容完全一样的刷屏
群里发通知,大家排队回「收到」:
首轮(建基线) 屏上=['通知', '收到', '收到', '收到']
又来 2 条收到(屏满) 判为新消息=['收到']
又来 1 条收到 判为新消息=[]改前改后结果一样,2 条只报出 1 条,后面那条直接漏了。原因是屏幕已经满了,前后两屏都是 4 个「收到」,单看内容根本分不出移动了几格。没有 id、没有时间戳、没有昵称,这在算法层面就是无解的,只能靠更多信号:比如每条消息的屏幕坐标和滚动条位置,或者会话列表里的未读数「[3条]」。好在「收到」本来也不该触发回复,我暂时接受这个漏报,在文档里写明。
几点体会
- 增量对齐一定要想清楚「对不上」时怎么办。默认当新消息,代价是重复回复;默认当旧消息,代价是漏回。我选的是宁可漏也不重复,但必须打日志。
- 同一个项目里有两套对齐(
diff_new和read_all_with_ids),只修了一套,另一套照样出事。我这次是对着实际调用链,从GetAllMessage一路往下看才发现的。 - 没有 Windows 也能测这类逻辑:把
uiautomation换成空壳模块、把读屏函数替换成返回列表,业务代码一行都不用改。
补丁和复现脚本都在本地,还没合进仓库。