📦 Web Scraper Pro — 智能网页抓取工具

智能网页抓取器，可将任何 URL 转换为干净可读的 Markdown 内容。采用多层回退策略确保最大兼容性。

💰 定价：每次网页抓取 0.001 USDT

⚠️ 重要：支付集成（必须先阅读）

本技能通过 SkillPay 实现商业化。在抓取任何 URL 之前，您必须执行支付验证步骤。

支付配置

# ═══════════════════════════════════════════════════
# SkillPay 计费集成
# ═══════════════════════════════════════════════════
import requests
BILLING_API_URL = 'https://skillpay.me'
BILLING_API_KEY = 'sk_d6d26f291dafc43acc8c2b6215b87cbc9b19c7d093aebdb2deeba42a3a0fea4b'
SKILL_ID = '4fb2d57e-e583-4ca3-8170-52df37a6572b'
HEADERS = {'X-API-Key': BILLING_API_KEY, 'Content-Type': 'application/json'}

计费 API 函数（必需）

⚠️ 此代码必须在任何网页抓取之前执行：

import requests
import os
BILLING_API_URL = 'https://skillpay.me'
BILLING_API_KEY = 'sk_d6d26f291dafc43acc8c2b6215b87cbc9b19c7d093aebdb2deeba42a3a0fea4b'
SKILL_ID = '4fb2d57e-e583-4ca3-8170-52df37a6572b'
HEADERS = {'X-API-Key': BILLING_API_KEY, 'Content-Type': 'application/json'}
# ① Check balance / 查余额
def check_balance(user_id: str) -> float:
    resp = requests.get(
        f'{BILLING_API_URL}/api/v1/billing/balance',
        params={'user_id': user_id},
        headers=HEADERS
    )
    return resp.json()['balance']  # USDT amount
# ② Charge per call / 每次调用扣费
def charge_user(user_id: str) -> dict:
    resp = requests.post(
        f'{BILLING_API_URL}/api/v1/billing/charge',
        headers=HEADERS,
        json={
            'user_id': user_id,
            'skill_id': SKILL_ID,
            'amount': 0.001,  # USDT per call
        }
    )
    data = resp.json()
    if data['success']:
        return {'ok': True, 'balance': data['balance']}
    # Insufficient balance → auto returns BNB Chain USDT payment link
    return {'ok': False, 'balance': data['balance'], 'payment_url': data.get('payment_url')}
# ③ Generate payment link / 生成充值链接
def get_payment_link(user_id: str, amount: float) -> str:
    resp = requests.post(
        f'{BILLING_API_URL}/api/v1/billing/payment-link',
        headers=HEADERS,
        json={'user_id': user_id, 'amount': amount}
    )
    return resp.json()['payment_url']  # BNB Chain USDT
USER_ID = os.environ.get("SKILLPAY_USER_ID", "anonymous_user")

支付流程集成

# ========================================
# STEP 0: CHARGE USER (MANDATORY)
# ========================================
# This must be the FIRST step before ANY web fetch
charge_result = charge_user(USER_ID)
if not charge_result['ok']:
    print(f"""
╔══════════════════════════════════════════════════════════════╗
║  💳 INSUFFICIENT BALANCE                                      ║
║                                                               ║
║  This web fetch costs 0.001 USDT.                            ║
║  Current balance: {charge_result['balance']:<41}║
║                                                               ║
║  Please top up at:                                            ║
║  {charge_result['payment_url']:<56}║
║                                                               ║
║  After payment, please retry your request.                   ║
╚══════════════════════════════════════════════════════════════╝
    """)
    raise SystemExit("Insufficient balance for web fetch")
print(f"✅ Charged 0.001 USDT. Remaining balance: {charge_result['balance']} USDT")

多层抓取策略 (Multi-Layer Fetch Strategy)

本 skill 采用智能多层回退策略，确保最大兼容性：

Layer 1: markdown.new（首选，最快）

Cloudflare 驱动的 URL→Markdown 转换服务，内置三层回退：

• 原生 Markdown: Accept: text/markdown 内容协商
• Workers AI: HTML→Markdown AI 转换
• 浏览器渲染: 无头浏览器处理 JS 重度页面

import requests
def fetch_via_markdown_new(url: str, method: str = "auto", retain_images: bool = True) -> str:
    """
    Layer 1: 使用 markdown.new 抓取网页
    
    Args:
        url: 目标网页 URL
        method: 转换方法 - "auto" | "ai" | "browser"
        retain_images: 是否保留图片链接
    
    Returns:
        str: Markdown 格式的网页内容
    """
    api_url = "https://markdown.new/"
    
    try:
        response = requests.post(
            api_url,
            headers={"Content-Type": "application/json"},
            json={
                "url": url,
                "method": method,
                "retain_images": retain_images
            },
            timeout=60
        )
        
        if response.status_code == 200:
            token_count = response.headers.get("x-markdown-tokens", "unknown")
            print(f"✅ [markdown.new] 抓取成功 (tokens: {token_count})")
            return response.text
        elif response.status_code == 429:
            print("⚠️ [markdown.new] 速率限制，切换到下一层...")
            return None
        else:
            print(f"⚠️ [markdown.new] 返回状态码 {response.status_code}，切换到下一层...")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"⚠️ [markdown.new] 请求失败: {e}，切换到下一层...")
        return None

支持的查询参数:

• method=auto|ai|browser - 指定转换方法
• retain_images=true|false - 是否保留图片
• 速率限制: 每 IP 每天 500 次请求

Layer 2: defuddle.md（备选方案）

开源的网页→Markdown 提取服务，由 Obsidian Web Clipper 创建者开发。

def fetch_via_defuddle(url: str) -> str:
    """
    Layer 2: 使用 defuddle.md 抓取网页
    
    Args:
        url: 目标网页 URL（不含 https:// 前缀亦可）
    
    Returns:
        str: 带有 YAML frontmatter 的 Markdown 内容
    """
    # defuddle 接受 URL 路径直接拼接
    clean_url = url.replace("https://", "").replace("http://", "")
    api_url = f"https://defuddle.md/{clean_url}"
    
    try:
        response = requests.get(api_url, timeout=60)
        
        if response.status_code == 200 and len(response.text.strip()) > 50:
            print(f"✅ [defuddle.md] 抓取成功")
            return response.text
        else:
            print(f"⚠️ [defuddle.md] 内容为空或失败 (status: {response.status_code})，切换到下一层...")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"⚠️ [defuddle.md] 请求失败: {e}，切换到下一层...")
        return None

Layer 3: Jina Reader（AI 内容提取）

Jina AI 的阅读器服务，擅长处理复杂页面。

def fetch_via_jina(url: str) -> str:
    """
    Layer 3: 使用 Jina Reader 抓取网页
    
    Args:
        url: 目标网页完整 URL
    
    Returns:
        str: 提取的主要文本内容
    """
    api_url = f"https://r.jina.ai/{url}"
    
    try:
        response = requests.get(
            api_url,
            headers={"Accept": "text/markdown"},
            timeout=60
        )
        
        if response.status_code == 200 and len(response.text.strip()) > 50:
            print(f"✅ [Jina Reader] 抓取成功")
            return response.text
        else:
            print(f"⚠️ [Jina Reader] 内容为空或失败，切换到下一层...")
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"⚠️ [Jina Reader] 请求失败: {e}，切换到下一层...")
        return None

Layer 4: Scrapling（终极兜底）

自适应爬虫库，具备反反爬能力。

def fetch_via_scrapling(url: str) -> str:
    """
    Layer 4: 使用 Scrapling 抓取网页（终极兜底）
    
    Args:
        url: 目标网页完整 URL
    
    Returns:
        str: 提取的文本内容
    """
    try:
        from scrapling import Scrapling
        
        scraper = Scrapling()
        result = scraper.fetch(
            url,
            mode="stealth",  # 使用隐身模式绕过反爬
            output_format="markdown"
        )
        
        if result and len(result.strip()) > 50:
            print(f"✅ [Scrapling] 抓取成功")
            return result
        else:
            print(f"⚠️ [Scrapling] 抓取失败，所有层都用尽")
            return None
            
    except ImportError:
        print("⚠️ [Scrapling] 未安装，跳过此层")
        return None
    except Exception as e:
        print(f"⚠️ [Scrapling] 请求失败: {e}")
        return None

主抓取函数

def smart_fetch(url: str, max_retries: int = 2) -> str:
    """
    智能网页抓取函数，自动选择最佳抓取层
    
    Args:
        url: 目标网页 URL
        max_retries: 最大重试次数
    
    Returns:
        str: Markdown 格式的网页内容，如果所有层都失败则返回 None
    """
    print(f"🎯 开始抓取: {url}")
    
    # 按优先级尝试各层
    layers = [
        ("markdown.new", fetch_via_markdown_new),
        ("defuddle.md", fetch_via_defuddle),
        ("Jina Reader", fetch_via_jina),
        ("Scrapling", fetch_via_scrapling),
    ]
    
    for layer_name, fetch_func in layers:
        for attempt in range(max_retries):
            try:
                result = fetch_func(url)
                if result:
                    return result
                break  # 当前层失败，尝试下一层
            except Exception as e:
                print(f"⚠️ [{layer_name}] 第 {attempt+1} 次尝试失败: {e}")
                if attempt < max_retries - 1:
                    time.sleep(1)  # 重试前等待
    
    print("❌ 所有抓取层均失败")
    return None
# 用法示例
if __name__ == "__main__":
    # ⚠️ 必须先执行支付验证（上方代码）
    
    url = "https://example.com/article"
    content = smart_fetch(url)
    
    if content:
        print("\n" + "="80)
        print("📄 抓取结果 (前500字符):")
        print("="80)
        print(content[:500] + "..." if len(content) > 500 else content)
    else:
        print("❌ 抓取失败")

使用场景示例

场景 1: 抓取普通网页

用户: 帮我抓取这个网页 https://example.com/article/123

执行流程:

• 支付验证 → 通过
• Layer 1 (markdown.new) → 成功
• 返回文章内容

场景 2: 抓取 GitHub README

用户: 我想看看这个库的介绍 https://github.com/D4Vinci/Scrapling

执行流程:

• 支付验证 → 通过
• Layer 1 (markdown.new) → GitHub 页面通常成功
• 返回 Scrapling 项目的 README 内容

场景 3: 抓取反爬网站

用户: 帮我抓取这个网页 https://某反爬网站.com/article/123

执行流程:

• 支付验证 → 通过
• Layer 1 → 失败
• Layer 2 → 失败
• Layer 3 → 失败
• Layer 4 (Scrapling Stealth) → 使用隐身模式绕过反爬
• 返回提取的内容

场景 4: 搜索信息（使用 Jina Search）

用户: 帮我搜一下 "Python asyncio best practices 2025"

def search_via_jina(query: str) -> str:
    """使用 Jina Search 搜索信息"""
    api_url = f"https://s.jina.ai/{query}"
    try:
        response = requests.get(api_url, timeout=60)
        if response.status_code == 200:
            return response.text
        return None
    except:
        return None
# 执行搜索
search_result = search_via_jina("Python asyncio best practices 2025")
print(search_result)

Prerequisites (按需安装)

基础依赖（Layer 1-3 只需 requests）

pip install requests

Scrapling 依赖（Layer 4 - 仅在需要时安装）

# 基础安装
pip install scrapling
# 完整安装（含浏览器和反反爬功能）
pip install "scrapling[fetchers]"
scrapling install

💰 Revenue & Analytics

在 SkillPay Dashboard 实时追踪您的收入。

• 每次抓取价格: 0.001 USDT
• 您的收入分成: 95%
• 结算: 即时（BNB Chain）

Powered by SkillPay - AI Skill Monetization Infrastructure