crawl_runner_template.py
7.29 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
#!/usr/bin/env python3
"""
crawl_runner.py — Template crawler for innovator-skill data acquisition.
Reads crawl_config.md to determine sources, then fetches and stores data.
Usage:
python crawl_runner.py # crawl all sources
python crawl_runner.py --source {name} # crawl specific source
python crawl_runner.py --dry-run # list what would be crawled
Output:
crawl/{source_name}/{YYYYMMDD}_{source}_{title}.md
crawl_log.md (appended)
"""
import os
import re
import time
import hashlib
import argparse
import requests
from datetime import datetime
from pathlib import Path
# ─── Configuration ──────────────────────────────────────────────────────────
SKILL_DIR = Path(__file__).parent # e.g., /innovator/xianxia_novel_skill/
CRAWL_DIR = SKILL_DIR / "crawl"
LOG_FILE = SKILL_DIR / "crawl_log.md"
CONFIG_FILE = SKILL_DIR / "crawl_config.md"
MAX_RETRIES = 3
RETRY_BACKOFF = 2 # seconds (doubled each retry)
REQUEST_DELAY = 1 # seconds between requests (be polite)
HEADERS = {
"User-Agent": "InnovatorSkillBot/1.0 (research crawler; contact: your@email.com)"
}
# ─── Source definitions (customize per skill) ───────────────────────────────
# These should be populated from crawl_config.md during skill creation.
# Format: list of dicts with keys: name, url, type, fetch_fn, industry_relevance
SOURCES = [
# Example structure — replace with real sources per skill:
# {
# "name": "source_name",
# "url": "https://example.com/rankings",
# "type": "unstructured", # structured | unstructured | experience
# "industry_relevance": "elite", # elite | average | broad
# "fetch_fn": "fetch_generic_page", # which fetch function to use
# "params": {} # extra params passed to fetch_fn
# },
]
# ─── Utility functions ───────────────────────────────────────────────────────
def get_content_hash(content: str) -> str:
return hashlib.md5(content.encode()).hexdigest()[:8]
def load_seen_hashes() -> set:
"""Load previously seen content hashes to skip duplicates."""
seen = set()
if LOG_FILE.exists():
for line in LOG_FILE.read_text().splitlines():
if line.startswith("hash:"):
seen.add(line.split("hash:")[1].strip())
return seen
def log(message: str, level: str = "INFO"):
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
line = f"[{ts}] [{level}] {message}"
print(line)
with open(LOG_FILE, "a") as f:
f.write(line + "\n")
def save_item(source_name: str, title: str, content: str, metadata: dict) -> Path:
"""Save a crawled item as a markdown file."""
date_str = datetime.now().strftime("%Y%m%d_%H%M%S")
safe_title = re.sub(r"[^\w\-]", "_", title)[:60]
filename = f"{date_str}_{source_name}_{safe_title}.md"
out_dir = CRAWL_DIR / source_name
out_dir.mkdir(parents=True, exist_ok=True)
out_path = out_dir / filename
md_content = f"""---
source: {source_name}
crawled_at: {datetime.now().isoformat()}
industry_relevance: {metadata.get('industry_relevance', 'unknown')}
data_type: {metadata.get('type', 'unstructured')}
url: {metadata.get('url', '')}
title: {title}
content_hash: {get_content_hash(content)}
---
{content}
"""
out_path.write_text(md_content, encoding="utf-8")
return out_path
# ─── Fetch functions (customize per skill) ──────────────────────────────────
def fetch_with_retry(url: str, retries: int = MAX_RETRIES) -> requests.Response | None:
"""Fetch URL with exponential backoff retry."""
for attempt in range(retries):
try:
resp = requests.get(url, headers=HEADERS, timeout=15)
resp.raise_for_status()
return resp
except requests.RequestException as e:
wait = RETRY_BACKOFF ** attempt
log(f"Attempt {attempt+1}/{retries} failed for {url}: {e}. Waiting {wait}s.", "WARN")
time.sleep(wait)
log(f"All retries exhausted for {url}", "ERROR")
return None
def fetch_generic_page(source: dict) -> list[dict]:
"""
Generic page fetcher — returns list of {title, content, url} dicts.
CUSTOMIZE THIS per skill. For example, parse a rankings page and
extract top items. This default just fetches raw page text.
"""
resp = fetch_with_retry(source["url"])
if not resp:
return []
return [{
"title": source["name"],
"content": resp.text[:10000], # truncate large pages
"url": source["url"]
}]
# ─── Main crawl loop ─────────────────────────────────────────────────────────
def crawl_source(source: dict, seen_hashes: set, dry_run: bool = False) -> int:
"""Crawl a single source. Returns count of new items saved."""
log(f"Crawling source: {source['name']} ({source['url']})")
if dry_run:
log(f"[DRY RUN] Would crawl: {source['name']}", "INFO")
return 0
fetch_fn_name = source.get("fetch_fn", "fetch_generic_page")
fetch_fn = globals().get(fetch_fn_name, fetch_generic_page)
items = fetch_fn(source)
saved = 0
for item in items:
content_hash = get_content_hash(item["content"])
if content_hash in seen_hashes:
log(f" Skipping duplicate: {item['title']} (hash: {content_hash})", "SKIP")
continue
path = save_item(
source_name=source["name"],
title=item["title"],
content=item["content"],
metadata={**source, "url": item.get("url", source["url"])}
)
# Record hash to prevent future duplicates
with open(LOG_FILE, "a") as f:
f.write(f"hash:{content_hash}\n")
seen_hashes.add(content_hash)
log(f" Saved: {path.name}")
saved += 1
time.sleep(REQUEST_DELAY) # polite delay
return saved
def main():
parser = argparse.ArgumentParser(description="Innovator-skill crawler")
parser.add_argument("--source", help="Crawl only this source name")
parser.add_argument("--dry-run", action="store_true", help="List what would be crawled")
args = parser.parse_args()
CRAWL_DIR.mkdir(parents=True, exist_ok=True)
seen_hashes = load_seen_hashes()
total_saved = 0
sources_to_crawl = SOURCES
if args.source:
sources_to_crawl = [s for s in SOURCES if s["name"] == args.source]
if not sources_to_crawl:
print(f"Source '{args.source}' not found in SOURCES list.")
return
log(f"Starting crawl run. Sources: {len(sources_to_crawl)}")
for source in sources_to_crawl:
try:
count = crawl_source(source, seen_hashes, dry_run=args.dry_run)
total_saved += count
except Exception as e:
log(f"Error crawling {source['name']}: {e}", "ERROR")
log(f"Crawl complete. New items saved: {total_saved}")
if __name__ == "__main__":
main()