crawl_docs.py

#!/usr/bin/env python3
"""
Crawl docs.status.network and save a local cache for agents.
Outputs: ~/.openclaw/shared/DOCS_CACHE.md (timestamped, with change detection)
Run before each pipeline cycle or daily via cron.
"""

import os
import re
import json
import hashlib
import urllib.request
import urllib.error
from datetime import datetime, timezone
from html.parser import HTMLParser

CACHE_PATH = os.path.expanduser("~/.openclaw/shared/DOCS_CACHE.md")
HASH_PATH = os.path.expanduser("~/.openclaw/shared/docs_hashes.json")

PAGES = [
    ("Home / Overview", "https://docs.status.network/"),
    ("Quick Start", "https://docs.status.network/quick-start"),
    ("Economic Model", "https://docs.status.network/tokenomics/economic-model"),
    ("Public Funding", "https://docs.status.network/tokenomics/public-funding"),
    ("Karmic Tokenomics", "https://docs.status.network/tokenomics/karmic-tokenomics"),
    ("SNT Staking", "https://docs.status.network/tokenomics/snt-staking"),
    ("Pre-Deposits", "https://docs.status.network/tokenomics/pre-deposits"),
    ("Network Details", "https://docs.status.network/general-info/network-details"),
    ("Gasless Transactions", "https://docs.status.network/general-info/gasless-transactions"),
    ("Add Status Network", "https://docs.status.network/general-info/add-status-network"),
    ("Bridge", "https://docs.status.network/general-info/bridge"),
    ("Bridging Testnet", "https://docs.status.network/general-info/bridge/bridging-testnet"),
    ("Contract Addresses: Tokens", "https://docs.status.network/general-info/contract-addresses/tokens"),
    ("Contract Addresses: Testnet", "https://docs.status.network/general-info/contract-addresses/testnet-contracts"),
    ("Contract Addresses: Pre-Deposit", "https://docs.status.network/general-info/contract-addresses/pre-deposit"),
    ("RPC", "https://docs.status.network/tools/rpc"),
    ("Testnet Faucets", "https://docs.status.network/tools/testnet-faucets"),
    ("Block Explorers", "https://docs.status.network/tools/block-explorers"),
    ("Deploy: Hardhat", "https://docs.status.network/tutorials/deploying-contracts/using-hardhat"),
    ("Deploy: Foundry", "https://docs.status.network/tutorials/deploying-contracts/using-foundry"),
    ("Deploy: Remix", "https://docs.status.network/tutorials/deploying-contracts/using-remix"),
    ("Running an RPC Node", "https://docs.status.network/tutorials/running-rpc-node"),
]


class HTMLTextExtractor(HTMLParser):
    """Strip HTML tags, keep text content. Skip script/style/nav/footer."""

    SKIP_TAGS = {"script", "style", "nav", "footer", "header", "noscript"}

    def __init__(self):
        super().__init__()
        self._pieces = []
        self._skip_depth = 0

    def handle_starttag(self, tag, attrs):
        if tag in self.SKIP_TAGS:
            self._skip_depth += 1

    def handle_endtag(self, tag):
        if tag in self.SKIP_TAGS and self._skip_depth > 0:
            self._skip_depth -= 1

    def handle_data(self, data):
        if self._skip_depth == 0:
            self._pieces.append(data)

    def get_text(self):
        raw = " ".join(self._pieces)
        raw = re.sub(r"\n{3,}", "\n\n", raw)
        raw = re.sub(r"[ \t]+", " ", raw)
        return raw.strip()


def fetch_page(url: str):
    """Fetch a URL and return the extracted text content."""
    req = urllib.request.Request(url, headers={"User-Agent": "StatusDocsBot/1.0"})
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            html = resp.read().decode("utf-8", errors="replace")
    except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError) as e:
        print(f"  WARN: Could not fetch {url}: {e}")
        return None

    parser = HTMLTextExtractor()
    parser.feed(html)
    return parser.get_text()


def load_previous_hashes() -> dict:
    if os.path.exists(HASH_PATH):
        with open(HASH_PATH, "r") as f:
            return json.load(f)
    return {}


def save_hashes(hashes: dict):
    os.makedirs(os.path.dirname(HASH_PATH), exist_ok=True)
    with open(HASH_PATH, "w") as f:
        json.dump(hashes, f, indent=2)


def main():
    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    prev_hashes = load_previous_hashes()
    new_hashes = {}
    changes = []

    sections = []
    total = len(PAGES)

    for i, (title, url) in enumerate(PAGES, 1):
        print(f"[{i}/{total}] Fetching: {title} ({url})")
        text = fetch_page(url)
        if text is None:
            sections.append(f"## {title}\n**URL**: {url}\n\n_Failed to fetch_\n")
            new_hashes[url] = prev_hashes.get(url, "error")
            continue

        content_hash = hashlib.sha256(text.encode()).hexdigest()[:16]
        new_hashes[url] = content_hash

        old_hash = prev_hashes.get(url)
        status_label = ""
        if old_hash is None:
            status_label = " (NEW)"
            changes.append(f"NEW: {title}")
        elif old_hash != content_hash:
            status_label = " (CHANGED)"
            changes.append(f"CHANGED: {title} ({url})")

        sections.append(
            f"## {title}{status_label}\n"
            f"**URL**: {url}\n"
            f"**Hash**: {content_hash}\n\n"
            f"{text}\n"
        )

    change_summary = "No changes detected." if not changes else "\n".join(f"- {c}" for c in changes)

    header = (
        f"# Status Network Docs Cache\n"
        f"**Crawled**: {now}\n"
        f"**Pages**: {total}\n"
        f"**Source**: https://docs.status.network/\n\n"
        f"## Change Summary\n{change_summary}\n\n"
        f"---\n\n"
    )

    os.makedirs(os.path.dirname(CACHE_PATH), exist_ok=True)
    with open(CACHE_PATH, "w") as f:
        f.write(header)
        f.write("\n---\n\n".join(sections))

    save_hashes(new_hashes)

    print(f"\nDone. Cache written to: {CACHE_PATH}")
    print(f"Crawl time: {now}")
    if changes:
        print(f"\nCHANGES DETECTED ({len(changes)}):")
        for c in changes:
            print(f"  {c}")
    else:
        print("No changes since last crawl.")


if __name__ == "__main__":
    main()