120 lines
3.3 KiB
Python
120 lines
3.3 KiB
Python
# -*- 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 []
|