Openakita/Skills@Apify Scraper

Apify Scraper — 网页数据抓取

When to Use

• 用户需要从网站抓取结构化数据（商品信息、社交媒体帖子、搜索结果等）
• 需要批量获取社交媒体平台数据（Instagram、TikTok、YouTube 等）
• 需要抓取 Google 搜索结果、地图信息、评价数据
• 需要定期监控网页变化
• 需要将非结构化网页内容转换为 JSON/CSV
• 需要从电商平台提取商品和价格信息

Prerequisites

必需配置

将 Token 添加到 .env 文件：

Copy code
APIFY_TOKEN=apify_api_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

必需依赖

可选依赖

验证配置

bashCopy code
curl -s "https://api.apify.com/v2/user/me?token=$APIFY_TOKEN" | python -m json.tool

Instructions

Apify Actor 快速概览

Apify 平台上有数千个 Actor（即预构建的爬虫/自动化程序）。本技能聚焦 55+ 个经过验证的、面向 AI 数据提取优化的 Actor。

平台 Actor 速查表

社交媒体

搜索引擎

电商平台

通用工具

Actor 自动选择策略

Agent 根据用户需求自动选择最合适的 Actor：

1. 解析用户意图 — 从用户描述中识别目标平台和数据类型
2. 匹配 Actor — 根据平台和功能匹配上方表格
3. 若无精确匹配 — 使用通用 Web Scraper 或 GPT Scraper
4. 确认方案 — 向用户确认将使用的 Actor 和预计数据量

Workflows

Workflow 1: 社交媒体数据抓取

步骤 1 — 确认需求

步骤 2 — 选择并配置 Actor

pythonCopy code
from apify_client import ApifyClient

client = ApifyClient(os.environ['APIFY_TOKEN'])

run_input = {
    "usernames": ["openai"],
    "resultsLimit": 100,
    "resultsType": "posts",
}

run = client.actor("apify/instagram-profile-scraper").call(run_input=run_input)

步骤 3 — 获取并处理结果

pythonCopy code
items = list(client.dataset(run["defaultDatasetId"]).iterate_items())

步骤 4 — 格式化输出

将数据转换为用户需要的格式（JSON/CSV/表格摘要）。

Workflow 2: 搜索引擎数据抓取

步骤 1 — 确认搜索参数

步骤 2 — 调用 Actor

pythonCopy code
run_input = {
    "queries": "AI agent 框架 2025",
    "maxPagesPerQuery": 3,
    "languageCode": "zh",
    "countryCode": "cn",
    "resultsPerPage": 10,
}

run = client.actor("apify/google-search-scraper").call(run_input=run_input)
items = list(client.dataset(run["defaultDatasetId"]).iterate_items())

步骤 3 — 提取关键字段

Workflow 3: 通用网页数据提取

当没有专用 Actor 时，使用 AI 驱动的通用提取：

方法 A — Web Scraper（基于选择器）

pythonCopy code
run_input = {
    "startUrls": [{"url": "https://example.com/products"}],
    "pageFunction": """
    async function pageFunction(context) {
        const $ = context.jQuery;
        const results = [];
        $('div.product-card').each((i, el) => {
            results.push({
                name: $(el).find('.title').text().trim(),
                price: $(el).find('.price').text().trim(),
                url: $(el).find('a').attr('href'),
            });
        });
        return results;
    }
    """,
    "maxRequestsPerCrawl": 100,
}

run = client.actor("apify/web-scraper").call(run_input=run_input)

方法 B — GPT Scraper（AI 智能提取）

pythonCopy code
run_input = {
    "startUrls": [{"url": "https://example.com/products"}],
    "instructions": "Extract all product names, prices, and descriptions from this page",
    "openaiApiKey": os.environ.get('OPENAI_API_KEY'),
    "maxRequestsPerCrawl": 10,
}

run = client.actor("drobnikj/gpt-scraper").call(run_input=run_input)

Workflow 4: 批量多平台抓取

同时从多个来源抓取数据：

步骤 1 — 列出所有抓取任务 步骤 2 — 并行启动多个 Actor 步骤 3 — 等待所有任务完成 步骤 4 — 合并结果并去重

pythonCopy code
import asyncio
from apify_client import ApifyClientAsync

async def batch_scrape(tasks):
    client = ApifyClientAsync(os.environ['APIFY_TOKEN'])
    results = {}

    async def run_actor(name, actor_id, input_data):
        run = await client.actor(actor_id).call(run_input=input_data)
        items = []
        async for item in client.dataset(run["defaultDatasetId"]).iterate_items():
            items.append(item)
        results[name] = items

    await asyncio.gather(*[
        run_actor(t['name'], t['actor_id'], t['input'])
        for t in tasks
    ])

    return results

Output Format

JSON 输出（默认）

jsonCopy code
{
  "metadata": {
    "actor": "apify/instagram-profile-scraper",
    "total_items": 42,
    "scraped_at": "2025-03-01T14:30:00Z",
    "run_id": "abc123",
    "cost_usd": 0.05
  },
  "data": [
    {
      "id": "post_12345",
      "text": "帖子内容...",
      "likes": 1234,
      "comments": 56,
      "timestamp": "2025-02-28T10:00:00Z",
      "url": "https://instagram.com/p/xxx"
    }
  ]
}

CSV 输出

pythonCopy code
import pandas as pd

df = pd.DataFrame(items)
df.to_csv('output.csv', index=False, encoding='utf-8-sig')

摘要表格

当数据量较大时，先输出摘要统计：

Copy code
📊 抓取完成
- Actor: Instagram Profile Scraper
- 总条数: 142 条帖子
- 时间范围: 2025-01-01 ~ 2025-03-01
- 平均点赞: 2,345
- 最高互动帖子: [URL]
- 费用: $0.12

Common Pitfalls

1. APIFY_TOKEN 未配置

症状：所有请求返回 401 解决：确认 .env 中的 APIFY_TOKEN 已正确设置

2. Actor 运行超时

症状：任务长时间未完成 解决：

• 减少 maxRequestsPerCrawl 或 resultsLimit
• 使用 memoryMbytes 增加内存分配
• 检查目标网站是否可达

3. 被目标网站封禁

症状：返回 403 或空结果 解决：

• 使用 Apify 的代理池（Actor 通常自带）
• 降低并发和频率
• 增加请求间隔

4. 数据格式不一致

不同 Actor 返回的数据结构不同。在处理数据前先检查字段：

pythonCopy code
if items:
    print("Available fields:", list(items[0].keys()))

5. 费用超预期

Apify 按计算单元（CU）收费。大规模抓取前：

• 先小量测试（resultsLimit: 10）确认结果质量
• 估算总费用：查看测试运行的 CU 消耗 × 总数据量倍数
• 设置账户费用上限

6. 社交平台反爬限制

Instagram、TikTok 等平台会动态调整反爬策略：

• 不要在短时间内大量抓取同一账号
• 使用平台提供的数据导出功能作为补充
• 遵守平台的 robots.txt 和 Terms of Service

伦理与合规指南

必须遵守

1. robots.txt — 尊重网站的 robots.txt 规则
2. Terms of Service — 不违反目标网站的服务条款
3. 隐私法规 — 遵守 GDPR、个人信息保护法等法规
4. 频率限制 — 不对目标网站造成过大负载
5. 数据用途 — 仅用于合法目的（分析、研究、商业决策）

禁止行为

1. 抓取个人隐私信息用于骚扰或监控
2. 大量抓取导致目标网站服务降级
3. 绕过付费墙或版权保护
4. 出售未经授权的第三方数据
5. 用于垃圾信息、虚假评论等恶意目的

建议做法

1. 缓存已获取数据，避免重复抓取
2. 使用增量抓取而非全量抓取
3. 在非高峰时段执行大规模任务
4. 提供 User-Agent 标识你的爬虫身份
5. 记录抓取日志以便审计

高级功能

定时任务

pythonCopy code
schedule_input = {
    "actorId": "apify/google-search-scraper",
    "cronExpression": "0 9 * * 1",  # 每周一早 9 点
    "input": {
        "queries": "竞品动态",
        "maxPagesPerQuery": 1,
    }
}

Webhook 通知

pythonCopy code
run = client.actor("apify/web-scraper").call(
    run_input=run_input,
    webhooks=[{
        "eventTypes": ["ACTOR.RUN.SUCCEEDED"],
        "requestUrl": "https://your-server.com/webhook",
    }]
)

EXTEND.md 扩展

用户可在技能同目录下创建 EXTEND.md 添加：

• 常用的 Actor ID 和预设配置
• 自定义数据处理管道
• 特定网站的抓取策略
• 代理配置和反封禁策略