import argparse import asyncio import json import os import random import re import shutil import subprocess import sys import time from pathlib import Path from patchright.async_api import async_playwright import asyncio from curl_cffi.requests import AsyncSession from cf_camoufox_captcha import solve_captcha from cf_camoufox_captcha.cloudflare.utils.detection import detect_cloudflare_challenge import logging from redis.retry import Retry from redis.backoff import ExponentialBackoff from redis.exceptions import ConnectionError, TimeoutError import redis.asyncio as redis from typing import Any, Dict, List, Optional, Union import orjson from logging.handlers import TimedRotatingFileHandler BASE_DIR = Path(__file__).resolve().parent # URL = "https://www.mbet.io/Home/raceMenu?sportId=9001&hours=24" # URL = "https://www.bet365.com/#/IP/" COOKIE_OUTPUT = BASE_DIR / "mbet_cookie_result.json" def create_logger_handler(log_path="logs.log",retention=5,): # 创建一个日志处理器，每天更换一次日志文件 handler = TimedRotatingFileHandler(log_path, when='midnight', interval=1, backupCount=3, encoding='utf-8') handler.suffix = "%Y-%m-%d" # 设置备份文件的后缀名，如log-2022-06-07.log handler.setLevel(logging.ERROR) # 只记录错误级别的日志 handler.setFormatter(logging.Formatter("%(asctime)s - %(filename)s - %(funcName)s - Line %(lineno)d - %(levelname)s - %(message)s", )) # 设置日志格式 logger_handler = logging.getLogger(__name__) # 创建一个logger logger_handler.addHandler(handler) # 添加处理器 return logger_handler logger_handler = create_logger_handler() class AsyncRedisPool: def __init__( self, redis_config, redis_db=None, log_handler=logging, max_connections: int = 100, ): if redis_db is None: redis_db = redis_config["redis_db"] redis_url = "redis://:pdmy888@127.0.0.1:6379/0" if not redis_config else f"redis://:{redis_config['redis_password']}@{redis_config['redis_host']}:{redis_config['redis_port']}/{redis_db}" self.pool = redis.ConnectionPool.from_url( redis_url, decode_responses=False, max_connections=max_connections, # 连接超时 socket_connect_timeout=5, # 读写超时 socket_timeout=5, # TCP keepalive socket_keepalive=True, # 空闲连接健康检查 health_check_interval=30, # 自动重试：连接断开/超时时重试 retry=Retry( ExponentialBackoff(base=1, cap=10), retries=3, ), ) self.redis = redis.Redis(connection_pool=self.pool) self.logger_handler = log_handler async def ping(self) -> bool: return await self.redis.ping() async def close(self): await self.redis.aclose() await self.pool.disconnect() async def set_json( self, key: Union[str, int], data: Union[dict, list], expire: Optional[int] = None, ) -> bool: if expire is not None and expire <= 0: raise ValueError("expire 必须是大于 0 的整数秒数") value = orjson.dumps(data) try: return await self.redis.set( str(key), value, ex=expire, ) except (ConnectionError, TimeoutError, ConnectionResetError) as set_json_err: print(f"Redis set_json 失败: {set_json_err}") self.logger_handler.error(f"delete>>>>>>{set_json_err}") return False async def get_json( self, key: Union[str, int], ) -> Union[dict, list, None]: try: value = await self.redis.get(str(key)) if value is None: return None return orjson.loads(value) except (ConnectionError, TimeoutError, ConnectionResetError) as get_json_err: print(f"Redis get_json 失败: {get_json_err}") self.logger_handler.error(f"delete>>>>>>{get_json_err}") return None async def delete(self, key: Union[str, int]) -> int: try: return await self.redis.delete(str(key)) except (ConnectionError, TimeoutError, ConnectionResetError) as delete_err: print(f"Redis delete 失败: {delete_err}") self.logger_handler.error(f"delete>>>>>>{delete_err}") return 0 async def batch_get_json( self, keys: List[Union[str, int]], ) -> Dict[str, Any]: if not keys: return {} redis_keys = [str(key) for key in keys] try: values = await self.redis.mget(redis_keys) result = {} for key, value in zip(redis_keys, values): if value is None: result[key] = None else: result[key] = orjson.loads(value) return result except (ConnectionError, TimeoutError, ConnectionResetError) as e: print(f"Redis batch_get_json 失败: {e}") return {} async def batch_set_json( self, data_map: Dict[Union[str, int], Union[dict, list]], expire: Optional[int] = None, batch_size: int = 1000, ) -> int: if not data_map: return 0 if expire is not None and expire <= 0: raise ValueError("expire 必须是大于 0 的整数秒数") items = list(data_map.items()) total = 0 try: for i in range(0, len(items), batch_size): batch = items[i:i + batch_size] if expire is None: mapping = { str(key): orjson.dumps(value) for key, value in batch } ok = await self.redis.mset(mapping) if ok: total += len(batch) else: async with self.redis.pipeline(transaction=False) as pipe: for key, value in batch: pipe.set( str(key), orjson.dumps(value), ex=expire, ) results = await pipe.execute() total += sum(1 for item in results if item is True) return total except (ConnectionError, TimeoutError, ConnectionResetError) as e: print(f"Redis batch_set_json 失败: {e}") return total class BrowserPageChecker: def __init__(self): self.print_status = True self.mouse_x = None self.mouse_y = None self.cf_success_seen = False self.xdotool_title_marker = f"PATCHRIGHT_AUTO_{os.getpid()}_{random.randint(100000, 999999)}" self.redis_config_dict = { "148.66.51.29": dict(redis_host='148.66.51.29', redis_port=6379, redis_username="", redis_password="pdmy888", redis_db=0, url="https://www.mbet.io/Home/raceMenu?sportId=9001&hours=24"), } self.connectionPool = {} # todo 连接redis async def connection_redis(self): for redis_host, redis_config in self.redis_config_dict.items(): if not self.connectionPool.get(redis_host): redis_server = AsyncRedisPool(redis_config, log_handler=logger_handler) self.connectionPool[redis_host] = redis_server def log(self, message): """ 日志输出函数 该函数用于输出指定的消息，并确保消息被立即刷新到输出设备 参数: message (str): 需要输出的消息内容 """ if self.print_status: print(message, flush=True) # 使用flush=True确保消息立即输出，不经过缓冲区 async def _sync_cursor_dot(self, page, x, y, pressed=False): try: await page.evaluate( """({x, y, pressed}) => { if (window.__moveCursorDot) { window.__moveCursorDot(x, y, pressed); } }""", {"x": x, "y": y, "pressed": pressed}, ) except Exception: pass async def _mouse_move(self, page, x, y): self.mouse_x = x self.mouse_y = y await page.mouse.move(x, y) await self._sync_cursor_dot(page, x, y) async def _mark_patchright_window(self, page): marker = self.xdotool_title_marker script = """marker => { window.__patchrightXdotoolMarker = marker; const applyMarker = () => { if (!document.title.includes(marker)) { document.title = `${marker} ${document.title || ''}`.trim(); } }; applyMarker(); if (!window.__patchrightXdotoolTitleTimer) { window.__patchrightXdotoolTitleTimer = setInterval(applyMarker, 500); } }""" try: await page.add_init_script(script, marker) except Exception: pass try: await page.evaluate(script, marker) except Exception: pass def _find_patchright_x_window(self): result = subprocess.run( ["xdotool", "search", "--name", self.xdotool_title_marker], capture_output=True, text=True, ) wids = [wid for wid in result.stdout.splitlines() if wid.strip()] return wids[-1] if wids else None async def _wait_patchright_x_window(self, page, timeout=3): try: await page.bring_to_front() except Exception: pass await self._mark_patchright_window(page) deadline = time.time() + timeout while time.time() < deadline: wid = self._find_patchright_x_window() if wid: return wid await asyncio.sleep(0.1) return None async def _page_to_screen_xy(self, page, viewport_x, viewport_y): window_id = await self._wait_patchright_x_window(page) metrics = await page.evaluate( """() => ({ screenX: window.screenX, screenY: window.screenY, outerWidth: window.outerWidth, outerHeight: window.outerHeight, innerWidth: window.innerWidth, innerHeight: window.innerHeight, dpr: window.devicePixelRatio || 1 })""" ) border_x = max(0, (metrics["outerWidth"] - metrics["innerWidth"]) / 2) toolbar_y = max(0, metrics["outerHeight"] - metrics["innerHeight"] - border_x) screen_x = int((metrics["screenX"] + border_x + viewport_x) * metrics["dpr"]) screen_y = int((metrics["screenY"] + toolbar_y + viewport_y) * metrics["dpr"]) try: cdp = await page.context.new_cdp_session(page) window_info = await cdp.send("Browser.getWindowForTarget") bounds_result = await cdp.send("Browser.getWindowBounds", {"windowId": window_info["windowId"]}) bounds = bounds_result.get("bounds", bounds_result) if "left" in bounds and "top" in bounds: screen_x = int((bounds["left"] + border_x + viewport_x) * metrics["dpr"]) screen_y = int((bounds["top"] + toolbar_y + viewport_y) * metrics["dpr"]) except Exception as exc: self.log(f"[xdotool] CDP窗口坐标获取失败: {exc!r}") if window_id is None: raise RuntimeError(f"未找到 patchright 窗口标题标记: {self.xdotool_title_marker}") return screen_x, screen_y, window_id def _xdotool_run(self, args, check=False): return subprocess.run(["xdotool", *args], capture_output=True, text=True, check=check) async def _xdotool_human_click(self, page, viewport_x, viewport_y): if not shutil.which("xdotool"): self.log("[xdotool] 未安装 xdotool，降级到 page.mouse") return False if not os.environ.get("DISPLAY"): self.log("[xdotool] DISPLAY 不存在，降级到 page.mouse") return False try: abs_x, abs_y, window_id = await self._page_to_screen_xy(page, viewport_x, viewport_y) self.log(f"[xdotool] 页面坐标=({viewport_x:.1f},{viewport_y:.1f}) 屏幕坐标=({abs_x},{abs_y}) window={window_id}") if window_id: self._xdotool_run(["windowactivate", str(window_id)]) await asyncio.sleep(random.uniform(0.15, 0.35)) cur = self._xdotool_run(["getmouselocation"]) mx = re.search(r"\bx:(-?\d+)\b", cur.stdout) my = re.search(r"\by:(-?\d+)\b", cur.stdout) start_x = int(mx.group(1)) if mx else abs_x + random.randint(120, 240) start_y = int(my.group(1)) if my else abs_y + random.randint(80, 180) self.log(f"[xdotool] 从屏幕坐标 ({start_x},{start_y}) 移动到 ({abs_x},{abs_y})") steps = random.randint(24, 36) for i in range(1, steps + 1): t = i / steps eased = t * t * (3 - 2 * t) jitter = max(0.2, 1.8 * (1 - t)) x = int(start_x + (abs_x - start_x) * eased + random.gauss(0, jitter)) y = int(start_y + (abs_y - start_y) * eased + random.gauss(0, jitter)) self._xdotool_run(["mousemove", str(x), str(y)]) await self._sync_cursor_dot(page, viewport_x, viewport_y) await asyncio.sleep(random.uniform(0.01, 0.028)) self._xdotool_run(["mousemove", "--sync", str(abs_x), str(abs_y)]) self.mouse_x = viewport_x self.mouse_y = viewport_y await self._sync_cursor_dot(page, viewport_x, viewport_y) for _ in range(random.randint(1, 3)): hover_x = abs_x + random.randint(-2, 2) hover_y = abs_y + random.randint(-2, 2) self._xdotool_run(["mousemove", "--sync", str(hover_x), str(hover_y)]) await asyncio.sleep(random.uniform(0.04, 0.12)) self._xdotool_run(["mousemove", "--sync", str(abs_x), str(abs_y)]) await asyncio.sleep(random.uniform(0.12, 0.28)) self._xdotool_run(["mousedown", "1"]) await self._sync_cursor_dot(page, viewport_x, viewport_y, pressed=True) await asyncio.sleep(random.uniform(0.08, 0.16)) self._xdotool_run(["mouseup", "1"]) await self._sync_cursor_dot(page, viewport_x, viewport_y, pressed=False) self.log("[xdotool] 系统级鼠标点击完成") return True except Exception as exc: self.log(f"[xdotool] 系统级点击失败，降级到 page.mouse: {exc!r}") return False async def _get_mouse_position(self, page, fallback_x=None, fallback_y=None): try: position = await page.evaluate( """({fallbackX, fallbackY}) => { const state = window.__cursorDotState || {}; return { x: Number.isFinite(state.x) ? state.x : fallbackX, y: Number.isFinite(state.y) ? state.y : fallbackY }; }""", {"fallbackX": fallback_x, "fallbackY": fallback_y}, ) except Exception: return fallback_x if fallback_x is not None else self.mouse_x, fallback_y if fallback_y is not None else self.mouse_y if position["x"] is None: position["x"] = fallback_x if fallback_x is not None else self.mouse_x if position["y"] is None: position["y"] = fallback_y if fallback_y is not None else self.mouse_y return position["x"], position["y"] async def _human_wander(self, page, target_x, target_y, start_x=None, start_y=None): """点击前模拟人类鼠标游走 2-5 秒，使行为更像真实用户""" import time as _time end = _time.time() + random.uniform(2, 5) # 随机游走持续 2-5 秒 if start_x is None or start_y is None: start_x, start_y = await self._get_mouse_position( page, target_x + random.uniform(-300, 300), target_y + random.uniform(-200, 200), ) cx = start_x cy = start_y await self._mouse_move(page, cx, cy) while _time.time() < end: nx = cx + random.uniform(-100, 100) # 每段随机目标点 ny = cy + random.uniform(-80, 80) steps = random.randint(5, 12) # 每段分多步移动 for i in range(steps): await self._mouse_move(page, cx + (nx - cx) * (i + 1) / steps + random.uniform(-2, 2), # 加微小抖动 cy + (ny - cy) * (i + 1) / steps + random.uniform(-2, 2)) await asyncio.sleep(random.uniform(0.02, 0.06)) cx, cy = nx, ny await asyncio.sleep(random.uniform(0.05, 0.2)) # 段间停顿 # 游走结束后平滑移向目标点 steps = random.randint(10, 20) for i in range(steps): await self._mouse_move(page, cx + (target_x - cx) * (i + 1) / steps + random.uniform(-1, 1), cy + (target_y - cy) * (i + 1) / steps + random.uniform(-1, 1)) await asyncio.sleep(random.uniform(0.02, 0.05)) async def _human_click(self, page, target_x, target_y, start_x=None, start_y=None): if await self._xdotool_human_click(page, target_x, target_y): return start_x, start_y = await self._get_mouse_position(page, start_x, start_y) await self._human_wander(page, target_x, target_y, start_x=start_x, start_y=start_y) await asyncio.sleep(random.uniform(0.1, 0.3)) await self._mouse_move(page, target_x, target_y) await self._sync_cursor_dot(page, target_x, target_y, pressed=True) await page.mouse.down() await asyncio.sleep(random.uniform(0.06, 0.16)) await page.mouse.up() await self._sync_cursor_dot(page, target_x, target_y, pressed=False) async def _find_clickable_checkbox(self, page, selectors, index, timeout): target = None used_selector = None for selector in selectors: try: locator = page.locator(selector).nth(index) await locator.wait_for(state="attached", timeout=timeout) count = await page.locator(selector).count() if count > index: target = locator used_selector = selector break except Exception: continue if target is None: return None, None, None try: await target.scroll_into_view_if_needed() except Exception: pass try: if used_selector == "input[type='checkbox']" and await target.is_checked(): return target, used_selector, "checked" except Exception: pass box = await target.bounding_box() if not box and used_selector == "input[type='checkbox']": try: label = target.locator("xpath=ancestor::label[1]") if await label.count() > 0: target = label box = await target.bounding_box() except Exception: pass return target, used_selector, box def _click_point_for_box(self, box, used_selector): if "iframe" in used_selector or "cloudflare" in used_selector or "managed" in used_selector: return ( box["x"] + random.uniform(20, min(35, max(20, box["width"] * 0.35))), box["y"] + box["height"] / 2 + random.uniform(-5, 5), ) return ( box["x"] + random.uniform(box["width"] * 0.3, box["width"] * 0.7), box["y"] + random.uniform(box["height"] * 0.3, box["height"] * 0.7), ) async def _find_cf_challenge_box(self, page): status = await self._collect_cf_status(page) text = status.get("text", "") looks_like_cf_page = ( "cloudflare" in text or "challenge" in text or "turnstile" in text or self._cf_waiting_phrase(text) or self._cf_success_phrase(text) ) if not looks_like_cf_page: self.log("[cf-click] 当前页面不像 Cloudflare 验证页，停止深度扫描") return None try: candidates = await page.evaluate( """() => { const out = []; const seen = new Set(); const keywords = [ 'cloudflare', 'challenge', 'turnstile', 'cf-turnstile', 'cf-challenge', 'verifying', 'verification', 'security verification', '正在验证', '安全验证' ]; const addNode = (el, source) => { if (!el || seen.has(el)) return; seen.add(el); const r = el.getBoundingClientRect(); if (!r || r.width < 20 || r.height < 20) return; const text = (el.innerText || el.textContent || '').toLowerCase(); const attrs = [ el.id || '', el.className || '', el.getAttribute && (el.getAttribute('src') || ''), el.getAttribute && (el.getAttribute('title') || ''), el.getAttribute && (el.getAttribute('name') || ''), el.getAttribute && (el.getAttribute('data-sitekey') || ''), el.getAttribute && (el.getAttribute('data-callback') || '') ].join(' ').toLowerCase(); const haystack = `${text} ${attrs}`; const tag = el.tagName; const keywordHit = keywords.some(k => haystack.includes(k)); const sizeHit = r.width >= 240 && r.width <= 420 && r.height >= 40 && r.height <= 120; const smallKeywordBox = keywordHit && r.width <= 520 && r.height <= 220; const isFrame = tag === 'IFRAME'; const viewportArea = Math.max(1, window.innerWidth * window.innerHeight); const areaRatio = (r.width * r.height) / viewportArea; if (tag === 'HTML' || tag === 'BODY') return; if (/^H[1-6]$/.test(tag)) return; if (haystack.includes('footer-wrapper') || haystack.includes('ray id:') || haystack.includes('performance and security by cloudflare')) return; if (!isFrame && areaRatio > 0.35) return; const inChallengeBand = r.y > 180 && r.y < window.innerHeight * 0.78; if (!isFrame && sizeHit && !keywordHit && !inChallengeBand) return; if (!isFrame && smallKeywordBox && !inChallengeBand) return; if (!isFrame && !sizeHit && !smallKeywordBox) return; out.push({ source, x: r.x, y: r.y, width: r.width, height: r.height, text: text.slice(0, 120), attrs: attrs.slice(0, 160), score: (sizeHit ? 30 : 0) + (isFrame ? 20 : 0) + (keywordHit ? 10 : 0) + (smallKeywordBox ? 8 : 0) }); }; const walk = root => { if (!root) return; const nodes = root.querySelectorAll ? root.querySelectorAll('*') : []; for (const el of nodes) { addNode(el, el.tagName.toLowerCase()); if (el.shadowRoot) walk(el.shadowRoot); } }; walk(document); return out .filter(item => item.x >= 0 && item.y >= 0) .sort((a, b) => b.score - a.score || (a.y - b.y)) .slice(0, 10); }""" ) except Exception as exc: self.log(f"[cf-click] JS深度扫描失败: {exc!r}") return None if candidates: self.log(f"[cf-click] 深度扫描候选: {candidates[:3]}") viewport = await page.evaluate("() => ({w: window.innerWidth, h: window.innerHeight})") best = None for item in candidates: text = (item.get("text") or "").lower() attrs = (item.get("attrs") or "").lower() source = (item.get("source") or "").lower() if source in {"h1", "h2", "h3", "h4", "h5", "h6"}: continue if "ray id:" in text or "footer-wrapper" in attrs or "performance and security by cloudflare" in text: continue if item["width"] * item["height"] > viewport["w"] * viewport["h"] * 0.35: continue best = item break if best is None: self.log("[cf-click] 深度扫描没有可点击的小方框候选") return await self._fallback_cf_challenge_box(page) if best["width"] * best["height"] > viewport["w"] * viewport["h"] * 0.35: self.log(f"[cf-click] 深度扫描候选过大，跳过: {best}") return None return best["source"], {"x": best["x"], "y": best["y"], "width": best["width"], "height": best["height"]} fallback = await self._fallback_cf_challenge_box(page) if fallback: return fallback return None async def _fallback_cf_challenge_box(self, page): try: info = await page.evaluate( """() => { const text = (document.body && (document.body.innerText || document.body.textContent) || '').toLowerCase(); return {w: window.innerWidth, h: window.innerHeight, text}; }""" ) except Exception: return None text = info.get("text", "") is_cf_page = ( ("performing security verification" in text or "security verification" in text or "正在进行安全验证" in text) and ("cloudflare" in text or "verifying" in text or "正在验证" in text) ) if not is_cf_page: return None width = min(302, max(260, info["w"] - 48)) height = 67 if info["w"] <= 1200: box = {"x": 24, "y": 304, "width": width, "height": height} else: box = {"x": max(24, (info["w"] - 896) / 2), "y": 280, "width": width, "height": height} self.log(f"[cf-click] 使用布局兜底方框: {box}") return "cloudflare-layout-fallback", box async def _collect_cf_status(self, page): try: return await page.evaluate( """() => { const parts = []; const addText = value => { if (value && typeof value === 'string') parts.push(value); }; const walk = root => { if (!root) return; addText(root.innerText || root.textContent || ''); const nodes = root.querySelectorAll ? root.querySelectorAll('*') : []; for (const el of nodes) { const id = el.id || ''; const cls = typeof el.className === 'string' ? el.className : ''; const aria = el.getAttribute && (el.getAttribute('aria-label') || ''); const title = el.getAttribute && (el.getAttribute('title') || ''); addText(`${id} ${cls} ${aria} ${title}`); if (el.shadowRoot) walk(el.shadowRoot); } }; walk(document.body || document.documentElement); let observedSuccess = null; try { observedSuccess = window.__cfSuccessTextSeen || JSON.parse(sessionStorage.getItem('__cf_success_seen') || 'null'); } catch (_) {} return { url: location.href, title: document.title || '', observedSuccess, text: parts.join('\\n').replace(/\\s+/g, ' ').trim().toLowerCase().slice(0, 12000) }; }""" ) except Exception as exc: return {"url": page.url, "title": "", "text": "", "error": repr(exc)} def _cf_success_phrase(self, text): success_phrases = [ "verification successful", "successfully verified", "successful verification", "验证成功", "已验证", "安全验证成功", "验证已成功", ] for phrase in success_phrases: if phrase in text: return phrase return None def _cf_waiting_phrase(self, text): waiting_phrases = [ "verifying", "checking", "performing security verification", "security verification", "please wait", "正在验证", "正在进行安全验证", "请稍候", ] for phrase in waiting_phrases: if phrase in text: return phrase return None def _cf_slow_verification_phrase(self, text): slow_phrases = [ "为什么验证时间较长", "验证时间较长", "电脑配置较旧", "网络连接较慢", "请稍等片刻", "刷新此页面", "故障排除文档", "why is verification taking so long", "verification is taking longer", "taking longer than expected", "refresh this page", ] for phrase in slow_phrases: if phrase in text: return phrase return None async def _is_cf_slow_verification_page(self, page): status = await self._collect_cf_status(page) text = status.get("text", "") phrase = self._cf_slow_verification_phrase(text) if phrase: return phrase return None async def _recover_cf_slow_verification(self, page, url, max_refreshes=2): for attempt in range(1, max_refreshes + 1): phrase = await self._is_cf_slow_verification_page(page) if not phrase: return True self.log(f"[cf-slow] 检测到验证耗时过长页面: {phrase}，第 {attempt}/{max_refreshes} 次重新加载验证页") try: await page.goto("about:blank", wait_until="domcontentloaded", timeout=10000) except Exception: pass await asyncio.sleep(random.uniform(1.5, 3.0)) await self.goto_with_retries(page, url, attempts=2) await asyncio.sleep(random.uniform(1.5, 3.0)) phrase = await self._is_cf_slow_verification_page(page) if phrase: self.log(f"[cf-slow] 多次刷新后仍停留在验证耗时过长页面: {phrase}") return False return True async def _is_business_page_loaded(self, page): try: status = await self._collect_cf_status(page) text = status.get("text", "") title = (status.get("title") or "").lower() url = status.get("url") or page.url if self._cf_slow_verification_phrase(text): return False if self._cf_waiting_phrase(text) or "performing security verification" in text: return False has_business_dom = await page.evaluate( """() => Boolean( document.querySelector('.race-item, .next-race-slider, .raceheaderbar, .betslipContainer, .betting-slip') )""" ) has_business_text = ( "mbet sport" in title or "ticket de pari" in text or "ticket de paris" in text or "paris ouverts" in text or "race-item" in text or "pur-sang" in text ) has_business_url = "/Home/raceMenu" in url or "/home/racemenu" in url.lower() return bool(has_business_dom or (has_business_url and has_business_text)) except Exception as exc: self.log(f"[business] 业务页面检测失败: {exc!r}") return False async def _wait_for_cf_success_text(self, page, timeout=18, poll_interval=0.4): deadline = time.time() + timeout last_waiting_phrase = None last_url = page.url while time.time() < deadline: status = await self._collect_cf_status(page) observed_success = status.get("observedSuccess") if observed_success: phrase = observed_success.get("phrase", "observed") self.cf_success_seen = True self.log(f"[cf-click] 捕获到验证成功文字: {phrase}") return True text = status.get("text", "") slow_phrase = self._cf_slow_verification_phrase(text) if slow_phrase: self.log(f"[cf-click] 检测到验证耗时过长页面: {slow_phrase}") return False success_phrase = self._cf_success_phrase(text) if success_phrase: self.cf_success_seen = True self.log(f"[cf-click] 发现验证成功文字: {success_phrase}") return True waiting_phrase = self._cf_waiting_phrase(text) if waiting_phrase and waiting_phrase != last_waiting_phrase: last_waiting_phrase = waiting_phrase self.log(f"[cf-click] 等待验证状态变化: {waiting_phrase}") if status.get("url") != last_url: last_url = status.get("url") self.log(f"[cf-click] 页面URL变化，继续等待成功文字: {last_url}") await asyncio.sleep(poll_interval + random.uniform(0, 0.25)) return False async def _wait_after_cf_click(self, page, timeout=24, require_success_text=False): deadline = time.time() + timeout while time.time() < deadline: slow_phrase = await self._is_cf_slow_verification_page(page) if slow_phrase: self.log(f"[cf-click] 点击后进入验证耗时过长页面: {slow_phrase}") return False if await self._wait_for_cf_success_text(page, timeout=0.8): return True if await self._is_business_page_loaded(page): self.log("[cf-click] 已跳转到 MBet 业务页面，点击判定成功") return True await asyncio.sleep(random.uniform(0.4, 0.8)) still_blocked = await detect_cloudflare_challenge(page, "interstitial") if not still_blocked and not require_success_text: self.log("[cf-click] 挑战已消失，但没有观察到验证成功文字") return True if not still_blocked: self.log("[cf-click] 挑战已消失，但没有出现验证成功文字，本轮不算成功") else: self.log("[cf-click] 点击后仍在验证页，本轮不算成功") return False async def _click_cf_box_when_visible(self, page, timeout=55, poll_interval=0.5, click_attempts=3, require_success_text=False, recover_url=None): selectors = [ "input[type='checkbox']", "[role='checkbox']", 'iframe[src*="challenges.cloudflare.com"]', 'iframe[title*="Cloudflare"]', 'iframe[title*="challenge"]', ".cf-turnstile", "[data-sitekey]", ] deadline = time.time() + timeout for attempt in range(1, click_attempts + 1): self.log(f"[cf-click] 第 {attempt}/{click_attempts} 次检测点击方框") clicked_this_attempt = False while time.time() < deadline: slow_phrase = await self._is_cf_slow_verification_page(page) if slow_phrase: if recover_url and await self._recover_cf_slow_verification(page, recover_url, max_refreshes=2): clicked_this_attempt = False continue self.log(f"[cf-click] 验证耗时过长页面无法恢复: {slow_phrase}") return False if self.cf_success_seen or await self._wait_for_cf_success_text(page, timeout=1): return True if await self._is_business_page_loaded(page): self.log("[cf-click] 已在 MBet 业务页面，停止继续点击") return True if not await detect_cloudflare_challenge(page, "interstitial"): self.log("[cf-click] Cloudflare 挑战已消失，未看到成功文字，停止继续点击") return False for selector in selectors: try: locator = page.locator(selector).first if await page.locator(selector).count() <= 0: continue box = await locator.bounding_box() if not box or box["width"] < 10 or box["height"] < 10: continue target_x, target_y = self._click_point_for_box(box, selector) start_x, start_y = await self._get_mouse_position(page) self.log( f"[cf-click] 方框已出现 selector={selector}, box={box}, " f"鼠标当前位置=({start_x}, {start_y}), 点击=({target_x:.1f}, {target_y:.1f})" ) await self._human_click(page, target_x, target_y, start_x=start_x, start_y=start_y) clicked_this_attempt = True if await self._wait_after_cf_click(page, timeout=random.uniform(18, 26), require_success_text=require_success_text): return True if not await detect_cloudflare_challenge(page, "interstitial"): return False break except Exception: continue if not clicked_this_attempt: found = await self._find_cf_challenge_box(page) if found: source, box = found target_x, target_y = self._click_point_for_box(box, source) start_x, start_y = await self._get_mouse_position(page) self.log( f"[cf-click] 深度扫描找到方框 source={source}, box={box}, " f"鼠标当前位置=({start_x}, {start_y}), 点击=({target_x:.1f}, {target_y:.1f})" ) await self._human_click(page, target_x, target_y, start_x=start_x, start_y=start_y) clicked_this_attempt = True if await self._wait_after_cf_click(page, timeout=random.uniform(18, 26), require_success_text=require_success_text): return True if not await detect_cloudflare_challenge(page, "interstitial"): return False if clicked_this_attempt: break await asyncio.sleep(poll_interval) if attempt < click_attempts: await asyncio.sleep(random.uniform(2, 4)) self.log("[cf-click] 超时未点击通过") return False async def optional_random_click_checkbox(self, page, selectors=None, timeout=8000, index=0, click_if_exists=True, click_attempts=3, retry_delay=3, require_success_text=False, recover_url=None): """ 可选点击 checkbox： - 等待 checkbox 出现并获取坐标 - 每次点击前重新读取当前鼠标位置和目标位置 - 失败或仍在挑战页时等待后重试 """ if selectors is None: # 依次尝试的 checkbox 选择器列表 selectors = [ "input[type='checkbox']", "label:has(input[type='checkbox'])", "[role='checkbox']", ".checkbox", ".check-box", 'iframe[src*="challenges.cloudflare.com"]', 'iframe[title*="Cloudflare"]', 'iframe[title*="challenge"]', ".cf-turnstile", "[data-sitekey]", ] if not click_if_exists: return True for attempt in range(1, click_attempts + 1): slow_phrase = await self._is_cf_slow_verification_page(page) if slow_phrase: if recover_url and await self._recover_cf_slow_verification(page, recover_url, max_refreshes=2): continue print(f"验证耗时过长页面无法恢复: {slow_phrase}") return False if self.cf_success_seen or await self._wait_for_cf_success_text(page, timeout=1): return True if await self._is_business_page_loaded(page): print("已在 MBet 业务页面，停止继续点击") return True if not await detect_cloudflare_challenge(page, "interstitial"): print("Cloudflare 挑战已消失，未看到成功文字，停止继续点击") return False target, used_selector, box = await self._find_clickable_checkbox(page, selectors, index, timeout) if not target: found = await self._find_cf_challenge_box(page) if found: used_selector, box = found target = True if box == "checked": print("checkbox 已经选中，不需要点击") return True if not target: print(f"第 {attempt}/{click_attempts} 次没有发现 checkbox，等待后重试") elif not box: print(f"第 {attempt}/{click_attempts} 次 checkbox 没有可点击坐标，等待后重试") else: print(f"第 {attempt}/{click_attempts} 次发现 checkbox，使用选择器: {used_selector}") target_x, target_y = self._click_point_for_box(box, used_selector) start_x, start_y = await self._get_mouse_position(page) print(f"鼠标当前位置: ({start_x}, {start_y}), 准备点击: ({target_x:.1f}, {target_y:.1f})") await self._human_click(page, target_x, target_y, start_x=start_x, start_y=start_y) if await self._wait_after_cf_click(page, timeout=random.uniform(14, 22), require_success_text=require_success_text): return True if not await detect_cloudflare_challenge(page, "interstitial"): return False print(f"第 {attempt}/{click_attempts} 次点击后仍在验证页，等待后重试") if attempt < click_attempts: await asyncio.sleep(retry_delay + random.uniform(0.5, 1.5)) print("checkbox 多次点击后仍未通过") return False def build_proxy_config(self, DEFAULT_PROXY): server = DEFAULT_PROXY["server"].strip() username = DEFAULT_PROXY["username"].strip() password = DEFAULT_PROXY["password"].strip() if not server: return None proxy = {"server": server} if username: proxy["username"] = username if password: proxy["password"] = password return proxy def proxy_to_curl_url(self, proxy): if not proxy: return None server = proxy["server"] if "://" not in server: server = f"http://{server}" scheme, rest = server.split("://", 1) username = proxy.get("username") password = proxy.get("password") if username and password and "@" not in rest: return f"{scheme}://{username}:{password}@{rest}" return f"{scheme}://{rest}" def is_navigation_network_error(self, exc): message = f"{type(exc).__name__}: {exc!s}" network_errors = [ "ERR_EMPTY_RESPONSE", "ERR_TUNNEL_CONNECTION_FAILED", "ERR_PROXY_CONNECTION_FAILED", "ERR_CONNECTION_CLOSED", "ERR_CONNECTION_RESET", "ERR_TIMED_OUT", ] return any(item in message for item in network_errors) async def goto_with_retries(self, page, url, attempts=3): last_exc = None for attempt in range(1, attempts + 1): try: return await page.goto(url, wait_until="domcontentloaded", timeout=60000) except Exception as exc: last_exc = exc if not self.is_navigation_network_error(exc) or attempt >= attempts: raise self.log(f"[goto] 第 {attempt}/{attempts} 次访问失败: {exc!r}，清空页面后重试") try: await page.goto("about:blank", wait_until="domcontentloaded", timeout=10000) except Exception: pass await asyncio.sleep(2 * attempt + random.uniform(0.5, 1.5)) raise last_exc def ensure_virtual_display(self,): if os.environ.get("DISPLAY"): self.log(f"[display] using existing DISPLAY={os.environ['DISPLAY']}") return None try: from pyvirtualdisplay import Display display = Display(visible=False, size=(1920, 1080)) display.start() self.log(f"[display] started pyvirtualdisplay DISPLAY={os.environ.get('DISPLAY')}") return display except ImportError: self.log("[display] pyvirtualdisplay not installed, starting Xvfb :99 directly") xvfb_cmd = ["Xvfb", ":99", "-screen", "0", "1920x1080x24"] proc = subprocess.Popen(xvfb_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) os.environ["DISPLAY"] = ":99" time.sleep(1) self.log("[display] started Xvfb DISPLAY=:99") return proc def stop_virtual_display(self, display): if display is None: return try: if hasattr(display, "stop"): display.stop() elif isinstance(display, subprocess.Popen) and display.poll() is None: display.terminate() display.wait(timeout=5) except Exception as exc: self.log(f"[display] cleanup warning: {exc!r}") def chrome_path_for(self, major): path_str = f"{BASE_DIR}/chrome_config/chrome{major}/chrome-linux64/chrome" path = Path(path_str) if not path.exists(): raise FileNotFoundError(f"Chrome executable not found: {path}") return str(path) def browser_ua(self, major): dd = [ f"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{major}.0.0.0 Safari/537.36", # f"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{major}.0.0.0 Safari/537.36 Edg/{major}.0.0.0" f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{major}.0.0.0 Safari/537.36" ] return random.choice(dd) def save_cookie_result(self, URL, major, ua, proxy, cookies, verify_status=None): cookie_header = "; ".join(f"{c['name']}={c['value']}" for c in cookies) data = {"saved_at": int(time.time()), "url": URL, "chrome_major": str(major), "user_agent": ua, "proxy_server": proxy["server"] if proxy else None, "verify_status": verify_status, "cookie_header": cookie_header, "cookies": cookies, } COOKIE_OUTPUT.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") self.log(f"[cookie] saved to {COOKIE_OUTPUT}") return cookie_header async def save_cf_failure_artifacts(self, page, chrome_versions, reason): prefix = BASE_DIR / f"mbet_cf_failed_chrome{chrome_versions}" screenshot_path = prefix.with_suffix(".png") html_path = prefix.with_suffix(".html") json_path = prefix.with_suffix(".json") status = await self._collect_cf_status(page) status["reason"] = reason status["success_seen"] = self.cf_success_seen try: await page.screenshot(path=str(screenshot_path), full_page=True) except Exception as exc: status["screenshot_error"] = repr(exc) try: html = await page.content() html_path.write_text(html, encoding="utf-8") except Exception as exc: status["html_error"] = repr(exc) try: json_path.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8") except Exception: pass self.log(f"[debug] failure artifacts: screenshot={screenshot_path}, html={html_path}, json={json_path}") return screenshot_path # todo 模拟鼠标控制 async def inject_cf_success_observer(self, page): script = """(() => { const phrases = [ 'verification successful', 'successfully verified', 'successful verification', '验证成功', '已验证', '安全验证成功', '验证已成功' ]; const collectText = () => { const parts = []; const walk = root => { if (!root) return; if (root.innerText || root.textContent) { parts.push(root.innerText || root.textContent || ''); } const nodes = root.querySelectorAll ? root.querySelectorAll('*') : []; for (const el of nodes) { if (el.shadowRoot) walk(el.shadowRoot); } }; walk(document.body || document.documentElement); return parts.join(' ').replace(/\\s+/g, ' ').toLowerCase(); }; const markIfSuccess = () => { const text = collectText(); const phrase = phrases.find(item => text.includes(item)); if (!phrase) return false; const payload = {phrase, at: Date.now(), url: location.href}; window.__cfSuccessTextSeen = payload; try { sessionStorage.setItem('__cf_success_seen', JSON.stringify(payload)); } catch (_) {} return true; }; if (window.__cfSuccessObserverInstalled) { markIfSuccess(); return; } window.__cfSuccessObserverInstalled = true; markIfSuccess(); const observer = new MutationObserver(markIfSuccess); observer.observe(document.documentElement, { childList: true, subtree: true, characterData: true, attributes: true, attributeFilter: ['class', 'style', 'aria-label', 'title'] }); window.__cfSuccessObserver = observer; window.__cfSuccessInterval = setInterval(markIfSuccess, 100); })()""" try: await page.add_init_script(script) except Exception: pass try: await page.evaluate(script) except Exception: pass async def inject_cursor_dot(self, page): await page.evaluate("""() => { let dot = document.getElementById('__cursor__'); if (!dot) { dot = document.createElement('div'); dot.id = '__cursor__'; dot.style.cssText = [ 'position:fixed', 'left:0', 'top:0', 'width:12px', 'height:12px', 'background:red', 'border:2px solid rgba(255,255,255,.95)', 'border-radius:50%', 'pointer-events:none', 'z-index:2147483647', 'transform:translate3d(-100px,-100px,0) translate(-50%,-50%)', 'transition:transform 45ms linear, width 80ms linear, height 80ms linear, opacity 80ms linear', 'box-shadow:0 0 0 4px rgba(255,0,0,.20)', 'opacity:.95' ].join(';'); document.body.appendChild(dot); } window.__cursorDotState = window.__cursorDotState || {x: null, y: null, pressed: false}; window.__moveCursorDot = (x, y, pressed = false) => { window.__cursorDotState = {x, y, pressed}; dot.style.transform = `translate3d(${x}px, ${y}px, 0) translate(-50%, -50%)`; dot.style.width = pressed ? '16px' : '12px'; dot.style.height = pressed ? '16px' : '12px'; dot.style.opacity = pressed ? '1' : '.95'; }; document.addEventListener('mousemove', e => { window.__moveCursorDot(e.clientX, e.clientY, false); }, true); document.addEventListener('mousedown', e => { window.__moveCursorDot(e.clientX, e.clientY, true); }, true); document.addEventListener('mouseup', e => { window.__moveCursorDot(e.clientX, e.clientY, false); }, true); document.addEventListener('mouseleave', () => { dot.style.opacity = '.35'; }); }""") # todo 启动是以代理和不同浏览器版本进行获取cf async def fetch_cookie_once(self, URL, chrome_versions, DEFAULT_PROXY, verify=True, require_success_text=False): self.cf_success_seen = False ua = self.browser_ua(chrome_versions) browser_path = self.chrome_path_for(chrome_versions) proxy = self.build_proxy_config(DEFAULT_PROXY) browser = None context = None launch_args = [ "--start-maximized", "--disable-blink-features=AutomationControlled", f"--user-agent={ua}", ] self.log(f"使用代理: {proxy['server'] if proxy else 'direct'}, 使用 浏览器版本: {ua}") async with async_playwright() as playwright: try: browser = await playwright.chromium.launch(executable_path=browser_path, headless=False, proxy=proxy, args=launch_args, ) context = await browser.new_context(no_viewport=True, user_agent=ua) page = await context.new_page() await self._mark_patchright_window(page) self.log(f"开始访问需要获取cf参数的网站") response = await self.goto_with_retries(page, URL, attempts=3) await self._mark_patchright_window(page) self.log(f"status: {response.status if response else None}, url: {page.url}") if not await self._recover_cf_slow_verification(page, URL, max_refreshes=2): screenshot_path = await self.save_cf_failure_artifacts( page, chrome_versions, "cloudflare slow verification page", ) raise RuntimeError(f"cloudflare slow verification page; screenshot={screenshot_path}") if response: headers = await response.request.all_headers() self.log(f"http ua: {headers.get('user-agent', ua)}") await asyncio.sleep(1) await self.inject_cursor_dot(page) viewport = await page.evaluate("() => ({w: window.innerWidth, h: window.innerHeight})") self.log(f"窗口大小: {viewport}") await self._human_wander(page, viewport["w"] / 2, viewport["h"] / 2) # 模拟鼠标 solved = False challenge_present = await detect_cloudflare_challenge(page, "interstitial") if challenge_present: solved = await self._click_cf_box_when_visible( page, timeout=55, poll_interval=0.5, click_attempts=3, require_success_text=require_success_text, recover_url=URL, ) else: solved = True if challenge_present and not solved: solved = await solve_captcha( page, captcha_type="cloudflare", challenge_type="interstitial", solve_attempts=1, wait_checkbox_attempts=4, wait_checkbox_delay=2, checkbox_click_attempts=2, ) if challenge_present and solved and require_success_text and not self.cf_success_seen: self.log("[challenge] 已点击/检测通过，但没有捕获验证成功文字，继续等待确认") solved = await self._wait_for_cf_success_text(page, timeout=12) if challenge_present and not solved: self.log(f"[challenge] automatic solve failed, trying optional checkbox click{solved}") solved = await self.optional_random_click_checkbox( page, timeout=1200, click_attempts=2, retry_delay=5, require_success_text=require_success_text, recover_url=URL, ) if require_success_text and not self.cf_success_seen: solved = False if not solved: screenshot_path = await self.save_cf_failure_artifacts( page, chrome_versions, "challenge not solved with required success text", ) raise RuntimeError(f"challenge still blocked or success text missing; screenshot={screenshot_path}") self.log("[challenge] solved after optional click and success text") else: if challenge_present: self.log("[challenge] solved, 已进入业务页面或捕获验证成功") else: self.log("[challenge] no challenge detected, 成功获取") cookies = await context.cookies() cookie_header = "; ".join(f"{c['name']}={c['value']}" for c in cookies) self.log(cookie_header) self.save_cookie_result(URL, chrome_versions, ua, proxy, cookies, verify_status=None) if verify: proxy_url = self.proxy_to_curl_url(proxy) # 调用from curl_cffi.requests import AsyncSession 脱离浏览器进行请求验证获取的cf是否正常 if await self.fetch_with_proxy(URL, {"User-Agent": ua, "Cookie": cookie_header}, proxy_url, impersonate="chrome"): print("脱离浏览器进行请求验证获取的cookie的cf参数正常") await asyncio.sleep(3) return True finally: if context: await context.close() if browser: await browser.close() # todo 是以异步进行请求验证参数是否正常 async def fetch_with_proxy(self, URL, headers, PROXY_URL, impersonate="chrome"): # 模拟Chrome126浏览器指纹，和Windows端浏览器行为一致 async with AsyncSession(impersonate=impersonate) as session: # 带代理请求 resp = await session.get(url=URL, headers=headers, proxy=PROXY_URL, timeout=15) self.log(f"验证获取的cf参数进行异步请求: status: {resp.status_code}") if resp.status_code == 200: # 验证正常就返回True return True # todo 启动自动化重试获取cookie def default_proxy_configs(self): return [ # {"server": "http://192.147.177.5:51523", "username": "lee047149", "password": "QVuDHPioJf"}, # {"server": "http://13.140.130.146:8400", "username": "pdmy", "password": "123456"}, # {"server": "http://148.66.51.30:8400", "username": "pdmy", "password": "123456"}, {"server": "http://148.66.51.29:8400", "username": "pdmy", "password": "123456"}, ] async def start_get_cookie(self, URL, chrome_versions, retries, verify, require_success_text=False): for proxy_index, DEFAULT_PROXY in enumerate(self.default_proxy_configs(), start=1): self.log(f"[proxy] chrome={chrome_versions} 使用代理 {proxy_index}: {DEFAULT_PROXY['server']}") for attempt in range(1, retries + 1): # 重试次数 try: self.log(f"\n[run] chrome={chrome_versions} proxy={proxy_index} 开始attempt={attempt}/{retries}") ok = await self.fetch_cookie_once( URL, chrome_versions, DEFAULT_PROXY, verify=verify, require_success_text=require_success_text, ) print(ok) if ok: self.log(f"[run] chrome={chrome_versions} proxy={proxy_index} 获取第{attempt}次成功获取结束") self.log(f"\n===============================================================================================================") break except Exception as exc: self.log(f"[error] chrome={chrome_versions} proxy={proxy_index} attempt={attempt}: {exc!r}") if self.is_navigation_network_error(exc): self.log(f"[run] chrome={chrome_versions} proxy={proxy_index} 访问阶段网络错误，切换下一个代理") break if attempt < retries: await asyncio.sleep(5) self.log(f"[run] chrome={chrome_versions} 所有代理均未成功，切换下一个浏览器版本") return def available_chrome_versions(self): versions = [] chrome_root = BASE_DIR / "chrome_config" for path in chrome_root.glob("chrome*"): match = re.fullmatch(r"chrome(\d+)", path.name) if not match: continue chrome_path = path / "chrome-linux64" / "chrome" if chrome_path.exists(): versions.append(int(match.group(1))) return sorted(set(versions)) def parse_chrome_versions_arg(self, value): if not value: return None versions = [] for item in value.split(","): item = item.strip() if not item: continue versions.append(int(item)) return versions def log_environment_check(self, chrome_versions_list): self.log( "[env] " f"DISPLAY={os.environ.get('DISPLAY')}, " f"xdotool={shutil.which('xdotool')}, " f"Xvfb={shutil.which('Xvfb')}, " f"chrome_versions={chrome_versions_list}" ) # 循环浏览器版本进行获取cf参数 async def run(self,retries, force_xvfb, no_xvfb,no_verify, success_text): # chrome_versions_list = [113, 114, 115, 116, 117, 118, 119, 120, 121, 122, 123, 124, 125, 126, 127, 128, 129, 130, 131, 132, 133, 134, 135, 136, 137, 138, 139, 140, 141, 142, 143, 144, 145, 146, 147, 148, 149, 150] chrome_versions_list = [118] self.log_environment_check(chrome_versions_list) for chrome_versions in chrome_versions_list: for host, redis_config in self.redis_config_dict.items(): url = redis_config.get("url") display = None original_display = os.environ.get("DISPLAY") try: if force_xvfb: os.environ.pop("DISPLAY", None) if not no_xvfb: display = self.ensure_virtual_display() await self.start_get_cookie( url, chrome_versions, retries, verify=not no_verify, require_success_text=success_text, ) finally: self.stop_virtual_display(display) if force_xvfb: if original_display is None: os.environ.pop("DISPLAY", None) else: os.environ["DISPLAY"] = original_display def main(self, ): asyncio.run(self.run(3,False, False, False, False)) if __name__ == "__main__": BrowserPageChecker().main()