You've already forked pruchase_jhw
zmq
116 lines
4.6 KiB
Python
116 lines
4.6 KiB
Python
from playwright.async_api import Page
|
|
import asyncio
|
|
import random
|
|
import re
|
|
|
|
async def human_scroll(page: Page, mode: str, value: int = 0):
|
|
"""
|
|
模拟人类行为的网页滚动方法
|
|
:param page: Playwright 的 Page 对象
|
|
:param mode: 滚动模式 -> 'top'(滚动到顶部), 'bottom'(滚动到底部), 'random'(随机上下滚动), 'distance'(滚动指定距离)
|
|
:param value: 当 mode='distance' 时，代表要滚动的像素距离（正数向下，负数向上）
|
|
"""
|
|
|
|
async def smooth_scroll_step(pixel_distance: int):
|
|
"""核心拟真算法：将一个大距离拆分成带有加速度和随机抖动的微小步长"""
|
|
if pixel_distance == 0:
|
|
return
|
|
|
|
direction = 1 if pixel_distance > 0 else -1
|
|
remaining = abs(pixel_distance)
|
|
|
|
while remaining > 0:
|
|
# 模拟人类滚动时越接近目标速度越慢的特征（减速缓冲）
|
|
if remaining > 100:
|
|
step = random.randint(30, 70)
|
|
elif remaining > 30:
|
|
step = random.randint(10, 25)
|
|
else:
|
|
step = remaining
|
|
|
|
remaining -= step
|
|
scroll_y = step * direction
|
|
|
|
# 执行单步微调滚动
|
|
await page.evaluate(f"window.scrollBy(0, {scroll_y});")
|
|
|
|
# 随机微小停顿，模拟手指滑动的摩擦阻力（20ms - 50ms）
|
|
await asyncio.sleep(random.uniform(0.02, 0.05))
|
|
|
|
# 0.5% 的极低概率触发人类的“视线停留”
|
|
if random.random() < 0.005:
|
|
await asyncio.sleep(random.uniform(0.3, 0.8))
|
|
|
|
# ==================== 模式 0：滚动到顶部 ====================
|
|
if mode == 'top':
|
|
# print("[HumanScroll] 正在模拟向上滚动至页面顶部...")
|
|
last_scroll = -1
|
|
while True:
|
|
# 获取当前滚动条位置
|
|
current_scroll = await page.evaluate("window.scrollY")
|
|
|
|
# 防死循环：如果位置不再变化，说明已经到顶或页面无法滚动
|
|
if current_scroll <= 0 or current_scroll == last_scroll:
|
|
break
|
|
|
|
last_scroll = current_scroll
|
|
chunk = random.randint(300, 700)
|
|
await smooth_scroll_step(-chunk)
|
|
await asyncio.sleep(random.uniform(0.5, 1.2)) # 每滑一下停顿看一眼
|
|
|
|
# ==================== 模式 1：滚动到底部 ====================
|
|
elif mode == 'bottom':
|
|
# print("[HumanScroll] 正在模拟向下滚动至页面底部...")
|
|
last_scroll = -1
|
|
while True:
|
|
# 合并 evaluate 请求，减少与浏览器的 IPC 通信开销
|
|
scroll_info = await page.evaluate("() => [window.scrollY + window.innerHeight, document.body.scrollHeight]")
|
|
current_scroll, total_height = scroll_info[0], scroll_info[1]
|
|
|
|
# 防死循环：如果位置不再变化，说明已经到底或页面无法滚动
|
|
if current_scroll >= total_height - 10 or current_scroll == last_scroll:
|
|
break
|
|
|
|
last_scroll = current_scroll
|
|
chunk = random.randint(300, 700)
|
|
await smooth_scroll_step(chunk)
|
|
await asyncio.sleep(random.uniform(0.5, 1.2)) # 每滑一下停顿看一眼
|
|
|
|
# ==================== 模式 2：随机上下滚动几下 ====================
|
|
elif mode == 'random':
|
|
# print("[HumanScroll] 正在执行人类迷惑行为：随机上下滚动...")
|
|
# 随机决定滚动 3 到 6 下
|
|
scroll_times = random.randint(3, 6)
|
|
|
|
for _ in range(scroll_times):
|
|
# 70% 概率向下，30% 概率向上
|
|
if random.random() > 0.3:
|
|
distance = random.randint(200, 500) # 向下滚
|
|
else:
|
|
distance = random.randint(-400, -150) # 向上回滚
|
|
|
|
await smooth_scroll_step(distance)
|
|
await asyncio.sleep(random.uniform(0.6, 1.5)) # 停顿
|
|
|
|
# ==================== 模式 3：滚动指定距离 ====================
|
|
elif mode == 'distance':
|
|
# print(f"[HumanScroll] 正在精准模拟滚动指定距离: {value} 像素")
|
|
await smooth_scroll_step(value)
|
|
|
|
else:
|
|
raise ValueError("未知的滚动模式！请选择 'top', 'bottom', 'random' 或 'distance'")
|
|
|
|
|
|
def format_japanese_price(price_str: str | None) -> int:
|
|
"""
|
|
将日本电商网站的价格字符串格式化为标准整数
|
|
支持的格式示例: "600円 (税込)", "1,250円", "¥ 3,000", "300"
|
|
"""
|
|
if not price_str:
|
|
return 0
|
|
|
|
digits = re.findall(r'\d', price_str)
|
|
if not digits:
|
|
return 0
|
|
|
|
return int("".join(digits)) |