coordinate/utils/mouse.py

# from DrissionPage import ChromiumPage
# # 假设上面的代码保存在 human_mouse.py
# from human_mouse import HumanMouse 

# # 初始化 DrissionPage
# page = ChromiumPage()
# page.get('https://example.com')

# # 初始化我们移植的鼠标控制器，开启 debug 模式可在页面上看到鼠标轨迹点
# mouse = HumanMouse(page, debug=True)

# # 1. 拟人化移动到一个坐标
# mouse.move(500, 300, humanize=True)

# # 2. 拟人化点击
# mouse.click(600, 400, humanize=True)

# # 3. 如果需要结合 DrissionPage 元素一起使用：
# ele = page.ele('@href=https://www.iana.org/domains/example')
# # 获取元素在屏幕中的中点坐标
# x, y = ele.rect.midpoint 
# mouse.click(x, y, humanize=True)

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

from DrissionPage import ChromiumPage

from utils.math_utils import (
    bezier_2d,
    fitts_duration,
    minimum_jerk,
    random_control_points,
)

logger = logging.getLogger(__name__)


class MouseButton(Enum):
    LEFT = "left"
    RIGHT = "right"
    MIDDLE = "middle"


class MouseEventType(Enum):
    MOUSE_MOVED = "mouseMoved"
    MOUSE_PRESSED = "mousePressed"
    MOUSE_RELEASED = "mouseReleased"


@dataclass(frozen=True)
class MouseTimingConfig:
    """模拟人类鼠标移动物理特性的配置"""

    fitts_a: float = 0.070
    fitts_b: float = 0.150

    frame_interval: float = 0.012
    frame_interval_variance: float = 0.004

    curvature_min: float = 0.10
    curvature_max: float = 0.28
    curvature_asymmetry: float = 0.6

    short_distance_threshold: float = 50.0

    tremor_amplitude: float = 0.85

    overshoot_probability: float = 0.22
    overshoot_distance_min: float = 0.03
    overshoot_distance_max: float = 0.10
    overshoot_speed_threshold: float = 260.0

    pre_click_pause_min: float = 0.04
    pre_click_pause_max: float = 0.16
    click_hold_min: float = 0.04
    click_hold_max: float = 0.12
    double_click_interval_min: float = 0.05
    double_click_interval_max: float = 0.09
    drag_start_pause_min: float = 0.06
    drag_start_pause_max: float = 0.16
    drag_end_pause_min: float = 0.04
    drag_end_pause_max: float = 0.12

    micro_pause_probability: float = 0.08
    micro_pause_min: float = 0.010
    micro_pause_max: float = 0.030

    min_duration: float = 0.08
    max_duration: float = 2.2


class HumanMouse:
    """
    DrissionPage 的高仿真人类鼠标控制器。
    提供移动、点击、双击和拖拽方法，采用贝塞尔曲线、Fitts 定律、生理性微颤和过冲校正。
    """

    _DEBUG_INIT_JS = """
    if (!document.getElementById('__dp_mouse_debug')) {
        const canvas = document.createElement('canvas');
        canvas.id = '__dp_mouse_debug';
        canvas.style.cssText = 'position:fixed;top:0;left:0;width:100vw;height:100vh;'
            + 'pointer-events:none;z-index:2147483647;';
        canvas.width = window.innerWidth;
        canvas.height = window.innerHeight;
        document.body.appendChild(canvas);
        window.__dp_debug_ctx = canvas.getContext('2d');
    }
    """

    _DEBUG_DOT_JS = """
    if (window.__dp_debug_ctx) {
        const ctx = window.__dp_debug_ctx;
        ctx.beginPath();
        ctx.arc(%s, %s, %s, 0, 2 * Math.PI);
        ctx.fillStyle = '%s';
        ctx.fill();
    }
    """

    def __init__(
        self,
        page: ChromiumPage,
        timing: Optional[MouseTimingConfig] = None,
        debug: bool = False,
    ):
        """
        初始化鼠标控制器
        :param page: DrissionPage 的 ChromiumPage 或 ChromiumTab 实例
        :param timing: 轨迹与时间配置
        :param debug: 是否在页面上绘制调试红蓝点
        """
        self._page = page
        self._timing = timing or MouseTimingConfig()
        self._position: Tuple[float, float] = self._read_initial_position()
        self._debug = debug
        self._debug_initialized = False
        self._session_profile = self._build_session_profile()

    @property
    def timing(self) -> MouseTimingConfig:
        return self._timing

    @timing.setter
    def timing(self, config: MouseTimingConfig) -> None:
        self._timing = config

    @property
    def debug(self) -> bool:
        return self._debug

    @debug.setter
    def debug(self, value: bool) -> None:
        self._debug = value
        self._debug_initialized = False

    def move(self, x: float, y: float, *, humanize: bool = False) -> None:
        """移动鼠标"""
        if humanize:
            self._move_humanized(x, y)
            return

        self._dispatch_move(x, y)
        
    def human_click_ele(self, element):
        """
        拟人化点击元素 (修复版)
        :param element: DrissionPage 的元素对象
        """
        # 1. 强制滚动，把元素尽量移到屏幕正中间，防止被顶部导航栏或底部悬浮窗遮盖
        element.scroll.to_see(center=True)
        time.sleep(0.5)  # 等待滚动动画完成，防止移动时坐标还在变化
        
        # 2. 【核心修复】获取相对于当前屏幕视口的坐标，而不是页面绝对坐标
        # 注意：这里使用的是 viewport_midpoint
        mid_x, mid_y = element.rect.viewport_midpoint 
        
        # 3. 如果元素太靠近边缘，可能会获取失败，做一个基础保护
        if mid_x is None or mid_y is None:
            logger.warning("Failed to get viewport midpoint, falling back to absolute midpoint.")
            mid_x, mid_y = element.rect.midpoint
            
        self.click(mid_x, mid_y, humanize=True)

    def click(
        self,
        x: float,
        y: float,
        *,
        button: MouseButton = MouseButton.LEFT,
        click_count: int = 1,
        humanize: bool = False,
    ) -> None:
        """点击鼠标"""
        if humanize:
            self._click_humanized(x, y, button, click_count)
            return

        self._dispatch_move(x, y)
        self._dispatch_button(MouseEventType.MOUSE_PRESSED, button, click_count)
        self._dispatch_button(MouseEventType.MOUSE_RELEASED, button, click_count)

    def double_click(
        self,
        x: float,
        y: float,
        *,
        button: MouseButton = MouseButton.LEFT,
        humanize: bool = False,
    ) -> None:
        """双击鼠标"""
        self.click(x, y, button=button, click_count=2, humanize=humanize)

    def down(self, button: MouseButton = MouseButton.LEFT) -> None:
        """按下鼠标按键"""
        self._dispatch_button(MouseEventType.MOUSE_PRESSED, button)

    def up(self, button: MouseButton = MouseButton.LEFT) -> None:
        """释放鼠标按键"""
        self._dispatch_button(MouseEventType.MOUSE_RELEASED, button)

    def drag(
        self,
        start_x: float,
        start_y: float,
        end_x: float,
        end_y: float,
        *,
        humanize: bool = False,
    ) -> None:
        """拖拽鼠标"""
        if humanize:
            self._drag_humanized(start_x, start_y, end_x, end_y)
            return

        self._dispatch_move(start_x, start_y)
        self._dispatch_button(MouseEventType.MOUSE_PRESSED, MouseButton.LEFT)
        self._dispatch_move(end_x, end_y)
        self._dispatch_button(MouseEventType.MOUSE_RELEASED, MouseButton.LEFT)

    def _move_humanized(self, target_x: float, target_y: float, custom_duration: Optional[float] = None) -> None:
        """拟人化移动核心逻辑"""
        start = self._position
        target = (target_x, target_y)
        distance = math.hypot(target_x - start[0], target_y - start[1])

        if distance < 1.0:
            self._dispatch_move(target_x, target_y)
            return

        config = self._timing

        if custom_duration is not None:
            duration = custom_duration
        else:
            duration = fitts_duration(distance, 20.0, config.fitts_a, config.fitts_b)
            duration = max(config.min_duration, min(duration, config.max_duration))
            duration *= self._session_profile["speed_scale"]

        should_overshoot = (
            distance > config.overshoot_speed_threshold
            and random.random() < config.overshoot_probability * self._session_profile["overshoot_bias"]
        )

        if should_overshoot:
            self._move_with_overshoot(start, target, duration)
        else:
            cp1, cp2 = self._get_control_points(start, target)
            self._perform_movement_loop(start, target, duration, cp1, cp2)

        self._perform_final_correction(target_x, target_y)

    def _move_with_overshoot(
        self,
        start: Tuple[float, float],
        target: Tuple[float, float],
        duration: float,
    ) -> None:
        """带过冲现象的移动"""
        config = self._timing
        overshoot_fraction = random.uniform(
            config.overshoot_distance_min, config.overshoot_distance_max
        )
        dx = target[0] - start[0]
        dy = target[1] - start[1]
        
        overshoot = (target[0] + dx * overshoot_fraction, target[1] + dy * overshoot_fraction)

        cp1, cp2 = self._get_control_points(start, overshoot)
        self._perform_movement_loop(start, overshoot, duration * 0.85, cp1, cp2)

        cp1, cp2 = self._get_control_points(overshoot, target)
        self._perform_movement_loop(overshoot, target, duration * 0.15, cp1, cp2)

    def _perform_movement_loop(
        self,
        start: Tuple[float, float],
        end: Tuple[float, float],
        duration: float,
        cp1: Tuple[float, float],
        cp2: Tuple[float, float],
    ) -> None:
        """按帧渲染并执行移动循环"""
        config = self._timing
        start_time = time.perf_counter()
        prev = (start[0], start[1], start_time)
        segment_pause_used = False

        while True:
            now = time.perf_counter()
            elapsed = now - start_time

            if elapsed >= duration:
                break

            progress = elapsed / duration
            eased = minimum_jerk(progress)
            x, y = bezier_2d(eased, start, cp1, cp2, end)

            sigma = self._compute_tremor_sigma(x, y, now, prev, config)
            x += random.gauss(0, sigma)
            y += random.gauss(0, sigma)

            self._dispatch_move(x, y)
            prev = (x, y, now)

            frame_delay = self._sample_frame_delay()
            time.sleep(max(0.001, frame_delay))

            if (not segment_pause_used) and random.random() < config.micro_pause_probability:
                pause = self._sample_micro_pause()
                time.sleep(pause)
                start_time += pause
                segment_pause_used = True

    @staticmethod
    def _compute_tremor_sigma(
        x: float,
        y: float,
        now: float,
        prev: Tuple[float, float, float],
        config: MouseTimingConfig,
    ) -> float:
        """动态计算手抖幅度"""
        dt = now - prev[2]
        if dt > 0:
            velocity = math.hypot(x - prev[0], y - prev[1]) / dt
            speed_factor = max(0.2, 1.0 - velocity / 500.0)
        else:
            speed_factor = 1.0
        return config.tremor_amplitude * speed_factor

    def _click_humanized(
        self,
        x: float,
        y: float,
        button: MouseButton,
        click_count: int,
    ) -> None:
        """拟人化点击"""
        self._move_humanized(x, y)
        self._micro_adjust_towards(x, y)

        pre_pause = self._sample_pre_click_pause()
        time.sleep(pre_pause)

        for i in range(click_count):
            current_count = i + 1
            self._dispatch_button(MouseEventType.MOUSE_PRESSED, button, current_count)

            hold = self._sample_click_hold()
            time.sleep(hold)

            self._dispatch_button(MouseEventType.MOUSE_RELEASED, button, current_count)

            if current_count < click_count:
                interval = self._sample_double_click_interval()
                time.sleep(interval)

    def _drag_humanized(
        self,
        start_x: float,
        start_y: float,
        end_x: float,
        end_y: float,
    ) -> None:
        """拟人化拖拽"""
        config = self._timing

        self._move_humanized(start_x, start_y)
        self._dispatch_button(MouseEventType.MOUSE_PRESSED, MouseButton.LEFT)

        drag_start_pause = random.uniform(config.drag_start_pause_min, config.drag_start_pause_max)
        time.sleep(drag_start_pause)

        start = self._position
        distance = math.hypot(end_x - start[0], end_y - start[1])
        duration = fitts_duration(distance, 20.0, config.fitts_a, config.fitts_b)
        duration = max(config.min_duration, min(duration, config.max_duration))

        cp1, cp2 = self._get_control_points(start, (end_x, end_y))
        self._perform_movement_loop(start, (end_x, end_y), duration, cp1, cp2)
        self._dispatch_move(end_x, end_y)

        drag_end_pause = random.uniform(config.drag_end_pause_min, config.drag_end_pause_max)
        time.sleep(drag_end_pause)

        self._dispatch_button(MouseEventType.MOUSE_RELEASED, MouseButton.LEFT)

    def _get_control_points(
        self,
        start: Tuple[float, float],
        end: Tuple[float, float],
    ) -> Tuple[Tuple[float, float], Tuple[float, float]]:
        config = self._timing
        return random_control_points(
            start,
            end,
            config.curvature_min,
            config.curvature_max,
            config.curvature_asymmetry,
            config.short_distance_threshold,
        )

    def _dispatch_move(self, x: float, y: float) -> None:
        """发送 CDP 鼠标移动指令"""
        self._page.run_cdp(
            "Input.dispatchMouseEvent",
            type=MouseEventType.MOUSE_MOVED.value,
            x=int(round(x)),
            y=int(round(y))
        )
        self._position = (x, y)

        if self._debug:
            self._debug_draw_dot(x, y, radius=2, color='rgba(0,150,255,0.6)')

    def _sample_frame_delay(self) -> float:
        config = self._timing
        base = config.frame_interval * self._session_profile["tempo_scale"]
        spread = config.frame_interval_variance * self._session_profile["tempo_jitter"]
        return random.gauss(base, max(0.001, spread))

    def _sample_pre_click_pause(self) -> float:
        config = self._timing
        return max(
            0.0,
            random.gauss(
                (config.pre_click_pause_min + config.pre_click_pause_max) / 2.0,
                (config.pre_click_pause_max - config.pre_click_pause_min) / 6.0,
            ),
        )

    def _sample_click_hold(self) -> float:
        config = self._timing
        return max(
            0.0,
            random.gauss(
                (config.click_hold_min + config.click_hold_max) / 2.0,
                (config.click_hold_max - config.click_hold_min) / 6.0,
            ),
        )

    def _sample_double_click_interval(self) -> float:
        config = self._timing
        return max(
            0.0,
            random.gauss(
                (config.double_click_interval_min + config.double_click_interval_max) / 2.0,
                (config.double_click_interval_max - config.double_click_interval_min) / 6.0,
            ),
        )

    def _sample_micro_pause(self) -> float:
        config = self._timing
        return max(
            0.0,
            random.gauss(
                (config.micro_pause_min + config.micro_pause_max) / 2.0,
                (config.micro_pause_max - config.micro_pause_min) / 6.0,
            ),
        )

    def _dispatch_button(
        self,
        event_type: MouseEventType,
        button: MouseButton,
        click_count: int = 1,
    ) -> None:
        """发送 CDP 鼠标按键指令"""
        self._page.run_cdp(
            "Input.dispatchMouseEvent",
            type=event_type.value,
            button=button.value,
            x=int(round(self._position[0])),
            y=int(round(self._position[1])),
            clickCount=click_count
        )

        if self._debug and event_type == MouseEventType.MOUSE_PRESSED:
            self._debug_draw_dot(
                self._position[0], self._position[1], radius=6, color='rgba(255,50,50,0.9)'
            )

    def _debug_draw_dot(self, x: float, y: float, radius: int, color: str) -> None:
        """绘制轨迹调试点"""
        if not self._debug_initialized:
            self._page.run_js(self._DEBUG_INIT_JS)
            self._debug_initialized = True

        script = self._DEBUG_DOT_JS % (int(round(x)), int(round(y)), radius, color)
        self._page.run_js(script)

    def _read_initial_position(self) -> Tuple[float, float]:
        try:
            data = self._page.run_js("return { x: window.screenX || 0, y: window.screenY || 0 };")
            if isinstance(data, dict):
                return (float(data.get("x", 0.0)), float(data.get("y", 0.0)))
        except Exception:
            pass
        return (0.0, 0.0)

    def _build_session_profile(self) -> dict:
        return {
            "speed_scale": random.uniform(0.90, 1.15),
            "tempo_scale": random.uniform(0.92, 1.10),
            "tempo_jitter": random.uniform(0.85, 1.25),
            "overshoot_bias": random.uniform(0.65, 1.15),
        }

    def _perform_final_correction(self, target_x: float, target_y: float) -> None:
        current_x, current_y = self._position
        distance = math.hypot(target_x - current_x, target_y - current_y)
        if distance <= 1.5:
            self._dispatch_move(target_x, target_y)
            return

        correction_end = (target_x, target_y)
        cp1, cp2 = self._get_control_points((current_x, current_y), correction_end)
        duration = min(self._timing.max_duration, max(self._timing.min_duration, 0.12 + distance / 900.0))
        self._perform_movement_loop((current_x, current_y), correction_end, duration, cp1, cp2)
        self._dispatch_move(target_x, target_y)

    def _micro_adjust_towards(self, x: float, y: float) -> None:
        current_x, current_y = self._position
        distance = math.hypot(x - current_x, y - current_y)
        if distance < 5.0:
            return
        offset_x = random.uniform(-1.0, 1.0)
        offset_y = random.uniform(-1.0, 1.0)
        self._dispatch_move(x + offset_x, y + offset_y)
        time.sleep(random.uniform(0.006, 0.018))
        self._dispatch_move(x, y)