coordinate/utils/cloudflare_bypass_for_scraping2.py

import random
import time
from typing import Any


class CloudflareBypasser:
    def __init__(self, driver: Any, log=True):
        self.driver = driver
        self.log = log
        
    def log_message(self, message):
        if self.log:
            print(message)
    
    def _normalize_page(self):
        # 兼容 TlsPlugin 中的 CamoufoxPageAdapter
        return getattr(self.driver, "_page", self.driver)

    def _is_challenge_frame(self, frame) -> bool:
        frame_name = (frame.name or "").lower()
        frame_url = (frame.url or "").lower()
        markers = (
            "turnstile",
            "challenges.cloudflare.com",
            "challenge",
            "cf-chl",
        )
        return any(m in frame_name or m in frame_url for m in markers)

    def _determine_challenge_type(self) -> str:
        try:
            page = self._normalize_page()
            title = (page.title() or "").lower()
            html = (page.content() or "").lower()
            if "please complete the captcha" in html or "turnstile" in html:
                return "turnstile"
            if "just a moment" in title or "checking your browser" in html:
                return "interstitial"
            return "none"
        except Exception as e:
            self.log_message(f"Error determining challenge type: {e}")
            return "unknown"

    def _click_checkbox_in_frame(self, frame) -> bool:
        selectors = [
            "input[type='checkbox']",
            "[role='checkbox']",
            "label.ctp-checkbox-label",
            "div.ctp-checkbox-label",
            "label[for*='cf']",
        ]
        for selector in selectors:
            try:
                loc = frame.locator(selector)
                if loc.count() <= 0:
                    continue
                target = loc.first
                target.click(timeout=2000)
                return True
            except Exception:
                continue

        # 在 frame 内做 open-shadow 递归查找（closed shadow 无法直接访问）
        try:
            clicked = frame.evaluate(
                """
                () => {
                    const selectors = [
                        "input[type='checkbox']",
                        "[role='checkbox']",
                        "label.ctp-checkbox-label",
                        "div.ctp-checkbox-label",
                        "label[for*='cf']"
                    ];
                    const seen = new WeakSet();
                    const stack = [document];

                    while (stack.length) {
                        const root = stack.pop();
                        if (!root || seen.has(root)) continue;
                        seen.add(root);

                        for (const sel of selectors) {
                            const hit = root.querySelector(sel);
                            if (hit) {
                                hit.click();
                                return true;
                            }
                        }

                        const nodes = root.querySelectorAll ? root.querySelectorAll("*") : [];
                        for (const node of nodes) {
                            if (node.shadowRoot) stack.push(node.shadowRoot);
                        }
                    }
                    return false;
                }
                """
            )
            return bool(clicked)
        except Exception:
            return False

    def _click_challenge_iframe_center(self) -> bool:
        page = self._normalize_page()
        for frame in page.frames:
            if not self._is_challenge_frame(frame):
                continue

            try:
                frame_el = frame.frame_element()
                box = frame_el.bounding_box()
                if not box:
                    continue

                # 人类化一点：点击中心附近随机偏移，避免固定坐标
                cx = box["x"] + box["width"] * (0.5 + random.uniform(-0.08, 0.08))
                cy = box["y"] + box["height"] * (0.5 + random.uniform(-0.08, 0.08))
                page.mouse.move(cx, cy, steps=10)
                time.sleep(random.uniform(0.15, 0.45))
                page.mouse.click(cx, cy, delay=random.randint(50, 180))
                return True
            except Exception:
                continue
        return False
    
    def click_verification_button(self, _is_dfs=False):
        try:
            page = self._normalize_page()
            for frame in page.frames:
                if not self._is_challenge_frame(frame):
                    continue
                if self._click_checkbox_in_frame(frame):
                    self.log_message("Challenge interaction succeeded by frame selector/evaluate.")
                    time.sleep(1)
                    return

            if self._click_challenge_iframe_center():
                self.log_message("Challenge interaction succeeded by iframe center click.")
                time.sleep(1)
                return

            self.log_message("Challenge click strategies exhausted.")

        except Exception as e:
            self.log_message(f"Error clicking verification button: {e}")

    def is_bypassed(self):
        try:
            page = self._normalize_page()
            title = (page.title() or "").lower()
            html = (page.content() or "").lower()
            blocked_markers = (
                "just a moment",
                "请稍候",
                "checking your browser",
                "cf-challenge",
                "please complete the captcha",
            )
            return not any(m in title or m in html for m in blocked_markers)
        except Exception as e:
            self.log_message(f"Error checking page title: {e}")
        return False

    def _collect_page_state(self) -> str:
        """
        采样当前页面状态，帮助定位卡在哪一轮挑战。
        """
        try:
            page = self._normalize_page()
            title = page.title()
            url = page.url
            challenge_type = self._determine_challenge_type()
            challenge_frames = 0
            for frame in page.frames:
                if self._is_challenge_frame(frame):
                    challenge_frames += 1
            return (
                f"title={title!r}, url={url!r}, challenge_type={challenge_type}, "
                f"challenge_frames={challenge_frames}"
            )
        except Exception as e:
            return f"state_collect_error={e}"

    def _collect_state_signature(self):
        try:
            page = self._normalize_page()
            title = (page.title() or "").lower()
            url = (page.url or "").lower()
            challenge_type = self._determine_challenge_type()
            challenge_frames = 0
            for frame in page.frames:
                if self._is_challenge_frame(frame):
                    challenge_frames += 1
            return (challenge_type, challenge_frames, title[:80], url[:120])
        except Exception:
            return ("unknown", -1, "", "")

    def bypass(self, max_retry=5):
        for i in range(max_retry):
            if self.is_bypassed():
                return True

            sig_before = self._collect_state_signature()
            state_before = self._collect_page_state()
            self.log_message(
                f"Verification page detected. try={i + 1}/{max_retry}, before_click: {state_before}"
            )
            self.click_verification_button(False)

            # 点击后短暂等待，再次检查是否通过
            time.sleep(1.2)
            if self.is_bypassed():
                self.log_message("Bypass success after click.")
                return True

            sig_after = self._collect_state_signature()
            no_transition = sig_before == sig_after
            if no_transition:
                self.log_message("No challenge state transition detected after click.")
                # 当页面状态完全不变时，做一次轻量刷新，触发 challenge 重新渲染
                try:
                    page = self._normalize_page()
                    page.reload(wait_until="domcontentloaded")
                    self.log_message("Page reloaded to retrigger challenge rendering.")
                    time.sleep(1.5)
                    if self.is_bypassed():
                        self.log_message("Bypass success after reload.")
                        return True
                except Exception as reload_err:
                    self.log_message(f"Reload failed: {reload_err}")

            # 递增退避，降低持续高频点击导致的风控风险
            wait_seconds = min(2 + i, 6)
            state_after = self._collect_page_state()
            self.log_message(
                f"Bypass not yet complete, sleeping {wait_seconds}s, after_click: {state_after}"
            )
            time.sleep(wait_seconds)

        final_ok = self.is_bypassed()
        if not final_ok:
            self.log_message(f"Bypass failed after retries. final_state: {self._collect_page_state()}")
        return final_ok