Playwright无法检测DevTools Cmd+F可见文本，如何实现浏览器级搜索匹配？

问题根源 #

Beacons.ai这类页面会通过动态异步加载、Shadow DOM或反爬策略隐藏内容：

• 常规的document.documentElement.outerHTML不会包含Shadow DOM内的内容
• 无头浏览器模式会被网站识别，导致内容不渲染
• 固定等待时间和networkidle状态不足以等待所有异步内容加载完成
• DevTools的全局搜索会遍历整个渲染树（包括Shadow DOM、动态生成的Script内容），而常规DOM抓取逻辑做不到这点

修复方案 #

1. 模拟真实浏览器环境 #

避免被网站识别为无头爬虫，添加浏览器指纹和禁用自动化标识：

# 修改browser和context的初始化代码
browser = await p.chromium.launch(
    headless=True,
    args=[
        "--disable-blink-features=AutomationControlled",
        "--start-maximized"
    ]
)
context = await browser.new_context(
    viewport=None,
    user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36"
)
# 禁用webdriver标识
await context.add_init_script("""
    Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
""")

2. 遍历整个渲染树（含Shadow DOM） #

实现和DevTools Cmd+F一致的全局搜索逻辑，递归遍历所有节点包括Shadow DOM：

# 替换check_link函数中的DOM搜索部分
onlyfans_found_bool = await page.evaluate("""() => {
    const target = 'onlyfans.com';
    function scanNode(node) {
        if (node.textContent?.toLowerCase().includes(target)) return true;
        if (node.shadowRoot && scanNode(node.shadowRoot)) return true;
        for (const child of node.children) {
            if (scanNode(child)) return true;
        }
        if (node.tagName === 'SCRIPT' && node.textContent.toLowerCase().includes(target)) return true;
        return false;
    }
    return scanNode(document.documentElement);
}""")

if onlyfans_found_bool:
    onlyfans_found = "x"
    # 截图逻辑保持不变

3. 优化等待策略 #

替换固定sleep为等待目标内容加载的动态等待：

# 替换await asyncio.sleep(WAIT_TIME)
try:
    await page.wait_for_function("""() => {
        const target = 'onlyfans.com';
        // 检查文本或Script内容
        return document.body.textContent.toLowerCase().includes(target) ||
               Array.from(document.querySelectorAll('script')).some(s => s.textContent.toLowerCase().includes(target));
    }""", timeout=15000)
except TimeoutError:
    # 超时后继续执行，避免终止任务
    pass

修改后的完整代码 #

import csv
import urllib.parse
import asyncio
import os
from playwright.async_api import async_playwright

INPUT_FILE = "input_links.csv"
OUTPUT_FILE = "onlyfans_deepsearch.csv"
SCREENSHOT_FOLDER = "screenshots"
CONCURRENT_TASKS = 5

os.makedirs(SCREENSHOT_FOLDER, exist_ok=True)

def resolve_redirect(link: str) -> str:
    if "l.instagram.com" in link and "?u=" in link:
        try:
            parsed = urllib.parse.urlparse(link)
            real_url = urllib.parse.parse_qs(parsed.query).get("u", [None])[0]
            return urllib.parse.unquote(real_url) if real_url else link
        except Exception:
            pass
    return link

async def check_link(context, identifier: str, raw_link: str) -> dict:
    if not raw_link.lower().startswith("http"):
        return {"ID": identifier, "Link": raw_link, "Onlyfans": ""}

    urls = [u for u in raw_link.split() if u.startswith("http")]
    onlyfans_found = ""

    for url in urls:
        resolved = resolve_redirect(url)
        try:
            page = await context.new_page()
            print(f"🔍 {identifier} → {resolved}")
            await page.goto(resolved, timeout=30000)
            await page.wait_for_load_state("networkidle")

            # 动态等待目标内容加载，超时不中断
            try:
                await page.wait_for_function("""() => {
                    const target = 'onlyfans.com';
                    return document.body.textContent.toLowerCase().includes(target) ||
                           Array.from(document.querySelectorAll('script')).some(s => s.textContent.toLowerCase().includes(target));
                }""", timeout=15000)
            except TimeoutError:
                pass

            # 模拟DevTools全局搜索，遍历所有节点（含Shadow DOM）
            onlyfans_found_bool = await page.evaluate("""() => {
                const target = 'onlyfans.com';
                function scanNode(node) {
                    if (node.textContent?.toLowerCase().includes(target)) return true;
                    if (node.shadowRoot && scanNode(node.shadowRoot)) return true;
                    for (const child of node.children) {
                        if (scanNode(child)) return true;
                    }
                    if (node.tagName === 'SCRIPT' && node.textContent.toLowerCase().includes(target)) return true;
                    return false;
                }
                return scanNode(document.documentElement);
            }""")

            if onlyfans_found_bool:
                onlyfans_found = "x"
                safe_id = "".join(c for c in identifier if c.isalnum() or c in ("_", "-"))
                await page.screenshot(path=f"{SCREENSHOT_FOLDER}/{safe_id}.png")

            await page.close()
            if onlyfans_found:
                break
        except Exception as e:
            print(f"❌ Error: {identifier} → {resolved}: {e}")
            continue

    return {"ID": identifier, "Link": raw_link, "Onlyfans": onlyfans_found}

async def process_all(rows):
    results = []
    sem = asyncio.Semaphore(CONCURRENT_TASKS)

    async with async_playwright() as p:
        # 模拟真实浏览器环境，避免反爬
        browser = await p.chromium.launch(
            headless=True,
            args=[
                "--disable-blink-features=AutomationControlled",
                "--start-maximized"
            ]
        )
        context = await browser.new_context(
            viewport=None,
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36"
        )
        # 禁用webdriver标识
        await context.add_init_script("""
            Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
        """)

        async def sem_task(row):
            async with sem:
                return await check_link(context, row[0].strip(), row[1].strip())

        tasks = [sem_task(row) for row in rows]
        for future in asyncio.as_completed(tasks):
            results.append(await future)

        await browser.close()
    return results

def load_input_csv(filepath):
    with open(filepath, 'r', encoding='utf-8') as f:
        reader = csv.reader(f)
        rows = []
        for row in reader:
            if len(row) >= 2:
                cleaned = [col.strip() for col in row]
                if len(cleaned) == 3 and cleaned[2].lower().strip() != "x":
                    cleaned[2] = ""
                rows.append(cleaned)
        return rows

def save_output_csv(filepath, data):
    with open(filepath, 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=["ID", "Link", "Onlyfans"])
        writer.writeheader()
        writer.writerows(data)

rows = load_input_csv(INPUT_FILE)
final_results = asyncio.run(process_all(rows))
save_output_csv(OUTPUT_FILE, final_results)
print(f"✅ Done. Results saved to {OUTPUT_FILE}")

关键改进点 #

• 模拟真实浏览器环境，绕过网站的无头检测
• 递归遍历所有DOM节点（包括Shadow DOM），实现和DevTools一致的搜索范围
• 动态等待目标内容加载，替代固定sleep提升可靠性
• 单独检查Script标签内容，覆盖动态注入的JSON数据场景

内容的提问来源于stack exchange，提问作者Jack Rhodes