import time import random from playwright.sync_api import Page from playwright.async_api import Page as AsyncPage from .human_like import HumanLikeTyper import asyncio from typing import Union, Optional, Dict, Any class HumanTypingWrapper: """ 人类化输入包装器，提供适度的人类化输入功能 支持同步和异步页面操作，设计更加简洁高效 """ def __init__(self, page: Union[Page, AsyncPage], config: Optional[Dict[str, Any]] = None): self.page = page self.is_async = hasattr(page, 'wait_for_timeout') # 检测是否为异步页面 # 初始化配置 - 使用更合理的默认值，简化设计 self.config = self._init_config(config) # 初始化human_like模块（仅用于同步页面） if not self.is_async: self.human_typer = HumanLikeTyper(page) def _init_config(self, config: Optional[Dict[str, Any]]) -> Dict[str, Any]: """初始化配置参数 采用行业标准的人类行为模拟设计： - 保持合理的输入速度变化 - 适度的停顿和错误修正 - 简化的参数设置，更易维护 """ default_config = { # 基础输入速度设置（更符合真实人类输入速度范围） 'min_typing_speed': 2.0, 'max_typing_speed': 5.0, # 适度的思考停顿设置 'pause_probability': 0.1, # 每10个字符左右停顿一次 'min_pause_duration': 0.2, 'max_pause_duration': 1.0, # 简化的错误修正设置 - 不要过多干扰正常输入 'correction_probability': 0.05, # 5%的概率进行错误修正 # 基本的点击前后停顿 'click_delay_before': (0.2, 0.5), 'click_delay_after': (0.3, 0.7), # 输入完成后的停顿 'finish_pause_probability': 0.6, 'finish_pause_duration': (0.5, 1.2), # 分段输入设置 - 更合理的分段长度 'chunk_input': True, 'max_chunk_length': 50, # 更长的段落长度，更符合人类阅读习惯 'chunk_pause_duration': (0.8, 2.0), # 最小延迟设置 - 确保不会有机器特征明显的快速输入 'min_char_delay': 0.08, # 确保字符间延迟不会过低 } if config: default_config.update(config) return default_config def _calculate_typing_delay(self) -> float: """简化的延迟计算方法 保持合理的随机变化，但避免过度复杂的计算 """ # 基础延迟计算 base_speed = random.uniform( self.config['min_typing_speed'], self.config['max_typing_speed'] ) # 基础延迟（字符间） base_delay = 1 / base_speed # 增加一些随机变化，但保持在合理范围内 variation = random.uniform(0.8, 1.3) final_delay = base_delay * variation # 确保最小延迟，避免机器特征 final_delay = max(self.config['min_char_delay'], final_delay) return final_delay def _should_pause(self) -> bool: """简化的停顿判断逻辑""" return random.random() < self.config['pause_probability'] def _should_correct(self) -> bool: """判断是否应该进行错误修正""" return random.random() < self.config['correction_probability'] def _get_pause_duration(self) -> float: """获取随机停顿时长""" return random.uniform( self.config['min_pause_duration'], self.config['max_pause_duration'] ) def _get_wrong_char(self, correct_char: str) -> str: """简化的错误字符生成 - 不需要完整的键盘映射表""" # 只保留几个常见的错误类型，减少复杂性 if correct_char.isalpha(): # 简单的字母错误（前后移位） char_code = ord(correct_char.lower()) if char_code > ord('a') and char_code < ord('z'): # 随机前后移位 wrong_offset = random.choice([-1, 1]) wrong_char = chr(char_code + wrong_offset) return wrong_char.upper() if correct_char.isupper() else wrong_char # 默认返回一个简单的错误字符 return random.choice('abcdefghijklmnopqrstuvwxyz') def _record_action(self, action_type: str): """空实现的_record_action方法，保持兼容性""" pass async def _sleep(self, duration: float): """统一的睡眠方法，兼容同步和异步""" if self.is_async: await asyncio.sleep(duration) else: time.sleep(duration) async def _type_char(self, char: str, delay: float): """简化的字符输入方法""" if self.is_async: await self.page.keyboard.type(char, delay=delay * 1000) else: self.page.keyboard.type(char, delay=delay * 1000) # 简单的微小停顿，增加自然感 await self._sleep(random.uniform(0.02, 0.05)) async def _press_key(self, key: str): """按键操作，兼容同步和异步""" if self.is_async: await self.page.keyboard.press(key) else: self.page.keyboard.press(key) # 记录按键动作 self._record_action(f"key_{key}") async def _click_element(self, selector: str): """点击元素，兼容同步和异步""" if self.is_async: await self.page.click(selector) else: self.page.click(selector) # 记录点击动作 self._record_action("click") async def _wait_for_selector(self, selector: str, timeout: int = 30000): """等待选择器，兼容同步和异步""" if self.is_async: await self.page.wait_for_selector(selector, timeout=timeout) else: self.page.wait_for_selector(selector, timeout=timeout) async def _perform_random_action(self): """简化的随机动作执行 - 不需要过多的动作类型""" # 简单的微小停顿就足够模拟人类行为的不确定性 await self._sleep(random.uniform(0.1, 0.3)) async def type_text_human(self, selector: str, text: str, clear_first: bool = True) -> bool: """ 以人类化方式在指定元素中输入文本 Args: selector: 元素选择器 text: 要输入的文本 clear_first: 是否先清空现有内容 Returns: bool: 是否成功输入 """ try: # 重置历史记录 self.last_delays = [] self.action_history = [] self.input_timestamps = [] # 等待元素并点击 await self._wait_for_selector(selector) # 点击前停顿 - 增加随机性和持续时间 delay_before = random.uniform(*self.config['click_delay_before']) await self._sleep(delay_before) # 模拟光标移动到元素的过程 await self._sleep(random.uniform(0.1, 0.3)) # 点击元素 await self._click_element(selector) # 点击后停顿 - 增加随机性和持续时间 delay_after = random.uniform(*self.config['click_delay_after']) await self._sleep(delay_after) # 清空现有内容 - 更自然的清空操作，使用更长的延迟 if clear_first: # 增加随机延迟，使清空操作更自然 await self._sleep(random.uniform(0.2, 0.4)) await self._press_key("Control+A") await self._sleep(random.uniform(0.15, 0.3)) await self._press_key("Delete") await self._sleep(random.uniform(0.3, 0.6)) # 分段输入长文本 if self.config['chunk_input'] and len(text) > self.config['max_chunk_length']: await self._type_text_in_chunks(text) else: await self._type_text_continuously(text) # 输入完成后可能停顿 - 更自然的结束停顿 if random.random() < self.config['finish_pause_probability']: pause_duration = random.uniform(*self.config['finish_pause_duration']) await self._sleep(pause_duration) # 输入完成后的最终停顿 await self._sleep(random.uniform(0.2, 0.5)) return True except Exception as e: print(f"人类化输入失败: {e}") return False async def _type_text_continuously(self, text: str): """简化的连续输入方法 保留必要的人类特征，但避免过度复杂化 """ for char in text: # 合理的停顿思考 if self._should_pause(): pause_duration = self._get_pause_duration() await self._sleep(pause_duration) # 适度的错误修正 if random.random() < self.config['correction_probability']: # 简化的错误修正 wrong_char = self._get_wrong_char(char) await self._type_char(wrong_char, self._calculate_typing_delay()) await self._sleep(random.uniform(0.1, 0.3)) # 发现错误后的短暂停顿 await self._press_key("Backspace") # 删除错误字符 await self._sleep(random.uniform(0.1, 0.2)) # 删除后的停顿 # 计算延迟并输入字符 delay = self._calculate_typing_delay() await self._type_char(char, delay) async def _type_text_in_chunks(self, text: str): """简化的分段输入方法 保持基本的段落分割逻辑，但简化实现""" chunk_size = self.config['max_chunk_length'] # 简单的分段处理 for i in range(0, len(text), chunk_size): # 获取当前段落，稍微随机化长度 actual_chunk_size = min(chunk_size, len(text) - i) # 如果不是最后一段，尽量在空格处分段 if i + actual_chunk_size < len(text): # 寻找段落内最近的空格 space_pos = text.rfind(' ', i, i + actual_chunk_size + 1) if space_pos > i: actual_chunk_size = space_pos - i + 1 # 输入当前段落 chunk = text[i:i + actual_chunk_size] await self._type_text_continuously(chunk) # 段落间停顿 if i + actual_chunk_size < len(text): pause_duration = random.uniform(*self.config['chunk_pause_duration']) await self._sleep(pause_duration) # 移除了专门的_simul ate_correction和_simul ate_backspace_retype方法 # 将这些功能合并到_type_text_continuously中，简化实现 async def click_and_type(self, selector: str, text: str, **kwargs) -> bool: """ 点击元素并输入文本的便捷方法 Args: selector: 元素选择器 text: 要输入的文本 **kwargs: 传递给type_text_human的额外参数 Returns: bool: 是否成功 """ return await self.type_text_human(selector, text, **kwargs) async def human_click(self, selector: str, wait_after: bool = True) -> bool: """ 人类化点击元素 Args: selector: 元素选择器 wait_after: 点击后是否等待 Returns: bool: 是否成功点击 """ try: await self._wait_for_selector(selector) # 点击前停顿 - 增加随机性 delay_before = random.uniform(*self.config['click_delay_before']) await self._sleep(delay_before) # 点击 await self._click_element(selector) # 点击后停顿 if wait_after: delay_after = random.uniform(*self.config['click_delay_after']) await self._sleep(delay_after) return True except Exception as e: print(f"人类化点击失败: {e}") return False async def human_scroll(self, direction: str = "down", amount: int = 3): """ 人类化滚动页面 Args: direction: 滚动方向 "up" 或 "down" amount: 滚动次数 """ key = "PageDown" if direction == "down" else "PageUp" for i in range(amount): await self._press_key(key) # 滚动间隔 - 增加随机性和一致性 pause = random.uniform(0.5, 1.2) # 增加滚动间隔 await self._sleep(pause) # 记录滚动动作 self._record_action(f"scroll_{direction}") def update_config(self, new_config: Dict[str, Any]): """更新配置""" self.config.update(new_config) def get_config(self) -> Dict[str, Any]: """获取当前配置""" return self.config.copy() # 便捷的工厂函数 def create_human_typer(page: Union[Page, AsyncPage], config: Optional[Dict[str, Any]] = None) -> HumanTypingWrapper: """ 创建人类化输入包装器的工厂函数 Args: page: Playwright页面对象（同步或异步） config: 可选的配置字典 Returns: HumanTypingWrapper实例 """ return HumanTypingWrapper(page, config)