305 lines
11 KiB
Python
305 lines
11 KiB
Python
"""UI 状态机
|
||
|
||
基于微信窗口数量和名称判断当前 UI 状态，提供状态恢复能力。
|
||
|
||
微信 v4.1.9 窗口模型：
|
||
- 1 个窗口（"微信"）：主聊天列表
|
||
- 2 个窗口（"微信" + 聊天名）：会话已打开
|
||
- 3+ 个窗口：可能有图片预览等额外窗口
|
||
"""
|
||
|
||
import logging
|
||
import time
|
||
from enum import Enum
|
||
|
||
from Cocoa import NSRunningApplication
|
||
|
||
from .ax_bridge import AXBridge
|
||
from .wechat_ui import WeChatUI
|
||
|
||
logger = logging.getLogger("wechat_clicker.state_machine")
|
||
|
||
|
||
class UIState(Enum):
|
||
UNKNOWN = "unknown"
|
||
MAIN_CHAT_LIST = "main_chat_list"
|
||
CONVERSATION_OPEN = "conversation_open"
|
||
MEDIA_PREVIEW = "media_preview"
|
||
WECHAT_NOT_RUNNING = "wechat_not_running"
|
||
WECHAT_MINIMIZED = "wechat_minimized"
|
||
WECHAT_ABNORMAL = "wechat_abnormal"
|
||
|
||
|
||
class StateMachine:
|
||
"""微信 UI 状态机"""
|
||
|
||
def __init__(self, ax: AXBridge, wechat_ui: WeChatUI):
|
||
self.ax = ax
|
||
self.ui = wechat_ui
|
||
self._current_state = UIState.UNKNOWN
|
||
self._conversation_name = None # 当前打开的会话名称
|
||
|
||
@property
|
||
def current_state(self) -> UIState:
|
||
return self._current_state
|
||
|
||
@property
|
||
def conversation_name(self) -> str:
|
||
return self._conversation_name or ""
|
||
|
||
# ----------------------------------------------------------------
|
||
# 状态检测
|
||
# ----------------------------------------------------------------
|
||
|
||
def detect_state(self) -> UIState:
|
||
"""检测当前微信 UI 状态。"""
|
||
windows = self.ui.get_all_windows()
|
||
|
||
if not windows:
|
||
if self.ax.is_app_running(self.ui.bundle_id):
|
||
self._current_state = UIState.WECHAT_MINIMIZED
|
||
self._conversation_name = None
|
||
logger.debug("微信在运行但无可见窗口（可能已最小化或隐藏）")
|
||
else:
|
||
self._current_state = UIState.WECHAT_NOT_RUNNING
|
||
self._conversation_name = None
|
||
logger.debug("微信未运行")
|
||
return self._current_state
|
||
|
||
# 检查主窗口是否最小化（AXWindows 会包含最小化的窗口）
|
||
main_window = None
|
||
main_window_area = 0
|
||
all_minimized = True
|
||
other_windows = []
|
||
for win in windows:
|
||
title = self.ax.get_title(win)
|
||
is_mini = self.ax.is_window_minimized(win)
|
||
if title == "微信":
|
||
size = self.ax.get_size(win)
|
||
area = size[0] * size[1]
|
||
if area > main_window_area:
|
||
if main_window is not None:
|
||
other_windows.append((main_window, "微信"))
|
||
main_window = win
|
||
main_window_area = area
|
||
else:
|
||
other_windows.append((win, title))
|
||
elif title:
|
||
other_windows.append((win, title))
|
||
if not is_mini:
|
||
all_minimized = False
|
||
|
||
if all_minimized:
|
||
self._current_state = UIState.WECHAT_MINIMIZED
|
||
self._conversation_name = None
|
||
logger.debug("微信所有窗口均已最小化")
|
||
return self._current_state
|
||
|
||
if main_window is None:
|
||
# 有窗口但标题都不是"微信"——可能是登录/更新界面、窗口正在加载
|
||
window_titles = [self.ax.get_title(w) or "(空标题)" for w in windows]
|
||
self._current_state = UIState.WECHAT_ABNORMAL
|
||
logger.warning(
|
||
f"微信窗口异常: {len(windows)} 个窗口但无'微信'主窗口, "
|
||
f"标题={window_titles}"
|
||
)
|
||
return self._current_state
|
||
|
||
window_count = len(windows)
|
||
|
||
if window_count == 1:
|
||
self._current_state = UIState.MAIN_CHAT_LIST
|
||
self._conversation_name = None
|
||
elif window_count == 2 and len(other_windows) == 1:
|
||
self._current_state = UIState.CONVERSATION_OPEN
|
||
self._conversation_name = other_windows[0][1]
|
||
elif window_count >= 3:
|
||
self._current_state = UIState.MEDIA_PREVIEW
|
||
if other_windows:
|
||
self._conversation_name = other_windows[0][1]
|
||
else:
|
||
self._current_state = UIState.UNKNOWN
|
||
|
||
other_titles = [t for _, t in other_windows]
|
||
logger.debug(
|
||
f"状态检测: {self._current_state.value}, "
|
||
f"窗口数={window_count}, 会话={self._conversation_name}, "
|
||
f"其他窗口={other_titles}"
|
||
)
|
||
return self._current_state
|
||
|
||
# ----------------------------------------------------------------
|
||
# 状态恢复
|
||
# ----------------------------------------------------------------
|
||
|
||
def recover_to_chat_list(self, max_retries: int = 3) -> bool:
|
||
"""恢复到主聊天列表状态。关闭所有会话和预览窗口。
|
||
|
||
Returns:
|
||
True 如果成功恢复到聊天列表。
|
||
"""
|
||
for attempt in range(max_retries):
|
||
state = self.detect_state()
|
||
|
||
if state == UIState.MAIN_CHAT_LIST:
|
||
return True
|
||
|
||
if state == UIState.WECHAT_NOT_RUNNING:
|
||
logger.error("微信未运行，无法恢复")
|
||
return False
|
||
|
||
if state == UIState.WECHAT_MINIMIZED:
|
||
logger.info("微信窗口已最小化，正在恢复...")
|
||
if self.ui.ensure_wechat_visible():
|
||
time.sleep(0.5)
|
||
continue
|
||
else:
|
||
logger.error("微信窗口恢复失败")
|
||
return False
|
||
|
||
if state == UIState.WECHAT_ABNORMAL:
|
||
# 有窗口但无"微信"主窗口——可能有遗留预览窗口挡住了
|
||
# 先尝试关闭非主窗口，再等待
|
||
logger.warning("微信窗口异常，尝试关闭多余窗口...")
|
||
self._close_non_main_windows()
|
||
self.ui.ensure_wechat_frontmost()
|
||
time.sleep(2.0)
|
||
continue
|
||
|
||
if state == UIState.MEDIA_PREVIEW:
|
||
# 先关闭预览（Escape）
|
||
logger.info("关闭媒体预览...")
|
||
self.ax.send_escape_key()
|
||
time.sleep(0.5)
|
||
|
||
if state in (UIState.CONVERSATION_OPEN, UIState.MEDIA_PREVIEW):
|
||
# 关闭会话窗口
|
||
self._close_all_conversations()
|
||
time.sleep(0.5)
|
||
|
||
if state == UIState.UNKNOWN:
|
||
# 可能有遗留的预览窗口（空标题），先关闭非主窗口
|
||
self._close_non_main_windows()
|
||
# Escape 关闭可能的弹窗（安全操作）
|
||
self.ax.send_escape_key()
|
||
time.sleep(0.3)
|
||
# 重新激活微信并点击侧边栏聊天按钮
|
||
self.ui.ensure_wechat_frontmost()
|
||
time.sleep(0.5)
|
||
self.ui.click_sidebar_chat_tab()
|
||
time.sleep(0.5)
|
||
|
||
logger.debug(f"恢复尝试 {attempt + 1}/{max_retries}")
|
||
|
||
# 最终检查
|
||
final_state = self.detect_state()
|
||
if final_state == UIState.MAIN_CHAT_LIST:
|
||
return True
|
||
|
||
logger.error(f"恢复失败，当前状态: {final_state.value}")
|
||
return False
|
||
|
||
def _close_all_conversations(self):
|
||
"""关闭所有会话窗口。"""
|
||
conv_windows = self.ui.get_conversation_windows()
|
||
for win in conv_windows:
|
||
title = self.ax.get_title(win)
|
||
close_btn = self.ui.get_close_button(win)
|
||
if close_btn:
|
||
logger.debug(f"关闭会话窗口: {title}")
|
||
self.ax.press(close_btn)
|
||
time.sleep(0.3)
|
||
else:
|
||
# 后备方案：Cmd+W
|
||
logger.debug(f"使用 Cmd+W 关闭窗口: {title}")
|
||
self.ax.send_cmd_w()
|
||
time.sleep(0.3)
|
||
|
||
def _close_non_main_windows(self):
|
||
"""关闭所有非主窗口（含空标题的遗留预览窗口和小尺寸浮窗）。"""
|
||
main_win = self.ui.get_main_window()
|
||
windows = self.ui.get_all_windows()
|
||
for win in windows:
|
||
if main_win is not None and win == main_win:
|
||
continue
|
||
title = self.ax.get_title(win) or "(空标题)"
|
||
close_btn = self.ui.get_close_button(win)
|
||
if close_btn:
|
||
logger.debug(f"关闭非主窗口: '{title}'")
|
||
self.ax.press(close_btn)
|
||
time.sleep(0.3)
|
||
else:
|
||
logger.debug(f"使用 Cmd+W 关闭非主窗口: '{title}'")
|
||
self.ui.ensure_wechat_frontmost()
|
||
time.sleep(0.2)
|
||
self.ax.send_cmd_w()
|
||
time.sleep(0.3)
|
||
|
||
def close_current_conversation(self) -> bool:
|
||
"""关闭当前打开的会话窗口。"""
|
||
conv_window = self.ui.get_conversation_window()
|
||
if conv_window is None:
|
||
return True # 没有会话窗口，视为成功
|
||
|
||
close_btn = self.ui.get_close_button(conv_window)
|
||
if close_btn:
|
||
self.ax.press(close_btn)
|
||
else:
|
||
self.ax.send_cmd_w()
|
||
|
||
time.sleep(0.5)
|
||
|
||
# 验证
|
||
state = self.detect_state()
|
||
return state == UIState.MAIN_CHAT_LIST
|
||
|
||
def dismiss_preview(self) -> bool:
|
||
"""关闭媒体预览。"""
|
||
state = self.detect_state()
|
||
if state != UIState.MEDIA_PREVIEW:
|
||
return True
|
||
|
||
self.ax.send_escape_key()
|
||
time.sleep(0.5)
|
||
|
||
state = self.detect_state()
|
||
if state == UIState.MEDIA_PREVIEW:
|
||
self.ax.send_escape_key()
|
||
time.sleep(0.5)
|
||
state = self.detect_state()
|
||
|
||
return state != UIState.MEDIA_PREVIEW
|
||
|
||
# ----------------------------------------------------------------
|
||
# Preview.app 处理
|
||
# ----------------------------------------------------------------
|
||
|
||
def close_preview_app(self) -> bool:
|
||
"""关闭 macOS Preview.app（预览）窗口。"""
|
||
preview_bundle = "com.apple.Preview"
|
||
apps = NSRunningApplication.runningApplicationsWithBundleIdentifier_(preview_bundle)
|
||
if not apps or len(apps) == 0:
|
||
return True
|
||
|
||
app = apps[0]
|
||
if app.isTerminated():
|
||
return True
|
||
|
||
logger.debug("检测到 Preview.app 正在运行，发送 Cmd+W 关闭窗口")
|
||
app.activateWithOptions_(0)
|
||
time.sleep(0.5)
|
||
self.ax.send_cmd_w()
|
||
time.sleep(0.5)
|
||
|
||
self.ui.ensure_wechat_frontmost()
|
||
time.sleep(0.3)
|
||
return True
|
||
|
||
def is_preview_app_running(self) -> bool:
|
||
"""检查 Preview.app 是否在运行。"""
|
||
preview_bundle = "com.apple.Preview"
|
||
apps = NSRunningApplication.runningApplicationsWithBundleIdentifier_(preview_bundle)
|
||
if not apps or len(apps) == 0:
|
||
return False
|
||
return not apps[0].isTerminated()
|