Kris
054c4676aa
主要变更： 1. 新增SSEVerifyBus实现基于推送的发送结果校验，替代原DB轮询方案 2. 移除FlowOrchestrator中的DB熔断器，避免雪崩风险 3. 重构SendTextFlow，改用SSE校验流程，删除原DB校验相关代码 4. 新增RoiSpec与ROI搜索支持，优化UI元素匹配精度 5. 更新所有分辨率配置文件，添加ROI配置与调整搜索框坐标 6. 新增WOC_DISABLE_OPENCV环境变量开关，支持纯几何定位模式 7. 清理FlowContext中废弃的before_msg_id字段
914 lines
35 KiB
Python
914 lines
35 KiB
Python
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,
|
||
ERROR_CODES,
|
||
)
|
||
from woc_bridge.ui.flows.send_file import _validate_file_path
|
||
|
||
logger = logging.getLogger("woc-bridge")
|
||
router = APIRouter()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 路由：POST /api/send/text
|
||
# ---------------------------------------------------------------------------
|
||
# display_name 解析缓存：wxid -> (display_name, timestamp)
|
||
# 联系人备注/昵称不常变，5 分钟 TTL 足够；最大 1024 条防内存膨胀
|
||
_display_name_cache: dict[str, tuple[str, float]] = {}
|
||
_DISPLAY_NAME_CACHE_TTL = 300.0 # 5 分钟
|
||
_DISPLAY_NAME_CACHE_MAX = 1024
|
||
_DISPLAY_NAME_WARMUP_LIMIT = 1000 # 启动时预热最近联系人数量
|
||
|
||
|
||
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 warm_display_name_cache(db_reader: Optional[DbReader]) -> int:
|
||
"""启动时异步预热 display_name 缓存。
|
||
|
||
从高并发场景看，首批 30-100 个请求可能同时到达，若缓存为空会并发查询
|
||
contact.db。预热最近 1000 个联系人可显著降低启动初期的 DB 压力。
|
||
|
||
Args:
|
||
db_reader: DbReader 实例
|
||
|
||
Returns:
|
||
预热写入的条目数
|
||
"""
|
||
if db_reader is None:
|
||
return 0
|
||
t0 = time.perf_counter()
|
||
try:
|
||
result = await asyncio.to_thread(
|
||
db_reader.get_contacts, "", _DISPLAY_NAME_WARMUP_LIMIT
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("warm_display_name_cache: 预热失败 %s", exc)
|
||
return 0
|
||
|
||
contacts = result.get("contacts", [])
|
||
warmed = 0
|
||
now = time.monotonic()
|
||
for contact in contacts:
|
||
wxid = contact.get("username") or contact.get("wxid")
|
||
if not wxid:
|
||
continue
|
||
resolved = contact.get("remark") or contact.get("nickname") or wxid
|
||
# 启动时批量写入，暂不触发 LRU 淘汰
|
||
_display_name_cache[wxid] = (resolved, now)
|
||
warmed += 1
|
||
|
||
logger.info(
|
||
"warm_display_name_cache: 预热 %d 条联系人 (%.0fms)",
|
||
warmed, (time.perf_counter() - t0) * 1000,
|
||
)
|
||
return warmed
|
||
|
||
|
||
async def _get_latest_create_time_for_talker(
|
||
db_reader: Optional[DbReader],
|
||
to_wxid: str,
|
||
) -> tuple[int, int]:
|
||
"""返回目标 talker 当前最新消息的 (create_time, local_id)，DB 不可用时返回 (0, 0)。"""
|
||
if db_reader is None:
|
||
return (0, 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:
|
||
ct = int(messages[0].get("create_time", 0))
|
||
lid = int(messages[0].get("local_id", 0))
|
||
return (ct, lid)
|
||
except Exception:
|
||
pass
|
||
return (0, 0)
|
||
|
||
|
||
async def _verify_sent_to_talker(
|
||
db_reader: Optional[DbReader],
|
||
to_wxid: str,
|
||
before_ct: int,
|
||
before_local_id: int = 0,
|
||
content: str = "",
|
||
max_wait_sec: float = 3.0,
|
||
) -> bool:
|
||
"""轮询 DB，确认目标 talker 在 (before_ct, before_local_id) 之后出现了新消息。
|
||
|
||
用于发送后校验消息是否真的落到了目标会话。DB 不可用时静默跳过。
|
||
默认最多等待 3 秒（P1 收紧，原 10s 过长会拖垮 L4 HTTP 超时），
|
||
轮询间隔 0.2s。
|
||
|
||
防串号校验（P1 关键修复）：
|
||
- 复合游标 (before_ct, before_local_id) 避免匹配到基线消息本身
|
||
（原 cursor_local_id=0 默认值会让 SQL `local_id > 0` 恒真，误把
|
||
基线消息当新消息）。
|
||
- 内容匹配校验：get_messages_by_session 按 talker 分表查询，返回消息的
|
||
talker 字段恒等于 to_wxid，无法通过 talker 字段检测串号。改为校验
|
||
是否有消息内容包含发送的 content；无 content 参数时回退到仅检查
|
||
新消息存在。
|
||
|
||
Note:
|
||
P3 阶段该校验已迁移到 SendTextFlow._wait_verify（基于 SSE 广播），
|
||
本函数仅 legacy 路径使用。
|
||
"""
|
||
if db_reader is None:
|
||
return True
|
||
if before_ct <= 0:
|
||
# 无基线时间戳（DB 不可用或会话无历史消息），无法准确校验
|
||
# 返回 True 避免假阳性（不阻塞发送），但记录 warning
|
||
logger.warning(
|
||
"send/text: DB 校验跳过，before_ct=%d 无法建立基线 (to_wxid=%s)",
|
||
before_ct, to_wxid,
|
||
)
|
||
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,
|
||
cursor_local_id=before_local_id,
|
||
limit=3,
|
||
direction="after",
|
||
is_sender=None,
|
||
)
|
||
messages = result.get("messages", [])
|
||
if messages:
|
||
# 内容匹配校验（防串号）
|
||
if content:
|
||
for msg in messages:
|
||
msg_content = msg.get("content", "")
|
||
if content in msg_content:
|
||
logger.info(
|
||
"send/text: DB 校验成功，内容匹配 (attempt=%d, %d msgs)",
|
||
attempts, len(messages),
|
||
)
|
||
return True
|
||
logger.warning(
|
||
"send/text: DB 校验：%d 条新消息但内容不匹配 (attempt=%d)",
|
||
len(messages), attempts,
|
||
)
|
||
# 继续轮询，可能是 DB 写入延迟
|
||
else:
|
||
# 无 content 参数，回退到仅检查新消息存在
|
||
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.2)
|
||
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:
|
||
"""发送文本消息。"""
|
||
# P3：根据 ui_backend 分支
|
||
ui_backend = _state.config.ui_backend if _state.config else "legacy"
|
||
|
||
if ui_backend == "flow" and _state.orchestrator is not None:
|
||
# 新架构：走 FlowOrchestrator
|
||
return await _send_text_via_flow(req)
|
||
|
||
# 旧架构：走 xdotool_driver（保持原有逻辑不变）
|
||
return await _send_text_via_legacy(req)
|
||
|
||
|
||
async def _send_text_via_legacy(req: SendTextRequest) -> SendResponse:
|
||
"""旧架构发送：走 xdotool_driver + send_queue + DB 校验。
|
||
|
||
流程：
|
||
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_<unix秒>_<随机>）
|
||
/ 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, before_local_id = await _get_latest_create_time_for_talker(_state.db_reader, req.to_wxid)
|
||
logger.info("send/text: 目标会话当前最新 create_time=%d local_id=%d", before_ct, before_local_id)
|
||
|
||
# 经发送队列串行执行
|
||
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, before_local_id, req.content)
|
||
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,
|
||
verified=verified,
|
||
error=None,
|
||
)
|
||
|
||
|
||
async def _send_text_via_flow(req: SendTextRequest) -> SendResponse:
|
||
"""新架构发送：经 FlowOrchestrator 编排（幂等/会话缓存/SSE 校验）。"""
|
||
orchestrator = _state.orchestrator
|
||
if orchestrator is None:
|
||
raise BridgeError(
|
||
code="BRIDGE_INTERNAL_ERROR", message="orchestrator not initialized"
|
||
)
|
||
|
||
t_total = time.perf_counter()
|
||
|
||
# 校验参数
|
||
if not req.to_wxid:
|
||
logger.warning("send/text: 参数校验失败 — to_wxid 为空 (flow 路径)")
|
||
raise BridgeError(code="INVALID_PARAMS", message="to_wxid 不能为空")
|
||
if not req.content:
|
||
logger.warning("send/text: 参数校验失败 — content 为空 (to_wxid=%s, flow 路径)", req.to_wxid)
|
||
raise BridgeError(code="INVALID_PARAMS", message="content 不能为空")
|
||
|
||
logger.info(
|
||
"send/text: 收到请求 (flow) to_wxid=%s content_len=%d display_name=%s client_request_id=%s",
|
||
req.to_wxid, len(req.content), req.display_name or "(未提供)",
|
||
req.client_request_id or "(none)",
|
||
)
|
||
|
||
# 检查登录态（与 legacy 路径一致）
|
||
t0 = time.perf_counter()
|
||
xdotool = _require_xdotool()
|
||
login_state = await xdotool.detect_login_state()
|
||
logger.info(
|
||
"send/text: 登录态检测 (flow) → %s (%.0fms)",
|
||
login_state, (time.perf_counter() - t0) * 1000,
|
||
)
|
||
if login_state != "logged_in":
|
||
logger.warning("send/text: 拒绝发送（flow 路径），登录态=%s", login_state)
|
||
raise BridgeError(
|
||
code="WECHAT_NOT_LOGGED_IN",
|
||
message=f"当前登录态为 {login_state}，无法发送消息",
|
||
)
|
||
|
||
# 解析 display_name（复用现有 _resolve_display_name）
|
||
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 解析 (flow) → %s (%.0fms)",
|
||
display_name, (time.perf_counter() - t0) * 1000,
|
||
)
|
||
|
||
# 调用 FlowOrchestrator
|
||
result = await orchestrator.send_text(
|
||
to_wxid=req.to_wxid,
|
||
content=req.content,
|
||
display_name=display_name,
|
||
client_request_id=req.client_request_id,
|
||
)
|
||
|
||
# FlowResult 转 SendResponse
|
||
if result.success:
|
||
logger.info(
|
||
"send/text: ✓ (flow) to=%s display_name=%s local_send_id=%s verified=%s skipped=%s (总耗时 %.0fms)",
|
||
req.to_wxid, display_name, result.local_id, result.verified, result.skipped,
|
||
(time.perf_counter() - t_total) * 1000,
|
||
)
|
||
return SendResponse(
|
||
success=True,
|
||
local_send_id=result.local_id,
|
||
placeholder=False,
|
||
verified=result.verified,
|
||
skipped=result.skipped,
|
||
)
|
||
|
||
# 失败：抛 BridgeError（保持与 legacy 一致的错误语义）
|
||
error_code = result.error_code or "SEND_FAILED"
|
||
logger.warning(
|
||
"send/text: ✗ (flow) to=%s failed code=%s error=%s (总耗时 %.0fms)",
|
||
req.to_wxid, error_code, result.error,
|
||
(time.perf_counter() - t_total) * 1000,
|
||
)
|
||
# 已知错误码直接透传（含可重试错误码 RATE_LIMITED/SEND_TIMEOUT 等），
|
||
# ERROR_CODES 表已补全所有 Flow 错误码的 HTTP 映射
|
||
if error_code in ERROR_CODES:
|
||
raise BridgeError(
|
||
code=error_code,
|
||
message=result.error or error_code.lower(),
|
||
details=result.details, # 传递 retry_after 等
|
||
)
|
||
# 未知错误码回落到 SEND_FAILED
|
||
raise BridgeError(code="SEND_FAILED", message=result.error or "send failed")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 路由：POST /api/send/image
|
||
# ---------------------------------------------------------------------------
|
||
@router.post("/api/send/image", response_model=SendResponse)
|
||
async def send_image(req: SendFileRequest) -> SendResponse:
|
||
"""发送图片消息。
|
||
|
||
flow 架构下走真实文件发送（系统文件选择器路径输入 + DB 校验，placeholder=False），
|
||
legacy 架构下为占位实现（仅发"[图片] 文件名"提示文本，placeholder=True）。
|
||
|
||
Args:
|
||
req: SendFileRequest（to_wxid + file_path + 可选 display_name / client_request_id）
|
||
|
||
Returns:
|
||
SendResponse：success / placeholder（flow=False, legacy=True） / verified
|
||
|
||
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:
|
||
"""发送文件消息。
|
||
|
||
flow 架构下走真实文件发送（系统文件选择器路径输入 + DB 校验，placeholder=False），
|
||
legacy 架构下为占位实现（仅发"[文件] 文件名"提示文本，placeholder=True）。
|
||
|
||
Args:
|
||
req: SendFileRequest（to_wxid + file_path + 可选 display_name / client_request_id）
|
||
|
||
Returns:
|
||
SendResponse：success / placeholder（flow=False, legacy=True） / verified
|
||
|
||
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:
|
||
"""发送文件/图片（自动选择 flow / legacy 路径）。
|
||
|
||
ui_backend == "flow" 且 orchestrator 可用时走真实文件发送（placeholder=False，
|
||
经 SendFileFlow 系统文件选择器路径输入 + DB 校验），
|
||
否则回退 legacy 占位实现（placeholder=True，仅发提示文本）。
|
||
"""
|
||
ui_backend = _state.config.ui_backend if _state.config else "legacy"
|
||
if ui_backend == "flow" and _state.orchestrator is not None:
|
||
return await _send_file_via_flow(req, is_image)
|
||
return await _send_file_via_legacy(req, is_image)
|
||
|
||
|
||
async def _send_file_via_flow(req: SendFileRequest, is_image: bool) -> SendResponse:
|
||
"""新架构文件发送：经 FlowOrchestrator 编排（真实发送，非 placeholder）。
|
||
|
||
流程：
|
||
1. 校验 to_wxid / file_path 非空
|
||
2. 路径白名单校验（提前返回 400，比入队后失败更友好）
|
||
3. 文件存在性校验（提前返回 400）
|
||
4. 登录态检测，非 logged_in 直接拒绝
|
||
5. 调 orchestrator.send_file（幂等 + 熔断 + 入队 + DB 校验）
|
||
6. 响应 placeholder=False + verified
|
||
|
||
Args:
|
||
req: SendFileRequest（to_wxid + file_path + 可选 display_name / client_request_id）
|
||
is_image: True=图片，False=普通文件（决定 file_type，仅用于日志）
|
||
|
||
Returns:
|
||
SendResponse：success / placeholder=False / verified / skipped
|
||
|
||
Raises:
|
||
BridgeError(INVALID_PARAMS): 参数为空 / 路径不在白名单 / 文件不存在（HTTP 400）
|
||
BridgeError(WECHAT_NOT_LOGGED_IN): 当前未登录（HTTP 401）
|
||
BridgeError(SEND_FAILED / SEND_TIMEOUT / ...): 发送失败（HTTP 500/503）
|
||
|
||
Notes:
|
||
- file_path 必须是容器内绝对路径且在白名单目录内
|
||
（/config/Desktop/ / /config/woc-uploads/ / /tmp/woc-files/）
|
||
- 真实文件传输经系统文件选择器（xdotool type 路径 + Return 提交）
|
||
- local_send_id 为空（文件发送不返回 local_send_id）
|
||
"""
|
||
orchestrator = _state.orchestrator
|
||
if orchestrator is None:
|
||
raise BridgeError(
|
||
code="BRIDGE_INTERNAL_ERROR", message="orchestrator not initialized"
|
||
)
|
||
|
||
# 校验参数
|
||
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 不能为空")
|
||
|
||
# 路径白名单校验（SendFileFlow 内部也会校验，路由层提前校验返回 400 更友好）
|
||
_validate_file_path(req.file_path)
|
||
|
||
# 文件存在性校验（提前返回 400，避免入队后才发现文件不存在）
|
||
if not os.path.isfile(req.file_path):
|
||
raise BridgeError(
|
||
code="INVALID_PARAMS", message=f"文件不存在: {req.file_path}"
|
||
)
|
||
|
||
# 检查登录态
|
||
xdotool = _require_xdotool()
|
||
login_state = await xdotool.detect_login_state()
|
||
if login_state != "logged_in":
|
||
logger.warning(
|
||
"send/file: 拒绝发送（flow 路径），登录态=%s", login_state
|
||
)
|
||
raise BridgeError(
|
||
code="WECHAT_NOT_LOGGED_IN",
|
||
message=f"当前登录态为 {login_state}，无法发送消息",
|
||
)
|
||
|
||
file_type = "image" if is_image else "file"
|
||
kind = "image" if is_image else "file"
|
||
logger.info(
|
||
"send/%s: 收到请求 to_wxid=%s file=%s (flow 路径)",
|
||
kind, req.to_wxid, req.file_path,
|
||
)
|
||
|
||
# P0 修复：解析 display_name 供 SendFileFlow 搜索会话使用。
|
||
# 部分联系人无法通过 wxid 搜索定位（微信搜索框对 wxid 不总是命中），
|
||
# 必须用备注/昵称搜索。_resolve_display_name 会查 contact.db 并缓存。
|
||
db_reader = _state.db_reader if _state else None
|
||
display_name = await _resolve_display_name(
|
||
db_reader, req.to_wxid, getattr(req, "display_name", None)
|
||
)
|
||
|
||
# 调用 FlowOrchestrator（幂等 + 熔断 + 入队 + DB 校验）
|
||
result = await orchestrator.send_file(
|
||
to_wxid=req.to_wxid,
|
||
file_path=req.file_path,
|
||
file_type=file_type,
|
||
client_request_id=req.client_request_id,
|
||
display_name=display_name,
|
||
)
|
||
|
||
if result["success"]:
|
||
logger.info(
|
||
"send/%s: to=%s file=%s → 成功 verified=%s (placeholder=False)",
|
||
kind, req.to_wxid, req.file_path, result["verified"],
|
||
)
|
||
return SendResponse(
|
||
success=True,
|
||
local_send_id="", # 文件发送无 local_send_id
|
||
placeholder=False,
|
||
verified=result["verified"],
|
||
skipped=result.get("skipped", False),
|
||
error=None,
|
||
)
|
||
|
||
# 失败：抛 BridgeError（保持与 _send_text_via_flow 一致的错误语义）
|
||
error_code = result.get("error_code") or "SEND_FAILED"
|
||
if error_code in ERROR_CODES:
|
||
raise BridgeError(
|
||
code=error_code,
|
||
message=result.get("error") or error_code.lower(),
|
||
)
|
||
raise BridgeError(
|
||
code="SEND_FAILED",
|
||
message=result.get("error") or "send file failed",
|
||
)
|
||
|
||
|
||
async def _send_file_via_legacy(req: SendFileRequest, is_image: bool) -> SendResponse:
|
||
"""send_image / send_file 共用实现（legacy 占位路径）。
|
||
|
||
流程：
|
||
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
|
||
（legacy 路径为占位实现，仅发提示文本；flow 路径返回 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 必须是容器内绝对路径（不是宿主机路径）
|
||
- legacy 路径不传输真实文件内容，仅发提示文本
|
||
- 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}",
|
||
)
|
||
|
||
# 微信撤回时限：发送后 2 分钟内（Task 11 收紧到 110s 留 10s UI 操作余量）
|
||
_REVOKE_WINDOW_SEC = 110
|
||
now_ts = int(time.time())
|
||
if now_ts - req.create_time > _REVOKE_WINDOW_SEC:
|
||
raise BridgeError(
|
||
code="REVOKE_WINDOW_EXPIRED",
|
||
message=f"消息已超过 2 分钟撤回时限（发送于 {req.create_time}，当前 {now_ts}，"
|
||
f"已为 UI 操作预留 10s 余量，阈值收紧到 {_REVOKE_WINDOW_SEC}s）",
|
||
)
|
||
|
||
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,
|
||
)
|
||
|
||
# Task 11: post_verify 校验撤回是否真正生效（DB 中出现撤回系统消息）
|
||
verified: bool | None = None
|
||
try:
|
||
verified = await xdotool._message.post_verify_revoke_message(
|
||
db_reader, req.talker, req.create_time
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"messages/revoke: post_verify 异常 (talker=%s): %s", req.talker, exc,
|
||
)
|
||
verified = False
|
||
return RevokeMessageResponse(success=True, error=None, verified=verified)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 路由：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
|
||
)
|
||
|
||
# P0 修复：UI 操作前捕获目标会话消息基线 (create_time, local_id)，
|
||
# 供 post_verify 用 direction="after" 只查新消息，避免历史 type=49 误报。
|
||
baseline_create_time = 0
|
||
baseline_local_id = 0
|
||
try:
|
||
baseline_create_time, baseline_local_id = (
|
||
await xdotool._message.capture_forward_baseline(
|
||
db_reader, req.target_display_name
|
||
)
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"messages/forward: 捕获基线异常 (target=%s): %s",
|
||
req.target_display_name, exc,
|
||
)
|
||
|
||
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,
|
||
)
|
||
|
||
# Task 11: post_verify 校验转发是否生效（目标会话出现 type=49 消息）
|
||
verified: bool | None = None
|
||
try:
|
||
verified = await xdotool._message.post_verify_forward_message(
|
||
db_reader, req.target_display_name,
|
||
baseline_create_time=baseline_create_time,
|
||
baseline_local_id=baseline_local_id,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"messages/forward: post_verify 异常 (target=%s): %s",
|
||
req.target_display_name, exc,
|
||
)
|
||
verified = False
|
||
return ForwardMessageResponse(success=True, error=None, verified=verified)
|