xhs-note-crawling/workplace/input-keyword/input-keyword.py

#!/usr/bin/env python3
"""关键词搜索流程：先读 ``save/project-config``；若有已保存坐标则跳过对应 OCR/模板匹配，否则现算并写回。
若输入框已有文案则 ``clear-input.start``；再 ``find_search_input.start``（未命中抛错）；
未清空时 ``move-mouse-to-pos.start`` 到占位文案；再 ``input-word-by-word.start`` → ``press-search-btn.start`` → ``then``。
新算出的坐标会合并写入配置（见 ``workplace/config-save-read.py``）。"""

from __future__ import annotations

import importlib.util
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Any

_ROOT = Path(__file__).resolve().parent.parent.parent
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))

from workplace import pyautogui as workplace_pyautogui  # noqa: E402
from workplace.screenshot import PillowImageGrabFullScreenScreenshotCaptureSaver  # noqa: E402


def _load_find_search_input_module():
    path = Path(__file__).resolve().parent / "find_search_input.py"
    spec = importlib.util.spec_from_file_location("workplace_find_search_input", path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _load_move_mouse_to_pos_module():
    path = Path(__file__).resolve().parent.parent / "move-mouse-to-pos.py"
    spec = importlib.util.spec_from_file_location("workplace_move_mouse_to_pos", path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _load_clear_input_module():
    path = Path(__file__).resolve().parent / "clear-input.py"
    spec = importlib.util.spec_from_file_location("workplace_clear_input", path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _load_input_word_by_word_module():
    path = Path(__file__).resolve().parent / "input-word-by-word.py"
    spec = importlib.util.spec_from_file_location("workplace_input_word_by_word", path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _load_press_search_btn_module():
    path = Path(__file__).resolve().parent / "press-search-btn.py"
    spec = importlib.util.spec_from_file_location("workplace_press_search_btn", path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _load_config_save_read_module():
    path = Path(__file__).resolve().parent.parent / "config-save-read.py"
    spec = importlib.util.spec_from_file_location("workplace_config_save_read", path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load {path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _wait_pointer_at(
    target_screen_x: int,
    target_screen_y: int,
    *,
    tolerance_pixels: int = 3,
) -> None:
    while True:
        current_pointer_x, current_pointer_y = workplace_pyautogui.raw.position()
        if (
            abs(int(current_pointer_x) - target_screen_x) <= tolerance_pixels
            and abs(int(current_pointer_y) - target_screen_y) <= tolerance_pixels
        ):
            return
        workplace_pyautogui.sleep_human_pointer_poll_step()


def _pointer_settled_callback(
    target_screen_x: int,
    target_screen_y: int,
    on_settled: Callable[[tuple[int, int]], Any],
) -> Any:
    """等指针落到目标附近后 ``on_settled((x, y))``，类似 Node 里异步完成后的 callback。"""
    _wait_pointer_at(target_screen_x, target_screen_y)
    workplace_pyautogui.sleep_human_after_pointer_settled_before_typing()
    return on_settled((target_screen_x, target_screen_y))


def init(
    page,
    keyword: str = "",
    *,
    project_config: dict[str, Any] | None = None,
    then: Callable[[tuple[int, int]], Any] | None = None,
    poll_interval_sec: float = 0.35,
) -> Any:
    """
    步骤概览：若已有文案则先清空；再 ``find_search_input`` 模块 ``start`` 一次；未命中则抛错。
    若未清空则移鼠到「搜索小红书」；若刚清空则指针留在清除钮附近；再输入 → 点搜索 → ``then``。
    """
    find_search_input = _load_find_search_input_module()
    move_mouse_to_pos = _load_move_mouse_to_pos_module()
    clear_input = _load_clear_input_module()
    input_word_by_word = _load_input_word_by_word_module()
    press_search_btn = _load_press_search_btn_module()
    config_save_read = _load_config_save_read_module()
    if project_config is None:
        from workplace.singleton import REPOSITORY_CONFIG_INI_SNAPSHOT

        project_config = REPOSITORY_CONFIG_INI_SNAPSHOT.project_config
    cfg = project_config
    clear_xy_preset = config_save_read.input_keyword_xy_pair_from_config(
        cfg,
        "clear_x_button_xy",
    )
    search_xy_preset = config_save_read.input_keyword_xy_pair_from_config(
        cfg,
        "ocr_search_xiaohongshu_xy",
    )
    full_screen_screenshot_capture_saver = PillowImageGrabFullScreenScreenshotCaptureSaver()

    # 步骤 1：若有非空文案则先清空；配置里有坐标则跳过 OCR
    clear_x_button_xy = clear_input.start(
        page,
        selector_search_input=find_search_input.selector_search_input,
        full_screen_screenshot_capture_saver=full_screen_screenshot_capture_saver,
        move_mouse_to_pos=move_mouse_to_pos,
        poll_interval_sec=poll_interval_sec,
        preset_clear_button_xy=clear_xy_preset,
    )
    cleared_search_input_this_iteration = clear_x_button_xy is not None
    if clear_x_button_xy is not None and clear_xy_preset is None:
        config_save_read.merge_input_keyword_config(clear_x_button_xy=clear_x_button_xy)

    # 步骤 2：「搜索小红书」中心；配置里有则跳过 OCR
    search_input_flow_result = find_search_input.start(
        page,
        keyword=keyword,
        preset_center_xy=search_xy_preset,
    )
    if search_input_flow_result.center_xy is None:
        raise RuntimeError(
            (search_input_flow_result.message or "").strip()
            or "OCR 未在全屏截图中找到「搜索小红书」。",
        )

    if search_xy_preset is None:
        config_save_read.merge_input_keyword_config(
            ocr_search_xiaohongshu_xy=search_input_flow_result.center_xy,
        )

    ocr_text_center_screen_x, ocr_text_center_screen_y = (
        search_input_flow_result.center_xy
    )
    if not cleared_search_input_this_iteration:
        move_mouse_to_pos.start(
            ocr_text_center_screen_x,
            ocr_text_center_screen_y,
        )
        pointer_settle_screen_x = ocr_text_center_screen_x
        pointer_settle_screen_y = ocr_text_center_screen_y
    else:
        cur_x, cur_y = workplace_pyautogui.raw.position()
        pointer_settle_screen_x = int(cur_x)
        pointer_settle_screen_y = int(cur_y)

    def on_pointer_settled(pointer_final_screen_xy: tuple[int, int]) -> Any:
        # 步骤 3：指针已在 OCR 目标点附近；聚焦用 PyAutoGUI 当前位置左键，见 input-word-by-word.type_keyword_human
        # 步骤 4：input-word-by-word.start → click_here + 逐字 insert_text + 校验
        input_word_by_word.start(page, keyword=keyword)
        # 步骤 5：press-search-btn.start（读配置坐标或 LoFTR 匹配后移鼠并点击顶栏搜索）
        _press_ok, _press_msg, press_search_button_xy = press_search_btn.start(
            page,
            project_config=project_config,
        )
        # 步骤 6：把最终屏幕坐标交给上层 ``then``；无 ``then`` 时返回占位码与 OCR 结果对象
        if then is not None:
            return then(pointer_final_screen_xy)
        return 0, search_input_flow_result

    return _pointer_settled_callback(
        pointer_settle_screen_x,
        pointer_settle_screen_y,
        on_pointer_settled,
    )


start = init

__all__ = ["init", "start"]