From 5d84df902a083e65b418bb10c7e5cdbfa88c2c59 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Fri, 10 Jul 2026 22:15:45 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E5=8F=91=E9=80=81=E6=96=B9=E5=90=91=E8=BF=87=E6=BB=A4=E3=80=81?= =?UTF-8?q?=E6=97=A5=E5=BF=97=E6=89=93=E7=82=B9=E4=B8=8E=E5=8F=91=E9=80=81?= =?UTF-8?q?=E7=A8=B3=E5=AE=9A=E6=80=A7=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 为消息拉取/搜索接口新增is_sender参数，支持按发送方向过滤消息 2. 为send_text接口添加全链路耗时打点与详细日志 3. 为发送队列添加限流、执行状态日志与耗时统计 4. 修复微信4.x自绘UI输入框焦点问题，新增点击输入框聚焦逻辑 5. 为xdotool命令添加超时保护，避免协程永久阻塞 6. 重构数据库查询逻辑，在SQL层提前过滤发送方向消息，减少数据传输 --- bridge/woc_bridge/db/reader.py | 156 ++++++++++++++++++---- bridge/woc_bridge/messaging/send_queue.py | 23 ++++ bridge/woc_bridge/routes/messages.py | 45 +++++-- bridge/woc_bridge/routes/send.py | 38 +++++- bridge/woc_bridge/ui/xdotool_driver.py | 104 +++++++++++++-- 5 files changed, 319 insertions(+), 47 deletions(-) diff --git a/bridge/woc_bridge/db/reader.py b/bridge/woc_bridge/db/reader.py index af1a78e..c331632 100644 --- a/bridge/woc_bridge/db/reader.py +++ b/bridge/woc_bridge/db/reader.py @@ -611,6 +611,46 @@ class DbReader: pass return id_to_username + def _resolve_self_rowid( + self, conn: sqlite3.Connection, self_wxid: str + ) -> Optional[int]: + """返回 self_wxid 在 Name2Id 表中的 rowid，用于 SQL 层按发送方向过滤。 + + 找不到时返回 None；调用方据此决定是否短路返回空结果 + （is_sender 非 None 但无法定位本人时无法准确过滤方向）。 + """ + if not self_wxid: + return None + try: + row = conn.execute( + "SELECT rowid FROM Name2Id WHERE user_name = ? LIMIT 1", + (self_wxid,), + ).fetchone() + if row is not None: + return int(row[0]) + except sqlite3.Error: + pass + return None + + @staticmethod + def _build_sender_filter( + is_sender: Optional[bool], self_rowid: Optional[int], + ) -> tuple[str, list]: + """构造发送方向 SQL 过滤片段。 + + Args: + is_sender: True=仅本人发送 / False=仅对方发送 / None=不过滤 + self_rowid: 当前账号在 Name2Id 中的 rowid + + Returns: + (sql_fragment, params)：sql_fragment 为 "" 表示无需过滤； + 否则形如 "real_sender_id = ?" 或 "real_sender_id != ?" + """ + if is_sender is None or self_rowid is None: + return "", [] + op = "=" if is_sender else "!=" + return f"real_sender_id {op} ?", [self_rowid] + def _resolve_msg_table_talkers(self, conn: sqlite3.Connection) -> dict[str, str]: """解析所有 Msg_* 分片表名到 talker username 的映射。 @@ -709,11 +749,24 @@ class DbReader: "session_type": session_type, } - def get_messages_since(self, cursor: int, limit: int = 50) -> dict: + def get_messages_since( + self, + cursor: int, + limit: int = 50, + is_sender: Optional[bool] = None, + ) -> dict: """读取 create_time > cursor 的增量消息。 WeChat 4.x 消息存储在 Msg_ 分片表中，每个会话一张表。 遍历所有分片表合并结果后按 create_time 排序返回。 + + Args: + cursor: 上次拉取的最大 create_time + limit: 最多返回条数 + is_sender: 发送方向过滤，True=仅本人发送 / False=仅对方发送 / + None=不过滤（默认）。SQL 层通过 real_sender_id 与 + Name2Id 中 self_wxid 的 rowid 比对实现，避免读出再裁剪。 + 当 is_sender 非 None 但无法定位本人 rowid 时返回空结果。 """ db_path = self._ensure_decrypted("message/message_0.db") conn = sqlite3.connect(db_path, isolation_level=None) @@ -728,15 +781,36 @@ class DbReader: id_to_username = self._load_name2id(conn) self_wxid = self._get_current_wxid() or "" + # 发送方向过滤：无法定位本人 rowid 时短路返回空 + sender_sql: str = "" + sender_params: list = [] + if is_sender is not None: + self_rowid = self._resolve_self_rowid(conn, self_wxid) + sender_sql, sender_params = self._build_sender_filter( + is_sender, self_rowid + ) + if not sender_sql: + return { + "messages": [], + "next_cursor": cursor, + "has_more": False, + } + all_rows: list[dict] = [] for table_name, talker in table_to_talker.items(): - cur = conn.execute( + where_parts = ["create_time > ?"] + params: list = [cursor] + if sender_sql: + where_parts.append(sender_sql) + params.extend(sender_params) + sql = ( f"SELECT local_id, local_type, create_time, real_sender_id, " f"message_content, WCDB_CT_message_content " - f"FROM [{table_name}] WHERE create_time > ? " - f"ORDER BY create_time ASC LIMIT ?", - (cursor, limit), + f"FROM [{table_name}] WHERE {' AND '.join(where_parts)} " + f"ORDER BY create_time ASC LIMIT ?" ) + params.append(limit) + cur = conn.execute(sql, tuple(params)) for row in cur.fetchall(): all_rows.append(self._row_to_message_dict( row, talker, id_to_username, self_wxid @@ -761,6 +835,7 @@ class DbReader: cursor: int = 0, limit: int = 50, direction: str = "before", + is_sender: Optional[bool] = None, ) -> dict: """按会话拉取历史消息（单会话，O(1) 定位分片表）。 @@ -774,6 +849,8 @@ class DbReader: after 模式取该时间之后的新消息 limit: 最多返回条数，1~200 direction: before（默认，往前翻历史）/ after（往后拉新消息） + is_sender: 发送方向过滤，True=仅本人发送 / False=仅对方发送 / + None=不过滤（默认）。无法定位本人 rowid 时返回空结果。 Returns: {"messages": [...], "next_cursor": int, "has_more": bool, "talker": str} @@ -803,32 +880,46 @@ class DbReader: id_to_username = self._load_name2id(conn) self_wxid = self._get_current_wxid() or "" + # 发送方向过滤：无法定位本人 rowid 时短路返回空 + sender_sql: str = "" + sender_params: list = [] + if is_sender is not None: + self_rowid = self._resolve_self_rowid(conn, self_wxid) + sender_sql, sender_params = self._build_sender_filter( + is_sender, self_rowid + ) + if not sender_sql: + return { + "messages": [], + "next_cursor": cursor, + "has_more": False, + "talker": talker, + } + + # 收集 WHERE 片段：时间过滤 + 发送方向过滤 + where_parts: list[str] = [] + time_params: list = [] if direction == "after": - where_sql = "create_time > ?" + where_parts.append("create_time > ?") + time_params.append(cursor) order_sql = "create_time ASC" else: # before 模式：cursor=0 时不加时间过滤（取最新 limit 条） if cursor > 0: - where_sql = "create_time < ?" - else: - where_sql = "1=1" + where_parts.append("create_time < ?") + time_params.append(cursor) order_sql = "create_time DESC" + if sender_sql: + where_parts.append(sender_sql) - if cursor > 0 or direction == "after": - sql = ( - f"SELECT local_id, local_type, create_time, real_sender_id, " - f"message_content, WCDB_CT_message_content " - f"FROM [{table_name}] WHERE {where_sql} " - f"ORDER BY {order_sql} LIMIT ?" - ) - params: tuple = (cursor, limit) - else: - sql = ( - f"SELECT local_id, local_type, create_time, real_sender_id, " - f"message_content, WCDB_CT_message_content " - f"FROM [{table_name}] ORDER BY {order_sql} LIMIT ?" - ) - params = (limit,) + where_clause = " AND ".join(where_parts) if where_parts else "1=1" + sql = ( + f"SELECT local_id, local_type, create_time, real_sender_id, " + f"message_content, WCDB_CT_message_content " + f"FROM [{table_name}] WHERE {where_clause} " + f"ORDER BY {order_sql} LIMIT ?" + ) + params: tuple = (*time_params, *sender_params, limit) cur = conn.execute(sql, params) rows = cur.fetchall() @@ -864,6 +955,7 @@ class DbReader: start_time: Optional[int] = None, end_time: Optional[int] = None, limit: int = 50, + is_sender: Optional[bool] = None, ) -> dict: """按关键词搜索历史消息。 @@ -876,6 +968,8 @@ class DbReader: start_time: 可选，起始时间戳（含） end_time: 可选，结束时间戳（含） limit: 最多返回条数，1~200 + is_sender: 发送方向过滤，True=仅本人发送 / False=仅对方发送 / + None=不过滤（默认）。无法定位本人 rowid 时返回空结果。 Returns: {"messages": [...], "total": int} @@ -900,6 +994,17 @@ class DbReader: id_to_username = self._load_name2id(conn) self_wxid = self._get_current_wxid() or "" + # 发送方向过滤：无法定位本人 rowid 时短路返回空 + sender_sql: str = "" + sender_params: list = [] + if is_sender is not None: + self_rowid = self._resolve_self_rowid(conn, self_wxid) + sender_sql, sender_params = self._build_sender_filter( + is_sender, self_rowid + ) + if not sender_sql: + return {"messages": [], "total": 0} + # 构建 WHERE 子句（LIKE 转义 % _ \ 为字面量） escaped = keyword.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") where_clauses = ["message_content LIKE ? ESCAPE '\\'"] @@ -910,6 +1015,9 @@ class DbReader: if end_time is not None: where_clauses.append("create_time <= ?") params.append(end_time) + if sender_sql: + where_clauses.append(sender_sql) + params.extend(sender_params) where_sql = " AND ".join(where_clauses) # 确定要搜索的表集合 diff --git a/bridge/woc_bridge/messaging/send_queue.py b/bridge/woc_bridge/messaging/send_queue.py index 0f236df..1ccb4af 100644 --- a/bridge/woc_bridge/messaging/send_queue.py +++ b/bridge/woc_bridge/messaging/send_queue.py @@ -7,12 +7,15 @@ from __future__ import annotations import asyncio +import logging import time from typing import Any, Awaitable, Callable from woc_bridge.models import BridgeError +logger = logging.getLogger("woc-bridge") + # 工厂类型：返回一个待执行的 coroutine CoroFactory = Callable[[], Awaitable[Any]] @@ -73,6 +76,7 @@ class SendQueue: loop = asyncio.get_running_loop() future: asyncio.Future = loop.create_future() await self._queue.put((coro_factory, future)) + logger.info("send_queue: 入队 (pending=%d)", self._queue.qsize()) return await future def pending_count(self) -> int: @@ -93,6 +97,10 @@ class SendQueue: # 计算建议等待秒数：最早一次调用距窗口边界还差多久 oldest = self._recent_call_times[0] retry_after = max(1, int(1.0 - (now - oldest)) + 1) + logger.warning( + "send_queue: 限流命中，拒绝执行 (recent=%d/%d, retry_after=%ds)", + len(self._recent_call_times), self.max_calls_per_sec, retry_after, + ) raise BridgeError( code="RATE_LIMITED", message=f"发送限流：每秒最多 {self.max_calls_per_sec} 次", @@ -108,8 +116,10 @@ class SendQueue: """ while True: coro_factory, future = await self._queue.get() + logger.info("send_queue: 出队，开始处理 (pending=%d)", self._queue.qsize()) # 标记任务是否真正开始执行（用于决定 finally 是否延时） executed = False + t_exec = time.perf_counter() try: # 执行前检查限流 self._check_rate_limit() @@ -117,7 +127,12 @@ class SendQueue: self._recent_call_times.append(time.monotonic()) executed = True # 执行任务 + logger.info("send_queue: 开始执行任务") result = await coro_factory() + logger.info( + "send_queue: 任务执行完成 (%.0fms)", + (time.perf_counter() - t_exec) * 1000, + ) if not future.done(): future.set_result(result) except asyncio.CancelledError: @@ -126,10 +141,18 @@ class SendQueue: future.cancel() raise except Exception as e: + logger.warning( + "send_queue: 任务执行抛异常 %s: %s (%.0fms)", + type(e).__name__, e, (time.perf_counter() - t_exec) * 1000, + ) if not future.done(): future.set_exception(e) finally: self._queue.task_done() # 仅在任务真正执行过时延时，限流失败的任务不延时 if executed: + logger.info( + "send_queue: 延时 %dms 后处理下一个", + self.send_delay_ms, + ) await asyncio.sleep(self.send_delay_ms / 1000.0) diff --git a/bridge/woc_bridge/routes/messages.py b/bridge/woc_bridge/routes/messages.py index 4cf93ab..60ade31 100644 --- a/bridge/woc_bridge/routes/messages.py +++ b/bridge/woc_bridge/routes/messages.py @@ -2,8 +2,9 @@ from __future__ import annotations import asyncio import logging +from typing import Optional -from fastapi import APIRouter, Request +from fastapi import APIRouter, Query, Request from fastapi.responses import StreamingResponse from woc_bridge.config import _require_db_reader, _require_message_streamer @@ -22,7 +23,14 @@ router = APIRouter() @router.get("/api/messages/since", response_model=MessagesResponse) @with_db_retry -async def get_messages_since(cursor: int = 0, limit: int = 50) -> MessagesResponse: +async def get_messages_since( + cursor: int = 0, + limit: int = 50, + is_sender: Optional[bool] = Query( + None, + description="按发送方向过滤：True=仅本人发送 / False=仅对方发送 / None=不过滤（默认）", + ), +) -> MessagesResponse: """增量消息拉取（按 create_time 游标）。 channels PullerAdapter 的核心接口：调用方维护本地 cursor（首次为 0）， @@ -33,6 +41,8 @@ async def get_messages_since(cursor: int = 0, limit: int = 50) -> MessagesRespon cursor: 上次拉取的最大 create_time，首次传 0（拉全量） limit: 最多返回条数，1~200，默认 50。建议用 /api/status 返回的 max_batch_size 作为上限参考 + is_sender: 按发送方向过滤，True=仅本人发送 / False=仅对方发送 / + None=不过滤（默认）。SQL 层过滤，无法定位本人时返回空 Returns: MessagesResponse：含 messages 列表 / next_cursor / has_more @@ -66,11 +76,13 @@ async def get_messages_since(cursor: int = 0, limit: int = 50) -> MessagesRespon await _check_db_readable() # DB 读取是同步阻塞操作，用 to_thread 包装 - result = await asyncio.to_thread(db_reader.get_messages_since, cursor, limit) + result = await asyncio.to_thread( + db_reader.get_messages_since, cursor, limit, is_sender + ) msg_count = len(result.get("messages", [])) logger.info( - "messages/since: cursor=%d limit=%d → 返回 %d 条, next_cursor=%s has_more=%s", - cursor, limit, msg_count, result.get("next_cursor"), result.get("has_more"), + "messages/since: cursor=%d limit=%d is_sender=%s → 返回 %d 条, next_cursor=%s has_more=%s", + cursor, limit, is_sender, msg_count, result.get("next_cursor"), result.get("has_more"), ) return MessagesResponse(**result) @@ -161,6 +173,10 @@ async def get_messages_by_session( cursor: int = 0, limit: int = 50, direction: str = "before", + is_sender: Optional[bool] = Query( + None, + description="按发送方向过滤：True=仅本人发送 / False=仅对方发送 / None=不过滤（默认）", + ), ) -> MessagesBySessionResponse: """按会话拉取历史消息（单会话，O(1) 定位分片表）。 @@ -173,6 +189,8 @@ async def get_messages_by_session( after 模式取该时间之后的新消息 limit: 最多返回条数，1~200，默认 50 direction: before（默认，往前翻历史）/ after（往后拉新消息） + is_sender: 按发送方向过滤，True=仅本人发送 / False=仅对方发送 / + None=不过滤（默认）。SQL 层过滤，无法定位本人时返回空 Returns: MessagesBySessionResponse：含 messages / next_cursor / has_more / talker @@ -210,12 +228,12 @@ async def get_messages_by_session( await _check_db_readable() result = await asyncio.to_thread( - db_reader.get_messages_by_session, talker, cursor, limit, direction + db_reader.get_messages_by_session, talker, cursor, limit, direction, is_sender ) msg_count = len(result.get("messages", [])) logger.info( - "messages/by_session: talker=%s cursor=%d direction=%s limit=%d → 返回 %d 条, next_cursor=%s has_more=%s", - talker, cursor, direction, limit, msg_count, + "messages/by_session: talker=%s cursor=%d direction=%s limit=%d is_sender=%s → 返回 %d 条, next_cursor=%s has_more=%s", + talker, cursor, direction, limit, is_sender, msg_count, result.get("next_cursor"), result.get("has_more"), ) return MessagesBySessionResponse(**result) @@ -232,6 +250,10 @@ async def search_messages( start_time: int = 0, end_time: int = 0, limit: int = 50, + is_sender: Optional[bool] = Query( + None, + description="按发送方向过滤：True=仅本人发送 / False=仅对方发送 / None=不过滤（默认）", + ), ) -> MessageSearchResponse: """按关键词搜索历史消息。 @@ -244,6 +266,8 @@ async def search_messages( start_time: 可选，起始时间戳（含），0 表示不限制 end_time: 可选，结束时间戳（含），0 表示不限制 limit: 最多返回条数，1~200，默认 50 + is_sender: 按发送方向过滤，True=仅本人发送 / False=仅对方发送 / + None=不过滤（默认）。SQL 层过滤，无法定位本人时返回空 Returns: MessageSearchResponse：含 messages / total @@ -279,10 +303,11 @@ async def search_messages( start_time if start_time > 0 else None, end_time if end_time > 0 else None, limit, + is_sender, ) msg_count = len(result.get("messages", [])) logger.info( - "messages/search: keyword=%r talker=%s → 命中 %d 条（返回 %d 条）", - keyword, talker or "(全部)", result.get("total", 0), msg_count, + "messages/search: keyword=%r talker=%s is_sender=%s → 命中 %d 条（返回 %d 条）", + keyword, talker or "(全部)", is_sender, result.get("total", 0), msg_count, ) return MessageSearchResponse(**result) diff --git a/bridge/woc_bridge/routes/send.py b/bridge/woc_bridge/routes/send.py index 0a68259..b3399c5 100644 --- a/bridge/woc_bridge/routes/send.py +++ b/bridge/woc_bridge/routes/send.py @@ -115,42 +115,76 @@ async def send_text(req: SendTextRequest) -> SendResponse: """ xdotool = _require_xdotool() send_queue = _require_send_queue() + t_total = time.perf_counter() # 校验请求体 if not req.to_wxid: + logger.warning("send/text: 参数校验失败 — to_wxid 为空") raise BridgeError(code="INVALID_PARAMS", message="to_wxid 不能为空") if not req.content: + logger.warning("send/text: 参数校验失败 — content 为空 (to_wxid=%s)", req.to_wxid) raise BridgeError(code="INVALID_PARAMS", message="content 不能为空") + logger.info( + "send/text: 收到请求 to_wxid=%s content_len=%d display_name=%s", + req.to_wxid, len(req.content), req.display_name or "(未提供)", + ) + # 检查登录态 + t0 = time.perf_counter() login_state = await xdotool.detect_login_state() + logger.info( + "send/text: 登录态检测 → %s (%.0fms)", + login_state, (time.perf_counter() - t0) * 1000, + ) if login_state != "logged_in": + logger.warning("send/text: 拒绝发送，登录态=%s", login_state) raise BridgeError( code="WECHAT_NOT_LOGGED_IN", message=f"当前登录态为 {login_state}，无法发送消息", ) # 解析显示名（优先备注/昵称，不再直接用 wxid 搜索） + t0 = time.perf_counter() display_name = await _resolve_display_name( _state.db_reader, req.to_wxid, req.display_name ) + logger.info( + "send/text: display_name 解析 → %s (%.0fms)", + display_name, (time.perf_counter() - t0) * 1000, + ) # 经发送队列串行执行 + logger.info("send/text: 入队 send_queue (pending=%d)", send_queue.pending_count()) try: + t0 = time.perf_counter() local_send_id = await send_queue.enqueue( lambda: xdotool.send_text(req.to_wxid, req.content, display_name) ) - except BridgeError: + logger.info( + "send/text: 队列执行完成 → local_send_id=%s (%.0fms)", + local_send_id, (time.perf_counter() - t0) * 1000, + ) + except BridgeError as e: + logger.warning( + "send/text: 发送失败 BridgeError code=%s msg=%s (%.0fms)", + e.code, e.message, (time.perf_counter() - t0) * 1000, + ) raise except Exception as e: + logger.error( + "send/text: 发送失败 %s: %s (%.0fms)", + type(e).__name__, e, (time.perf_counter() - t0) * 1000, + ) raise BridgeError( code="SEND_FAILED", message=f"发送失败: {e}", ) logger.info( - "send/text: to=%s display_name=%s content_len=%d → 成功 local_send_id=%s", + "send/text: ✓ to=%s display_name=%s content_len=%d → local_send_id=%s (总耗时 %.0fms)", req.to_wxid, display_name, len(req.content), local_send_id, + (time.perf_counter() - t_total) * 1000, ) return SendResponse(success=True, local_send_id=local_send_id, placeholder=False, error=None) diff --git a/bridge/woc_bridge/ui/xdotool_driver.py b/bridge/woc_bridge/ui/xdotool_driver.py index 71b0b1d..e8f3287 100644 --- a/bridge/woc_bridge/ui/xdotool_driver.py +++ b/bridge/woc_bridge/ui/xdotool_driver.py @@ -21,6 +21,13 @@ from woc_bridge.models import BridgeError, LoginState # 模块级 logger：与 server.py 同名，便于 s6 统一收日志 logger = logging.getLogger("woc-bridge") +# 单条 xdotool/pgrep/kill 子进程命令的整体超时（秒）。 +# 这类命令本身均为毫秒级，15s 足够兜底；超时不重试，退避交给上层 outbox。 +_CMD_TIMEOUT_SEC = 15.0 +# send_text 中“粘贴内容 + 短暂等待 + 回车发送”三步整体的超时（秒）。 +# 配合 _open_session_by_name 的 10s，send_text 总上限约 25s，留给 HTTP 往返与中间件 5s。 +_SEND_BODY_TIMEOUT_SEC = 15.0 + class XdotoolDriver: """xdotool/xclip 异步驱动。 @@ -54,9 +61,16 @@ class XdotoolDriver: ) -> tuple[int, bytes, bytes]: """执行一条命令并返回 (returncode, stdout, stderr)。 + xdotool / pgrep / kill 均为毫秒级命令，统一用 _CMD_TIMEOUT_SEC 兜底， + 防止子进程挂死导致协程永久阻塞。超时不在此重试（退避交给上层 outbox， + 与 httpx.NetworkError 只重试一次的策略对齐，避免双重重试放大延迟）。 + Args: args: 命令及其参数列表，如 ["xdotool", "search", "--name", "微信"] input_bytes: 需要写入 stdin 的字节 + + Raises: + BridgeError: 命令超时（code=SEND_FAILED，message 指出超时的命令名） """ proc = await asyncio.create_subprocess_exec( *args, @@ -65,7 +79,20 @@ class XdotoolDriver: stderr=asyncio.subprocess.PIPE, env=self._env(), ) - stdout, stderr = await proc.communicate(input=input_bytes) + try: + stdout, stderr = await asyncio.wait_for( + proc.communicate(input=input_bytes), + timeout=_CMD_TIMEOUT_SEC, + ) + except asyncio.TimeoutError as exc: + # 先杀子进程再收尸，避免僵尸进程残留占住管道/资源 + proc.kill() + await proc.wait() + logger.error("[ui] command timeout: %s", args[0]) + raise BridgeError( + code="SEND_FAILED", + message=f"命令超时: {args[0]}", + ) from exc return proc.returncode, stdout, stderr async def _key(self, key: str) -> None: @@ -325,6 +352,42 @@ class XdotoolDriver: ) logger.info("[ui] type done rc=%s", rc) + # ------------------------------------------------------------------ + # 鼠标点击聊天输入框（强制焦点） + # ------------------------------------------------------------------ + # 微信 4.x 自绘 UI 进入会话后焦点不在输入框，xdotool type 会打到错误位置。 + # 实测点击窗口底部中间区域 (800, 800) 能稳定聚焦到聊天输入框。 + # 坐标基于窗口几何 1435x876，若窗口尺寸变化需重新校准。 + _INPUT_BOX_X = 800 + _INPUT_BOX_Y = 800 + + async def _click_input_box(self) -> None: + """鼠标点击聊天输入框，强制焦点落到输入框。 + + 微信 4.x 自绘 UI 通过搜索进入会话后，焦点不在聊天输入框， + 直接 xdotool type 会导致文字打到错误位置。必须先鼠标点击 + 输入框区域强制聚焦。 + """ + logger.info("[ui] click input box (%d, %d)", self._INPUT_BOX_X, self._INPUT_BOX_Y) + rc, _, stderr = await self._run([ + "xdotool", "mousemove", str(self._INPUT_BOX_X), str(self._INPUT_BOX_Y), + ]) + if rc != 0: + err = stderr.decode(errors="ignore").strip() if stderr else "unknown" + raise BridgeError( + code="SEND_FAILED", + message=f"鼠标移动到输入框失败 (code={rc}): {err}", + ) + rc, _, stderr = await self._run(["xdotool", "click", "1"]) + if rc != 0: + err = stderr.decode(errors="ignore").strip() if stderr else "unknown" + raise BridgeError( + code="SEND_FAILED", + message=f"点击输入框失败 (code={rc}): {err}", + ) + logger.info("[ui] click input box done rc=%s", rc) + await self._sleep(0.3) + # ------------------------------------------------------------------ # 会话定位（Ctrl+F 搜索） # ------------------------------------------------------------------ @@ -441,19 +504,36 @@ class XdotoolDriver: 本地生成的 channel_msg_id，格式 local__<随机> """ name = display_name if display_name else to_wxid - logger.info("[send_text] to=%s name=%s content_len=%s", to_wxid, name, len(content)) - # 1. 定位会话 + t_total = time.perf_counter() + logger.info("[send_text] 开始 to=%s name=%s content_len=%d", to_wxid, name, len(content)) + # 1. 定位会话（_open_session_by_name 内部已含 10s 整体超时） + t0 = time.perf_counter() await self._open_session_by_name(name) - logger.info("[send_text] session opened") - # 2. 粘贴内容并发送 - await self._paste_via_xclip(content) - logger.info("[send_text] content pasted") - await self._sleep(0.2) - await self._key("Return") - logger.info("[send_text] return pressed") + logger.info("[send_text] 会话定位完成 (%.0fms)", (time.perf_counter() - t0) * 1000) + # 1.5 鼠标点击聊天输入框强制焦点 + # 微信 4.x 自绘 UI 进入会话后焦点不在输入框，xdotool type 会打到错误位置。 + # 实测点击 (800, 800) 能稳定聚焦到输入框（窗口几何 1435x876）。 + await self._click_input_box() + # 2. 输入内容并发送（三步整体超时保护，与定位会话的 10s 合计最坏 25s） + async def _send_body() -> None: + await self._paste_via_xclip(content) + logger.info("[send_text] 文本已输入") + await self._sleep(0.2) + await self._key("Return") + try: + t0 = time.perf_counter() + await asyncio.wait_for(_send_body(), timeout=_SEND_BODY_TIMEOUT_SEC) + logger.info("[send_text] 回车已发送 (%.0fms)", (time.perf_counter() - t0) * 1000) + except asyncio.TimeoutError as exc: + logger.error("[send_text] 输入/发送步骤超时") + raise BridgeError( + code="SEND_FAILED", + message="文本输入/发送步骤超时", + ) from exc # 3. 生成 local_send_id（本地 ID，非微信原生 msg_id） local_send_id = f"local_{int(time.time())}_{random.randint(0, 0xFFFFFF):06x}" - logger.info("[send_text] done local_send_id=%s", local_send_id) + logger.info("[send_text] 完成 local_send_id=%s (总耗时 %.0fms)", + local_send_id, (time.perf_counter() - t_total) * 1000) return local_send_id # ------------------------------------------------------------------ @@ -497,6 +577,8 @@ class XdotoolDriver: # 1. 定位会话 await self._open_session_by_name(display_name if display_name else to_wxid) + # 1.5 鼠标点击聊天输入框强制焦点（与 send_text 同理） + await self._click_input_box() # 2. MVP：发送文件路径提示文本（真实文件传输留给后续完善） filename = os.path.basename(file_path)