140 lines
4.9 KiB
Python
140 lines
4.9 KiB
Python
"""
|
||
使用 Nodriver 访问 Cloudflare 保护的 audiences.me/torrents.php，
|
||
启动时从根目录 cookies.txt 加载 Cookie，采用等待自动通过 + 选择器点击 Turnstile 过验证。
|
||
每 30 分钟执行一次检查。
|
||
"""
|
||
|
||
import asyncio
|
||
import nodriver as uc
|
||
from cookies_loader import (
|
||
get_cookies_path,
|
||
cookies_file_exists,
|
||
is_netscape_format,
|
||
load_cookies_nodriver_native,
|
||
inject_netscape_cookies,
|
||
)
|
||
|
||
TARGET_URL = "https://audiences.me/torrents.php"
|
||
|
||
|
||
async def get_search_btn_value(page):
|
||
"""获取 #search_btn 的 value，用于判断是否需要验证。"""
|
||
try:
|
||
el = await page.query_selector("#search_btn")
|
||
print(f"el: {el}")
|
||
if el is None:
|
||
return None
|
||
# nodriver Element 用 el["value"] 或 get_js_attributes 取属性，无 get_attribute
|
||
raw = el["value"]
|
||
if raw is None:
|
||
attrs = await el.get_js_attributes()
|
||
raw = attrs.get("value") if attrs else None
|
||
return (raw or "").strip()
|
||
except Exception as e:
|
||
print(f"Exception: {e}")
|
||
return None
|
||
|
||
|
||
async def try_click_cf_turnstile(page):
|
||
"""
|
||
通过截图+模板匹配定位 CF Turnstile 复选框并点击（与语言无关）。
|
||
复选框在跨域 iframe + Shadow DOM 内，用选择器点不到，需用坐标点击。
|
||
依赖：pip install opencv-python
|
||
"""
|
||
try:
|
||
# 用自定义模板 box.png 在截图中匹配复选框区域，返回中心坐标（截图是设备像素）
|
||
pos = await page.template_location(template_image="box.png")
|
||
if pos is None:
|
||
return
|
||
x, y = int(pos[0]), int(pos[1])
|
||
# 保存带点击位置标记的截图（便于调试）
|
||
try:
|
||
await page.save_screenshot("click_position.png")
|
||
import cv2
|
||
im = cv2.imread("click_position.png")
|
||
if im is not None:
|
||
r = max(15, min(im.shape[:2]) // 30)
|
||
cv2.circle(im, (x, y), r, (0, 0, 255), 3)
|
||
cv2.line(im, (x - r - 5, y), (x + r + 5, y), (0, 0, 255), 2)
|
||
cv2.line(im, (x, y - r - 5), (x, y + r + 5), (0, 0, 255), 2)
|
||
cv2.imwrite("click_position.png", im)
|
||
except Exception:
|
||
pass
|
||
# 截图坐标是设备像素，CDP 鼠标事件要 CSS 像素，需除以 devicePixelRatio
|
||
try:
|
||
raw = await page.evaluate("window.devicePixelRatio")
|
||
dpr = float(getattr(raw, "value", raw) if hasattr(raw, "value") else raw) if raw is not None else 1.0
|
||
except Exception:
|
||
dpr = 1.0
|
||
if dpr <= 0:
|
||
dpr = 1.0
|
||
x_css, y_css = x / dpr, y / dpr
|
||
# 先模拟人类移动再点击，提高通过率
|
||
await page.mouse_move(x_css, y_css, steps=10)
|
||
await page.sleep(0.3)
|
||
await page.mouse_click(x_css, y_css)
|
||
await page.sleep(3)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
INTERVAL_MINUTES = 30
|
||
|
||
|
||
async def run_one_check():
|
||
"""执行一次检查：打开浏览器、过 CF 验证，完成后关闭浏览器。"""
|
||
# 建议非无头模式以提高 CF 通过率
|
||
browser = await uc.start(headless=False)
|
||
path = get_cookies_path()
|
||
|
||
if cookies_file_exists():
|
||
if is_netscape_format(path):
|
||
# Netscape 格式：先打开目标页再注入 cookies，再重新打开使 cookies 生效（避免 reload 导致 Not attached）
|
||
page = await browser.get(TARGET_URL)
|
||
await inject_netscape_cookies(page, path)
|
||
page = await browser.get(TARGET_URL)
|
||
else:
|
||
# 先尝试 nodriver 原生格式加载，再打开目标页
|
||
await load_cookies_nodriver_native(browser, path)
|
||
page = await browser.get(TARGET_URL)
|
||
else:
|
||
page = await browser.get(TARGET_URL)
|
||
|
||
while True:
|
||
# 进入页面后等待一下再判断
|
||
await page.sleep(2)
|
||
val = await get_search_btn_value(page)
|
||
print(f"val: {val}")
|
||
if val and "给我搜" in val:
|
||
# 不需要验证，直接结束本次检查
|
||
browser.stop()
|
||
return
|
||
|
||
# 需要验证：等待 CF 自动通过后尝试点击 Turnstile
|
||
await page.sleep(15)
|
||
await try_click_cf_turnstile(page)
|
||
await page.sleep(3)
|
||
|
||
# 验证后等 10 秒看是否变为「给我搜」
|
||
await page.sleep(10)
|
||
val_after = await get_search_btn_value(page)
|
||
print(f"val_after: {val_after}")
|
||
if val_after and "给我搜" in val_after:
|
||
# 已通过，关闭浏览器，结束本次检查
|
||
browser.stop()
|
||
return
|
||
|
||
# 仍未通过，刷新页面重新走验证
|
||
page = await browser.get(TARGET_URL)
|
||
|
||
|
||
async def main():
|
||
"""每 30 分钟执行一次检查，循环运行。"""
|
||
while True:
|
||
await run_one_check()
|
||
await asyncio.sleep(INTERVAL_MINUTES * 60)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
uc.loop().run_until_complete(main())
|