#!/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()