75 KiB
75 KiB
微信好友申请自动通过设计方案
版本：v1.2（深度审查修正版 · 第二轮） 日期：2026-07-16 范围：
bridge/woc_bridge全链路 目标：实现好友申请的自动监听、规则匹配、UI 自动通过、DB 校验闭环v1.1 修正摘要（基于源码逐行验证，共修正 15 项）：
- [P0] IdemCache 签名误用：
get(flow_name, to_wxid, content, request_id)/set(flow_name, to_wxid, content, value, request_id)— value 是 set 的第 4 参数- [P0] WCDB_CT_message_content 列硬编码矛盾：SELECT 硬编码 + row.keys() 兜底矛盾，改为 PRAGMA 预探测
- [P0] @stranger 假设未验证：仅注释提及，无代码验证，补充 fallback 策略与环境验证清单
- [P0] BridgeError 未捕获：
_ensure_decrypted抛DB_ENCRYPTED/DB_NEED_INIT，仅 catchsqlite3.Error会泄漏- [P1] lambda 闭包捕获 bug：循环中
req变量延迟绑定，需默认参数捕获- [P1]
_enabled_event未初始化：__init__缺少asyncio.Event创建- [P1]
AcceptRuleEngine缺get_config方法：API 路由调用但未定义- [P1]
FriendRequestWatcher缺状态属性：is_running/is_enabled/processed_count等未定义- [P1]
_parse_list未定义：config.py 中不存在此函数- [P1]
models/__init__.py导出遗漏：新增 6 个模型需追加导出- [P2]
_clickvs_click_at：统一用_click_at（链式命令更高效）- [P2] 游标缺复合 tie-breaker：
create_time >改为(create_time, local_id)复合游标- [P2]
verify_friend_accepted双查询优化：合并为单条 SQL- [P2]
local_type=4未识别：陌生人被误判为 person，影响get_contacts结果- [P2]
get_contacts未过滤 @stranger：陌生人混入联系人列表v1.2 修正摘要（第二轮深度审查，新增 9 项修正）：
- [P0]
_poll_once复合游标调用缺失：原get_friend_requests_since, self._cursor, limit=50缺cursor_local_id，且单一_cursor无法支撑复合游标 → 改为_cursor_create_time+_cursor_local_id双属性- [P0]
_poll_once返回字段名错误：原result["next_cursor"]与 3.3.1 节返回的next_create_time/next_local_id不一致- [P0]
_poll_once缺解析步骤：原requests: list[FriendRequestInfo] = result["requests"]类型错误，DB 返回 list[dict]，需经parse_friend_request解析- [P0]
_init_cursor未设置_cursor_inited：导致_watch_loop每轮都重新初始化游标，无法推进- [P0]
list_friend_requests路由参数错误：原get_friend_requests_since, 0, limit把 limit 当成 cursor_local_id → 改为0, 0, limit- [P1]
_handle_request缺状态统计更新：_processed_count/_accepted_count/_rejected_count/_last_processed_time未更新- [P1]
_handle_request异常透传不全：原统一except Exception吞掉 BridgeError 错误码 → 拆分except BridgeError透传- [P1] lifespan 异常捕获冗余：
except (asyncio.TimeoutError, Exception)中 Exception 已涵盖 TimeoutError → 统一为except Exception- [P1] lifespan 启动顺序未说明：补充现有启动/停止顺序，明确 friend_watcher 在 message_streamer 之后启动、之前停止
1. 背景与现状
1.1 现有能力缺口
| 能力 | 现状 | 文件位置 |
|---|---|---|
| 主动添加好友 | POST /api/friends/add，xdotool UI 自动化，experimental |
routes/contacts.py:257 |
| 修改好友备注 | POST /api/contacts/{wxid}/remark，experimental |
routes/contacts.py:191 |
| 接收/通过好友申请 | 不存在 | — |
| 好友申请监听 | 不存在 | — |
1.2 现有架构可复用基础
| 组件 | 能力 | 文件 |
|---|---|---|
MessageStreamer |
双层增量检测（DB mtime + (create_time, local_id) 复合游标），1s/5s 自适应轮询，SSE 广播 |
messaging/streamer.py |
SendQueue |
单 worker 串行执行，滑动窗口限流，队列满抛 RATE_LIMITED |
messaging/send_queue.py |
IdemCache |
幂等去重（namespace + key + content + request_id） | ui/idem_cache.py |
CircuitBreaker |
熔断器（failure_threshold / recovery_timeout） | ui/circuit_breaker.py |
XdotoolDriver |
窗口激活、坐标计算、_step 超时模型（L1-L4）、add_friend 参考实现 |
ui/xdotool_driver.py |
DbReader |
动态探测表名/列名，_ensure_decrypted 解密缓存，_SYSTEM_WXIDS 含 fmessage |
db/reader.py |
FlowOrchestrator |
幂等 + 熔断 + 会话缓存 + 入队编排（仅 send_text） |
ui/orchestrator.py |
1.3 微信 4.x Linux 好友申请的存在形式
验证状态说明：以下结论基于代码分析推断，部分假设（标注
[需环境验证]）需在真实微信 4.x Linux 环境中用sqlite3直接确认 DB schema 后方可作为实现依据。
来源 A：消息 DB 系统消息（代码已验证）
- DB：
message/message_0.db - 分片表：
Msg_<MD5("fmessage")>（表名计算逻辑已验证，reader.py:677、reader.py:899） talker = "fmessage"（系统账号，已在_SYSTEM_WXIDS白名单中，reader.py:1391）local_type = 10000（系统消息，映射render_type="system"，reader.py:38-46）message_content为 XML 格式的 sysmsg[需环境验证]：具体 XML 结构（节点名/属性名）需在真实 fmessage 消息上确认，设计方案中的<sysmsg type="verifyUser"><Link>...是基于微信协议常识的推断<!-- 推断格式，需环境验证实际节点名 --> <sysmsg type="verifyUser" ...> <Link ...> <UserName>xxx@stranger</UserName> <NickName>申请人昵称</NickName> <Content>验证消息内容</Content> <Scene>...</Scene> </Link> </sysmsg>- 当前 bridge 不解析此 XML，
content字段原样返回给客户端（reader.py:688-709_decompress_msg_content仅做 zstd 解压 + UTF-8 解码）
来源 B：联系人 DB（部分假设 [需环境验证]）
- DB：
contact/contact.db - 表：
contact/contacts/rcontact（动态探测已验证，reader.py:496-512） username列名动态探测（候选["username", "wxid"]，reader.py:1364）- 申请人以
username = xxx@stranger形式写入[需环境验证]：代码中@stranger仅出现在注释（reader.py:1389、reader.py:1416），实际逻辑用wxid.split("@", 1)[0]取 base 部分，对任意@xxx后缀都生效。真实后缀名需在 contact.db 中确认 local_type = 4[需环境验证]：微信协议中通常代表陌生人/未通过验证，但项目当前未识别此值（_infer_contact_type仅处理 2 和 512，reader.py:1421-1425），陌生人会被默认推断为person（fallback），与正常好友混淆
环境验证清单（实施前必做）：
- 在真实微信 4.x Linux 环境中，用
sqlite3打开contact.db，执行SELECT username, local_type FROM contact WHERE local_type NOT IN (1,2,512)确认陌生人后缀格式和 local_type 值 - 收到一条好友申请后，用
sqlite3打开message_0.db，执行SELECT message_content FROM Msg_<MD5('fmessage')> ORDER BY create_time DESC LIMIT 1确认 XML 结构 - 通过好友申请后，再次查 contact.db 确认
@stranger后缀是否消失、local_type是否变化
1.4 截图 UI 布局分析
基于 ui/profiles/templates/screenshot/ 下的截图：
| 截图 | 布局描述 |
|---|---|
04-好友申请界面.png |
左侧"新的朋友"申请列表，每项含头像/昵称/验证消息 + 右侧"前往验证"按钮 |
05-好友申请通过界面.png |
"通过朋友验证"弹窗，含申请人信息 + 底部"确定"按钮 |
UI 导航路径：主界面 → 通讯录图标 → 新的朋友 → 列表项"前往验证" → 弹窗"确定" → 完成
2. 总体架构
2.1 全链路数据流
┌─────────────────────────────────────────────────────────────────────┐
│ 监听层（双源） │
│ │
│ ┌─────────────────────┐ ┌──────────────────────────────┐ │
│ │ FriendRequestWatcher │ │ MessageStreamer (已有) │ │
│ │ (新增后台协程) │ │ _watch_loop → _poll_once │ │
│ │ │ │ │ │
│ │ 轮询 message DB: │ │ 新消息事件广播 │ │
│ │ Msg_<MD5(fmessage)> │ │ (talker=fmessage, │ │
│ │ 增量游标检测 │ │ local_type=10000) │ │
│ └────────┬────────────┘ └──────────┬───────────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ FriendRequestParser (新增) │ │
│ │ XML sysmsg 解析 → FriendRequestInfo │ │
│ └─────────────────────────┬───────────────────────────────────┘ │
└─────────────────────────────┼───────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 规则匹配引擎（新增） │
│ │
│ ┌─────────────┐ ┌──────────────┐ ┌─────────────┐ ┌──────────┐ │
│ │ 白名单匹配 │ │ 关键词匹配 │ │ 黑名单过滤 │ │ auto_accept│
│ │ (wxid/昵称) │ │ (验证消息) │ │ (wxid/昵称) │ │ (全局开关) │
│ └──────┬──────┘ └──────┬───────┘ └──────┬──────┘ └────┬─────┘ │
│ └────────────────┼─────────────────┼──────────────┘ │
│ ▼ ▼ │
│ ┌──────────────┐ │
│ │ 决策结果 │ │
│ │ accept/reject │ │
│ │ /skip │ │
│ └──────┬───────┘ │
└──────────────────────────┼────────────────────────────────────────┘
│ accept
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 去重 + 入队层 │
│ │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ IdemCache (已有) │ │ SendQueue (已有) │ │
│ │ namespace= │ │ 串行执行 │ │
│ │ "friend_accept" │────▶│ enqueue(lambda: │ │
│ │ key=stranger_wxid│ │ accept_flow) │ │
│ │ TTL=300s │ │ │ │
│ └──────────────────┘ └────────┬─────────┘ │
└─────────────────────────────────────┼───────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ UI 自动化层 │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ accept_friend_request (新增于 xdotool_driver.py) │ │
│ │ │ │
│ │ 1. 激活窗口 │ │
│ │ 2. 点击通讯录图标 │ │
│ │ 3. 点击"新的朋友"入口 │ │
│ │ 4. 定位目标申请项 → 点击"前往验证" │ │
│ │ 5. 弹窗 → 点击"确定" │ │
│ │ 6. Esc 返回主界面 │ │
│ └─────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────┬─────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 校验 + 状态层 │
│ │
│ ┌──────────────────────┐ ┌──────────────────┐ │
│ │ DB 校验 (新增) │ │ CircuitBreaker │ │
│ │ contact 表中 │ │ (已有, 复用) │ │
│ │ @stranger 消失验证 │ │ accept_verify │ │
│ └──────────┬───────────┘ │ failure_threshold │ │
│ │ │ =5, recovery=30s │ │
│ ▼ └──────────────────┘ │
│ ┌──────────────────────┐ │
│ │ 结果记录 │ │
│ │ (日志 + metrics) │ │
│ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
2.2 分层职责
| 层 | 组件 | 职责 | 新增/复用 |
|---|---|---|---|
| L0 监听 | FriendRequestWatcher |
轮询 message DB 检测 fmessage 新消息 | 新增 |
| L0 解析 | FriendRequestParser |
解析 sysmsg XML 提取申请人信息 | 新增 |
| L1 匹配 | AcceptRuleEngine |
白名单/关键词/黑名单/auto_accept 规则决策 | 新增 |
| L2 去重 | IdemCache |
按 stranger_wxid 去重，TTL=300s | 复用 |
| L3 入队 | SendQueue |
串行化 UI 操作，避免 xdotool 命令冲突 | 复用 |
| L4 UI | XdotoolDriver.accept_friend_request |
6 步 UI 操作流程 | 新增 |
| L5 校验 | DbReader.verify_friend_accepted |
contact 表 @stranger 消失验证 | 新增 |
| L6 熔断 | CircuitBreaker |
UI 操作连续失败保护 | 复用 |
| L7 API | routes/contacts.py |
手动查询/通过/配置接口 | 新增路由 |
3. 详细设计
3.1 监听层：FriendRequestWatcher
3.1.1 设计决策：独立 Watcher vs 复用 MessageStreamer
| 方案 | 优点 | 缺点 |
|---|---|---|
| A. 复用 MessageStreamer + handler 注册 | 统一轮询，无重复 DB 读取 | MessageStreamer 当前无 handler 注册机制，需改造其架构 |
| B. 独立 FriendRequestWatcher | 不侵入现有 streamer，独立游标/频率控制 | 多一个 DB 轮询（但仅查 fmessage 单表，开销极小） |
选择方案 B：独立 Watcher。理由：
- MessageStreamer 设计为"纯队列广播模型，无 handler 注册"（streamer.py:14-20），注入 handler 会破坏其架构纯净性
- fmessage 轮询单表
Msg_<MD5("fmessage")>且有独立游标，SQL 开销 <1ms，不会对 DB 造成压力 - 独立 Watcher 可按需启停（auto_accept 关闭时不轮询），不影响消息流
3.1.2 数据结构
# 新文件：messaging/friend_watcher.py
@dataclass
class FriendRequestInfo:
"""好友申请信息（从 fmessage sysmsg XML 解析）。"""
stranger_wxid: str # 申请人 wxid（含 @stranger 后缀）
nickname: str # 申请人昵称
verify_message: str # 验证消息内容
scene: str # 来源场景（如群聊/搜索/二维码）
raw_xml: str # 原始 XML（调试用）
create_time: int # 消息时间戳
msg_local_id: int # 消息 local_id（游标用）
class FriendRequestWatcher:
"""好友申请监听器：轮询 fmessage 系统消息，解析后触发自动通过。"""
def __init__(
self,
db_reader: "DbReader",
rule_engine: "AcceptRuleEngine",
send_queue: "SendQueue",
idem_cache: "IdemCache",
xdotool_driver: "XdotoolDriver",
breaker: "CircuitBreaker",
poll_interval: float = 3.0, # 轮询间隔（秒）
cursor_create_time: int = 0, # 初始游标 create_time（0 = 从最新开始）
cursor_local_id: int = 0, # 初始游标 local_id（tie-breaker）
) -> None:
self._db_reader = db_reader
self._rule_engine = rule_engine
self._send_queue = send_queue
self._idem_cache = idem_cache
self._xdotool = xdotool_driver
self._breaker = breaker
self._poll_interval = poll_interval
# 复合游标 (create_time, local_id)，与 get_friend_requests_since 返回值对齐
self._cursor_create_time: int = cursor_create_time
self._cursor_local_id: int = cursor_local_id
self._last_db_mtime: float = 0.0
self._last_wal_mtime: float = 0.0
self._watcher: Optional[asyncio.Task] = None
self._enabled_event: asyncio.Event = asyncio.Event() # auto_accept 开关事件
self._cursor_inited: bool = False
# 状态统计（供 /api/friends/auto_accept/status 查询）
self._processed_count: int = 0
self._accepted_count: int = 0
self._rejected_count: int = 0
self._last_processed_time: Optional[int] = None
@property
def is_running(self) -> bool:
return self._watcher is not None and not self._watcher.done()
@property
def is_enabled(self) -> bool:
return self._enabled_event.is_set()
@property
def processed_count(self) -> int:
return self._processed_count
@property
def accepted_count(self) -> int:
return self._accepted_count
@property
def rejected_count(self) -> int:
return self._rejected_count
@property
def last_processed_time(self) -> Optional[int]:
return self._last_processed_time
async def start(self) -> None:
if self._watcher is None or self._watcher.done():
self._watcher = asyncio.create_task(self._watch_loop())
logger.info("FriendRequestWatcher 已启动")
async def stop(self) -> None:
if self._watcher is not None and not self._watcher.done():
self._watcher.cancel()
try:
await self._watcher
except asyncio.CancelledError:
pass
self._watcher = None
def set_enabled(self, enabled: bool) -> None:
if enabled:
# 开启时重置游标，避免批量处理积压申请
self._cursor_inited = False
self._enabled_event.set()
else:
self._enabled_event.clear()
logger.info("FriendRequestWatcher enabled=%s", enabled)
3.1.3 轮询逻辑
复用 MessageStreamer 的双层增量检测模式：
async def _poll_once(self) -> None:
# 1. mtime 感知（一级过滤，避免空 SQL）
mtimes = await asyncio.to_thread(
self._db_reader.get_db_mtime, "message/message_0.db"
)
if mtimes is None:
return
db_mtime, wal_mtime = mtimes
if db_mtime == self._last_db_mtime and wal_mtime == self._last_wal_mtime:
return
self._last_db_mtime = db_mtime
self._last_wal_mtime = wal_mtime
# 2. 查询 fmessage 分片表增量消息（复合游标）
result = await asyncio.to_thread(
self._db_reader.get_friend_requests_since,
self._cursor_create_time,
self._cursor_local_id,
50, # limit
)
if result is None:
# DB 不可读（_ensure_decrypted 抛 BridgeError）
return
raw_items: list[dict] = result["requests"]
# 3. 解析 XML + 逐条处理
for item in raw_items:
info = parse_friend_request(
item["content"], item["create_time"], item["local_id"]
)
if info is None:
# 非 verifyUser 类型或解析失败，跳过
continue
await self._handle_request(info)
# 4. 推进复合游标（与 get_friend_requests_since 返回字段对齐）
self._cursor_create_time = result["next_create_time"]
self._cursor_local_id = result["next_local_id"]
修正说明：
- 原代码
get_friend_requests_since, self._cursor, limit=50缺少cursor_local_id，且单一self._cursor无法支撑复合游标- 原代码
result["next_cursor"]字段名错误，3.3.1 节返回的是next_create_time/next_local_id- 原代码
requests: list[FriendRequestInfo] = result["requests"]类型错误，DB 返回的是 list[dict]，需经parse_friend_request解析才得到FriendRequestInfo- 增加
parse_friend_request解析步骤，过滤非 verifyUser 类型的 fmessage 消息
3.1.4 游标初始化策略
| 启动时机 | 游标对齐策略 | 理由 |
|---|---|---|
| Bridge 首次启动 | cursor_create_time = get_max_create_time_for_talker("fmessage")，cursor_local_id = 0 |
跳过历史申请，仅处理启动后的新申请 |
| auto_accept 从关闭切换为开启 | 同上（set_enabled 重置 _cursor_inited=False） |
避免开启瞬间批量处理积压申请 |
| auto_accept 持续运行 | 持续推进复合游标，不重置 | 正常增量处理 |
async def _init_cursor(self) -> bool:
"""对齐游标到当前 fmessage 表最大 create_time。
Returns:
True 表示已对齐（或 DB 不可读但标记为已初始化，避免无限重试）
False 表示异常，下一轮重试
"""
try:
max_ct = await asyncio.to_thread(
self._db_reader.get_max_create_time_for_talker, "fmessage"
)
if max_ct is None:
# DB 不可读（_ensure_decrypted 抛 BridgeError 已被吞掉返回 None）
# 不标记 _cursor_inited，下一轮重试
logger.warning("FriendRequestWatcher 游标初始化: DB 不可读，将在下轮重试")
return False
if max_ct and max_ct > 0:
self._cursor_create_time = max_ct
self._cursor_local_id = 0
logger.info(
"FriendRequestWatcher 游标对齐到 create_time=%d（跳过历史申请）",
self._cursor_create_time,
)
# 无论 max_ct 是否为 0（空表），都标记为已初始化，避免空表时无限重试
self._cursor_inited = True
return True
except Exception as e:
logger.warning("FriendRequestWatcher 游标初始化失败: %s", e)
return False
修正说明：
- 原代码未设置
self._cursor_inited = True，会导致_watch_loop每轮都重新初始化游标，无法推进get_max_create_time_for_talker返回None（DB 不可读）时不标记已初始化，下一轮重试；返回0（空表）时标记已初始化- 复合游标：
_cursor_create_time对齐到 max，_cursor_local_id重置为 0（同秒内的历史消息不处理）
3.1.5 轮询间隔
| 状态 | 间隔 | 理由 |
|---|---|---|
| auto_accept 开启 | 3s | 好友申请低频事件，3s 足够及时且无 DB 压力 |
| auto_accept 关闭 | 不轮询 | Watcher 协程暂停在 asyncio.Event 上 |
| DB_ENCRYPTED | 3s 重试 _init_cursor |
DB 不可读时 _cursor_inited 保持 False，每轮重试初始化（_init_cursor 内部 catch BridgeError，不会抛异常风暴） |
| 异常 | 5s 退避 | 避免 _poll_once 抛未预期异常时风暴 |
async def _watch_loop(self) -> None:
while True:
try:
# auto_accept 关闭时挂起，避免空轮询
await self._enabled_event.wait()
# 游标未初始化时先对齐（DB 恢复可读后自动补齐）
# DB_ENCRYPTED 时 _init_cursor 返回 False，下一轮仍会重试
if not self._cursor_inited:
await self._init_cursor()
if not self._cursor_inited:
# DB 仍不可读，本轮跳过 _poll_once
await asyncio.sleep(self._poll_interval)
continue
await asyncio.sleep(self._poll_interval)
await self._poll_once()
except asyncio.CancelledError:
raise
except Exception as e:
logger.exception("FriendRequestWatcher 异常: %s", e)
await asyncio.sleep(5.0)
修正说明：
- 原表格 "DB_ENCRYPTED 10s 退避" 与代码实现不一致（代码无 10s 退避逻辑）→ 改为 3s 重试
_init_cursor_watch_loop增加if not self._cursor_inited: continue跳过_poll_once，避免 DB 不可读时无效查询_init_cursor返回 False 时（DB 不可读），_cursor_inited保持 False，下一轮重试
3.2 解析层：FriendRequestParser
3.2.1 XML 解析
fmessage 系统消息的 message_content 是 XML 字符串。解析逻辑独立为无状态函数，便于单测：
# 新文件：messaging/friend_parser.py
import xml.etree.ElementTree as ET
import re
def parse_friend_request(content: str, create_time: int, msg_local_id: int) -> Optional[FriendRequestInfo]:
"""解析 fmessage 系统消息 XML，提取好友申请信息。
Args:
content: message_content 原始文本（XML）
create_time: 消息时间戳
msg_local_id: 消息 local_id
Returns:
FriendRequestInfo 或 None（解析失败 / 非 verifyUser 类型）
"""
if not content or not content.strip():
return None
try:
root = ET.fromstring(content)
except ET.ParseError:
# 部分消息可能不是合法 XML（如纯文本通知），跳过
return None
# 检查是否为 verifyUser 类型 sysmsg
msg_type = root.get("type", "")
if msg_type != "verifyUser":
return None
# 提取 Link 节点中的申请人信息
link = root.find(".//Link")
if link is None:
return None
stranger_wxid = _text(link, "UserName")
nickname = _text(link, "NickName")
verify_message = _text(link, "Content")
scene = _text(link, "Scene")
if not stranger_wxid:
return None
return FriendRequestInfo(
stranger_wxid=stranger_wxid,
nickname=nickname,
verify_message=verify_message,
scene=scene,
raw_xml=content,
create_time=create_time,
msg_local_id=msg_local_id,
)
def _text(parent: ET.Element, tag: str) -> str:
el = parent.find(tag)
return el.text.strip() if el is not None and el.text else ""
3.2.2 容错策略
| 异常 | 处理 |
|---|---|
| XML 解析失败 | 返回 None，记 warning 日志，不中断轮询 |
| 非 verifyUser 类型 | 返回 None（fmessage 也承载好友通过/拒绝通知，需过滤） |
| UserName 缺失 | 返回 None（无法定位申请人） |
| NickName 缺失 | 空字符串兜底（不影响通过操作） |
| Content 含非 UTF-8 字符 | errors="replace" 兜底（已在 _decompress_msg_content 处理） |
3.3 DB 读取层：DbReader 新增方法
3.3.1 get_friend_requests_since
# 在 db/reader.py 中新增
def get_friend_requests_since(
self, cursor_create_time: int, cursor_local_id: int = 0, limit: int = 50
) -> Optional[dict]:
"""查询 fmessage 会话中指定游标之后的好友申请消息。
直接查 Msg_<MD5("fmessage")> 单表，避免遍历所有分片。
复合游标 (create_time, local_id) 作为 tie-breaker，避免同秒消息丢失。
Args:
cursor_create_time: 上次处理到的 create_time（0 = 从最新开始）
cursor_local_id: 同 create_time 下已处理的 local_id（tie-breaker）
limit: 单次读取上限
Returns:
{"requests": list[dict], "next_create_time": int, "next_local_id": int}
或 None（DB 不可读）
Notes:
- _ensure_decrypted 可能抛 BridgeError(DB_ENCRYPTED / DB_NEED_INIT)，
调用方需捕获 BridgeError 而非仅 sqlite3.Error
- WCDB_CT_message_content 列可能不存在（非 WCDB 表），
用 PRAGMA 预探测而非硬编码 SELECT
"""
empty = {
"requests": [],
"next_create_time": cursor_create_time,
"next_local_id": cursor_local_id,
}
try:
db_path = self._ensure_decrypted("message/message_0.db")
except BridgeError:
# DB 加密/无 key：返回 None 让调用方知道 DB 不可读
return None
conn = sqlite3.connect(db_path, isolation_level=None)
conn.row_factory = sqlite3.Row
try:
# 计算 fmessage 分片表名
table_name = f"Msg_{hashlib.md5(b'fmessage').hexdigest()}"
# 验证表存在
cur = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
(table_name,),
)
if cur.fetchone() is None:
return empty
# 动态探测 WCDB_CT_message_content 列是否存在
cols = self._table_columns(conn, table_name)
has_ct_col = "WCDB_CT_message_content" in cols
ct_col = "WCDB_CT_message_content" if has_ct_col else "0 AS WCDB_CT_message_content"
# 复合游标查询（与 get_messages_since 一致的 tie-breaker）
sql = (
f"SELECT local_id, create_time, message_content, {ct_col} "
f"FROM [{table_name}] "
f"WHERE (create_time > ? OR (create_time = ? AND local_id > ?)) "
f"ORDER BY create_time ASC, local_id ASC LIMIT ?"
)
rows = conn.execute(
sql, (cursor_create_time, cursor_create_time, cursor_local_id, limit)
).fetchall()
requests = []
next_ct = cursor_create_time
next_lid = cursor_local_id
for row in rows:
content = self._decompress_msg_content(
row["message_content"],
row["WCDB_CT_message_content"],
)
requests.append({
"local_id": row["local_id"],
"create_time": row["create_time"],
"content": content,
})
next_ct = row["create_time"]
next_lid = row["local_id"]
return {
"requests": requests,
"next_create_time": next_ct,
"next_local_id": next_lid,
}
except sqlite3.Error:
return empty
finally:
conn.close()
3.3.2 verify_friend_accepted
# 在 db/reader.py 中新增
def verify_friend_accepted(self, stranger_wxid: str) -> bool:
"""校验好友申请是否已通过：contact 表中 @stranger 后缀消失。
通过前：username = "wxid_xxx@stranger"（[需环境验证] 后缀名）
通过后：username = "wxid_xxx"（后缀被移除，local_type 可能变为 1）
策略：用 base_wxid 查 contact 表，若存在 base_wxid 记录且不存在
base_wxid@stranger 记录，则判定已通过。合并为单条 SQL 避免 2 次查询。
Args:
stranger_wxid: 含 @ 后缀的 wxid
Returns:
True 表示已通过，False 表示仍待验证或 DB 不可读
Notes:
- _ensure_decrypted 抛 BridgeError 时返回 False（不影响 UI 操作结果）
- @stranger 后缀是假设，若真实后缀不同需调整 SQL LIKE 模式
"""
try:
db_path = self._ensure_decrypted("contact/contact.db")
except BridgeError:
return False
conn = sqlite3.connect(db_path, isolation_level=None)
conn.row_factory = sqlite3.Row
try:
table = self._find_contact_table(conn)
if table is None:
return False
# 动态探测 username 列名（候选: username / wxid）
cols = self._table_columns(conn, table)
username_col = self._pick_column(cols, ["username", "wxid"])
if username_col is None:
return False
base_wxid = stranger_wxid.split("@")[0]
# 单条 SQL：base_wxid 存在且不含 @ 的记录数 > 0
# 同时检查是否有 @stranger 后缀的记录残留
sql = (
f"SELECT "
f" SUM(CASE WHEN {username_col} = ? THEN 1 ELSE 0 END) AS base_cnt, "
f" SUM(CASE WHEN {username_col} LIKE ? THEN 1 ELSE 0 END) AS stranger_cnt "
f"FROM {table}"
)
cur = conn.execute(sql, (base_wxid, f"{base_wxid}@%"))
row = cur.fetchone()
if row is None:
return False
base_cnt = row["base_cnt"] or 0
stranger_cnt = row["stranger_cnt"] or 0
# base_wxid 存在（已变为好友）且无 @stranger 残留
return base_cnt > 0 and stranger_cnt == 0
except sqlite3.Error:
return False
finally:
conn.close()
3.3.3 get_max_create_time_for_talker
# 在 db/reader.py 中新增
def get_max_create_time_for_talker(self, talker: str) -> Optional[int]:
"""获取指定会话的消息最大 create_time（游标初始化用）。
直接查 Msg_<MD5(talker)> 单表，避免遍历所有分片。
返回 None 表示 DB 不可读（与 get_max_create_time 语义一致），
返回 0 表示表为空或不存在。
"""
try:
db_path = self._ensure_decrypted("message/message_0.db")
except BridgeError:
return None
conn = sqlite3.connect(db_path, isolation_level=None)
try:
table_name = f"Msg_{hashlib.md5(talker.encode()).hexdigest()}"
cur = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
(table_name,),
)
if cur.fetchone() is None:
return 0
cur = conn.execute(f"SELECT MAX(create_time) FROM [{table_name}]")
row = cur.fetchone()
return int(row[0]) if row and row[0] else 0
except sqlite3.Error:
return 0
finally:
conn.close()
3.4 规则匹配引擎：AcceptRuleEngine
3.4.1 配置模型
# 在 models/contact.py 中新增
class AcceptRuleConfig(BaseModel):
"""自动通过规则配置。"""
enabled: bool = Field(default=False, description="全局开关")
accept_all: bool = Field(default=False, description="通过所有申请（忽略以下规则）")
# 白名单：匹配任一规则即通过
whitelist_wxids: list[str] = Field(
default_factory=list, description="白名单 wxid 列表"
)
whitelist_nicknames: list[str] = Field(
default_factory=list, description="白名单昵称列表（精确匹配）"
)
# 关键词：验证消息含任一关键词即通过
keywords: list[str] = Field(
default_factory=list, description="验证消息关键词列表（子串匹配）"
)
# 黑名单：匹配任一规则即拒绝（优先级高于白名单和关键词）
blacklist_wxids: list[str] = Field(
default_factory=list, description="黑名单 wxid 列表"
)
blacklist_nicknames: list[str] = Field(
default_factory=list, description="黑名单昵称列表"
)
# 场景过滤：仅通过指定来源场景的申请
allow_scenes: list[str] = Field(
default_factory=list, description="允许的场景（空=不限制）"
)
class AcceptDecision(enum.Enum):
"""规则匹配决策结果。"""
ACCEPT = "accept"
REJECT = "reject"
SKIP = "skip" # 不匹配任何规则，跳过
class AcceptRuleEngine:
"""好友申请自动通过规则引擎。"""
def __init__(self, config: AcceptRuleConfig) -> None:
self._config = config
self._lock = asyncio.Lock() # 配置热更新保护
async def get_config(self) -> AcceptRuleConfig:
"""获取当前规则配置（供 API 查询）。"""
async with self._lock:
return self._config
async def evaluate(self, req: FriendRequestInfo) -> AcceptDecision:
async with self._lock:
cfg = self._config
if not cfg.enabled:
return AcceptDecision.SKIP
if cfg.accept_all:
return AcceptDecision.ACCEPT
# 黑名单优先
if req.stranger_wxid in cfg.blacklist_wxids:
return AcceptDecision.REJECT
if req.nickname in cfg.blacklist_nicknames:
return AcceptDecision.REJECT
# 场景过滤
if cfg.allow_scenes and req.scene not in cfg.allow_scenes:
return AcceptDecision.SKIP
# 白名单
if req.stranger_wxid in cfg.whitelist_wxids:
return AcceptDecision.ACCEPT
if req.nickname in cfg.whitelist_nicknames:
return AcceptDecision.ACCEPT
# 关键词
if cfg.keywords:
for kw in cfg.keywords:
if kw in req.verify_message:
return AcceptDecision.ACCEPT
# 不匹配任何规则
return AcceptDecision.SKIP
async def update_config(self, config: AcceptRuleConfig) -> None:
async with self._lock:
self._config = config
3.4.2 匹配优先级
黑名单 (wxid / 昵称) ──→ REJECT（最高优先级）
│ 不命中
▼
场景过滤 (allow_scenes) ──→ SKIP（不在允许场景内）
│ 通过
▼
白名单 (wxid / 昵称) ──→ ACCEPT
│ 不命中
▼
关键词 (验证消息子串) ──→ ACCEPT
│ 不命中
▼
默认 ──→ SKIP（不处理，等待人工介入）
3.5 UI 自动化层：accept_friend_request
3.5.1 实现方案
在 xdotool_driver.py 中新增 accept_friend_request 方法，复用现有的 _step 超时模型和窗口几何计算：
# 在 ui/xdotool_driver.py 中新增
# 通讯录图标和新的朋友入口的几何常量（基于 1920x1080，坐标为估算值需实测校准）
_CONTACT_ICON_X_RATIO = 0.04 # 通讯录图标 X 比例（左侧导航栏第 2 个图标）
_CONTACT_ICON_Y_RATIO = 0.15 # 通讯录图标 Y 比例
_NEW_FRIEND_ENTRY_Y_RATIO = 0.22 # "新的朋友"入口 Y 比例（通讯录页顶部）
_VERIFY_BUTTON_X_OFFSET = -80 # "前往验证"按钮距右侧偏移（向左为负）
_VERIFY_BUTTON_Y_RATIO = 0.30 # 第一条申请项"前往验证"按钮 Y 比例
_CONFIRM_BUTTON_Y_OFFSET = -60 # "确定"按钮距底部偏移（向上为负）
async def accept_friend_request(
self,
stranger_wxid: str = "",
nickname: str = "",
timeout_sec: float = 25.0,
) -> bool:
"""通过好友申请（experimental）。
UI 路径（基于截图 04/05）：
1. 激活微信窗口
2. 点击左侧导航栏"通讯录"图标
3. 点击"新的朋友"入口
4. 定位目标申请项（按昵称匹配，空则取第一条）
5. 点击该项右侧"前往验证"按钮
6. 在弹窗中点击"确定"
7. Esc 返回主界面
Args:
stranger_wxid: 申请人 wxid（用于日志，当前版本不参与 UI 定位）
nickname: 申请人昵称（用于在列表中匹配定位，空则取第一条）
timeout_sec: 整体超时
Returns:
True 表示 UI 操作流程执行完成
Raises:
BridgeError(WINDOW_NOT_FOUND / SEND_FAILED)
Notes:
- 使用 _click_at（链式命令）而非 _click（两条命令），减少子进程创建
- 当前版本仅点击列表第一条申请项，昵称匹配待后续实现
- 所有坐标均为估算值，需在目标分辨率实测调优
"""
deadline = time.monotonic() + timeout_sec
async def _step(coro: Awaitable[None], desc: str) -> None:
remaining = deadline - time.monotonic()
if remaining <= 0:
if asyncio.iscoroutine(coro):
coro.close()
raise BridgeError(
code="SEND_FAILED",
message=f"通过好友申请超时: {desc}",
)
try:
await asyncio.wait_for(
coro, timeout=max(_STEP_MIN_TIMEOUT_SEC, remaining)
)
except asyncio.TimeoutError as exc:
raise BridgeError(
code="SEND_FAILED",
message=f"通过好友申请步骤超时: {desc}",
) from exc
# 1. 激活窗口
window_id = await self.find_wechat_window()
if window_id is None:
raise BridgeError(
code="WINDOW_NOT_FOUND",
message="未找到微信窗口，无法通过好友申请",
)
await _step(self._activate_window_fast(window_id), "激活窗口")
await _step(self._sleep(0.3), "等待窗口激活")
# 关闭可能存在的弹窗
await _step(self._key("Escape"), "关闭弹窗")
await _step(self._sleep(0.2), "等待 Esc 生效")
# 2. 获取窗口几何
win_x, win_y, win_w, win_h = await self._get_window_geometry()
# 3. 点击通讯录图标（左侧导航栏第 2 个）
contact_x = win_x + int(win_w * _CONTACT_ICON_X_RATIO)
contact_y = win_y + int(win_h * _CONTACT_ICON_Y_RATIO)
await _step(self._click_at(contact_x, contact_y, "通讯录图标"), "点击通讯录图标")
await _step(self._sleep(0.8), "等待通讯录页面加载")
# 4. 点击"新的朋友"入口
new_friend_y = win_y + int(win_h * _NEW_FRIEND_ENTRY_Y_RATIO)
await _step(self._click_at(contact_x, new_friend_y, "新的朋友"), "点击新的朋友")
await _step(self._sleep(0.8), "等待新的朋友列表加载")
# 5. 定位并点击目标申请项的"前往验证"按钮
# 当前版本取第一条申请项（昵称匹配待后续实现）
verify_x = win_x + win_w + _VERIFY_BUTTON_X_OFFSET # 负偏移 = 向左
verify_y = win_y + int(win_h * _VERIFY_BUTTON_Y_RATIO)
await _step(self._click_at(verify_x, verify_y, "前往验证"), "点击前往验证")
await _step(self._sleep(1.0), "等待验证弹窗打开")
# 6. 点击弹窗"确定"按钮
confirm_x = win_x + win_w // 2
confirm_y = win_y + win_h + _CONFIRM_BUTTON_Y_OFFSET # 负偏移 = 向上
await _step(self._click_at(confirm_x, confirm_y, "确定"), "点击确定")
await _step(self._sleep(0.8), "等待通过操作完成")
# 7. 返回主界面
await _step(self._key("Escape"), "返回主界面")
await _step(self._sleep(0.3), "等待返回")
logger.info(
"accept_friend_request: wxid=%s nickname=%s → UI 操作完成",
stranger_wxid, nickname,
)
return True
3.5.2 Profile 扩展
在 ui/profiles/ YAML 中新增好友申请相关元素定位：
# 追加到 default.yaml / wechat_4.0_1280x720.yaml / wechat_4.0_1920x1080.yaml
contact_icon:
by_geom:
relative_to: window
x_ratio: 0.04
y_ratio: 0.15
x_offset: 0
y_offset: 0
threshold: 0.80
require_image: false
description: 左侧导航栏通讯录图标
new_friend_entry:
by_geom:
relative_to: window
x_ratio: 0.04
y_ratio: 0.22
x_offset: 0
y_offset: 0
threshold: 0.80
require_image: false
description: 通讯录页"新的朋友"入口
verify_button:
by_geom:
relative_to: window_bottom_right
x_ratio: 1.0
y_ratio: 0.30
x_offset: -80
y_offset: 0
threshold: 0.75
require_image: false
description: 好友申请项"前往验证"按钮
confirm_button:
by_geom:
relative_to: window_bottom_right
x_ratio: 0.5
y_ratio: 1.0
x_offset: 0
y_offset: -60
threshold: 0.75
require_image: false
description: "通过朋友验证"弹窗确定按钮
3.6 去重与入队
3.6.1 去重策略
使用 IdemCache 按 stranger_wxid 去重，防止同一申请被重复处理。
关键接口签名（idem_cache.py:61-67、idem_cache.py:98-105）：
def get(self, flow_name: str, to_wxid: str, content: str, client_request_id: str = "") -> Optional[Any] def set(self, flow_name: str, to_wxid: str, content: str, value: Any, client_request_id: str = "") -> None注意：
set的value在client_request_id之前，不能按位置传client_request_id，否则会被当成value。SendQueue.enqueue 是延迟执行（send_queue.py:65-94）：
coro_factory入队后由 worker 协程_run调用await coro_factory()（send_queue.py:145）。enqueue的await实际等待的是future（worker 设置结果），不是coro_factory()本身。 因此循环中创建的 lambda 若直接捕获循环变量，会在 worker 真正执行时拿到被覆盖的值 —— 必须用默认参数捕获。
async def _handle_request(self, req: FriendRequestInfo) -> None:
self._processed_count += 1
self._last_processed_time = req.create_time
# 1. 规则匹配
decision = await self._rule_engine.evaluate(req)
if decision == AcceptDecision.REJECT:
self._rejected_count += 1
logger.info(
"friend_request: wxid=%s nickname=%s → decision=%s（拒绝）",
req.stranger_wxid, req.nickname, decision.value,
)
return
if decision != AcceptDecision.ACCEPT:
logger.info(
"friend_request: wxid=%s nickname=%s → decision=%s（跳过）",
req.stranger_wxid, req.nickname, decision.value,
)
return
# 2. 幂等去重（TTL=300s，5 分钟内不重复处理同一申请人）
# content 参数传空串（去重 key 是 stranger_wxid，无内容维度）
cached = self._idem_cache.get(
"friend_accept", req.stranger_wxid, "", ""
)
if cached is not None:
logger.info(
"friend_request: wxid=%s → 幂等命中，跳过",
req.stranger_wxid,
)
return
# 3. 熔断检查
if not self._breaker.allow():
logger.warning(
"friend_request: wxid=%s → 熔断器 OPEN，跳过",
req.stranger_wxid,
)
return
# 4. 入队执行（lambda 用默认参数捕获 req，避免延迟执行时变量被覆盖）
try:
result = await self._send_queue.enqueue(
lambda req=req: self._xdotool.accept_friend_request(
stranger_wxid=req.stranger_wxid,
nickname=req.nickname,
),
delay_ms=None, # 使用 send_queue 默认间隔
)
# 5. 等待微信 DB WAL 刷盘后校验（微信 DB 写入有 1-3s 延迟）
await asyncio.sleep(2.0)
verified = await asyncio.to_thread(
self._db_reader.verify_friend_accepted, req.stranger_wxid
)
if verified:
# set 签名: (flow_name, to_wxid, content, value, client_request_id)
self._idem_cache.set(
"friend_accept", req.stranger_wxid, "",
{"success": True, "verified": True}, "",
)
self._breaker.record_success()
self._accepted_count += 1
logger.info(
"friend_request: wxid=%s nickname=%s → 通过并验证成功",
req.stranger_wxid, req.nickname,
)
else:
# UI 操作完成但 DB 校验未通过（可能有延迟）
self._idem_cache.set(
"friend_accept", req.stranger_wxid, "",
{"success": True, "verified": False}, "",
)
self._breaker.record_failure()
logger.warning(
"friend_request: wxid=%s → UI 操作完成但 DB 校验未通过",
req.stranger_wxid,
)
except BridgeError as e:
# 透传 BridgeError 错误码（RATE_LIMITED / SEND_FAILED / WINDOW_NOT_FOUND 等）
self._breaker.record_failure()
logger.error(
"friend_request: wxid=%s → BridgeError code=%s: %s",
req.stranger_wxid, getattr(e, "code", "UNKNOWN"), e,
)
except Exception as e:
self._breaker.record_failure()
logger.error(
"friend_request: wxid=%s → 失败: %s",
req.stranger_wxid, e,
)
3.6.2 DB 校验延迟
微信 DB 写入有 WAL 延迟（通常 1-3 秒）。DB 校验应在 UI 操作完成后等待 2 秒再查（已整合到 3.6.1 的 _handle_request 中，见上方 await asyncio.sleep(2.0)）。
注意：
SendQueue.enqueue是await future语义 ——enqueue返回时 UI 操作已完成（或抛错）。因此sleep(2.0)放在enqueue之后即可，无需在 lambda 内部 sleep。
3.7 API 路由层
3.7.1 新增路由
# 在 routes/contacts.py 中新增
@router.get("/api/friends/requests", response_model=FriendRequestsResponse)
@with_db_retry
async def list_friend_requests(limit: int = 50) -> FriendRequestsResponse:
"""查询待处理的好友申请列表（从 fmessage 系统消息解析）。
Args:
limit: 1~200，默认 50
Returns:
FriendRequestsResponse：含 requests 列表 / total
"""
if limit < 1 or limit > 200:
raise BridgeError(
code="INVALID_PARAMS",
message=f"limit 必须在 1~200 之间，收到 {limit}",
)
db_reader = _require_db_reader()
await _check_db_readable()
# get_friend_requests_since(cursor_create_time, cursor_local_id, limit)
# 查询全部待处理申请：cursor_create_time=0, cursor_local_id=0
result = await asyncio.to_thread(
db_reader.get_friend_requests_since, 0, 0, limit
)
if result is None:
# DB 不可读（_ensure_decrypted 抛 BridgeError 已被 _check_db_readable 拦截，
# 此处兜底防御）
return FriendRequestsResponse(requests=[], total=0)
# 解析 XML 提取结构化信息
requests = []
for item in result.get("requests", []):
info = parse_friend_request(
item["content"], item["create_time"], item["local_id"]
)
if info is not None:
requests.append(FriendRequestItem(
stranger_wxid=info.stranger_wxid,
nickname=info.nickname,
verify_message=info.verify_message,
scene=info.scene,
create_time=info.create_time,
))
return FriendRequestsResponse(requests=requests, total=len(requests))
@router.post("/api/friends/accept", response_model=AcceptFriendResponse)
async def accept_friend(req: AcceptFriendRequest) -> AcceptFriendResponse:
"""手动通过好友申请（experimental）。
Args:
req: AcceptFriendRequest
Returns:
AcceptFriendResponse
"""
if not req.stranger_wxid:
raise BridgeError(code="INVALID_PARAMS", message="stranger_wxid 不能为空")
xdotool = _require_xdotool()
send_queue = _require_send_queue()
login_state = await xdotool.detect_login_state()
if login_state != "logged_in":
raise BridgeError(
code="WECHAT_NOT_LOGGED_IN",
message=f"当前登录态为 {login_state}，无法通过好友申请",
)
try:
# lambda 用默认参数捕获 req（SendQueue.enqueue 延迟执行，见 3.6.1 说明）
await send_queue.enqueue(
lambda req=req: xdotool.accept_friend_request(
stranger_wxid=req.stranger_wxid,
nickname=req.nickname or "",
)
)
except BridgeError:
# 透传 BridgeError（RATE_LIMITED / SEND_FAILED / WINDOW_NOT_FOUND 等）
raise
except Exception as e:
raise BridgeError(code="SEND_FAILED", message=f"通过好友申请失败: {e}")
logger.info(
"friends/accept: wxid=%s nickname=%s → UI 操作完成",
req.stranger_wxid, req.nickname,
)
return AcceptFriendResponse(success=True, error=None)
@router.get("/api/friends/auto_accept/config", response_model=AcceptRuleConfig)
async def get_auto_accept_config() -> AcceptRuleConfig:
"""查询自动通过规则配置。"""
rule_engine = _require_rule_engine()
return await rule_engine.get_config()
@router.put("/api/friends/auto_accept/config", response_model=AcceptRuleConfig)
async def update_auto_accept_config(config: AcceptRuleConfig) -> AcceptRuleConfig:
"""更新自动通过规则配置（热生效）。"""
rule_engine = _require_rule_engine()
await rule_engine.update_config(config)
# 同步更新 watcher 的 enabled 状态
watcher = _require_friend_watcher()
watcher.set_enabled(config.enabled)
return config
@router.get("/api/friends/auto_accept/status", response_model=AutoAcceptStatus)
async def get_auto_accept_status() -> AutoAcceptStatus:
"""查询自动通过运行状态。"""
watcher = _require_friend_watcher()
return AutoAcceptStatus(
running=watcher.is_running,
enabled=watcher.is_enabled,
processed_count=watcher.processed_count,
accepted_count=watcher.accepted_count,
rejected_count=watcher.rejected_count,
last_processed_time=watcher.last_processed_time,
)
3.7.2 新增模型
# 在 models/contact.py 中新增
class FriendRequestItem(BaseModel):
"""好友申请条目。"""
stranger_wxid: str
nickname: str = ""
verify_message: str = ""
scene: str = ""
create_time: int
class FriendRequestsResponse(BaseModel):
"""好友申请列表响应。"""
requests: list[FriendRequestItem] = Field(default_factory=list)
total: int = 0
class AcceptFriendRequest(BaseModel):
"""手动通过好友申请请求。"""
stranger_wxid: str = Field(description="申请人 wxid（含 @stranger 后缀）")
nickname: Optional[str] = Field(default=None, description="申请人昵称（UI 定位用）")
class AcceptFriendResponse(BaseModel):
"""通过好友申请响应。"""
success: bool = Field(default=False)
error: Optional[str] = Field(default=None)
class AutoAcceptStatus(BaseModel):
"""自动通过运行状态。"""
running: bool
enabled: bool
processed_count: int = 0
accepted_count: int = 0
rejected_count: int = 0
last_processed_time: Optional[int] = None
3.8 配置层
3.8.1 环境变量
# .env 新增
# 自动通过全局开关（false=关闭，true=开启）
WOC_AUTO_ACCEPT_ENABLED=false
# 通过所有申请（忽略规则，危险！仅测试用）
WOC_AUTO_ACCEPT_ALL=false
# 白名单 wxid（逗号分隔）
WOC_AUTO_ACCEPT_WHITELIST_WXIDS=
# 白名单昵称（逗号分隔）
WOC_AUTO_ACCEPT_WHITELIST_NICKNAMES=
# 验证消息关键词（逗号分隔）
WOC_AUTO_ACCEPT_KEYWORDS=
# 黑名单 wxid（逗号分隔）
WOC_AUTO_ACCEPT_BLACKLIST_WXIDS=
# 黑名单昵称（逗号分隔）
WOC_AUTO_ACCEPT_BLACKLIST_NICKNAMES=
# 允许场景（逗号分隔，空=不限制）
WOC_AUTO_ACCEPT_ALLOW_SCENES=
# 轮询间隔（秒）
WOC_AUTO_ACCEPT_POLL_INTERVAL=3
3.8.2 BridgeConfig 扩展
# 在 config.py BridgeConfig 中新增字段
@dataclass
class BridgeConfig:
# ... 现有字段 ...
# 好友自动通过配置
auto_accept_enabled: bool = False
auto_accept_all: bool = False
auto_accept_whitelist_wxids: list[str] = field(default_factory=list)
auto_accept_whitelist_nicknames: list[str] = field(default_factory=list)
auto_accept_keywords: list[str] = field(default_factory=list)
auto_accept_blacklist_wxids: list[str] = field(default_factory=list)
auto_accept_blacklist_nicknames: list[str] = field(default_factory=list)
auto_accept_allow_scenes: list[str] = field(default_factory=list)
auto_accept_poll_interval: float = 3.0
@classmethod
def from_args_and_env(cls, args):
return cls(
# ... 现有字段 ...
auto_accept_enabled=os.environ.get(
"WOC_AUTO_ACCEPT_ENABLED", "false"
).lower() in ("true", "1", "yes", "on"),
auto_accept_all=os.environ.get(
"WOC_AUTO_ACCEPT_ALL", "false"
).lower() in ("true", "1", "yes", "on"),
auto_accept_whitelist_wxids=_parse_list(
os.environ.get("WOC_AUTO_ACCEPT_WHITELIST_WXIDS", "")
),
auto_accept_whitelist_nicknames=_parse_list(
os.environ.get("WOC_AUTO_ACCEPT_WHITELIST_NICKNAMES", "")
),
auto_accept_keywords=_parse_list(
os.environ.get("WOC_AUTO_ACCEPT_KEYWORDS", "")
),
auto_accept_blacklist_wxids=_parse_list(
os.environ.get("WOC_AUTO_ACCEPT_BLACKLIST_WXIDS", "")
),
auto_accept_blacklist_nicknames=_parse_list(
os.environ.get("WOC_AUTO_ACCEPT_BLACKLIST_NICKNAMES", "")
),
auto_accept_allow_scenes=_parse_list(
os.environ.get("WOC_AUTO_ACCEPT_ALLOW_SCENES", "")
),
auto_accept_poll_interval=float(
os.environ.get("WOC_AUTO_ACCEPT_POLL_INTERVAL", "3")
),
)
def _parse_list(s: str) -> list[str]:
"""逗号分隔字符串转列表。"""
if not s.strip():
return []
return [item.strip() for item in s.split(",") if item.strip()]
3.9 生命周期与初始化
3.9.1 AppState 扩展
# 在 config.py AppState 中新增字段
class AppState:
# ... 现有字段 ...
# 好友自动通过组件
friend_watcher: Optional[Any] = None # FriendRequestWatcher
rule_engine: Optional[Any] = None # AcceptRuleEngine
accept_breaker: Optional[Any] = None # CircuitBreaker
3.9.2 初始化序列
# 在 app.py _init_state 中新增
def _init_state(config: BridgeConfig) -> None:
# ... 现有初始化 ...
# 好友自动通过规则引擎
accept_config = AcceptRuleConfig(
enabled=config.auto_accept_enabled,
accept_all=config.auto_accept_all,
whitelist_wxids=config.auto_accept_whitelist_wxids,
whitelist_nicknames=config.auto_accept_whitelist_nicknames,
keywords=config.auto_accept_keywords,
blacklist_wxids=config.auto_accept_blacklist_wxids,
blacklist_nicknames=config.auto_accept_blacklist_nicknames,
allow_scenes=config.auto_accept_allow_scenes,
)
_state.rule_engine = AcceptRuleEngine(accept_config)
# 熔断器
_state.accept_breaker = CircuitBreaker(
"accept_verify", failure_threshold=5, recovery_timeout=60.0,
)
# 好友申请监听器
_state.friend_watcher = FriendRequestWatcher(
db_reader=_state.db_reader,
rule_engine=_state.rule_engine,
send_queue=_state.send_queue,
idem_cache=_state.idem_cache,
xdotool_driver=_state.xdotool,
breaker=_state.accept_breaker,
poll_interval=config.auto_accept_poll_interval,
)
3.9.3 lifespan 启停
现有 lifespan 启动顺序（app.py:64-129）：
send_queue.start()→message_streamer.start()→watchdog.start()→resource_reaper.start()→xdotool._full_cleanup_on_startup()→yield停止顺序（反序）：
message_streamer.stop()→send_queue.stop()→watchdog.stop()→resource_reaper.stop()，每个均包asyncio.wait_for(..., timeout=5.0)。FriendRequestWatcher 应在
message_streamer.start()之后启动（不依赖 streamer，但保持"基础设施先于业务"顺序），在message_streamer.stop()之前停止（先停业务再停基础设施）。
# 在 app.py lifespan 中新增
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# ... 现有启动（send_queue / message_streamer / watchdog / resource_reaper / xdotool cleanup）...
# 启动好友申请监听器（仅当 auto_accept 开启时）
# 放在 message_streamer 之后，保持"基础设施先于业务"顺序
if _state.friend_watcher is not None and _state.config.auto_accept_enabled:
try:
await _state.friend_watcher.start()
except Exception as e:
logger.warning("[lifespan] friend_watcher.start() 失败: %s", e)
try:
yield
finally:
# 先停 friend_watcher（业务），再停基础设施
if _state.friend_watcher is not None:
try:
await asyncio.wait_for(
_state.friend_watcher.stop(), timeout=5.0
)
except Exception:
# Exception 已涵盖 TimeoutError，无需单独列
logger.warning("[lifespan] friend_watcher.stop() 超时或异常")
# ... 现有停止（message_streamer / send_queue / watchdog / resource_reaper）...
修正说明：
- 原
except (asyncio.TimeoutError, Exception)冗余 ——Exception已涵盖TimeoutError，统一为except Exception- 启动时增加异常捕获，避免 friend_watcher 启动失败阻塞 lifespan
- 启动顺序：在
message_streamer.start()之后（friend_watcher 不依赖 streamer，但保持基础设施优先）- 停止顺序：在
message_streamer.stop()之前（先停业务再停基础设施）
3.9.4 依赖检查函数
# 在 config.py 中新增
def _require_rule_engine() -> AcceptRuleEngine:
if _state.rule_engine is None:
raise BridgeError(
code="BRIDGE_INTERNAL_ERROR",
message="rule_engine 未初始化",
)
return _state.rule_engine
def _require_friend_watcher() -> FriendRequestWatcher:
if _state.friend_watcher is None:
raise BridgeError(
code="BRIDGE_INTERNAL_ERROR",
message="friend_watcher 未初始化",
)
return _state.friend_watcher
3.9.5 模型导出
当前
models/__init__.py（models/init.py:68-87）共导出 36 个名字，不包含任何好友申请相关模型。新增的 6 个模型必须追加到__all__列表，否则from woc_bridge.models import AcceptRuleConfig等导入会失败。
# 在 models/__init__.py 中新增导入和导出
from woc_bridge.models.contact import ( # 假设新增模型定义在 contact.py
AcceptRuleConfig,
AcceptDecision,
FriendRequestItem,
FriendRequestsResponse,
AcceptFriendRequest,
AcceptFriendResponse,
AutoAcceptStatus,
)
# 追加到 __all__ 列表（按字母序插入联系人域之后）
__all__ = [
# ... 现有 36 个名字 ...
# 好友自动通过域
"AcceptRuleConfig",
"AcceptDecision",
"FriendRequestItem",
"FriendRequestsResponse",
"AcceptFriendRequest",
"AcceptFriendResponse",
"AutoAcceptStatus",
]
注意：
AcceptDecision是枚举，若 API 不直接返回可不导出（但路由内部会用到，建议导出）- 模型定义文件选择：优先追加到现有
models/contact.py（与好友/联系人语义一致），避免新建文件routes/contacts.py现有 import 风格是from woc_bridge.models import (...)，新增模型需在同一 import 语句中追加
4. 关键设计决策
4.1 为什么不修改 MessageStreamer
| 考量 | 说明 |
|---|---|
| 架构纯净性 | MessageStreamer 设计为"纯队列广播模型"（streamer.py:14-20），注入 handler 会破坏其设计 |
| 职责单一 | Streamer 负责消息分发，不负责业务逻辑（如 XML 解析、规则匹配） |
| 独立游标 | fmessage 查询单表，开销极小（<1ms），独立游标避免与消息流游标耦合 |
| 独立频率 | 好友申请低频事件，3s 轮询足够；消息流需要 1s 高频 |
| 独立启停 | auto_accept 关闭时 Watcher 挂起，不影响消息流 |
4.2 为什么用 DB 轮询而非 UI 监听
| 方案 | 优点 | 缺点 |
|---|---|---|
| DB 轮询（选择） | 精确、可靠、不依赖 UI 状态、可解析结构化信息 | 有 WAL 延迟（1-3s） |
| UI 轮询 | 实时性好 | 微信 4.x 自绘 UI 无法检测"新的朋友"角标变化；UI 状态不可靠 |
4.3 为什么 UI 操作走 SendQueue
- 避免 xdotool 命令冲突：所有 UI 操作（发消息、通过好友、发朋友圈）共享一个微信窗口，必须串行
- 限流保护：SendQueue 的滑动窗口限流防止高频操作触发微信风控
- 队列满保护：
RATE_LIMITED错误让调用方知道系统繁忙，可稍后重试
4.4 为什么 DB 校验而非 UI 截图校验
- 可靠性：DB 是微信真实状态，UI 截图受分辨率/主题/动画影响
- 开销：DB 查询单条记录 <1ms，截图 + OpenCV 匹配 100-300ms
- 已有基础设施：
_ensure_decrypted+sqlite3已有完整解密缓存机制
4.5 熔断器参数
| 参数 | 值 | 理由 |
|---|---|---|
failure_threshold |
5 | 连续 5 次 UI 操作失败（可能是微信未登录/窗口异常），停止自动通过 |
recovery_timeout |
60s | 1 分钟后尝试恢复（比 send_text 的 30s 更长，因为好友通过低频） |
5. 异常处理矩阵
| 异常场景 | 处理策略 | 错误码 |
|---|---|---|
| DB 加密 | Watcher 暂停，10s 后重试，等待 key 提取 | — |
| DB 不可读 | Watcher 暂停，5s 后重试 | — |
| XML 解析失败 | 跳过该消息，记 warning，不中断轮询 | — |
| 非 verifyUser 类型 | 跳过（fmessage 也承载通过/拒绝通知） | — |
| 规则不匹配 | 跳过，记 info 日志 | — |
| 幂等命中 | 跳过，记 info 日志 | — |
| 熔断器 OPEN | 跳过，记 warning | — |
| SendQueue 满 | 抛 RATE_LIMITED，Watcher 记 error，下一轮重试 |
RATE_LIMITED |
| UI 操作超时 | 抛 SEND_FAILED，熔断器记录失败 |
SEND_FAILED |
| UI 操作失败 | 抛 SEND_FAILED，熔断器记录失败 |
SEND_FAILED |
| DB 校验未通过 | UI 成功但校验失败，熔断器记录失败，记 warning | — |
| 微信未登录 | detect_login_state 检测，抛 WECHAT_NOT_LOGGED_IN |
WECHAT_NOT_LOGGED_IN |
| 窗口未找到 | 抛 WINDOW_NOT_FOUND |
WINDOW_NOT_FOUND |
| Watcher 协程异常 | 记 exception 日志，5s 后继续轮询 | — |
6. 安全考虑
6.1 防止恶意批量加好友
- SendQueue 限流：
max_calls_per_sec=10，send_delay_ms=3000，即每 3 秒最多通过 1 个好友 - IdemCache 去重：同一
stranger_wxid5 分钟内不重复处理 - 熔断器：连续 5 次失败暂停 60 秒
6.2 规则配置安全
accept_all=true是危险配置，仅用于测试环境- 黑名单优先级最高，防止恶意用户反复申请
- API 修改配置需 Bearer Token 认证（已有
WOC_BRIDGE_API_TOKEN机制）
6.3 日志审计
所有自动通过操作记录完整日志：
friend_request: wxid=xxx@stranger nickname=张三 → decision=accept
friend_request: wxid=xxx@stranger nickname=张三 → UI 操作完成
friend_request: wxid=xxx@stranger nickname=张三 → 通过并验证成功
7. 测试策略
7.1 单元测试
| 模块 | 测试重点 |
|---|---|
friend_parser.py |
XML 解析各种格式（正常/缺失字段/非 verifyUser/非法 XML） |
AcceptRuleEngine |
白名单/关键词/黑名单/场景过滤的优先级与组合 |
DbReader.get_friend_requests_since |
游标推进、空表、表不存在 |
DbReader.verify_friend_accepted |
@stranger 存在/消失、base_wxid 存在/不存在 |
7.2 集成测试
| 场景 | 验证点 |
|---|---|
| 正常自动通过 | fmessage 消息 → 规则匹配 → UI 操作 → DB 校验 → 幂等缓存 |
| 规则不匹配 | fmessage 消息 → 规则匹配 → SKIP → 不执行 UI 操作 |
| 重复申请 | 同一 stranger_wxid 第二次 → 幂等命中 → 跳过 |
| 熔断触发 | 连续 5 次 UI 失败 → 熔断器 OPEN → 后续申请跳过 |
| auto_accept 开关切换 | 关闭 → 开启 → 游标重置 → 处理新申请 |
7.3 UI 测试
需在真实微信 4.x Linux 环境验证：
- 通讯录图标坐标准确性
- "新的朋友"入口坐标准确性
- "前往验证"按钮坐标准确性
- "确定"按钮坐标准确性
- 多条申请时列表滚动定位
8. 文件变更清单
| 文件 | 变更类型 | 说明 |
|---|---|---|
messaging/friend_watcher.py |
新增 | FriendRequestWatcher 后台监听器 |
messaging/friend_parser.py |
新增 | XML sysmsg 解析器 |
db/reader.py |
修改 | 新增 3 个方法：get_friend_requests_since、verify_friend_accepted、get_max_create_time_for_talker |
ui/xdotool_driver.py |
修改 | 新增 accept_friend_request 方法 + UI 常量 |
ui/circuit_breaker.py |
不变 | 复用现有 CircuitBreaker |
models/contact.py |
修改 | 新增 6 个模型：FriendRequestItem、FriendRequestsResponse、AcceptFriendRequest、AcceptFriendResponse、AcceptRuleConfig、AutoAcceptStatus（+ AcceptDecision 枚举） |
models/__init__.py |
修改 | 追加 7 个名字到 __all__：AcceptRuleConfig、AcceptDecision、FriendRequestItem、FriendRequestsResponse、AcceptFriendRequest、AcceptFriendResponse、AutoAcceptStatus（详见 3.9.5） |
routes/contacts.py |
修改 | 新增 5 个路由：GET /api/friends/requests、POST /api/friends/accept、GET/PUT /api/friends/auto_accept/config、GET /api/friends/auto_accept/status；import 追加新模型 + _require_rule_engine + _require_friend_watcher + parse_friend_request |
config.py |
修改 | BridgeConfig 新增 10 个字段，AppState 新增 3 个字段，新增 _parse_list 函数 + 2 个 _require_* 函数 |
app.py |
修改 | _init_state 新增组件初始化（rule_engine / accept_breaker / friend_watcher），lifespan 新增 Watcher 启停（在 message_streamer 之后启动、之前停止） |
ui/profiles/default.yaml |
修改 | 新增 4 个元素定位（contact_icon / new_friend_entry / verify_button / confirm_button） |
ui/profiles/wechat_4.0_1280x720.yaml |
修改 | 同上 |
ui/profiles/wechat_4.0_1920x1080.yaml |
修改 | 同上 |
9. API 接口汇总
| 方法 | 路径 | 说明 | 认证 |
|---|---|---|---|
| GET | /api/friends/requests |
查询待处理好友申请列表 | Bearer Token |
| POST | /api/friends/accept |
手动通过好友申请 | Bearer Token |
| GET | /api/friends/auto_accept/config |
查询自动通过规则配置 | Bearer Token |
| PUT | /api/friends/auto_accept/config |
更新自动通过规则配置（热生效） | Bearer Token |
| GET | /api/friends/auto_accept/status |
查询自动通过运行状态 | Bearer Token |
10. 后续演进
10.1 P3 架构迁移
当前方案中 accept_friend_request 直接在 xdotool_driver.py 中实现（P1 模式）。后续可按 send_text 的 P3 迁移路径重构：
- 新增
AcceptFriendFlow（继承Flow，自定义AcceptFlowState枚举） FlowOrchestrator新增accept_friend编排方法（幂等 + 熔断 + 入队）WeChatCapabilities新增accept_friend能力 API
10.2 好友申请拒绝
当前方案仅实现自动通过（ACCEPT），后续可扩展自动拒绝（REJECT）：
xdotool_driver.py新增reject_friend_request方法- 规则引擎
AcceptDecision.REJECT时入队执行拒绝操作
10.3 消息通知
自动通过成功后，可通过 MessageStreamer 的 SSE 通道推送 friend_accepted 事件给客户端，实现实时通知。
10.4 统计与监控
在 ui/metrics.py 中新增 Prometheus 指标：
woc_friend_requests_total{decision="accept|reject|skip"}woc_friend_accept_success_totalwoc_friend_accept_failure_total{reason="ui_timeout|db_verify_failed|..."}