from __future__ import annotations import asyncio import logging import os import time from typing import Optional from fastapi import APIRouter from woc_bridge.config import _state, _require_xdotool, _require_send_queue, _require_db_reader from woc_bridge.db import DbReader from woc_bridge.db.coordinator import with_db_retry from woc_bridge.models import ( SendTextRequest, SendFileRequest, SendResponse, RevokeMessageRequest, RevokeMessageResponse, ForwardMessageRequest, ForwardMessageResponse, BridgeError, ) logger = logging.getLogger("woc-bridge") router = APIRouter() # --------------------------------------------------------------------------- # 路由：POST /api/send/text # --------------------------------------------------------------------------- # display_name 解析缓存：wxid -> (display_name, timestamp) # 联系人备注/昵称不常变，5 分钟 TTL 足够；最大 256 条防内存膨胀 _display_name_cache: dict[str, tuple[str, float]] = {} _DISPLAY_NAME_CACHE_TTL = 300.0 # 5 分钟 _DISPLAY_NAME_CACHE_MAX = 256 async def _resolve_display_name( db_reader: Optional[DbReader], to_wxid: str, display_name: Optional[str], ) -> str: """解析用于微信搜索框定位会话的显示名。 优先级： 1. 调用方显式传入的 display_name（strip 后非空才用） 2. 缓存命中（5 分钟 TTL） 3. 从 contact.db 读取该 wxid 的备注（remark） 4. 从 contact.db 读取该 wxid 的昵称（nickname） 5. 回退到 wxid 本身 DB 不可用时静默回退到 wxid，避免阻塞发送流程。 """ # 显式传入优先 if display_name and display_name.strip(): return display_name.strip() # 缓存命中 cached = _display_name_cache.get(to_wxid) if cached and time.monotonic() - cached[1] < _DISPLAY_NAME_CACHE_TTL: return cached[0] if db_reader is None: return to_wxid try: contact = await asyncio.to_thread(db_reader.get_contact_detail, to_wxid) except Exception: return to_wxid if contact: resolved = contact.get("remark") or contact.get("nickname") or to_wxid else: resolved = to_wxid # 写入缓存（简单 LRU：超限时删最旧条目） if len(_display_name_cache) >= _DISPLAY_NAME_CACHE_MAX: oldest = min(_display_name_cache, key=lambda k: _display_name_cache[k][1]) del _display_name_cache[oldest] _display_name_cache[to_wxid] = (resolved, time.monotonic()) return resolved async def _get_latest_create_time_for_talker( db_reader: Optional[DbReader], to_wxid: str, ) -> int: """返回目标 talker 当前最新消息的 create_time，DB 不可用时返回 0。""" if db_reader is None: return 0 try: result = await asyncio.to_thread( db_reader.get_messages_by_session, to_wxid, cursor=0, limit=1, direction="before", is_sender=None, ) messages = result.get("messages", []) if messages: return int(messages[0].get("create_time", 0)) except Exception: pass return 0 async def _verify_sent_to_talker( db_reader: Optional[DbReader], to_wxid: str, before_ct: int, max_wait_sec: float = 10.0, ) -> bool: """轮询 DB，确认目标 talker 在 before_ct 之后出现了新消息。 用于发送后校验消息是否真的落到了目标会话。DB 不可用时静默跳过。 默认最多等待 10 秒，兼容 WeChat 4.x 异步写入 DB 的延迟。 """ if db_reader is None: return True deadline = time.monotonic() + max_wait_sec attempts = 0 while time.monotonic() < deadline: attempts += 1 try: result = await asyncio.to_thread( db_reader.get_messages_by_session, to_wxid, cursor=before_ct, limit=3, direction="after", is_sender=None, ) messages = result.get("messages", []) if messages: logger.info( "send/text: DB 校验成功，目标 talker 发现 %d 条新消息 (attempt=%d)", len(messages), attempts, ) return True logger.debug("send/text: DB 校验第 %d 次轮询未找到新消息", attempts) except Exception as e: logger.debug("send/text: DB 校验第 %d 次轮询异常: %s", attempts, e) await asyncio.sleep(0.5) logger.warning( "send/text: DB 校验超时，%ds 内目标 talker 未出现新消息 (attempts=%d)", max_wait_sec, attempts, ) return False @router.post("/api/send/text", response_model=SendResponse) @with_db_retry async def send_text(req: SendTextRequest) -> SendResponse: """发送文本消息（channels OutboundAdapter 核心接口）。 流程： 1. 校验 to_wxid / content 非空 2. 解析 display_name：显式传入 > 备注 > 昵称 > wxid 3. 检测登录态，非 logged_in 直接拒绝（避免 UI 操作引发不可预期行为） 4. 记录目标会话当前最新 create_time 5. 经 send_queue 串行执行：activate_window → 鼠标点击左侧搜索框 → 输入 display_name → 点击第一个搜索结果 → 点击输入框 → 输入 content → 回车发送 6. 发送后轮询 DB 校验目标会话是否出现新消息，未出现则抛 SEND_FAILED 7. 返回本地生成的 local_send_id（不等待微信发送回执） Args: req: SendTextRequest（to_wxid + content + 可选 display_name） Returns: SendResponse：success=true / local_send_id（格式 local__<随机>） / placeholder=false（文本发送为真实发送，非占位） Raises: BridgeError(INVALID_PARAMS): to_wxid 或 content 为空（HTTP 400） BridgeError(WECHAT_NOT_LOGGED_IN): 当前未登录（HTTP 401） BridgeError(SEND_FAILED): xdotool/xclip 操作失败（HTTP 500） Notes: - 发送队列串行化避免搜索框/鼠标点击竞态 - local_send_id 由 bridge 本地生成，不对应微信真实 msg_id； 调用方应用它做幂等去重，**禁止用它到 /api/media/{msg_id} 查媒体** - 发送队列的发送间隔与限频由 .env 的 WOC_BRIDGE_SEND_DELAY_MS / WOC_BRIDGE_MAX_CALLS_PER_SEC 控制 - DB 校验失败时消息可能已经发出（可能到错误会话），调用方需人工排查 """ 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, ) # 发送前记录目标会话最新消息时间，用于发送后校验 before_ct = await _get_latest_create_time_for_talker(_state.db_reader, req.to_wxid) logger.info("send/text: 目标会话当前最新 create_time=%d", before_ct) # 经发送队列串行执行 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) ) 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}", ) # 发送后校验：目标会话是否出现了新消息 verified = await _verify_sent_to_talker(_state.db_reader, req.to_wxid, before_ct) if not verified: logger.error( "send/text: 发送后 DB 校验失败，消息可能未进入目标会话 to=%s before_ct=%d", req.to_wxid, before_ct, ) raise BridgeError( code="SEND_FAILED", message="发送后校验失败：消息未到达目标会话", ) logger.info( "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) # --------------------------------------------------------------------------- # 路由：POST /api/send/image # --------------------------------------------------------------------------- @router.post("/api/send/image", response_model=SendResponse) async def send_image(req: SendFileRequest) -> SendResponse: """发送图片消息（MVP 简化版）。 当前 MVP 实现仅校验文件存在 + 进入会话 + 发送"[图片] 文件名"提示文本， 真实图片传输（拖拽 / Ctrl+V 粘贴文件路径）留给后续完善（W-02）。 响应中 placeholder=true 标识当前为占位实现。 Args: req: SendFileRequest（to_wxid + file_path + 可选 display_name） Returns: SendResponse：success=true / local_send_id / placeholder=true Raises: 同 _send_file：INVALID_PARAMS / WECHAT_NOT_LOGGED_IN / SEND_FAILED """ return await _send_file(req, is_image=True) # --------------------------------------------------------------------------- # 路由：POST /api/send/file # --------------------------------------------------------------------------- @router.post("/api/send/file", response_model=SendResponse) @with_db_retry async def send_file(req: SendFileRequest) -> SendResponse: """发送文件消息（MVP 简化版）。 当前 MVP 实现仅校验文件存在 + 进入会话 + 发送"[文件] 文件名"提示文本， 真实文件传输（拖拽 / 文件选择器）留给后续完善（W-02）。 响应中 placeholder=true 标识当前为占位实现。 Args: req: SendFileRequest（to_wxid + file_path + 可选 display_name） Returns: SendResponse：success=true / local_send_id / placeholder=true Raises: 同 _send_file：INVALID_PARAMS / WECHAT_NOT_LOGGED_IN / SEND_FAILED """ return await _send_file(req, is_image=False) async def _send_file(req: SendFileRequest, is_image: bool) -> SendResponse: """send_image / send_file 共用实现。 流程： 1. 校验 to_wxid / file_path 非空，文件存在 2. 解析 display_name：显式传入 > 备注 > 昵称 > wxid 3. 检测登录态，非 logged_in 直接拒绝 4. 经 send_queue 串行执行：activate → Ctrl+F 搜索会话（按 display_name） → 选中并进入 → xclip 粘贴"[图片/文件] 文件名" → 回车发送 Args: req: SendFileRequest（to_wxid + file_path + 可选 display_name） is_image: True=图片，False=普通文件（仅影响提示文本前缀） Returns: SendResponse：success=true / local_send_id / placeholder=true （W-02 完成前为占位实现，仅发提示文本；完成后改为 placeholder=false） Raises: BridgeError(INVALID_PARAMS): to_wxid/file_path 为空或文件不存在（HTTP 400） BridgeError(WECHAT_NOT_LOGGED_IN): 当前未登录（HTTP 401） BridgeError(SEND_FAILED): xdotool/xclip 操作失败（HTTP 500） Notes: - file_path 必须是容器内绝对路径（不是宿主机路径） - MVP 阶段不传输真实文件内容，仅发提示文本；后续完善时需考虑 拖拽到聊天窗口 / Ctrl+V 粘贴文件 / 文件选择器对话框等方案 - local_send_id 不对应微信原生 msg_id，禁止用于 /api/media/{msg_id} """ xdotool = _require_xdotool() send_queue = _require_send_queue() if not req.to_wxid: raise BridgeError(code="INVALID_PARAMS", message="to_wxid 不能为空") if not req.file_path: raise BridgeError(code="INVALID_PARAMS", message="file_path 不能为空") if not os.path.isfile(req.file_path) or not os.access(req.file_path, os.R_OK): raise BridgeError( code="INVALID_PARAMS", message=f"文件不存在或不可读: {req.file_path}", ) # 检查登录态 login_state = await xdotool.detect_login_state() if login_state != "logged_in": raise BridgeError( code="WECHAT_NOT_LOGGED_IN", message=f"当前登录态为 {login_state}，无法发送消息", ) # 解析显示名（优先备注/昵称，不再直接用 wxid 搜索） display_name = await _resolve_display_name( _state.db_reader, req.to_wxid, req.display_name ) try: local_send_id = await send_queue.enqueue( lambda: xdotool.send_file( req.to_wxid, req.file_path, is_image=is_image, display_name=display_name ) ) except BridgeError: raise except Exception as e: raise BridgeError( code="SEND_FAILED", message=f"发送失败: {e}", ) kind = "image" if is_image else "file" logger.info( "send/%s: to=%s display_name=%s file=%s → 成功 local_send_id=%s (placeholder)", kind, req.to_wxid, display_name, req.file_path, local_send_id, ) # W-02 完成前，图片/文件发送为占位实现（仅发提示文本） return SendResponse(success=True, local_send_id=local_send_id, placeholder=True, error=None) # --------------------------------------------------------------------------- # 路由：POST /api/messages/revoke (experimental) # --------------------------------------------------------------------------- @router.post("/api/messages/revoke", response_model=RevokeMessageResponse) async def revoke_message(req: RevokeMessageRequest) -> RevokeMessageResponse: """撤回指定会话中的最近消息（experimental）。 UI 自动化路径：定位会话 → 长按最近消息 → 选择撤回 → 确认。 微信撤回时限：发送后 2 分钟内。 Args: req: RevokeMessageRequest（talker + create_time） Returns: RevokeMessageResponse Raises: BridgeError(INVALID_PARAMS): talker 为空（HTTP 400） BridgeError(WECHAT_NOT_LOGGED_IN): 未登录（HTTP 401） BridgeError(REVOKE_WINDOW_EXPIRED): 超过 2 分钟撤回时限（HTTP 409） BridgeError(WINDOW_NOT_FOUND): 找不到微信窗口（HTTP 503） BridgeError(SEND_FAILED): UI 操作失败或超时（HTTP 500） Notes: - experimental：消息气泡坐标为估算值，需在目标分辨率实测调优 - 无法校验撤回是否真正生效（无 UI 元素检测） """ if not req.talker: raise BridgeError(code="INVALID_PARAMS", message="talker 不能为空") if req.create_time <= 0: raise BridgeError( code="INVALID_PARAMS", message=f"create_time 必须 > 0，收到 {req.create_time}", ) xdotool = _require_xdotool() send_queue = _require_send_queue() db_reader = _require_db_reader() login_state = await xdotool.detect_login_state() if login_state != "logged_in": raise BridgeError( code="WECHAT_NOT_LOGGED_IN", message=f"当前登录态为 {login_state}，无法撤回消息", ) # 解析 display_name（优先 req.display_name，其次 DB 查 remark/nickname，回退 wxid） display_name = await _resolve_display_name( db_reader, req.talker, req.display_name ) try: await send_queue.enqueue( lambda: xdotool.revoke_message( req.talker, req.create_time, display_name ) ) except BridgeError: raise except Exception as e: raise BridgeError( code="SEND_FAILED", message=f"撤回失败: {e}", ) logger.info( "messages/revoke: talker=%s create_time=%d → UI 操作完成", req.talker, req.create_time, ) return RevokeMessageResponse(success=True, error=None) # --------------------------------------------------------------------------- # 路由：POST /api/messages/forward (experimental) # --------------------------------------------------------------------------- @router.post("/api/messages/forward", response_model=ForwardMessageResponse) async def forward_message(req: ForwardMessageRequest) -> ForwardMessageResponse: """转发消息到指定会话（experimental）。 UI 自动化路径：定位源会话 → 长按消息 → 选择转发 → 搜索目标 → 确认。 Args: req: ForwardMessageRequest Returns: ForwardMessageResponse Raises: BridgeError(INVALID_PARAMS): 参数为空（HTTP 400） BridgeError(WECHAT_NOT_LOGGED_IN): 未登录（HTTP 401） BridgeError(WINDOW_NOT_FOUND / SEND_FAILED) Notes: - experimental：坐标均为估算值 - 转发目标选择界面是独立窗口，坐标可能与主窗口不同 """ if not req.talker: raise BridgeError(code="INVALID_PARAMS", message="talker 不能为空") if not req.target_display_name: raise BridgeError(code="INVALID_PARAMS", message="target_display_name 不能为空") xdotool = _require_xdotool() send_queue = _require_send_queue() db_reader = _require_db_reader() login_state = await xdotool.detect_login_state() if login_state != "logged_in": raise BridgeError( code="WECHAT_NOT_LOGGED_IN", message=f"当前登录态为 {login_state}，无法转发消息", ) # 解析源会话 display_name（优先 req.source_display_name，其次 DB 查，回退 wxid） source_display_name = await _resolve_display_name( db_reader, req.talker, req.source_display_name ) try: await send_queue.enqueue( lambda: xdotool.forward_message( req.talker, req.target_display_name, source_display_name ) ) except BridgeError: raise except Exception as e: raise BridgeError( code="SEND_FAILED", message=f"转发失败: {e}", ) logger.info( "messages/forward: talker=%s target=%s → UI 操作完成", req.talker, req.target_display_name, ) return ForwardMessageResponse(success=True, error=None)