Trafilatura + CloakBrowser：全文采集的完整流水线

┌──────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│  亿牛云代理       │     │  CloakBrowser    │     │  Trafilatura     │
│  住宅出口 IP      │ ──→ │  指纹伪装+渲染   │ ──→ │  正文/元数据提取  │
│  geoip 时区匹配   │     │  humanize 行为   │     │  Markdown/JSON   │
└──────────────────┘     └──────────────────┘     └──────────────────┘
                                                          │
                                                   ┌──────┴──────┐
                                                   │  存储/输出    │
                                                   │  文件/数据库  │
                                                   └─────────────┘

import trafilatura
from cloakbrowser import launch
from lxml import html
 
def render_and_extract(url, proxy=None):
    """使用 CloakBrowser 渲染页面，再用 Trafilatura 提取正文"""
 
    # 1. 启动 CloakBrowser
    browser = launch(
        proxy=proxy,
        headless=False,
        geoip=True,
        humanize=True,
    )
    page = browser.new_page()
 
    try:
        # 2. 访问页面
        page.goto(url, wait_until="networkidle")
 
        # 3. 等待内容渲染完成
        page.wait_for_timeout(3000)
 
        # 4. 获取完整 HTML
        html_content = page.content()
 
        # 5. 用 Trafilatura 提取正文
        result = trafilatura.extract(
            html_content,
            output_format="markdown",
            with_metadata=True,
            include_tables=True,
        )
 
        return result
 
    finally:
        browser.close()
 
# 使用
result = render_and_extract(
    "https://example.16yun.cn/article",
    proxy="http://user:pass@proxy.16yun.cn:8888",
)
print(result)

import trafilatura
import json
from cloakbrowser import launch_persistent_context
 
def scrape_cloudflare_protected_site():
    """采集 Cloudflare 保护的技术博客"""
 
    # 使用持久化上下文——首次访问通过 Turnstile 后，后续复用
    ctx = launch_persistent_context(
        "./profiles/tech-blog",
        headless=False,
        proxy="http://user:pass@proxy.16yun.cn:8888",
        geoip=True,
        humanize=True,
    )
 
    page = ctx.new_page()
 
    # 访问首页——可能触发 Turnstile
    page.goto("https://example.16yun.cn", wait_until="networkidle")
 
    # 首次访问可能需要手动处理 Turnstile（CloakBrowser 自动通过）
    page.wait_for_timeout(5000)
 
    # 获取文章链接列表
    article_links = page.eval_on_selector_all(
        "a[href*='/blog/']",
        "elements => elements.map(el => el.href)"
    )
    print(f"发现 {len(article_links)} 篇文章")
 
    # 逐篇提取
    for i, article_url in enumerate(article_links[:10]):
        print(f"[{i+1}/{len(article_links[:10])}] {article_url}")
        page.goto(article_url, wait_until="networkidle")
        page.wait_for_timeout(2000)
 
        html_content = page.content()
        result = trafilatura.extract(
            html_content,
            output_format="json",
            with_metadata=True,
            include_tables=True,
        )
 
        if result:
            data = json.loads(result)
            data["url"] = article_url
 
            filename = f"articles/article-{i+1}.json"
            with open(filename, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            print(f"  SAVED:: {data.get('title', 'untitled')}")
 
    ctx.close()
 
scrape_cloudflare_protected_site()

import trafilatura
import json
from cloakbrowser import launch
from concurrent.futures import ThreadPoolExecutor, as_completed
import os
 
class RenderingExtractor:
    """渲染 + 提取管线"""
 
    def __init__(self, proxy, profile_dir="./profiles"):
        self.proxy = proxy
        self.profile_dir = profile_dir
 
    def extract(self, url, profile_name="default"):
        ctx = launch_persistent_context(
            os.path.join(self.profile_dir, profile_name),
            headless=False,
            proxy=self.proxy,
            geoip=True,
            humanize=True,
        )
        page = ctx.new_page()
 
        try:
            page.goto(url, wait_until="networkidle")
            page.wait_for_timeout(3000)
 
            html_content = page.content()
            result = trafilatura.extract(
                html_content,
                output_format="json",
                with_metadata=True,
                include_tables=True,
            )
            return json.loads(result) if result else None
        finally:
            ctx.close()
 
# 使用
extractor = RenderingExtractor(
    proxy="http://user:pass@proxy.16yun.cn:8888"
)
 
urls = [
    "https://example.16yun.cn/article-1",
    "https://example.16yun.cn/article-2",
]
 
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {
        executor.submit(extractor.extract, url, f"site-{i}"): url
        for i, url in enumerate(urls)
    }
    for future in as_completed(futures):
        result = future.result()
        if result:
            print(f"OK: {result.get('title')}")

# 等待特定元素出现，而非固定时间
page.wait_for_selector("article", timeout=15000)
 
# 或等待网络空闲
page.goto(url, wait_until="networkidle")