diff --git a/app/sender/wechat_native.py b/app/sender/wechat_native.py index ff4ee98..25cf76f 100644 --- a/app/sender/wechat_native.py +++ b/app/sender/wechat_native.py @@ -72,6 +72,12 @@ class NativeActionResult: diagnostics: dict[str, object] = field(default_factory=dict) +class _SendLayoutTransition(RuntimeError): + def __init__(self, boxes): + super().__init__("微信输入区与消息列表正在过渡，尚不可截图") + self.boxes = boxes + + def _now() -> str: return datetime.now().isoformat() @@ -1309,33 +1315,49 @@ def _read_send_region_boxes(self) -> tuple[tuple[int, int, int, int], tuple[int, raise RuntimeError(f"无法唯一定位微信 {automation_id}，已停止发送") boxes.append(matches[0]) composer, chat = boxes - if chat[3] > composer[1] or min(chat[2], composer[2]) <= max(chat[0], composer[0]): + if min(chat[2], composer[2]) <= max(chat[0], composer[0]): raise RuntimeError("微信输入区与消息列表位置异常，已停止发送") + if chat[3] > composer[1]: + raise _SendLayoutTransition((composer, chat)) return composer, chat def _capture_send_regions(self): from PIL import ImageGrab - boxes = self._read_send_region_boxes() - window_box = self._window_rect(self._window) - if self._send_region_boxes is not None: + deadline = time.monotonic() + self.stage_timeout + for attempt in range(self._poll_rounds(self.stage_timeout)): + overlapping = False + try: + boxes = self._read_send_region_boxes() + except _SendLayoutTransition as exc: + boxes, overlapping = exc.boxes, True + window_box = self._window_rect(self._window) + if self._send_region_boxes is None: + if overlapping: + raise RuntimeError("微信初始输入区布局尚未稳定，已停止发送") + self._send_region_boxes = boxes + self._send_window_box = window_box + break old_composer, old_chat = self._send_region_boxes composer_box, chat_box = boxes - # 粘贴长文字或图片会向上展开编辑区并同步缩短消息列表。 - # 只允许两者的公共分界线变化；移动、缩放、换列、标题变化仍停止。 - vertical_expansion = ( + # 图片编辑区与消息列表的动画可能先后更新；只等待固定窗口内 + # 公共分界线的过渡，绝不忽略移动、缩放、换列或标题区域变化。 + anchored = ( window_box == self._send_window_box and (composer_box[0], composer_box[2], composer_box[3]) == (old_composer[0], old_composer[2], old_composer[3]) and chat_box[:3] == old_chat[:3] and composer_box[1] <= old_composer[1] - and abs((composer_box[1] - chat_box[3]) - (old_composer[1] - old_chat[3])) <= 2 + and chat_box[3] <= old_chat[3] ) - if not vertical_expansion: + if not anchored: raise RuntimeError("发送核验期间微信布局或窗口位置已变化，无法确认提交结果") - else: - self._send_region_boxes = boxes - self._send_window_box = window_box + gap_delta = (composer_box[1] - chat_box[3]) - (old_composer[1] - old_chat[3]) + if not overlapping and abs(gap_delta) <= 2: + break + if attempt + 1 >= self._poll_rounds(self.stage_timeout) or time.monotonic() >= deadline: + raise RuntimeError("微信图片编辑区展开/收起未完成，无法确认提交结果") + time.sleep(self.poll_interval) composer = ImageGrab.grab(bbox=boxes[0], all_screens=True).convert("RGB") chat = ImageGrab.grab(bbox=boxes[1], all_screens=True).convert("RGB") return composer, chat diff --git a/tests/test_v2_wechat_native.py b/tests/test_v2_wechat_native.py index 63367b4..3f9ad28 100644 --- a/tests/test_v2_wechat_native.py +++ b/tests/test_v2_wechat_native.py @@ -19,6 +19,7 @@ _select_group_search_match, _select_uia_group_search_match, _title_matches, + _SendLayoutTransition, ) import pytest @@ -1200,11 +1201,90 @@ def test_capture_accepts_editor_expansion_and_collapse_with_stable_window(tmp_pa def test_unrelated_layout_change_is_not_editor_expansion(tmp_path, monkeypatch, boxes): driver = WindowsWechatDriver(_settings(tmp_path)) driver._window = 123 + monkeypatch.setattr("app.sender.wechat_native.time.sleep", lambda *args: None) monkeypatch.setattr(driver, "_window_rect", lambda hwnd: (0, 0, 1100, 950)) baseline = ((350, 710, 1090, 880), (350, 140, 1100, 700)) monkeypatch.setattr(driver, "_read_send_region_boxes", lambda: baseline) monkeypatch.setattr("PIL.ImageGrab.grab", lambda **kwargs: Image.new("RGB", (200, 80), "white")) driver._capture_send_regions() monkeypatch.setattr(driver, "_read_send_region_boxes", lambda: boxes) - with pytest.raises(RuntimeError, match="布局或窗口位置已变化"): + with pytest.raises(RuntimeError, match="布局或窗口位置已变化|展开/收起未完成"): driver._capture_send_regions() + + +@pytest.mark.parametrize("editor_first", [False, True]) +def test_capture_waits_for_both_sides_of_expansion_animation(tmp_path, monkeypatch, editor_first): + driver = WindowsWechatDriver(_settings(tmp_path)) + driver._window = 123 + monkeypatch.setattr(driver, "_window_rect", lambda hwnd: (0, 0, 1100, 950)) + monkeypatch.setattr("app.sender.wechat_native.time.sleep", lambda *args: None) + baseline = ((350, 710, 1090, 880), (350, 140, 1100, 700)) + expanded = ((350, 510, 1090, 880), (350, 140, 1100, 500)) + partial = ((350, 510, 1090, 880), (350, 140, 1100, 650)) if editor_first else ((350, 650, 1090, 880), (350, 140, 1100, 500)) + pending = iter([baseline, partial, expanded]) + reads, captured = [], [] + def boxes(): + value = next(pending) + reads.append(value) + if value == partial and editor_first: + raise _SendLayoutTransition(value) + return value + def grab(*, bbox, all_screens): + captured.append(bbox) + return Image.new("RGB", (bbox[2]-bbox[0], bbox[3]-bbox[1]), "white") + monkeypatch.setattr(driver, "_read_send_region_boxes", boxes) + monkeypatch.setattr("PIL.ImageGrab.grab", grab) + driver._capture_send_regions() + driver._capture_send_regions() + assert reads == [baseline, partial, expanded] + assert captured == [*baseline, *expanded] + + +def test_expansion_that_never_finishes_is_not_captured(tmp_path, monkeypatch): + driver = WindowsWechatDriver(_settings(tmp_path)) + driver._window = 123 + driver._send_window_box = (0, 0, 1100, 950) + driver._send_region_boxes = ((350, 710, 1090, 880), (350, 140, 1100, 700)) + monkeypatch.setattr(driver, "_window_rect", lambda hwnd: driver._send_window_box) + monkeypatch.setattr(driver, "_read_send_region_boxes", lambda: ((350, 650, 1090, 880), (350, 140, 1100, 500))) + monkeypatch.setattr("app.sender.wechat_native.time.sleep", lambda *args: None) + monkeypatch.setattr("PIL.ImageGrab.grab", lambda **kwargs: pytest.fail("unstable layout must not be captured")) + with pytest.raises(RuntimeError, match="展开/收起未完成"): + driver._capture_send_regions() + + +def test_image_waits_for_expansion_then_verifies_collapse_after_enter(tmp_path, monkeypatch): + driver = WindowsWechatDriver(_settings(tmp_path)) + driver._window = 123 + monkeypatch.setattr(driver, "_window_rect", lambda hwnd: (0, 0, 1100, 950)) + monkeypatch.setattr("app.sender.wechat_native.time.sleep", lambda *args: None) + monkeypatch.setattr(driver, "_focus_composer", lambda: None) + monkeypatch.setattr(driver, "_composer_is_empty", lambda: (True, "empty")) + monkeypatch.setattr(driver, "_set_clipboard_image", lambda *args: None) + state = {"phase": "empty", "partial_seen": False} + normal = ((350, 710, 1090, 880), (350, 140, 1100, 700)) + partial = ((350, 650, 1090, 880), (350, 140, 1100, 500)) + expanded = ((350, 510, 1090, 880), (350, 140, 1100, 500)) + def boxes(): + if state["phase"] == "animating": + state["partial_seen"] = True + state["phase"] = "staged" + return partial + return expanded if state["phase"] == "staged" else normal + def grab(*, bbox, all_screens): + image = Image.new("RGB", (bbox[2]-bbox[0], bbox[3]-bbox[1]), (250, 250, 250)) + if bbox == expanded[0] and state["phase"] == "staged": + ImageDraw.Draw(image).rectangle((20, 10, 200, 310), fill="blue") + if bbox == normal[1] and state["phase"] == "sent": + ImageDraw.Draw(image).rectangle((20, 400, 200, 550), fill="blue") + return image + def enter(name): + assert name == "enter" and state["phase"] == "staged" and state["partial_seen"] + state["phase"] = "sent" + monkeypatch.setattr(driver, "_read_send_region_boxes", boxes) + monkeypatch.setattr("PIL.ImageGrab.grab", grab) + monkeypatch.setattr(driver, "_hotkey", lambda *args: state.update(phase="animating")) + monkeypatch.setattr(driver, "_key", enter) + result = driver.paste_image(tmp_path / "image.png") + assert result.success and result.submitted and not result.outcome_unknown + assert result.verification_level == "ui_observed"