# -*- coding: utf-8 -*- """ DrissionPage 浏览器控制基础封装。 提供连接浏览器、拟人化操作等通用能力,供各任务处理器复用。 """ from __future__ import annotations import logging import random import time from typing import Optional logger = logging.getLogger("worker.browser_control") try: from DrissionPage import Chromium except ImportError: Chromium = None # type: ignore # ─── 拟人化参数 ─── HUMAN_DELAY_CLICK = (0.2, 0.6) HUMAN_DELAY_BETWEEN = (0.1, 0.4) HUMAN_MOVE_DURATION = (0.25, 0.7) HUMAN_CLICK_OFFSET = 12 def ensure_drission(): """确保 DrissionPage 已安装。""" if Chromium is None: raise RuntimeError("请先安装: pip install DrissionPage") def connect_browser(port: int = None, addr: str = None): """使用 DrissionPage 连接浏览器。""" ensure_drission() if port is not None: return Chromium(port) if addr: return Chromium(addr) raise ValueError("需要 port 或 addr 以连接浏览器") def human_delay(low: float = 0.2, high: float = 0.6): """随机延迟,模拟人类操作间隔。""" time.sleep(random.uniform(low, high)) def human_click(tab, ele, scroll_first: bool = True) -> bool: """ 拟人化点击:先滚动到可见(可选),再动作链「移动 → 短暂停顿 → 点击」, 移动带随机耗时与元素内随机偏移。 """ if ele is None: return False try: if scroll_first and hasattr(ele, "scroll") and getattr(ele.scroll, "to_see", None): ele.scroll.to_see() human_delay(0.15, 0.4) ox = random.randint(-HUMAN_CLICK_OFFSET, HUMAN_CLICK_OFFSET) oy = random.randint(-HUMAN_CLICK_OFFSET, HUMAN_CLICK_OFFSET) duration = random.uniform(*HUMAN_MOVE_DURATION) tab.actions.move_to(ele, offset_x=ox, offset_y=oy, duration=duration) tab.actions.wait(*HUMAN_DELAY_BETWEEN) tab.actions.click() return True except Exception: try: ele.click() return True except Exception: return False def safe_click(tab, ele) -> bool: """多种方式尝试点击元素(拟人 → 普通 → JS),确保触发成功。""" if ele is None: return False # 方式1: 拟人点击 if human_click(tab, ele): return True # 方式2: 普通模拟点击 try: ele.click(by_js=False) return True except Exception: pass # 方式3: JS 点击 try: ele.click(by_js=True) return True except Exception: pass return False def find_element(tab, selectors: list[str], timeout: int = 3): """按优先级尝试多个选择器,返回第一个找到的元素。""" for sel in selectors: try: ele = tab.ele(sel, timeout=timeout) if ele: return ele except Exception: continue return None def find_elements(tab, selectors: list[str], timeout: int = 3) -> list: """按优先级尝试多个选择器,返回第一组找到的元素列表。""" for sel in selectors: try: eles = tab.eles(sel, timeout=timeout) if eles and len(eles) > 0: return eles except Exception: continue return []