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字段
443 lines
17 KiB
Python
443 lines
17 KiB
Python
"""L5 FlowOrchestrator：编排幂等/会话缓存/队列/Flow。
|
||
|
||
send_text 流程：
|
||
1. 幂等检查（命中返回 skipped=true）
|
||
2. 入队（自适应 delay_ms：同联系人 1000 / 不同 config.send_delay_ms）
|
||
3. 执行 SendTextFlow（内部用 SSEVerifyBus 校验）
|
||
4. 成功且 verified 才写幂等缓存
|
||
5. verified=False 不阻塞发送、不熔断（仅记 warning，调用方按需重试）
|
||
6. 失败时 session_cache.invalidate(to_wxid)（仅失效当前联系人）
|
||
|
||
熔断器已移除：SSE 校验超时不会导致雪崩，单次 verify 失败不应阻塞全局发送。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import time
|
||
from collections import OrderedDict
|
||
from typing import Optional, TYPE_CHECKING
|
||
|
||
from woc_bridge.models import BridgeError
|
||
|
||
if TYPE_CHECKING:
|
||
from woc_bridge.ui.flows.send_file import SendFileFlow
|
||
from woc_bridge.ui.flows.send_text import SendTextFlow
|
||
from woc_bridge.ui.errors import FlowError
|
||
from woc_bridge.ui.flow import FlowContext, FlowResult, FlowState
|
||
from woc_bridge.ui.idem_cache import IdemCache
|
||
|
||
logger = logging.getLogger("woc-bridge")
|
||
|
||
# 同联系人发送间隔（毫秒），比默认 send_delay_ms 短
|
||
_SAME_CONTACT_DELAY_MS = 1000
|
||
|
||
|
||
class SessionCache:
|
||
"""会话缓存：LRU 多会话，30s TTL。
|
||
|
||
命中时 SendTextFlow 跳过 activate/click_search_box/type_query/open_session 四步，
|
||
直接校验会话仍打开后进入 focus_input。
|
||
多会话缓存提升 30-100 并发人群中重复联系人的命中率。
|
||
"""
|
||
|
||
def __init__(self, ttl: float = 30.0, max_size: int = 16) -> None:
|
||
self.ttl = ttl
|
||
self.max_size = max_size
|
||
# wxid -> expire_at，按最近使用排序
|
||
self._cache: OrderedDict[str, float] = OrderedDict()
|
||
|
||
def get(self, to_wxid: str) -> bool:
|
||
"""查询指定 wxid 是否在缓存且未过期。命中时移到队尾（最近使用）。"""
|
||
expire_at = self._cache.get(to_wxid)
|
||
if expire_at is None:
|
||
return False
|
||
if time.monotonic() >= expire_at:
|
||
self._cache.pop(to_wxid, None)
|
||
return False
|
||
self._cache.move_to_end(to_wxid)
|
||
return True
|
||
|
||
def set(self, to_wxid: str) -> None:
|
||
"""更新缓存。"""
|
||
self._cache[to_wxid] = time.monotonic() + self.ttl
|
||
self._cache.move_to_end(to_wxid)
|
||
# 超限淘汰最久未使用
|
||
while len(self._cache) > self.max_size:
|
||
self._cache.popitem(last=False)
|
||
|
||
def invalidate(self, to_wxid: Optional[str] = None) -> None:
|
||
"""失效缓存。
|
||
|
||
Args:
|
||
to_wxid: 为 None 时清除全部缓存；为具体 wxid 时仅清除该条目
|
||
（传空字符串不会误清全部）。
|
||
"""
|
||
if to_wxid is None:
|
||
self._cache.clear()
|
||
else:
|
||
self._cache.pop(to_wxid, None)
|
||
|
||
|
||
class FlowOrchestrator:
|
||
"""L5 Flow 编排器。
|
||
|
||
整合幂等缓存、会话缓存、发送队列、Flow 执行、SSE 发送结果校验。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
send_queue,
|
||
idem_cache: IdemCache,
|
||
actions,
|
||
db_reader,
|
||
config,
|
||
send_text_flow: Optional[SendTextFlow] = None,
|
||
send_file_flow: Optional[SendFileFlow] = None,
|
||
verify_bus=None,
|
||
) -> None:
|
||
"""初始化编排器。
|
||
|
||
Args:
|
||
send_queue: SendQueue 实例
|
||
idem_cache: IdemCache 实例
|
||
actions: L3 Actions 实例
|
||
db_reader: DbReader 实例
|
||
config: BridgeConfig 实例（含 send_delay_ms 等）
|
||
send_text_flow: 可选的 SendTextFlow 实例（用于测试注入）
|
||
send_file_flow: 可选的 SendFileFlow 实例（必须由 app._init_state 注入，
|
||
因 SendFileFlow 依赖 XdotoolDriver，orchestrator 无法惰性创建）
|
||
verify_bus: SSE 发送结果校验总线（SSEVerifyBus 实例）。
|
||
惰性创建 SendTextFlow 时注入，用于 SSE 推送校验。
|
||
"""
|
||
self.send_queue = send_queue
|
||
self.idem_cache = idem_cache
|
||
self.actions = actions
|
||
self.db_reader = db_reader
|
||
self.config = config
|
||
# LRU 多会话缓存：最多 16 个会话，30s TTL
|
||
self.session_cache = SessionCache(ttl=30.0, max_size=16)
|
||
self._send_text_flow = send_text_flow
|
||
self._send_file_flow = send_file_flow
|
||
self._verify_bus = verify_bus
|
||
|
||
def _get_send_text_flow(self):
|
||
"""获取 SendTextFlow 实例（惰性创建或使用注入的）。"""
|
||
if self._send_text_flow is not None:
|
||
return self._send_text_flow
|
||
# 惰性导入避免循环依赖
|
||
from woc_bridge.ui.flows.send_text import SendTextFlow
|
||
return SendTextFlow(
|
||
actions=self.actions,
|
||
db_reader=self.db_reader,
|
||
session_cache=self.session_cache,
|
||
verify_bus=self._verify_bus,
|
||
)
|
||
|
||
def _get_send_file_flow(self):
|
||
"""获取 SendFileFlow 实例（必须由 app._init_state 注入）。
|
||
|
||
SendFileFlow 依赖 XdotoolDriver（用于 xdotool type 路径），
|
||
orchestrator 未持有 XdotoolDriver 引用，无法惰性创建。
|
||
未注入时抛 BridgeError(BRIDGE_INTERNAL_ERROR)。
|
||
"""
|
||
if self._send_file_flow is not None:
|
||
return self._send_file_flow
|
||
raise BridgeError(
|
||
code="BRIDGE_INTERNAL_ERROR",
|
||
message="send_file_flow 未初始化（需在 app._init_state 中注入 SendFileFlow）",
|
||
)
|
||
|
||
async def send_text(
|
||
self,
|
||
to_wxid: str,
|
||
content: str,
|
||
display_name: str = "",
|
||
client_request_id: str = "",
|
||
) -> FlowResult:
|
||
"""发送文本消息。
|
||
|
||
流程：
|
||
1. 幂等检查
|
||
2. 自适应延时入队
|
||
3. 执行 SendTextFlow（内部用 SSEVerifyBus 校验）
|
||
4. 成功且 verified 写幂等缓存
|
||
5. verified=False 不阻塞、不熔断（调用方按需重试）
|
||
6. 失败时失效会话缓存
|
||
|
||
Args:
|
||
to_wxid: 目标 wxid
|
||
content: 消息内容
|
||
display_name: 显示名（搜索用）
|
||
client_request_id: 客户端幂等 ID
|
||
|
||
Returns:
|
||
FlowResult
|
||
"""
|
||
started_at = time.monotonic()
|
||
|
||
logger.info(
|
||
"[orchestrator] send_text START to_wxid=%s content_len=%d display_name=%s client_request_id=%s",
|
||
to_wxid, len(content), display_name or "(empty)", client_request_id or "(none)",
|
||
)
|
||
|
||
# 1. 幂等检查
|
||
cached = self.idem_cache.get(
|
||
"send_text", to_wxid, content, client_request_id
|
||
)
|
||
if cached is not None:
|
||
logger.info(
|
||
"[orchestrator] idem cache hit, skip send (to_wxid=%s)",
|
||
to_wxid,
|
||
)
|
||
# 返回缓存的 FlowResult，标记 skipped
|
||
if isinstance(cached, FlowResult):
|
||
return FlowResult(
|
||
success=True,
|
||
local_id=cached.local_id,
|
||
verified=cached.verified,
|
||
skipped=True,
|
||
state=cached.state,
|
||
duration_ms=(time.monotonic() - started_at) * 1000,
|
||
)
|
||
# 兜底：构造一个 skipped 结果
|
||
return FlowResult(
|
||
success=True,
|
||
skipped=True,
|
||
duration_ms=(time.monotonic() - started_at) * 1000,
|
||
)
|
||
|
||
# 2. 自适应延时：同联系人用短延时
|
||
is_same_contact = self.session_cache.get(to_wxid)
|
||
delay_ms = _SAME_CONTACT_DELAY_MS if is_same_contact else None
|
||
# None 表示用 send_queue 的默认 send_delay_ms
|
||
logger.info(
|
||
"[orchestrator] send_text enqueue: same_contact=%s delay_ms=%s to_wxid=%s",
|
||
is_same_contact, delay_ms, to_wxid,
|
||
)
|
||
|
||
# 3. 构造 FlowContext
|
||
ctx = FlowContext(
|
||
to_wxid=to_wxid,
|
||
content=content,
|
||
display_name=display_name,
|
||
client_request_id=client_request_id,
|
||
)
|
||
|
||
# 4. 入队执行（带队列等待超时，避免无界排队）
|
||
flow = self._get_send_text_flow()
|
||
try:
|
||
result: FlowResult = await self.send_queue.enqueue(
|
||
lambda: flow.run(ctx),
|
||
delay_ms=delay_ms,
|
||
wait_timeout_ms=self.config.send_queue_wait_timeout_ms,
|
||
)
|
||
except Exception as exc:
|
||
logger.error(
|
||
"[orchestrator] send_queue exception: %s: %s (to_wxid=%s)",
|
||
type(exc).__name__, exc, to_wxid,
|
||
)
|
||
# 失败时仅失效当前联系人会话缓存，避免误伤其它缓存命中
|
||
self.session_cache.invalidate(to_wxid)
|
||
# 保留 BridgeError 的原始错误码（如 RATE_LIMITED），
|
||
# 避免限流/超时等可重试错误被吞成 SEND_FAILED
|
||
error_code = "SEND_FAILED"
|
||
if hasattr(exc, "code") and exc.code:
|
||
error_code = exc.code
|
||
# 传递 details（如 retry_after）供路由层设置 Retry-After 头
|
||
exc_details = None
|
||
if hasattr(exc, "details") and exc.details:
|
||
exc_details = exc.details
|
||
return FlowResult(
|
||
success=False,
|
||
error=str(exc),
|
||
error_code=error_code,
|
||
state=FlowState.FAILED,
|
||
duration_ms=(time.monotonic() - started_at) * 1000,
|
||
details=exc_details,
|
||
)
|
||
|
||
# 5. 结果处理
|
||
if result.success and result.verified:
|
||
# 成功且 verified：写幂等缓存
|
||
self.idem_cache.set(
|
||
"send_text", to_wxid, content, result, client_request_id
|
||
)
|
||
logger.info(
|
||
"[orchestrator] send_text DONE ✓ to_wxid=%s local_id=%s verified=%s (%.0fms)",
|
||
to_wxid, result.local_id, result.verified,
|
||
(time.monotonic() - started_at) * 1000,
|
||
)
|
||
elif result.success and not result.verified:
|
||
# 发送成功但 SSE 校验未匹配（超时/消息未到/发错会话）
|
||
# 不熔断、不阻塞：调用方按需重试或通过 SSE 自行确认
|
||
logger.warning(
|
||
"[orchestrator] send success but verify unmatched (to_wxid=%s, local_id=%s)",
|
||
to_wxid, result.local_id,
|
||
)
|
||
else:
|
||
# 发送失败：仅失效当前联系人会话缓存，避免误伤其它缓存命中
|
||
self.session_cache.invalidate(to_wxid)
|
||
logger.warning(
|
||
"[orchestrator] send failed: %s (code=%s, to_wxid=%s)",
|
||
result.error, result.error_code, to_wxid,
|
||
)
|
||
|
||
return result
|
||
|
||
async def send_file(
|
||
self,
|
||
to_wxid: str,
|
||
file_path: str,
|
||
file_type: str = "file",
|
||
client_request_id: Optional[str] = None,
|
||
display_name: Optional[str] = None,
|
||
) -> dict:
|
||
"""文件发送 Flow 编排：幂等检查 + 入队 + 校验。
|
||
|
||
流程：
|
||
1. 幂等检查（命中返回 skipped=True）
|
||
2. 入队执行 SendFileFlow.execute（路径白名单 + 文件选择器输入 + post_verify）
|
||
3. 成功且 verified 写幂等缓存
|
||
4. verified=False 不熔断、不阻塞（仅记 warning）
|
||
5. 失败时失效会话缓存
|
||
|
||
幂等缓存键：(send_file, to_wxid, file_path, client_request_id)
|
||
（IdemCache API 仅支持 4 段 key，file_path 作为 content 槽位传入）
|
||
|
||
Args:
|
||
to_wxid: 目标会话 wxid
|
||
file_path: 文件绝对路径（必须在白名单目录内）
|
||
file_type: "image" / "file" / "video"（仅用于日志，不影响流程）
|
||
client_request_id: 客户端幂等 ID（None 或空字符串表示不参与幂等）
|
||
display_name: 用于微信搜索框定位会话的显示名（备注/昵称），
|
||
None 时 SendFileFlow 回退到 to_wxid 搜索
|
||
|
||
Returns:
|
||
{"success": bool, "verified": bool, "placeholder": False,
|
||
"skipped": bool, "error": Optional[str], "error_code": Optional[str]}
|
||
|
||
Raises:
|
||
BridgeError(BRIDGE_INTERNAL_ERROR): send_file_flow 未注入
|
||
"""
|
||
started_at = time.monotonic()
|
||
# client_request_id 归一化：None -> ""
|
||
idem_id = client_request_id or ""
|
||
|
||
logger.info(
|
||
"[orchestrator] send_file START to_wxid=%s file=%s type=%s display_name=%s client_request_id=%s",
|
||
to_wxid, file_path, file_type, display_name or "(none)", idem_id or "(none)",
|
||
)
|
||
|
||
# 1. 幂等检查
|
||
cached = self.idem_cache.get("send_file", to_wxid, file_path, idem_id)
|
||
if cached is not None:
|
||
logger.info(
|
||
"[orchestrator] send_file idem cache hit, skip (to_wxid=%s)",
|
||
to_wxid,
|
||
)
|
||
if isinstance(cached, dict):
|
||
return {
|
||
"success": cached.get("success", True),
|
||
"verified": cached.get("verified", False),
|
||
"placeholder": False,
|
||
"skipped": True,
|
||
"error": None,
|
||
"error_code": None,
|
||
}
|
||
# 兜底：构造一个 skipped 结果
|
||
return {
|
||
"success": True, "verified": False, "placeholder": False,
|
||
"skipped": True, "error": None, "error_code": None,
|
||
}
|
||
|
||
# 2. 获取 SendFileFlow 实例（未注入时抛 BridgeError）
|
||
flow = self._get_send_file_flow()
|
||
|
||
# 3. 入队执行（SendFileFlow.execute 内部完成路径校验 + 文件选择器输入 + post_verify）
|
||
logger.info(
|
||
"[orchestrator] send_file enqueue (pending=%d)",
|
||
self.send_queue.pending_count(),
|
||
)
|
||
try:
|
||
result = await self.send_queue.enqueue(
|
||
lambda: flow.execute(
|
||
to_wxid, file_path, file_type, display_name=display_name
|
||
),
|
||
wait_timeout_ms=self.config.send_queue_wait_timeout_ms,
|
||
)
|
||
except BridgeError as exc:
|
||
logger.error(
|
||
"[orchestrator] send_file BridgeError code=%s msg=%s (to_wxid=%s)",
|
||
exc.code, exc.message, to_wxid,
|
||
)
|
||
# 失败时失效当前联系人会话缓存
|
||
self.session_cache.invalidate(to_wxid)
|
||
return {
|
||
"success": False, "verified": False, "placeholder": False,
|
||
"skipped": False,
|
||
"error": exc.message,
|
||
"error_code": exc.code,
|
||
}
|
||
except Exception as exc:
|
||
logger.error(
|
||
"[orchestrator] send_file exception: %s: %s (to_wxid=%s)",
|
||
type(exc).__name__, exc, to_wxid,
|
||
)
|
||
self.session_cache.invalidate(to_wxid)
|
||
return {
|
||
"success": False, "verified": False, "placeholder": False,
|
||
"skipped": False,
|
||
"error": str(exc),
|
||
"error_code": "SEND_FAILED",
|
||
}
|
||
|
||
# 4. 结果处理
|
||
# SendFileFlow.execute 返回 {"success": bool, "verified": Optional[bool], "error": Optional[str]}
|
||
success = result.get("success", False)
|
||
verified = result.get("verified", False)
|
||
|
||
if success and verified:
|
||
# 成功且 verified：写幂等缓存
|
||
self.idem_cache.set(
|
||
"send_file", to_wxid, file_path, result, idem_id
|
||
)
|
||
# 更新会话缓存（send_file 打开了 to_wxid 会话）
|
||
self.session_cache.set(to_wxid)
|
||
logger.info(
|
||
"[orchestrator] send_file DONE ✓ to=%s file=%s verified=%s (%.0fms)",
|
||
to_wxid, file_path, verified,
|
||
(time.monotonic() - started_at) * 1000,
|
||
)
|
||
elif success and verified is False:
|
||
# 发送成功但校验失败：不熔断、不阻塞
|
||
self.session_cache.set(to_wxid)
|
||
logger.warning(
|
||
"[orchestrator] send_file success but verify unmatched (to_wxid=%s)",
|
||
to_wxid,
|
||
)
|
||
elif success and verified is None:
|
||
# 发送成功但无法校验：不写幂等缓存，不熔断
|
||
self.session_cache.set(to_wxid)
|
||
logger.info(
|
||
"[orchestrator] send_file success but verify skipped (to_wxid=%s)",
|
||
to_wxid,
|
||
)
|
||
else:
|
||
# 发送失败：仅失效当前联系人会话缓存
|
||
self.session_cache.invalidate(to_wxid)
|
||
logger.warning(
|
||
"[orchestrator] send_file failed: %s (to_wxid=%s)",
|
||
result.get("error"), to_wxid,
|
||
)
|
||
|
||
return {
|
||
"success": success,
|
||
"verified": verified,
|
||
"placeholder": False,
|
||
"skipped": False,
|
||
"error": result.get("error"),
|
||
"error_code": None,
|
||
}
|