coordinate/utils/keyboard.py

# from DrissionPage import ChromiumPage
# # 假设上述代码存为 human_keyboard.py
# from human_keyboard import HumanKeyboard, Key

# page = ChromiumPage()
# page.get('https://www.baidu.com')

# # 初始化高度拟人化键盘
# keyboard = HumanKeyboard(page)

# # 1. 点击搜索框以获得输入焦点
# ele = page.ele('#kw')
# ele.click()

# # 2. 以极度拟人化的方式输入文本（有概率打错字又自己倒退回删并重新打）
# # 因为开了 humanize，你会看到打字速度变化、可能突然思考停顿等效果。
# keyboard.type_text("Hello World, let's test humanized typing!!!", humanize=True)

# # 3. 按下回车键
# keyboard.press(Key.ENTER)

# # 4. 执行快捷键 (如 Ctrl + A 全选)
# keyboard.hotkey(Key.CONTROL, Key.A)

import logging
import random
import time
import warnings
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Tuple, List

# 引入 DrissionPage 的基础类以进行类型提示
from DrissionPage import ChromiumPage

logger = logging.getLogger(__name__)

# ==========================================
# 常量、枚举与键盘布局配置
# ==========================================

DEFAULT_TYPO_PROBABILITY = 0.02

# 模拟错别字时需要的 QWERTY 键盘相邻按键映射表
QWERTY_NEIGHBORS = {
    'q': ['w', 'a', 's'], 'w': ['q', 'e', 'a', 's', 'd'], 'e': ['w', 'r', 's', 'd', 'f'],
    'r': ['e', 't', 'd', 'f', 'g'], 't': ['r', 'y', 'f', 'g', 'h'], 'y': ['t', 'u', 'g', 'h', 'j'],
    'u': ['y', 'i', 'h', 'j', 'k'], 'i': ['u', 'o', 'j', 'k', 'l'], 'o': ['i', 'p', 'k', 'l'],
    'p': ['o', 'l'], 'a': ['q', 'w', 's', 'z', 'x'], 's': ['q', 'w', 'e', 'a', 'd', 'z', 'x', 'c'],
    'd': ['w', 'e', 'r', 's', 'f', 'x', 'c', 'v'], 'f': ['e', 'r', 't', 'd', 'g', 'c', 'v', 'b'],
    'g': ['r', 't', 'y', 'f', 'h', 'v', 'b', 'n'], 'h': ['t', 'y', 'u', 'g', 'j', 'b', 'n', 'm'],
    'j': ['y', 'u', 'i', 'h', 'k', 'n', 'm'], 'k': ['u', 'i', 'o', 'j', 'l', 'm'],
    'l': ['i', 'o', 'p', 'k'], 'z': ['a', 's', 'x'], 'x': ['a', 's', 'd', 'z', 'c'],
    'c': ['s', 'd', 'f', 'x', 'v'], 'v': ['d', 'f', 'g', 'c', 'b'],
    'b': ['f', 'g', 'h', 'v', 'n'], 'n': ['g', 'h', 'j', 'b', 'm'],
    'm': ['h', 'j', 'k', 'n']
}


class TypoType(Enum):
    """错字类型"""
    ADJACENT = "adjacent"          # 按到相邻键
    TRANSPOSE = "transpose"        # 字母顺序颠倒
    DOUBLE = "double"              # 重复按键
    SKIP = "skip"                  # 漏按
    MISSED_SPACE = "missed_space"  # 漏按空格


class Key(Enum):
    """
    底层 CDP 需要的特殊按键定义。
    格式: (key_name, windows_virtual_key_code)
    """
    BACKSPACE = ("Backspace", 8)
    TAB = ("Tab", 9)
    ENTER = ("Enter", 13)
    SHIFT = ("Shift", 16)
    CONTROL = ("Control", 17)
    ALT = ("Alt", 18)
    ESCAPE = ("Escape", 27)
    SPACE = (" ", 32)
    META = ("Meta", 91)  # Win / Cmd 键


@dataclass(frozen=True)
class TypoResult:
    """错字生成结果"""
    typo_type: TypoType
    wrong_char: str = ''


@dataclass(frozen=True)
class TimingConfig:
    """高仿真打字节奏时间配置 (单位：秒)"""
    keystroke_min: float = 0.03
    keystroke_max: float = 0.12
    punctuation_min: float = 0.08
    punctuation_max: float = 0.18
    thinking_probability: float = 0.02
    thinking_min: float = 0.3
    thinking_max: float = 0.7
    distraction_probability: float = 0.005
    distraction_min: float = 0.5
    distraction_max: float = 1.2
    mistake_realize_min: float = 0.1
    mistake_realize_max: float = 0.25
    after_correction_min: float = 0.03
    after_correction_max: float = 0.08
    double_press_min: float = 0.02
    double_press_max: float = 0.05
    hesitation_min: float = 0.15
    hesitation_max: float = 0.3


@dataclass(frozen=True)
class TypoConfig:
    """各类错字发生的权重比例"""
    adjacent_weight: float = 0.55
    transpose_weight: float = 0.20
    double_weight: float = 0.12
    skip_weight: float = 0.08
    missed_space_weight: float = 0.05


# ==========================================
# 核心控制类
# ==========================================

class HumanKeyboard:
    """
    DrissionPage 的高仿真人类键盘控制器。
    提供：
    - 公共按键操作（press, down, up, hotkey）
    - 私有文本输入，包含高度拟人的错别字产生、自我纠正、思考停顿与节奏变化。
    """

    PAUSE_CHARS = frozenset(' .,!?;:\n')

    def __init__(
        self,
        page: ChromiumPage,
        timing: Optional[TimingConfig] = None,
        typo_config: Optional[TypoConfig] = None,
    ):
        """
        初始化键盘控制器。

        :param page: DrissionPage 的 ChromiumPage 或 ChromiumTab 实例
        :param timing: 节奏时间配置
        :param typo_config: 错别字权重配置
        """
        self._page = page
        self._timing = timing or TimingConfig()
        self._typo_config = typo_config or TypoConfig()

    def press(
        self,
        key: Key,
        modifiers: Optional[int] = None,
        interval: float = 0.1,
    ):
        """
        按下一个键并释放。
        
        :param key: 要按的键 (来自 Key enum)
        :param modifiers: 组合键掩码 (Alt=1, Ctrl=2, Meta=4, Shift=8)
        :param interval: 按下和释放之间的延迟
        """
        logger.debug(f'Pressing key: {key.name} with modifiers: {modifiers}')
        self.down(key, modifiers)
        time.sleep(interval)
        self.up(key)

    def down(self, key: Key, modifiers: Optional[int] = None):
        """按下按键（不释放）"""
        key_name, code = key.value
        logger.debug(f'Key down: {key_name}')
        
        params = {
            "type": "keyDown",
            "key": key_name,
            "windowsVirtualKeyCode": code,
            "nativeVirtualKeyCode": code,
        }
        if modifiers is not None:
            params["modifiers"] = modifiers
            
        self._page.run_cdp("Input.dispatchKeyEvent", **params)

    def up(self, key: Key):
        """释放按键"""
        key_name, code = key.value
        logger.debug(f'Key up: {key_name}')
        
        self._page.run_cdp(
            "Input.dispatchKeyEvent",
            type="keyUp",
            key=key_name,
            windowsVirtualKeyCode=code,
            nativeVirtualKeyCode=code,
        )

    def hotkey(self, key1: Key, key2: Key, key3: Optional[Key] = None):
        """
        执行组合快捷键（最多支持3个键）。
        例如：keyboard.hotkey(Key.CONTROL, Key.C)
        """
        logger.debug(f'Hotkey: {key1.name} + {key2.name}' + (f' + {key3.name}' if key3 else ''))
        keys = [key1, key2]
        if key3 is not None:
            keys.append(key3)

        modifiers, non_modifiers = self._split_modifiers_and_keys(keys)
        modifier_value = self._calculate_modifier_value(modifiers)

        for key in non_modifiers:
            self.down(key, modifiers=modifier_value)
            time.sleep(0.05)

        time.sleep(0.1)

        for key in reversed(non_modifiers):
            self.up(key)
            time.sleep(0.05)

    def type_text(
        self,
        text: str,
        humanize: bool = False,
        interval: Optional[float] = None,
    ):
        """
        逐字输入文本。

        :param text: 要输入的文本
        :param humanize: 是否开启拟人化（变速、犯错后退格重打、思考停顿）
        :param interval: (已弃用) 字符输入间隔
        """
        if interval is not None:
            warnings.warn(
                '"interval" 参数已被弃用。请使用 "humanize=True" 开启真实的人类打字模拟。',
                DeprecationWarning,
                stacklevel=2,
            )

        if humanize:
            self._type_text_humanized(text)
            return

        for current_char in text:
            self._type_char(current_char)
            time.sleep(0.05)

    def _type_text_humanized(self, text: str):
        """高拟人化文本输入核心逻辑"""
        char_index = 0
        while char_index < len(text):
            current_char = text[char_index]
            next_char = text[char_index + 1] if char_index + 1 < len(text) else None

            # 判断是否打错字并执行错字修补逻辑，返回 True 表示下个字应跳过
            should_skip_next = self._process_char_with_typo(current_char, next_char)

            if should_skip_next:
                char_index += 1

            self._apply_realistic_delay(current_char)
            char_index += 1

    def _type_char(self, char: str):
        """通过底层 CDP 输入单个普通字符"""
        # 下拉字符事件
        self._page.run_cdp(
            "Input.dispatchKeyEvent",
            type="keyDown",
            key=char,
            text=char,
            unmodifiedText=char
        )
        # 抬起字符事件
        self._page.run_cdp(
            "Input.dispatchKeyEvent",
            type="keyUp",
            key=char
        )

    def _type_backspace(self):
        """输入退格键（用来模拟人打错字回删）"""
        self.press(Key.BACKSPACE)

    def _process_char_with_typo(
        self,
        current_char: str,
        next_char: Optional[str],
    ) -> bool:
        """处理错别字产生与修复。返回 True 表示下个字符需要跳过处理"""
        if not self._should_make_typo():
            self._type_char(current_char)
            return False

        typo = self._generate_typo(current_char, next_char)
        return self._handle_typo(current_char, next_char, typo)

    def _handle_typo(
        self,
        current_char: str,
        next_char: Optional[str],
        typo: TypoResult,
    ) -> bool:
        """根据错别字类型执行具体的打错和修正行为"""
        if typo.typo_type == TypoType.ADJACENT:
            self._do_adjacent_typo(current_char, typo.wrong_char)
            return False

        if typo.typo_type == TypoType.TRANSPOSE and next_char:
            self._do_transpose_typo(current_char, next_char)
            return True

        if typo.typo_type == TypoType.DOUBLE:
            self._do_double_typo(current_char)
            return False

        if typo.typo_type == TypoType.SKIP:
            self._do_skip_typo(current_char)
            return False

        if typo.typo_type == TypoType.MISSED_SPACE and current_char == ' ' and next_char:
            self._do_missed_space_typo(current_char, next_char)
            return True

        self._type_char(current_char)
        return False

    def _do_adjacent_typo(self, correct_char: str, wrong_char: str):
        """打错相邻键 -> 停顿 -> 退格删除 -> 补上正确的"""
        timing = self._timing
        self._type_char(wrong_char)
        time.sleep(random.uniform(timing.mistake_realize_min, timing.mistake_realize_max))
        self._type_backspace()
        time.sleep(random.uniform(timing.after_correction_min, timing.after_correction_max))
        self._type_char(correct_char)

    def _do_transpose_typo(self, current_char: str, next_char: str):
        """字母顺序打反 -> 停顿 -> 删两格 -> 重新打正确的顺序"""
        timing = self._timing
        self._type_char(next_char)
        time.sleep(random.uniform(timing.keystroke_min, timing.keystroke_max))
        self._type_char(current_char)

        time.sleep(random.uniform(timing.mistake_realize_min, timing.mistake_realize_max))
        self._type_backspace()
        self._type_backspace()
        time.sleep(random.uniform(timing.after_correction_min, timing.after_correction_max))

        self._type_char(current_char)
        time.sleep(random.uniform(timing.keystroke_min, timing.keystroke_max))
        self._type_char(next_char)

    def _do_double_typo(self, current_char: str):
        """不小心连按两次 -> 停顿 -> 退格删掉多余的一个"""
        timing = self._timing
        self._type_char(current_char)
        time.sleep(random.uniform(timing.double_press_min, timing.double_press_max))
        self._type_char(current_char)
        time.sleep(random.uniform(timing.mistake_realize_min, timing.mistake_realize_max))
        self._type_backspace()

    def _do_skip_typo(self, current_char: str):
        """大脑卡壳漏打：犹豫停顿 -> 接着打正常的"""
        timing = self._timing
        time.sleep(random.uniform(timing.hesitation_min, timing.hesitation_max))
        self._type_char(current_char)

    def _do_missed_space_typo(self, space_char: str, next_char: str):
        """漏敲空格 -> 接着敲了下一个字 -> 发现错误 -> 退格删掉 -> 补空格 -> 重打下个字"""
        timing = self._timing
        self._type_char(next_char)
        time.sleep(random.uniform(timing.mistake_realize_min, timing.mistake_realize_max))
        self._type_backspace()
        time.sleep(random.uniform(timing.after_correction_min, timing.after_correction_max))
        self._type_char(space_char)
        time.sleep(random.uniform(timing.after_correction_min, timing.after_correction_max))
        self._type_char(next_char)

    def _apply_realistic_delay(self, typed_char: str):
        """根据输入的不同字符和人的状态，加入随机延迟"""
        timing = self._timing
        delay = random.uniform(timing.keystroke_min, timing.keystroke_max)

        # 遇到标点符号，延迟加长
        if typed_char in self.PAUSE_CHARS:
            delay += random.uniform(timing.punctuation_min, timing.punctuation_max)

        # 模拟突然思考
        if random.random() < timing.thinking_probability:
            delay += random.uniform(timing.thinking_min, timing.thinking_max)

        # 模拟被打断分心
        if random.random() < timing.distraction_probability:
            delay += random.uniform(timing.distraction_min, timing.distraction_max)

        time.sleep(delay)

    @staticmethod
    def _should_make_typo() -> bool:
        """决定此时是否发生手误"""
        return random.random() < DEFAULT_TYPO_PROBABILITY

    def _generate_typo(self, current_char: str, next_char: Optional[str]) -> TypoResult:
        """生成一个具体的错误结果"""
        typo_type = self._select_typo_type()
        return self._create_typo(typo_type, current_char, next_char)

    def _select_typo_type(self) -> TypoType:
        """通过权重随机挑选错误类型"""
        config = self._typo_config
        typo_types = [
            TypoType.ADJACENT,
            TypoType.TRANSPOSE,
            TypoType.DOUBLE,
            TypoType.SKIP,
            TypoType.MISSED_SPACE,
        ]
        typo_weights = [
            config.adjacent_weight,
            config.transpose_weight,
            config.double_weight,
            config.skip_weight,
            config.missed_space_weight,
        ]
        return random.choices(typo_types, weights=typo_weights, k=1)[0]

    def _create_typo(
        self,
        typo_type: TypoType,
        current_char: str,
        next_char: Optional[str],
    ) -> TypoResult:
        """构建错字返回结果"""
        if typo_type == TypoType.ADJACENT:
            return self._create_adjacent_typo(current_char)
        elif typo_type == TypoType.TRANSPOSE:
            return self._create_transpose_typo(current_char, next_char)
        elif typo_type == TypoType.MISSED_SPACE:
            return self._create_missed_space_typo(current_char)
        elif typo_type == TypoType.DOUBLE:
            return TypoResult(typo_type=TypoType.DOUBLE, wrong_char=current_char)
        else:
            return TypoResult(typo_type=TypoType.SKIP)

    def _create_transpose_typo(self, current_char: str, next_char: Optional[str]) -> TypoResult:
        """顺序颠倒错误（如果下个不是字母则降级为按错相邻键）"""
        if next_char and next_char.isalpha():
            return TypoResult(typo_type=TypoType.TRANSPOSE, wrong_char=next_char)
        return self._create_adjacent_typo(current_char)

    def _create_missed_space_typo(self, current_char: str) -> TypoResult:
        """漏打空格错误（如果当前本不是空格则降级为按错相邻键）"""
        if current_char == ' ':
            return TypoResult(typo_type=TypoType.MISSED_SPACE)
        return self._create_adjacent_typo(current_char)

    @staticmethod
    def _create_adjacent_typo(original_char: str) -> TypoResult:
        """生成按错相邻键的错误"""
        lowercase_char = original_char.lower()

        # 如果字典中没有对应相邻键，降级成连按两下
        if lowercase_char not in QWERTY_NEIGHBORS:
            return TypoResult(typo_type=TypoType.DOUBLE, wrong_char=original_char)

        adjacent_char = random.choice(QWERTY_NEIGHBORS[lowercase_char])

        if original_char.isupper():
            adjacent_char = adjacent_char.upper()

        return TypoResult(typo_type=TypoType.ADJACENT, wrong_char=adjacent_char)

    @staticmethod
    def _split_modifiers_and_keys(keys: List[Key]) -> Tuple[List[Key], List[Key]]:
        """分离修饰键(Ctrl/Shift等)和普通键"""
        modifier_keys = {Key.CONTROL, Key.SHIFT, Key.ALT, Key.META}
        modifiers = [key for key in keys if key in modifier_keys]
        non_modifiers = [key for key in keys if key not in modifier_keys]
        return modifiers, non_modifiers

    @staticmethod
    def _calculate_modifier_value(modifiers: List[Key]) -> Optional[int]:
        """计算 CDP 事件所需的 modifiers 掩码值"""
        if not modifiers:
            return None

        modifier_map = {
            Key.ALT: 1,
            Key.CONTROL: 2,
            Key.META: 4,
            Key.SHIFT: 8,
        }
        value = sum(modifier_map.get(mod, 0) for mod in modifiers)
        return value if value > 0 else None