Web Intelligence Skill

Role

You are an elite web intelligence engineer. You extract structured, clean data

from the open web using a suite of tools — from simple requests to full headless

browser automation. You respect robots.txt, avoid hammering servers, and build

robust pipelines that handle errors, redirects, and anti-bot measures gracefully.

Part 1: Setup

pip install requests beautifulsoup4 lxml playwright httpx
playwright install chromium

Part 2: Simple Requests + BeautifulSoup

Basic Scraper with Headers

import requests
from bs4 import BeautifulSoup
import time
import random

# Always set a real User-Agent
HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/120.0.0.0 Safari/537.36"
    ),
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-GB,en;q=0.5",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
}

def fetch_page(url: str, timeout: int = 10, retries: int = 3) -> BeautifulSoup | None:
    """Fetch a URL and return a BeautifulSoup object."""
    session = requests.Session()
    session.headers.update(HEADERS)

    for attempt in range(retries):
        try:
            response = session.get(url, timeout=timeout)
            response.raise_for_status()

            # Polite delay between requests
            time.sleep(random.uniform(1.0, 2.5))

            return BeautifulSoup(response.content, "lxml")

        except requests.HTTPError as e:
            if e.response.status_code == 429:
                wait = 30 * (attempt + 1)
                print(f"Rate limited. Waiting {wait}s...")
                time.sleep(wait)
            elif e.response.status_code == 403:
                print(f"Access denied: {url}")
                return None
            else:
                print(f"HTTP error {e.response.status_code}: {url}")

        except requests.RequestException as e:
            print(f"Request error attempt {attempt + 1}: {e}")
            time.sleep(2 ** attempt)

    return None

Data Extraction Patterns

def extract_text_blocks(soup: BeautifulSoup, selector: str) -> list[str]:
    """Extract text from all matching CSS selectors."""
    elements = soup.select(selector)
    return [el.get_text(strip=True) for el in elements if el.get_text(strip=True)]


def extract_links(soup: BeautifulSoup, base_url: str,
                   filter_pattern: str = None) -> list[str]:
    """Extract all links from a page, with optional pattern filter."""
    from urllib.parse import urljoin, urlparse
    import re

    links = []
    for a in soup.find_all("a", href=True):
        href = urljoin(base_url, a["href"])
        # Only include same-domain links
        if urlparse(href).netloc == urlparse(base_url).netloc:
            if filter_pattern is None or re.search(filter_pattern, href):
                links.append(href)

    return list(set(links))  # Deduplicate


def extract_structured_table(soup: BeautifulSoup,
                               table_selector: str = "table") -> list[dict]:
    """Extract the first matching table as a list of dicts."""
    table = soup.select_one(table_selector)
    if not table:
        return []

    headers = [th.get_text(strip=True) for th in table.find_all("th")]
    rows = []
    for tr in table.find_all("tr")[1:]:  # Skip header row
        cells = [td.get_text(strip=True) for td in tr.find_all(["td", "th"])]
        if cells and len(cells) == len(headers):
            rows.append(dict(zip(headers, cells)))

    return rows

Part 3: Playwright for JavaScript-Rendered Pages

Use Playwright when sites use React/Vue/Angular or load content via JS.

from playwright.sync_api import sync_playwright
import time

def fetch_js_page(url: str, wait_for: str = None,
                   timeout: int = 30000) -> str | None:
    """Fetch a JavaScript-rendered page using Playwright."""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context(
            user_agent=(
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                "AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36"
            ),
            viewport={"width": 1280, "height": 800}
        )
        page = context.new_page()

        try:
            page.goto(url, timeout=timeout, wait_until="networkidle")

            if wait_for:
                page.wait_for_selector(wait_for, timeout=timeout)

            html = page.content()
            browser.close()
            return html

        except Exception as e:
            print(f"Playwright error: {e}")
            browser.close()
            return None


def scroll_and_extract(url: str, item_selector: str,
                        max_scrolls: int = 10) -> list[str]:
    """Scroll a page to load lazy content and extract items."""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        page.goto(url, wait_until="networkidle")

        items = set()
        for _ in range(max_scrolls):
            # Get current items
            elements = page.query_selector_all(item_selector)
            for el in elements:
                text = el.inner_text().strip()
                if text:
                    items.add(text)

            # Scroll to bottom
            page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
            time.sleep(1.5)

        browser.close()
        return list(items)

Part 4: Structured Data Extraction

JSON-LD and Schema.org

import json

def extract_json_ld(soup: BeautifulSoup) -> list[dict]:
    """Extract JSON-LD structured data from a page."""
    results = []
    for script in soup.find_all("script", type="application/ld+json"):
        try:
            data = json.loads(script.string)
            results.append(data)
        except (json.JSONDecodeError, TypeError):
            pass
    return results


def extract_meta_tags(soup: BeautifulSoup) -> dict:
    """Extract Open Graph and Twitter card meta tags."""
    meta = {}
    for tag in soup.find_all("meta"):
        name = tag.get("property") or tag.get("name") or tag.get("itemprop")
        content = tag.get("content")
        if name and content:
            meta[name] = content
    return meta

RSS / XML Feed Parser

import feedparser

def parse_rss_feed(url: str) -> list[dict]:
    """Parse an RSS or Atom feed."""
    feed = feedparser.parse(url)
    articles = []

    for entry in feed.entries:
        articles.append({
            "title": entry.get("title", ""),
            "link": entry.get("link", ""),
            "published": entry.get("published", ""),
            "summary": entry.get("summary", ""),
            "tags": [t.term for t in entry.get("tags", [])]
        })

    return articles

Part 5: Crawling Pipeline

from collections import deque
from urllib.parse import urlparse
import time

class SiteCrawler:
    """Polite single-domain web crawler."""

    def __init__(self, start_url: str, max_pages: int = 100,
                  delay: float = 1.5):
        self.start_url = start_url
        self.domain = urlparse(start_url).netloc
        self.max_pages = max_pages
        self.delay = delay
        self.visited: set[str] = set()
        self.queue = deque([start_url])
        self.results: list[dict] = []

    def crawl(self) -> list[dict]:
        while self.queue and len(self.visited) < self.max_pages:
            url = self.queue.popleft()

            if url in self.visited:
                continue

            self.visited.add(url)
            soup = fetch_page(url)

            if soup:
                # Extract data
                result = self.extract(url, soup)
                if result:
                    self.results.append(result)

                # Queue new links
                new_links = extract_links(soup, url)
                for link in new_links:
                    if link not in self.visited:
                        self.queue.append(link)

            time.sleep(self.delay)

        return self.results

    def extract(self, url: str, soup: BeautifulSoup) -> dict | None:
        """Override this to customise what gets extracted per page."""
        title = soup.find("title")
        h1 = soup.find("h1")
        return {
            "url": url,
            "title": title.text.strip() if title else "",
            "h1": h1.text.strip() if h1 else "",
            "text_length": len(soup.get_text())
        }

Part 6: IMI Brand Monitoring

class IMIBrandMonitor:
    """Monitor brand mentions across configured web sources."""

    def __init__(self, brand: str, sources: list[str]):
        self.brand = brand
        self.brand_lower = brand.lower()
        self.sources = sources

    def scan_sources(self) -> list[dict]:
        """Scan all sources for brand mentions."""
        mentions = []
        for url in self.sources:
            soup = fetch_page(url)
            if not soup:
                continue

            # Find paragraphs mentioning brand
            for p in soup.find_all(["p", "h1", "h2", "h3", "li"]):
                text = p.get_text(strip=True)
                if self.brand_lower in text.lower():
                    mentions.append({
                        "source": url,
                        "text": text[:500],
                        "element": p.name,
                    })

        return mentions

    def get_sentiment_snippets(self) -> list[dict]:
        """Get brand mentions with surrounding context."""
        results = []
        for mention in self.scan_sources():
            results.append({
                **mention,
                "brand": self.brand,
                "word_count": len(mention["text"].split())
            })
        return results

Output Standards

• ALWAYS check robots.txt before crawling: https://domain.com/robots.txt
• ALWAYS add delays between requests (minimum 1 second)
• Never scrape personal data without legal basis
• Log all HTTP errors; never silently swallow them
• Store raw HTML alongside extracted data for reproducibility
• Use lxml parser for BeautifulSoup (faster and more lenient than html.parser)
• Handle encoding explicitly: response.encoding = response.apparent_encoding

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)