data-scraper-agent

Data Scraper Agent

建立一個可上線、由 AI 驅動的資料收集代理，適用於任何公開資料來源。
可排程執行，使用免費 LLM 豐富結果，儲存至資料庫，並隨著時間持續改進。

技術棧：Python · Gemini Flash（免費）· GitHub Actions（免費）· Notion / Sheets / Supabase

何時啟用

• 使用者想要收集或監控任何公開網站或 API
• 使用者說「建立一個機器人來檢查...」、「幫我監控 X」、「從...收集資料」
• 使用者想要追蹤職缺、價格、新聞、儲存庫、運動比分、活動、列表
• 使用者詢問如何在不付費託管的情況下自動化資料收集
• 使用者想要一個能根據他們的決策變得越來越聰明的代理

核心概念

三層架構

每個資料收集代理都有三層：

收集 → 豐富 → 儲存
  │       │       │
爬蟲    AI (LLM)  資料庫
排程    評分/     Notion /
執行    摘要      Sheets /
        與分類    Supabase

免費技術棧

層級	工具	原因
爬取	requests + BeautifulSoup	免費，涵蓋 80% 的公開網站
JS 渲染網站	playwright（免費）	當 HTML 擷取失敗時使用
AI 豐富	透過 REST API 的 Gemini Flash	每天 500 次請求、100 萬 tokens — 免費
儲存	Notion API	免費方案，適合檢視的 UI
排程	GitHub Actions cron	公開儲存庫免費
學習	儲存庫中的 JSON 回饋檔案	零基礎設施，保留在 git 中

AI 模型降級鏈

建立代理以在配額耗盡時自動降級 Gemini 模型：

gemini-2.0-flash-lite (30 RPM) →
gemini-2.0-flash (15 RPM) →
gemini-2.5-flash (10 RPM) →
gemini-flash-lite-latest (降級)

批次 API 呼叫以提高效率

永遠不要對每個項目單獨呼叫 LLM。始終批次處理：

# 錯誤：33 個項目需要 33 次 API 呼叫
for item in items:
    result = call_ai(item)  # 33 次呼叫 → 達到速率限制

# 正確：33 個項目只需 7 次 API 呼叫（批次大小 5）
for batch in chunks(items, size=5):
    results = call_ai(batch)  # 7 次呼叫 → 保持在免費方案內

---

工作流程

步驟 1：了解目標

詢問使用者：

1. 要收集什麼：「什麼資料來源？URL / API / RSS / 公開端點？」
2. 要提取什麼：「哪些欄位重要？標題、價格、URL、日期、分數？」
3. 如何儲存：「結果要去哪裡？Notion、Google Sheets、Supabase 還是本地檔案？」
4. 如何豐富：「你想要 AI 對每個項目評分、摘要、分類或比對嗎？」
5. 頻率：「多久執行一次？每小時、每天、每週？」

常見範例提示：

• 求職板 → 根據履歷評分相關性
• 產品價格 → 價格下降時發出警報
• GitHub 儲存庫 → 摘要新版本
• 新聞 feed → 按主題 + 情緒分類
• 運動結果 → 提取統計資料到追蹤器
• 活動行事曆 → 按興趣篩選

---

步驟 2：設計收集架構

為使用者產生此目錄結構：

my-agent/
├── config.yaml              # 使用者自訂（關鍵字、篩選器、偏好）
├── profile/
│   └── context.md           # AI 使用的使用者背景（履歷、興趣、條件）
├── scraper/
│   ├── __init__.py
│   ├── main.py              # 協調器：爬取 → 豐富 → 儲存
│   ├── filters.py           # 基於規則的預先篩選（快速，在 AI 之前）
│   └── sources/
│       ├── __init__.py
│       └── source_name.py   # 每個資料來源一個檔案
├── ai/
│   ├── __init__.py
│   ├── client.py            # 具有模型降級的 Gemini REST 客戶端
│   ├── pipeline.py          # 批次 AI 分析
│   ├── jd_fetcher.py        # 從 URL 擷取完整內容（選用）
│   └── memory.py            # 從使用者回饋中學習
├── storage/
│   ├── __init__.py
│   └── notion_sync.py       # 或 sheets_sync.py / supabase_sync.py
├── data/
│   └── feedback.json        # 使用者決策歷史（自動更新）
├── .env.example
├── setup.py                 # 一次性資料庫/結構建立
├── enrich_existing.py       # 對舊資料列回填 AI 分數
├── requirements.txt
└── .github/
    └── workflows/
        └── scraper.yml      # GitHub Actions 排程

---

步驟 3：建立來源連接器

任何資料來源的範本：

# scraper/sources/my_source.py
"""
[來源名稱] — 從 [哪裡] 收集 [什麼]。
方法：[REST API / HTML 爬取 / RSS feed]
"""
import requests
from bs4 import BeautifulSoup
from datetime import datetime, timezone
from scraper.filters import is_relevant

HEADERS = {
    "User-Agent": "Mozilla/5.0 (compatible; research-bot/1.0)",
}


def fetch() -> list[dict]:
    """
    回傳具有一致結構的項目列表。
    每個項目至少必須有：name, url, date_found。
    """
    results = []

    # ---- REST API 來源 ----
    resp = requests.get("https://api.example.com/items", headers=HEADERS, timeout=15)
    if resp.status_code == 200:
        for item in resp.json().get("results", []):
            if not is_relevant(item.get("title", "")):
                continue
            results.append(_normalise(item))

    return results


def _normalise(raw: dict) -> dict:
    """將原始 API/HTML 資料轉換為標準結構。"""
    return {
        "name": raw.get("title", ""),
        "url": raw.get("link", ""),
        "source": "MySource",
        "date_found": datetime.now(timezone.utc).date().isoformat(),
        # 在此加入領域特定欄位
    }

HTML 擷取模式：

soup = BeautifulSoup(resp.text, "lxml")
for card in soup.select("[class*='listing']"):
    title = card.select_one("h2, h3").get_text(strip=True)
    link = card.select_one("a")["href"]
    if not link.startswith("http"):
        link = f"https://example.com{link}"

RSS feed 模式：

import xml.etree.ElementTree as ET
root = ET.fromstring(resp.text)
for item in root.findall(".//item"):
    title = item.findtext("title", "")
    link = item.findtext("link", "")

---

步驟 4：建立 Gemini AI 客戶端

# ai/client.py
import os, json, time, requests

_last_call = 0.0

MODEL_FALLBACK = [
    "gemini-2.0-flash-lite",
    "gemini-2.0-flash",
    "gemini-2.5-flash",
    "gemini-flash-lite-latest",
]


def generate(prompt: str, model: str = "", rate_limit: float = 7.0) -> dict:
    """呼叫 Gemini，遇到 429 時自動降級。回傳解析後的 JSON 或 {}。"""
    global _last_call

    api_key = os.environ.get("GEMINI_API_KEY", "")
    if not api_key:
        return {}

    elapsed = time.time() - _last_call
    if elapsed < rate_limit:
        time.sleep(rate_limit - elapsed)

    models = [model] + [m for m in MODEL_FALLBACK if m != model] if model else MODEL_FALLBACK
    _last_call = time.time()

    for m in models:
        url = f"https://generativelanguage.googleapis.com/v1beta/models/{m}:generateContent?key={api_key}"
        payload = {
            "contents": [{"parts": [{"text": prompt}]}],
            "generationConfig": {
                "responseMimeType": "application/json",
                "temperature": 0.3,
                "maxOutputTokens": 2048,
            },
        }
        try:
            resp = requests.post(url, json=payload, timeout=30)
            if resp.status_code == 200:
                return _parse(resp)
            if resp.status_code in (429, 404):
                time.sleep(1)
                continue
            return {}
        except requests.RequestException:
            return {}

    return {}


def _parse(resp) -> dict:
    try:
        text = (
            resp.json()
            .get("candidates", [{}])[0]
            .get("content", {})
            .get("parts", [{}])[0]
            .get("text", "")
            .strip()
        )
        if text.startswith("```"):
            text = text.split("\n", 1)[-1].rsplit("```", 1)[0]
        return json.loads(text)
    except (json.JSONDecodeError, KeyError):
        return {}

---

步驟 5：建立 AI 管線（批次）

# ai/pipeline.py
import json
import yaml
from pathlib import Path
from ai.client import generate

def analyse_batch(items: list[dict], context: str = "", preference_prompt: str = "") -> list[dict]:
    """批次分析項目。回傳已豐富 AI 欄位的項目。"""
    config = yaml.safe_load((Path(__file__).parent.parent / "config.yaml").read_text())
    model = config.get("ai", {}).get("model", "gemini-2.5-flash")
    rate_limit = config.get("ai", {}).get("rate_limit_seconds", 7.0)
    min_score = config.get("ai", {}).get("min_score", 0)
    batch_size = config.get("ai", {}).get("batch_size", 5)

    batches = [items[i:i + batch_size] for i in range(0, len(items), batch_size)]
    print(f"  [AI] {len(items)} 個項目 → {len(batches)} 次 API 呼叫")

    enriched = []
    for i, batch in enumerate(batches):
        print(f"  [AI] 批次 {i + 1}/{len(batches)}...")
        prompt = _build_prompt(batch, context, preference_prompt, config)
        result = generate(prompt, model=model, rate_limit=rate_limit)

        analyses = result.get("analyses", [])
        for j, item in enumerate(batch):
            ai = analyses[j] if j < len(analyses) else {}
            if ai:
                score = max(0, min(100, int(ai.get("score", 0))))
                if min_score and score < min_score:
                    continue
                enriched.append({**item, "ai_score": score, "ai_summary": ai.get("summary", ""), "ai_notes": ai.get("notes", "")})
            else:
                enriched.append(item)

    return enriched


def _build_prompt(batch, context, preference_prompt, config):
    priorities = config.get("priorities", [])
    items_text = "\n\n".join(
        f"項目 {i+1}: {json.dumps({k: v for k, v in item.items() if not k.startswith('_')})}"
        for i, item in enumerate(batch)
    )

    return f"""分析以下 {len(batch)} 個項目並回傳一個 JSON 物件。

# 項目
{items_text}

# 使用者背景
{context[:800] if context else "未提供"}

# 使用者優先順序
{chr(10).join(f"- {p}" for p in priorities)}

{preference_prompt}

# 指示
回傳：{{"analyses": [{{"score": <0-100>, "summary": "<2 句話>", "notes": "<為什麼符合或不符合>"}} for each item in order]}}
請簡潔。分數 90+=極佳符合，70-89=良好，50-69=普通，<50=弱。"""

---

步驟 6：建立回饋學習系統

# ai/memory.py
"""從使用者決策中學習以改善未來評分。"""
import json
from pathlib import Path

FEEDBACK_PATH = Path(__file__).parent.parent / "data" / "feedback.json"


def load_feedback() -> dict:
    if FEEDBACK_PATH.exists():
        try:
            return json.loads(FEEDBACK_PATH.read_text())
        except (json.JSONDecodeError, OSError):
            pass
    return {"positive": [], "negative": []}


def save_feedback(fb: dict):
    FEEDBACK_PATH.parent.mkdir(parents=True, exist_ok=True)
    FEEDBACK_PATH.write_text(json.dumps(fb, indent=2))


def build_preference_prompt(feedback: dict, max_examples: int = 15) -> str:
    """將回饋歷史轉換為提示中的偏好偏誤區段。"""
    lines = []
    if feedback.get("positive"):
        lines.append("# 使用者喜歡的項目（正向訊號）：")
        for e in feedback["positive"][-max_examples:]:
            lines.append(f"- {e}")
    if feedback.get("negative"):
        lines.append("\n# 使用者跳過/拒絕的項目（負向訊號）：")
        for e in feedback["negative"][-max_examples:]:
            lines.append(f"- {e}")
    if lines:
        lines.append("\n使用這些模式來偏誤新項目的評分。")
    return "\n".join(lines)

**與儲存層整合：**每次執行後，查詢資料庫中狀態為正向/負向的項目，並使用提取的模式呼叫 save_feedback()。

---

步驟 7：建立儲存（Notion 範例）

# storage/notion_sync.py
import os
from notion_client import Client
from notion_client.errors import APIResponseError

_client = None

def get_client():
    global _client
    if _client is None:
        _client = Client(auth=os.environ["NOTION_TOKEN"])
    return _client

def get_existing_urls(db_id: str) -> set[str]:
    """擷取所有已儲存的 URL — 用於去重。"""
    client, seen, cursor = get_client(), set(), None
    while True:
        resp = client.databases.query(database_id=db_id, page_size=100, **{"start_cursor": cursor} if cursor else {})
        for page in resp["results"]:
            url = page["properties"].get("URL", {}).get("url", "")
            if url: seen.add(url)
        if not resp["has_more"]: break
        cursor = resp["next_cursor"]
    return seen

def push_item(db_id: str, item: dict) -> bool:
    """將一個項目推送到 Notion。成功時回傳 True。"""
    props = {
        "Name": {"title": [{"text": {"content": item.get("name", "")[:100]}}]},
        "URL": {"url": item.get("url")},
        "Source": {"select": {"name": item.get("source", "Unknown")}},
        "Date Found": {"date": {"start": item.get("date_found")}},
        "Status": {"select": {"name": "New"}},
    }
    # AI 欄位
    if item.get("ai_score") is not None:
        props["AI Score"] = {"number": item["ai_score"]}
    if item.get("ai_summary"):
        props["Summary"] = {"rich_text": [{"text": {"content": item["ai_summary"][:2000]}}]}
    if item.get("ai_notes"):
        props["Notes"] = {"rich_text": [{"text": {"content": item["ai_notes"][:2000]}}]}

    try:
        get_client().pages.create(parent={"database_id": db_id}, properties=props)
        return True
    except APIResponseError as e:
        print(f"[notion] 推送失敗：{e}")
        return False

def sync(db_id: str, items: list[dict]) -> tuple[int, int]:
    existing = get_existing_urls(db_id)
    added = skipped = 0
    for item in items:
        if item.get("url") in existing:
            skipped += 1; continue
        if push_item(db_id, item):
            added += 1; existing.add(item["url"])
        else:
            skipped += 1
    return added, skipped

---

步驟 8：在 main.py 中協調

# scraper/main.py
import os, sys, yaml
from pathlib import Path
from dotenv import load_dotenv

load_dotenv()

from scraper.sources import my_source          # 加入你的來源

# 注意：此範例使用 Notion。如果 storage.provider 是 "sheets" 或 "supabase"，
# 請將此匯入替換為 storage.sheets_sync 或 storage.supabase_sync，並相應更新
# env 變數和 sync() 呼叫。
from storage.notion_sync import sync

SOURCES = [
    ("My Source", my_source.fetch),
]

def ai_enabled():
    return bool(os.environ.get("GEMINI_API_KEY"))

def main():
    config = yaml.safe_load((Path(__file__).parent.parent / "config.yaml").read_text())
    provider = config.get("storage", {}).get("provider", "notion")

    # 根據 provider 從環境變數解析儲存目標識別碼
    if provider == "notion":
        db_id = os.environ.get("NOTION_DATABASE_ID")
        if not db_id:
            print("錯誤：未設定 NOTION_DATABASE_ID"); sys.exit(1)
    else:
        # 在此擴展以支援 sheets (SHEET_ID) 或 supabase (SUPABASE_TABLE) 等。
        print(f"錯誤：provider '{provider}' 尚未在 main.py 中連接"); sys.exit(1)

    config = yaml.safe_load((Path(__file__).parent.parent / "config.yaml").read_text())
    all_items = []

    for name, fetch_fn in SOURCES:
        try:
            items = fetch_fn()
            print(f"[{name}] {len(items)} 個項目")
            all_items.extend(items)
        except Exception as e:
            print(f"[{name}] 失敗：{e}")

    # 依 URL 去重
    seen, deduped = set(), []
    for item in all_items:
        if (url := item.get("url", "")) and url not in seen:
            seen.add(url); deduped.append(item)

    print(f"不重複項目：{len(deduped)}")

    if ai_enabled() and deduped:
        from ai.memory import load_feedback, build_preference_prompt
        from ai.pipeline import analyse_batch

        # load_feedback() 讀取由你的回饋同步腳本寫入的 data/feedback.json。
        # 為保持最新，請實作一個獨立的 feedback_sync.py，查詢你的
        # 儲存提供者中狀態為正向/負向的項目，並呼叫 save_feedback()。
        feedback = load_feedback()
        preference = build_preference_prompt(feedback)
        context_path = Path(__file__).parent.parent / "profile" / "context.md"
        context = context_path.read_text() if context_path.exists() else ""
        deduped = analyse_batch(deduped, context=context, preference_prompt=preference)
    else:
        print("[AI] 已跳過 — 未設定 GEMINI_API_KEY")

    added, skipped = sync(db_id, deduped)
    print(f"完成 — {added} 個新增，{skipped} 個已存在")

if __name__ == "__main__":
    main()

---

步驟 9：GitHub Actions 工作流程

# .github/workflows/scraper.yml
name: Data Scraper Agent

on:
  schedule:
    - cron: "0 */3 * * *"  # 每 3 小時 — 根據需求調整
  workflow_dispatch:        # 允許手動觸發

permissions:
  contents: write   # 回饋歷史提交步驟需要

jobs:
  scrape:
    runs-on: ubuntu-latest
    timeout-minutes: 20

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
          cache: "pip"

      - run: pip install -r requirements.txt

      # 如果 requirements.txt 中啟用了 Playwright，請取消註解
      # - name: Install Playwright browsers
      #   run: python -m playwright install chromium --with-deps

      - name: Run agent
        env:
          NOTION_TOKEN: ${{ secrets.NOTION_TOKEN }}
          NOTION_DATABASE_ID: ${{ secrets.NOTION_DATABASE_ID }}
          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
        run: python -m scraper.main

      - name: Commit feedback history
        run: |
          git config user.name "github-actions[bot]"
          git config user.email "github-actions[bot]@users.noreply.github.com"
          git add data/feedback.json || true
          git diff --cached --quiet || git commit -m "chore: update feedback history"
          git push

---

步驟 10：config.yaml 範本

# 自訂此檔案 — 無需修改程式碼

# 要收集的內容（AI 前的預先篩選）
filters:
  required_keywords: []      # 項目必須包含至少一個
  blocked_keywords: []       # 項目不得包含任何一個

# 你的優先順序 — AI 用於評分
priorities:
  - "範例優先順序 1"
  - "範例優先順序 2"

# 儲存
storage:
  provider: "notion"         # notion | sheets | supabase | sqlite

# 回饋學習
feedback:
  positive_statuses: ["Saved", "Applied", "Interested"]
  negative_statuses: ["Skip", "Rejected", "Not relevant"]

# AI 設定
ai:
  enabled: true
  model: "gemini-2.5-flash"
  min_score: 0               # 過濾掉低於此分數的項目
  rate_limit_seconds: 7      # API 呼叫之間的秒數
  batch_size: 5              # 每次 API 呼叫的項目數

---

常見爬取模式

模式 1：REST API（最簡單）

resp = requests.get(url, params={"q": query}, headers=HEADERS, timeout=15)
items = resp.json().get("results", [])

模式 2：HTML 爬取

soup = BeautifulSoup(resp.text, "lxml")
for card in soup.select(".listing-card"):
    title = card.select_one("h2").get_text(strip=True)
    href = card.select_one("a")["href"]

模式 3：RSS Feed

import xml.etree.ElementTree as ET
root = ET.fromstring(resp.text)
for item in root.findall(".//item"):
    title = item.findtext("title", "")
    link = item.findtext("link", "")
    pub_date = item.findtext("pubDate", "")

模式 4：分頁 API

page = 1
while True:
    resp = requests.get(url, params={"page": page, "limit": 50}, timeout=15)
    data = resp.json()
    items = data.get("results", [])
    if not items:
        break
    for item in items:
        results.append(_normalise(item))
    if not data.get("has_more"):
        break
    page += 1

模式 5：JS 渲染頁面（Playwright）

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch()
    page = browser.new_page()
    page.goto(url)
    page.wait_for_selector(".listing")
    html = page.content()
    browser.close()

soup = BeautifulSoup(html, "lxml")

---

應避免的反模式

反模式	問題	修正
每個項目一次 LLM 呼叫	立即達到速率限制	每批次 5 個項目
程式碼中硬編碼關鍵字	不可重複使用	將所有設定移至 config.yaml
爬取時不限制速率	IP 被封鎖	在請求之間加入 time.sleep(1)
在程式碼中儲存機密	安全風險	始終使用 .env + GitHub Secrets
無去重	重複資料列堆積	在推送前始終檢查 URL
忽略 robots.txt	法律/道德風險	遵守爬取規則；盡可能使用公開 API
使用 requests 處理 JS 渲染網站	空回應	使用 Playwright 或尋找底層 API
maxOutputTokens 太低	JSON 截斷，解析錯誤	批次回應使用 2048+

---

免費方案限制參考

服務	免費限制	典型使用量
Gemini Flash Lite	30 RPM, 1500 RPD	每 3 小時間隔約 56 次請求/天
Gemini 2.0 Flash	15 RPM, 1500 RPD	良好的降級選項
Gemini 2.5 Flash	10 RPM, 500 RPD	謹慎使用
GitHub Actions	無限（公開儲存庫）	約 20 分鐘/天
Notion API	無限	約 200 次寫入/天
Supabase	500MB 資料庫, 2GB 傳輸	對大多數代理足夠
Google Sheets API	300 次請求/分鐘	適用於小型代理

---

需求範本

requests==2.31.0
beautifulsoup4==4.12.3
lxml==5.1.0
python-dotenv==1.0.1
pyyaml==6.0.2
notion-client==2.2.1   # 如果使用 Notion
# playwright==1.40.0   # 如果使用 JS 渲染網站，請取消註解

---

品質檢查清單

在標記代理完成前：

• [ ] config.yaml 控制所有使用者面向設定 — 無硬編碼值
• [ ] profile/context.md 存放使用者特定背景，供 AI 比對使用
• [ ] 每次儲存推送前依 URL 去重
• [ ] Gemini 客戶端具有模型降級鏈（4 個模型）
• [ ] 批次大小 ≤ 5 個項目/每次 API 呼叫
• [ ] maxOutputTokens ≥ 2048
• [ ] .env 在 .gitignore 中
• [ ] 提供 .env.example 供入門使用
• [ ] setup.py 在首次執行時建立資料庫結構
• [ ] enrich_existing.py 對舊資料列回填 AI 分數
• [ ] GitHub Actions 工作流程在每次執行後提交 feedback.json
• [ ] README 涵蓋：5 分鐘內完成設定、所需機密、自訂化

---

實際範例

"建立一個代理，監控 Hacker News 上 AI 新創募資的新聞"
"從 3 個電子商務網站爬取產品價格，並在降價時發出警報"
"追蹤標記為 'llm' 或 'agents' 的新 GitHub 儲存庫 — 摘要每個"
"從 LinkedIn 和 Cutshort 收集營運長職缺列表到 Notion"
"監控一個 subreddit 中提及我公司的貼文 — 分類情緒"
"每天爬取 arXiv 上我有興趣主題的新學術論文"
"追蹤運動賽事結果並在 Google Sheets 中維持即時表格"
"建立一個房地產列表監控器 — 新物件低於 1000 萬盧比時發出警報"

---

參考實作

一個使用此確切架構建立的完整運作代理，將從 4 個以上來源收集資料，批次 Gemini 呼叫，從儲存在 Notion 中的已應徵/已拒絕決策中學習，並在 GitHub Actions 上 100% 免費執行。按照上述步驟 1–9 建立你自己的代理。