data-scraper-agent

# Data Scraper Agent 为任何公开数据源构建生产就绪、AI 驱动的数据采集智能体。 按计划运行，使用免费 LLM 丰富结果，存储到数据库，并持续改进。 **技术栈：Python · Gemini Flash（免费）· GitHub Actions（免费）· Notion / Sheets / Supabase** ## 何时激活 - 用户想要抓取或监控任何公开网站或 API - 用户说"构建一个检查...的机器人"、"帮我监控 X"、"从...收集数据" - 用户想要跟踪职位、价格、新闻、仓库、体育比分、事件、列表 - 用户询问如何在免 hosting 费用的情况下自动化数据采集 - 用户想要一个能根据自身决策变得越来越聪明的智能体 ## 核心概念 ### 三层架构 每个数据采集智能体都有三层： ``` COLLECT → ENRICH → STORE │ │ │ 抓取器 AI (LLM) 数据库 按计划 评分/ Notion / 运行 摘要 Sheets / & 分类 Supabase ``` ### 免费技术栈 | 层 | 工具 | 原因 | |---|---|---| | **抓取** | `requests` + `BeautifulSoup` | 零成本，覆盖 80% 的公开网站 | | **JS 渲染站点** | `playwright`（免费） | 当 HTML 抓取失败时使用 | | **AI 丰富** | Gemini Flash REST API | 500 请求/天，1M token/天 — 免费 | | **存储** | 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。始终批量处理： ```python # 差：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 仓库 → 摘要新版本 - 新闻源 → 按主题 + 情感分类 - 体育结果 → 提取统计数据到追踪器 - 活动日历 → 按兴趣筛选 --- ### 步骤 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 # 一次性 DB/schema 创建 ├── enrich_existing.py # 为旧行回填 AI 评分 ├── requirements.txt └── .github/ └── workflows/ └── scraper.yml # GitHub Actions 调度 ``` --- ### 步骤 3：构建抓取器数据源 适用于任何数据源的模板： ```python # scraper/sources/my_source.py """ [数据源名称] — 从 [哪里] 抓取 [什么]。 方法：[REST API / HTML 抓取 / RSS 源] """ 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]: """ 返回具有一致 schema 的项目列表。 每个项目至少包含：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 数据转换为标准 schema。""" return { "name": raw.get("title", ""), "url": raw.get("link", ""), "source": "MySource", "date_found": datetime.now(timezone.utc).date().isoformat(), # 在此添加领域特定字段 } ``` **HTML 抓取模式：** ```python 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 源模式：** ```python 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 客户端 ```python # 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 管道（批量） ```python # 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"Item {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": "<为什么匹配或不匹配>"}} 按顺序对应每个项目]}} 简洁为要。评分 90+=优秀匹配，70-89=良好，50-69=一般，<50=较弱。""" ``` --- ### 步骤 6：构建反馈学习系统 ```python # 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 示例） ```python # 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: