Build a fully automated AI-powered data collection agent for any public source — job boards, prices, news, GitHub, sports, anything. Scrapes on a schedule, enriches data with a free LLM (Gemini Flash), stores results in Notion/Sheets/Supabase, and learns from user feedback. Runs 100% free on GitHub Actions. Use when the user wants to monitor, collect, or track any public data automatically.
Build a fully automated AI-powered data collection agent for any public source — job boards, prices, news, GitHub, sports, anything. Scrapes on a schedule, enriches data with a free LLM (Gemini Flash), stores results in Notion/Sheets/Supabase, and learns from user feedback. Runs 100% free on GitHub Actions. Use when the user wants to monitor, collect, or track any public data automatically.
origin
community
Data Scraper Agent
あらゆる公開データソースに対応した、本番運用可能な AI 搭載データ収集エージェントを構築します。
スケジュールで実行し、無料の LLM で結果をエンリッチメントし、データベースに保存し、時間とともに改善されます。
# scraper/sources/my_source.py"""
[Source Name] — scrapes [what] from [where].
Method: [REST API / HTML scraping / 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)",
}
deffetch() -> list[dict]:
"""
Returns a list of items with consistent schema.
Each item must have at minimum: name, url, date_found.
"""
results = []
# ---- REST API source ----
resp = requests.get("https://api.example.com/items", headers=HEADERS, timeout=15)
if resp.status_code == 200:
for item in resp.json().get("results", []):
ifnot is_relevant(item.get("title", "")):
continue
results.append(_normalise(item))
return results
def_normalise(raw: dict) -> dict:
"""Convert raw API/HTML data to the standard schema."""return {
"name": raw.get("title", ""),
"url": raw.get("link", ""),
"source": "MySource",
"date_found": datetime.now(timezone.utc).date().isoformat(),
# add domain-specific fields here
}
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"]
ifnot link.startswith("http"):
link = f"https://example.com{link}"
RSS フィードパターン：
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.pyimport 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",
]
defgenerate(prompt: str, model: str = "", rate_limit: float = 7.0) -> dict:
"""Call Gemini with auto-fallback on 429. Returns parsed JSON or {}."""global _last_call
api_key = os.environ.get("GEMINI_API_KEY", "")
ifnot 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)
continuereturn {}
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.pyimport json
import yaml
from pathlib import Path
from ai.client import generate
defanalyse_batch(items: list[dict], context: str = "", preference_prompt: str = "") -> list[dict]:
"""Analyse items in batches. Returns items enriched with AI fields."""
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 inrange(0, len(items), batch_size)]
print(f" [AI] {len(items)} items → {len(batches)} API calls")
enriched = []
for i, batch inenumerate(batches):
print(f" [AI] Batch {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 inenumerate(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() ifnot k.startswith('_')})}"for i, item inenumerate(batch)
)
returnf"""Analyse these {len(batch)} items and return a JSON object.
# Items
{items_text}
# User Context
{context[:800] if context else"Not provided"}
# User Priorities
{chr(10).join(f"- {p}"for p in priorities)}{preference_prompt}
# Instructions
Return: {{"analyses": [{{"score": <0-100>, "summary": "<2 sentences>", "notes": "<why this matches or doesn't>"}} for each item in order]}}
Be concise. Score 90+=excellent match, 70-89=good, 50-69=ok, <50=weak."""
ステップ 6: フィードバック学習システムの構築
# ai/memory.py"""Learn from user decisions to improve future scoring."""import json
from pathlib import Path
FEEDBACK_PATH = Path(__file__).parent.parent / "data" / "feedback.json"defload_feedback() -> dict:
if FEEDBACK_PATH.exists():
try:
return json.loads(FEEDBACK_PATH.read_text())
except (json.JSONDecodeError, OSError):
passreturn {"positive": [], "negative": []}
defsave_feedback(fb: dict):
FEEDBACK_PATH.parent.mkdir(parents=True, exist_ok=True)
FEEDBACK_PATH.write_text(json.dumps(fb, indent=2))
defbuild_preference_prompt(feedback: dict, max_examples: int = 15) -> str:
"""Convert feedback history into a prompt bias section."""
lines = []
if feedback.get("positive"):
lines.append("# Items the user LIKED (positive signal):")
for e in feedback["positive"][-max_examples:]:
lines.append(f"- {e}")
if feedback.get("negative"):
lines.append("\n# Items the user SKIPPED/REJECTED (negative signal):")
for e in feedback["negative"][-max_examples:]:
lines.append(f"- {e}")
if lines:
lines.append("\nUse these patterns to bias scoring on new items.")
return"\n".join(lines)
ストレージレイヤーとの統合： 各実行後に DB からポジティブ/ネガティブステータスのアイテムをクエリし、抽出したパターンで save_feedback() を呼び出します。
ステップ 7: ストレージの構築（Notion の例）
# storage/notion_sync.pyimport os
from notion_client import Client
from notion_client.errors import APIResponseError
_client = Nonedefget_client():
global _client
if _client isNone:
_client = Client(auth=os.environ["NOTION_TOKEN"])
return _client
defget_existing_urls(db_id: str) -> set[str]:
"""Fetch all URLs already stored — used for deduplication."""
client, seen, cursor = get_client(), set(), NonewhileTrue:
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)
ifnot resp["has_more"]: break
cursor = resp["next_cursor"]
return seen
defpush_item(db_id: str, item: dict) -> bool:
"""Push one item to Notion. Returns True on success."""
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 fieldsif item.get("ai_score") isnotNone:
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)
returnTrueexcept APIResponseError as e:
print(f"[notion] Push failed: {e}")
returnFalsedefsync(db_id: str, items: list[dict]) -> tuple[int, int]:
existing = get_existing_urls(db_id)
added = skipped = 0for item in items:
if item.get("url") in existing:
skipped += 1; continueif push_item(db_id, item):
added += 1; existing.add(item["url"])
else:
skipped += 1return added, skipped
ステップ 8: main.py でのオーケストレーション
# scraper/main.pyimport os, sys, yaml
from pathlib import Path
from dotenv import load_dotenv
load_dotenv()
from scraper.sources import my_source # add your sources# NOTE: This example uses Notion. If storage.provider is "sheets" or "supabase",# replace this import with storage.sheets_sync or storage.supabase_sync and update# the env var and sync() call accordingly.from storage.notion_sync import sync
SOURCES = [
("My Source", my_source.fetch),
]
defai_enabled():
returnbool(os.environ.get("GEMINI_API_KEY"))
defmain():
config = yaml.safe_load((Path(__file__).parent.parent / "config.yaml").read_text())
provider = config.get("storage", {}).get("provider", "notion")
# Resolve the storage target identifier from env based on providerif provider == "notion":
db_id = os.environ.get("NOTION_DATABASE_ID")
ifnot db_id:
print("ERROR: NOTION_DATABASE_ID not set"); sys.exit(1)
else:
# Extend here for sheets (SHEET_ID) or supabase (SUPABASE_TABLE) etc.print(f"ERROR: provider '{provider}' not yet wired in 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)} items")
all_items.extend(items)
except Exception as e:
print(f"[{name}] FAILED: {e}")
# Deduplicate by URL
seen, deduped = set(), []
for item in all_items:
if (url := item.get("url", "")) and url notin seen:
seen.add(url); deduped.append(item)
print(f"Unique items: {len(deduped)}")
if ai_enabled() and deduped:
from ai.memory import load_feedback, build_preference_prompt
from ai.pipeline import analyse_batch
# load_feedback() reads data/feedback.json written by your feedback sync script.# To keep it current, implement a separate feedback_sync.py that queries your# storage provider for items with positive/negative statuses and calls 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] Skipped — GEMINI_API_KEY not set")
added, skipped = sync(db_id, deduped)
print(f"Done — {added} new, {skipped} existing")
if __name__ == "__main__":
main()
ステップ 9: GitHub Actions ワークフロー
# .github/workflows/scraper.ymlname:DataScraperAgenton:schedule:-cron:"0 */3 * * *"# every 3 hours — adjust to your needsworkflow_dispatch:# allow manual triggerpermissions:contents:write# required for the feedback-history commit stepjobs:scrape:runs-on:ubuntu-latesttimeout-minutes:20steps:-uses:actions/checkout@v4-uses:actions/setup-python@v5with:python-version:"3.11"cache:"pip"-run:pipinstall-rrequirements.txt# Uncomment if Playwright is enabled in requirements.txt# - name: Install Playwright browsers# run: python -m playwright install chromium --with-deps-name:Runagentenv:NOTION_TOKEN:${{secrets.NOTION_TOKEN}}NOTION_DATABASE_ID:${{secrets.NOTION_DATABASE_ID}}GEMINI_API_KEY:${{secrets.GEMINI_API_KEY}}run:python-mscraper.main-name:Commitfeedbackhistoryrun:|
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 テンプレート
# Customise this file — no code changes needed# What to collect (pre-filter before AI)filters:required_keywords: [] # item must contain at least oneblocked_keywords: [] # item must not contain any# Your priorities — AI uses these for scoringpriorities:-"example priority 1"-"example priority 2"# Storagestorage:provider:"notion"# notion | sheets | supabase | sqlite# Feedback learningfeedback:positive_statuses: ["Saved", "Applied", "Interested"]
negative_statuses: ["Skip", "Rejected", "Not relevant"]
# AI settingsai:enabled:truemodel:"gemini-2.5-flash"min_score:0# filter out items below this scorerate_limit_seconds:7# seconds between API callsbatch_size:5# items per API call
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 フィード
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", "")
