Kris
be93906635
1. 新增is_window_active接口并在xdotool后端实现，用于校验窗口焦点 2. 为xdotool后端添加主窗口ID缓存，减少重复xdotool调用 3. 精简click命令流程，移除--sync参数避免VNC环境死等 4. 简化post_verify执行逻辑，减少不必要的点击操作 5. 为缓存命中分支添加会话校验回退机制，处理窗口失焦/消失场景
252 lines
9.6 KiB
Python
252 lines
9.6 KiB
Python
"""L4 Flow 基类与状态机基础设施。
|
||
|
||
Flow 是 L4 层的核心抽象：编排 L3 Actions 完成完整业务流程（如发送文本、
|
||
发送文件、发布朋友圈）。每个 Flow 由若干 transition 组成，每个 transition
|
||
对应一个状态转换。
|
||
|
||
状态机（9 个状态）：
|
||
INITIAL → WINDOW_ACTIVATED → SEARCH_BOX_CLICKED → QUERY_TYPED →
|
||
SESSION_OPENED → INPUT_FOCUSED → TEXT_TYPED → SENT → CLEANED_UP
|
||
任何状态失败 → FAILED
|
||
|
||
异常处理：FlowError / TimeoutError 时调 _reset_to_idle 恢复 UI 状态，
|
||
避免残留脏状态影响下次流程。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import enum
|
||
import logging
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from typing import Optional
|
||
|
||
from woc_bridge.ui.backends.base import WindowGeometry
|
||
from woc_bridge.ui.errors import FlowError, PermanentError
|
||
|
||
logger = logging.getLogger("woc-bridge")
|
||
|
||
# 整流程超时（与 backends/xdotool.py 的 _FLOW_TIMEOUT_SEC 对齐）
|
||
_FLOW_TIMEOUT_SEC = 30.0
|
||
|
||
|
||
class FlowState(enum.Enum):
|
||
"""Flow 状态机 9 个状态。"""
|
||
INITIAL = "initial"
|
||
WINDOW_ACTIVATED = "window_activated"
|
||
SEARCH_BOX_CLICKED = "search_box_clicked"
|
||
QUERY_TYPED = "query_typed"
|
||
SESSION_OPENED = "session_opened"
|
||
INPUT_FOCUSED = "input_focused"
|
||
TEXT_TYPED = "text_typed"
|
||
SENT = "sent"
|
||
CLEANED_UP = "cleaned_up"
|
||
FAILED = "failed"
|
||
|
||
|
||
@dataclass
|
||
class FlowContext:
|
||
"""Flow 执行上下文（贯穿整个流程，各 transition 共享）。
|
||
|
||
所有字段均有默认值，避免构造时必填字段顺序问题。
|
||
"""
|
||
to_wxid: str = ""
|
||
content: str = ""
|
||
display_name: str = ""
|
||
client_request_id: str = ""
|
||
win_geom: Optional[WindowGeometry] = None
|
||
local_send_id: str = ""
|
||
verified: bool = False
|
||
image_confidence: float = 0.0
|
||
# 内部状态
|
||
current_state: FlowState = FlowState.INITIAL
|
||
started_at: float = field(default_factory=time.monotonic)
|
||
|
||
|
||
@dataclass
|
||
class FlowResult:
|
||
"""Flow 执行结果。"""
|
||
success: bool = False
|
||
local_id: str = ""
|
||
verified: bool = False
|
||
skipped: bool = False
|
||
error: str = ""
|
||
error_code: str = ""
|
||
state: FlowState = FlowState.INITIAL
|
||
image_confidence: float = 0.0
|
||
duration_ms: float = 0.0
|
||
details: Optional[dict] = None
|
||
|
||
|
||
class Flow:
|
||
"""L4 Flow 基类。
|
||
|
||
子类需实现 run(ctx) 方法，编排各 transition。
|
||
基类提供 _reset_to_idle / _verify_main_view / _full_cleanup_on_startup
|
||
等通用恢复与校验方法。
|
||
"""
|
||
|
||
def __init__(self, actions, db_reader=None) -> None:
|
||
"""初始化 Flow。
|
||
|
||
Args:
|
||
actions: L3 Actions 实例
|
||
db_reader: DB 读取器（可选，用于校验）
|
||
"""
|
||
self.actions = actions
|
||
self.db_reader = db_reader
|
||
|
||
# ------------------------------------------------------------------
|
||
# 状态恢复
|
||
# ------------------------------------------------------------------
|
||
async def _reset_to_idle(self, win_geom: Optional[WindowGeometry] = None) -> bool:
|
||
"""多策略清场：Esc×3 + 点击安全空白 + BackSpace×30 + Esc×2 + 校验。
|
||
|
||
Flow 失败或启动时调用，确保 UI 回到干净状态。
|
||
点击位置采用比例坐标，固定落在右侧聊天区域，避免命中左侧联系人列表、
|
||
分割线或底部输入框导致的误操作。
|
||
|
||
异常路径下应尽量减少 xdotool 调用次数，避免恢复路径放大延迟。
|
||
去掉 click 的 --sync 后不会卡死，但每条命令仍需 ~100ms，
|
||
故精简为 4 步：Esc×3 → click 安全空白 → BackSpace×30 → Esc×2。
|
||
post_verify 仅在 OpenCV 模式下有意义（纯坐标模式立即返回）。
|
||
|
||
Args:
|
||
win_geom: 窗口几何（可选，用于点击空白区域与 post_verify）
|
||
|
||
Returns:
|
||
True 表示清场成功（post_verify 通过或无需校验），False 表示可能仍残留脏状态
|
||
"""
|
||
try:
|
||
# 1. Esc×3 关闭可能的弹窗/搜索框/联系人详情
|
||
await self.actions.backend.key_press("Escape", repeat=3)
|
||
await asyncio.sleep(0.2)
|
||
|
||
# 2. 点击右侧聊天区域的安全空白处（避开左侧联系人面板与底部输入区）
|
||
if win_geom is not None and win_geom.width >= 200 and win_geom.height >= 200:
|
||
# 左侧联系人面板一般占宽度 30% 以内，右侧安全区取 65%-85%
|
||
# 顶部工具栏约 8%，底部输入区约 15%-20%，安全 y 取 25%-45%
|
||
safe_x_ratio = 0.75
|
||
safe_y_ratio = 0.40
|
||
blank_x = win_geom.x + int(win_geom.width * safe_x_ratio)
|
||
blank_y = win_geom.y + int(win_geom.height * safe_y_ratio)
|
||
# 兜底保护：若计算点仍落在左侧 42% 区域内，则强制右移
|
||
left_panel_max_x = win_geom.x + int(win_geom.width * 0.42)
|
||
if blank_x <= left_panel_max_x:
|
||
blank_x = win_geom.x + int(win_geom.width * 0.75)
|
||
await self.actions.backend.click(blank_x, blank_y)
|
||
logger.debug(
|
||
"[flow] _reset_to_idle click safe blank (%d,%d) ratio=(%.2f,%.2f)",
|
||
blank_x, blank_y, safe_x_ratio, safe_y_ratio,
|
||
)
|
||
await asyncio.sleep(0.2)
|
||
|
||
# 3. BackSpace×30 清空可能残留的输入框/搜索框内容
|
||
await self.actions.backend.key_press("BackSpace", repeat=30)
|
||
await asyncio.sleep(0.15)
|
||
|
||
# 4. Esc×2 再次关闭可能因 BackSpace 触发的弹窗
|
||
await self.actions.backend.key_press("Escape", repeat=2)
|
||
await asyncio.sleep(0.15)
|
||
|
||
# 5. post_verify：尝试用 main_view 模板校验
|
||
# 纯坐标模式下 wait_for 立即返回（纯算术），不阻塞
|
||
if win_geom is not None:
|
||
try:
|
||
elem = await self.actions.wait_for(
|
||
"main_view", win_geom, timeout=2.0, interval=0.3
|
||
)
|
||
if elem is not None:
|
||
logger.info("[flow] _reset_to_idle post_verify OK")
|
||
return True
|
||
logger.warning("[flow] _reset_to_idle post_verify: main_view not found")
|
||
return False
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"[flow] _reset_to_idle post_verify exception: %s", exc
|
||
)
|
||
return False
|
||
logger.info("[flow] _reset_to_idle done (no post_verify, win_geom=None)")
|
||
return True
|
||
except Exception as exc:
|
||
logger.warning("[flow] _reset_to_idle exception: %s", exc)
|
||
return False
|
||
|
||
async def _verify_main_view(self, win_geom: WindowGeometry) -> bool:
|
||
"""校验当前是否在主视图（用 main_view 模板）。
|
||
|
||
Args:
|
||
win_geom: 窗口几何
|
||
|
||
Returns:
|
||
True 表示在主视图
|
||
"""
|
||
try:
|
||
elem = await self.actions.wait_for(
|
||
"main_view", win_geom, timeout=3.0, interval=0.3
|
||
)
|
||
return elem is not None
|
||
except Exception:
|
||
return False
|
||
|
||
async def _full_cleanup_on_startup(self) -> bool:
|
||
"""启动时全量清场：3 次重试 _reset_to_idle。
|
||
|
||
上次崩溃可能残留脏状态（搜索框打开/输入框有内容），
|
||
启动时调用此方法确保干净起点。
|
||
|
||
Returns:
|
||
True 表示清场成功，False 表示 3 次都失败（可能需人工 VNC 接入）
|
||
"""
|
||
for attempt in range(3):
|
||
logger.info(
|
||
"[flow] startup cleanup attempt %d/3", attempt + 1
|
||
)
|
||
try:
|
||
# 启动时无 win_geom，用 None（跳过 post_verify）
|
||
ok = await self._reset_to_idle(win_geom=None)
|
||
if ok:
|
||
logger.info(
|
||
"[flow] startup cleanup OK at attempt %d", attempt + 1
|
||
)
|
||
return True
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"[flow] startup cleanup attempt %d exception: %s",
|
||
attempt + 1, exc,
|
||
)
|
||
await asyncio.sleep(1.0)
|
||
logger.warning("[flow] startup cleanup failed 3 attempts")
|
||
return False
|
||
|
||
# ------------------------------------------------------------------
|
||
# 子类实现
|
||
# ------------------------------------------------------------------
|
||
def _ensure_valid_geometry(self, geom: WindowGeometry) -> WindowGeometry:
|
||
"""校验窗口几何有效（尺寸 >= 200x200），无效则抛 PermanentError。
|
||
|
||
微信启动过程中可能出现隐藏/加载小窗（如 1x1），虽然 backend 的
|
||
_find_main_window_id 已做 min_area 过滤，这里再做一次防御性校验，
|
||
避免后续 click_element 用错误几何计算坐标导致点到窗口外。
|
||
|
||
Args:
|
||
geom: 待校验的窗口几何
|
||
|
||
Returns:
|
||
原始 geom（校验通过）
|
||
|
||
Raises:
|
||
PermanentError: geom 无效（width/height < 200）
|
||
"""
|
||
if geom.width < 200 or geom.height < 200:
|
||
raise PermanentError(
|
||
code="WINDOW_NOT_FOUND",
|
||
message=f"窗口几何无效: {geom.width}x{geom.height}（需 >=200x200）",
|
||
)
|
||
return geom
|
||
|
||
async def run(self, ctx: FlowContext) -> FlowResult:
|
||
"""子类实现：编排各 transition，返回 FlowResult。"""
|
||
raise NotImplementedError("subclass must implement run()")
|