DEV Community

coreclaw
coreclaw

Posted on

Web Data for Market Research: How to Build a Competitive Intelligence Pipeline with Python

#api #python #webscraping #data

The fastest way to build competitive intelligence is to treat public web data as a scheduled dataset, not a one-off search session. A small Python pipeline that collects, normalizes, and stores competitor pricing, product launches, and messaging changes will outperform manual research within a week.

TL;DR: Define the signals you care about (price, product, messaging), write a small Python scheduler that fetches public pages, normalize the results into a JSON or database record, and diff each run against the previous one. The code below uses environment variables so you can plug in any endpoint or proxy config you already have.

Why Manual Market Research Stops Working

Spreadsheets and browser tabs work for three competitors. They collapse at ten. The problems compound quickly:

  • Stale data. A price you recorded on Monday may change by Wednesday. Manual checks cannot keep up with dynamic sites.
  • Hidden changes. Competitors update hero copy, add testimonials, or restructure navigation without announcing it. You only notice what you remember to look for.
  • No history. A single snapshot tells you what a site looked like today. It does not tell you when a feature was added, when pricing shifted, or when a page was removed.
  • Human inconsistency. Two team members record the same field differently. One rounds prices, another copies the full string. Normalization becomes impossible.

A pipeline replaces these problems with repeatable runs, structured output, and historical diffs.

What a Competitive Intelligence Pipeline Actually Is

At its core, the pipeline is three steps:

  1. Collect public pages on a schedule.
  2. Extract structured fields from each page.
  3. Store and compare results across runs to surface changes.

The input is a list of URLs or search queries. The output is a timestamped record set you can query, diff, or feed into a dashboard. You do not need a complex orchestrator to start. A scheduled Python script, a JSONL file, and a simple diff function are enough for most teams.

Building the Pipeline

Step 1 — Define Your Signals

Before writing code, list the specific signals that matter to your business. Examples:

Signal Source Frequency
Pricing page values Competitor pricing pages Daily
Product feature lists Product or solutions pages Weekly
Job postings Careers pages Weekly
Customer testimonials Case study or review pages Monthly
Meta descriptions and titles Homepage and landing pages Weekly

Limit your first version to three signals. Scope creep is the main reason these projects stall.

Step 2 — Configure Environment Variables

Never hard-code endpoints, keys, or target URLs in the script. Use environment variables so the same code works across development, staging, and production without edits.

import os

# Data collection config
ENDPOINT = os.environ.get("SCRAPER_ENDPOINT")
API_KEY = os.environ.get("SCRAPER_API_KEY")
TARGET_URLS = os.environ.get("TARGET_URLS", "").split(",")
SCHEDULE_HOURS = int(os.environ.get("SCHEDULE_HOURS", "24"))

# Storage config
OUTPUT_DIR = os.environ.get("OUTPUT_DIR", "./snapshots")
Enter fullscreen mode Exit fullscreen mode

Set these in a .env file or your deployment environment. If you are using a managed scraper API, copy the endpoint from the provider console. If you are self-hosting, point ENDPOINT to your local or cloud scraper service.

Step 3 — Fetch and Normalize

The script below fetches each target URL, extracts a small set of fields, and writes a timestamped JSON file. It is designed to work with any HTTP client or scraper API that returns HTML or structured JSON.

import json
import os
import hashlib
from datetime import datetime, timezone
from pathlib import Path
import requests
from bs4 import BeautifulSoup

ENDPOINT = os.environ.get("SCRAPER_ENDPOINT")
API_KEY = os.environ.get("SCRAPER_API_KEY")
TARGET_URLS = [u.strip() for u in os.environ.get("TARGET_URLS", "").split(",") if u.strip()]
OUTPUT_DIR = Path(os.environ.get("OUTPUT_DIR", "./snapshots"))
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.0",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.5",
    "Accept-Encoding": "gzip, deflate, br",
    "DNT": "1",
    "Connection": "keep-alive",
}


def fetch_page(url: str) -> str:
    """Fetch raw HTML. If an endpoint is configured, route through it."""
    if ENDPOINT:
        resp = requests.post(
            ENDPOINT,
            headers={"Authorization": f"Bearer {API_KEY}"} if API_KEY else {},
            json={"url": url, "render": True},
            timeout=60,
        )
    else:
        resp = requests.get(url, headers=HEADERS, timeout=30)
    resp.raise_for_status()
    return resp.text


def extract_signals(html: str, url: str) -> dict:
    """Extract a small, consistent set of fields from HTML."""
    soup = BeautifulSoup(html, "html.parser")

    title = soup.title.string.strip() if soup.title else ""
    meta_desc = ""
    meta_tag = soup.find("meta", attrs={"name": "description"})
    if meta_tag:
        meta_desc = meta_tag.get("content", "")

    # Example: collect all visible h2 headings as feature signals
    headings = [h.get_text(strip=True) for h in soup.find_all("h2")]

    # Example: collect all paragraph text for NLP later
    paragraphs = [p.get_text(strip=True) for p in soup.find_all("p") if len(p.get_text(strip=True)) > 40]

    return {
        "url": url,
        "crawled_at": datetime.now(timezone.utc).isoformat(),
        "title": title,
        "meta_description": meta_desc,
        "headings": headings[:20],  # cap to avoid bloat
        "paragraphs": paragraphs[:10],
        "content_hash": hashlib.sha256(html.encode("utf-8")).hexdigest()[:16],
    }


def save_snapshot(record: dict) -> Path:
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    safe_url = record["url"].replace("https://", "").replace("/", "_")[:80]
    filename = f"{timestamp}_{safe_url}.json"
    path = OUTPUT_DIR / filename
    path.write_text(json.dumps(record, indent=2, ensure_ascii=False), encoding="utf-8")
    return path


def run_pipeline():
    if not TARGET_URLS:
        raise SystemExit("No TARGET_URLS configured.")

    for url in TARGET_URLS:
        try:
            html = fetch_page(url)
            record = extract_signals(html, url)
            path = save_snapshot(record)
            print(f"Saved snapshot: {path}")
        except Exception as e:
            print(f"Failed for {url}: {e}")


if __name__ == "__main__":
    run_pipeline()
Enter fullscreen mode Exit fullscreen mode

Step 4 — Detect Changes Across Runs

Collecting data is only half the job. The value comes from knowing what changed. Add a diff function that compares the latest snapshot against the previous one for the same URL.

def load_latest_snapshot(url: str) -> dict | None:
    safe_url = url.replace("https://", "").replace("/", "_")[:80]
    matches = sorted(OUTPUT_DIR.glob(f"*_{safe_url}.json"))
    if len(matches) >= 2:
        return json.loads(matches[-2].read_text(encoding="utf-8"))
    return None


def diff_records(current: dict, previous: dict | None) -> dict:
    if previous is None:
        return {"status": "first_run", "changes": []}

    changes = []
    for field in ["title", "meta_description"]:
        if current.get(field) != previous.get(field):
            changes.append({
                "field": field,
                "before": previous.get(field),
                "after": current.get(field),
            })

    if current.get("content_hash") != previous.get("content_hash"):
        # Hash changed; check headings
        old_h = set(previous.get("headings", []))
        new_h = set(current.get("headings", []))
        added = list(new_h - old_h)
        removed = list(old_h - new_h)
        if added or removed:
            changes.append({"field": "headings", "added": added, "removed": removed})

    return {
        "status": "changed" if changes else "unchanged",
        "changes": changes,
    }
Enter fullscreen mode Exit fullscreen mode

Insert the diff call inside run_pipeline() after save_snapshot():

previous = load_latest_snapshot(url)
diff = diff_records(record, previous)
print(json.dumps(diff, indent=2))
Enter fullscreen mode Exit fullscreen mode

What the Output Looks Like

A single snapshot file is a JSON record:

{
  "url": "https://example.com/pricing",
  "crawled_at": "2026-08-28T06:00:00+00:00",
  "title": "Pricing - Example SaaS",
  "meta_description": "Transparent pricing for teams of all sizes.",
  "headings": ["Starter", "Pro", "Enterprise", "FAQ"],
  "paragraphs": ["Starter includes up to 5 seats...", "Pro adds API access..."],
  "content_hash": "a3f7c2d8e1b90425"
}
Enter fullscreen mode Exit fullscreen mode

A diff record for the same URL on the next run might look like this:

{
  "status": "changed",
  "changes": [
    {
      "field": "headings",
      "added": ["New AI Features"],
      "removed": []
    }
  ]
}
Enter fullscreen mode Exit fullscreen mode

This tells you exactly when a competitor added a new section to their pricing page.

Business Use Cases

SaaS pricing teams run daily checks on competitor pricing pages. When a diff shows a price change, they receive a Slack alert and decide whether to respond.

Product marketers track messaging changes. If three competitors start using the same feature phrase, it signals a market shift worth investigating.

Sales operations monitor job postings. A sudden hiring spike in a competitor's sales department often precedes a territory push or new product launch.

Investors and analysts collect public landing-page data at scale to build sentiment signals without relying solely on earnings calls.

Build vs Buy vs Platform

Approach Best for Maintenance burden Scaling path
Self-hosted script (above) One to three signals, technical team High: proxies, parsing, storage Add workers or containers
Managed scraper API Teams that want structured data without infrastructure Low: provider handles proxy rotation and parsing Increase request volume
Ready-made marketplace worker Common sources (maps, e-commerce, search) Very low: pre-built extractor and schema Run multiple workers in parallel
Full platform Multiple signals, multiple teams, scheduling, alerts Lowest: UI-based configuration Add seats and worker concurrency

A self-hosted script is the right starting point when you need full control over extraction logic. As signal volume grows, moving to a managed scraper API or a ready-made worker marketplace reduces the time you spend on proxy management and parser maintenance.

The CoreClaw Workers platform lets you deploy scrapers without managing servers. If you prefer pre-built extractors for common sources, the CoreClaw scraper marketplace offers ready-made workers you can run immediately.

Limitations and Compliance

  • Terms of service. Respect the target site's terms. Some sites explicitly prohibit automated access. Review robots.txt and site policies before adding a URL to your target list.
  • Rate limits. Public sites may throttle or block repeated requests. Implement exponential backoff and respect Retry-After headers.
  • Layout fragility. The BeautifulSoup selectors above are simple. If a site redesigns, your extraction logic may break. For production use, prefer structured data or API responses when available, or use more robust selector strategies.
  • Data freshness. A snapshot is only as fresh as your schedule. Daily collection catches most pricing changes. Real-time monitoring requires more infrastructure.
  • Legal boundaries. Do not collect personal data, login-protected content, or data behind access controls. Focus on public-facing business information only.

Public-web-data compliance is an evolving area. Review applicable laws in your jurisdiction and the target site's terms before running any automated collection workflow.

FAQ

Is there an official API for competitor data?

There is no single official API that covers all competitors. Most teams build pipelines like the one above, or use a managed scraper API that returns structured data from public pages.

What data fields should I start with?

Start with title, meta description, visible headings, and any pricing text. These fields change when messaging or pricing shifts, and they are easy to extract reliably.

How often should the workflow run?

Daily for pricing and job postings. Weekly for product pages and testimonials. Monthly for deep content audits. Adjust based on how quickly your market moves.

Can this connect to a CRM or BI tool?

Yes. Replace the JSON file output with a database insert, webhook, or API call. Most teams send diffs to Slack or email first, then graduate to a CRM or BI dashboard.

What happens when a page layout changes?

Your extraction logic may return empty or incorrect fields. Monitor for unexpected empty results and alert when a page's content hash changes but your structured fields do not.

How do I avoid getting blocked?

Use a rotating proxy, respect rate limits, vary request timing, and send realistic headers. If you do not want to manage proxies yourself, use a scraper API that includes residential proxy rotation.

What should I verify before production use?

Confirm that every target URL permits automated access, test your extraction logic against multiple page variants, and set up alerting for pipeline failures. Always verify current pricing and quotas on your provider's official page. You can check the latest CoreClaw pricing details for pay-per-result and subscription options.

Summary and Next Steps

A competitive intelligence pipeline does not need to be complex. Define your signals, write a scheduled fetch-and-diff script, and store timestamped snapshots. Within a week you will have a historical dataset that manual research cannot match.

If you are ready to move beyond self-hosted scripts, explore the CoreClaw scraper marketplace for ready-made workers, or start building a custom pipeline on the CoreClaw Workers platform.

Related Reading

Top comments (1)

Collapse
 
topstar_ai profile image
Luis Cruz •

The idea of treating public web data as a scheduled dataset is a game changer for competitive intelligence. By automating the collection and normalization of competitor signals, you significantly reduce human error and maintain a more accurate historical record. One improvement I’d suggest is incorporating a versioning system for your snapshots to easily track changes over time, which could help streamline analysis further. If you're considering enhancing the pipeline's architecture or integrating more complex data processing, I’d be happy to discuss a paid collaboration to contribute to that aspect. What other signals do you see as valuable to add in future iterations?