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 []
|