Build a Competitor Intelligence Dashboard with WebShot API

pip install requests schedule Pillow slack-sdk jinja2

import requests
import os
from datetime import datetime
from pathlib import Path
 
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://apivult.com/webshot/v1"
 
# Define the pages you want to track
COMPETITOR_PAGES = [
    {
        "id": "competitor_a_pricing",
        "company": "Competitor A",
        "url": "https://competitor-a.com/pricing",
        "description": "Pricing page",
        "importance": "critical"   # critical | high | medium
    },
    {
        "id": "competitor_a_features",
        "company": "Competitor A",
        "url": "https://competitor-a.com/features",
        "description": "Features overview",
        "importance": "high"
    },
    {
        "id": "competitor_b_pricing",
        "company": "Competitor B",
        "url": "https://competitor-b.com/pricing",
        "description": "Pricing page",
        "importance": "critical"
    },
    {
        "id": "competitor_b_integrations",
        "company": "Competitor B",
        "url": "https://competitor-b.com/integrations",
        "description": "Integrations list",
        "importance": "high"
    },
    {
        "id": "competitor_c_homepage",
        "company": "Competitor C",
        "url": "https://competitor-c.com",
        "description": "Homepage hero",
        "importance": "medium"
    }
]

def capture_screenshot(
    url: str,
    width: int = 1440,
    full_page: bool = True,
    wait_ms: int = 2000
) -> bytes:
    """
    Capture a full-page screenshot via WebShot API.
    Returns raw PNG bytes.
    """
    response = requests.post(
        f"{BASE_URL}/capture",
        headers={
            "X-API-Key": API_KEY,
            "Content-Type": "application/json"
        },
        json={
            "url": url,
            "viewport": {"width": width, "height": 900},
            "full_page": full_page,
            "format": "png",
            "wait_ms": wait_ms,
            "block_ads": True,
            "block_trackers": True,
            "quality": 90
        },
        timeout=60
    )
 
    if response.status_code != 200:
        raise ValueError(f"WebShot error {response.status_code}: {response.text}")
 
    return response.content
 
 
def save_screenshot(page_id: str, png_bytes: bytes, base_dir: str = "./screenshots") -> str:
    """Save screenshot with timestamp, return file path."""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    page_dir = Path(base_dir) / page_id
    page_dir.mkdir(parents=True, exist_ok=True)
 
    filename = page_dir / f"{timestamp}.png"
    filename.write_bytes(png_bytes)
    return str(filename)
 
 
def get_latest_screenshot(page_id: str, base_dir: str = "./screenshots") -> str | None:
    """Return the most recent screenshot path for a page, or None."""
    page_dir = Path(base_dir) / page_id
    if not page_dir.exists():
        return None
 
    screenshots = sorted(page_dir.glob("*.png"), reverse=True)
    return str(screenshots[0]) if screenshots else None

from PIL import Image, ImageChops
import math
import struct
 
 
def calculate_image_diff(path_a: str, path_b: str) -> dict:
    """
    Compare two screenshots and calculate visual difference.
    Returns diff percentage and a diff image.
    """
    img_a = Image.open(path_a).convert("RGB")
    img_b = Image.open(path_b).convert("RGB")
 
    # Resize to same dimensions for comparison
    if img_a.size != img_b.size:
        # Use the older image's dimensions as reference
        img_b = img_b.resize(img_a.size, Image.LANCZOS)
 
    diff = ImageChops.difference(img_a, img_b)
 
    # Calculate percentage of pixels that changed
    pixels = list(diff.getdata())
    total_pixels = len(pixels)
    changed_pixels = sum(
        1 for r, g, b in pixels
        if math.sqrt(r**2 + g**2 + b**2) > 30  # threshold for "significant" change
    )
 
    diff_percentage = (changed_pixels / total_pixels) * 100
 
    # Save diff image
    diff_path = path_b.replace(".png", "_diff.png")
    diff.save(diff_path)
 
    return {
        "diff_percentage": round(diff_percentage, 2),
        "changed_pixels": changed_pixels,
        "total_pixels": total_pixels,
        "diff_image_path": diff_path,
        "change_level": classify_change(diff_percentage)
    }
 
 
def classify_change(diff_pct: float) -> str:
    """Classify the magnitude of change."""
    if diff_pct > 20:
        return "major"    # Major redesign or new content block
    elif diff_pct > 5:
        return "significant"  # Content changes, new features, pricing updates
    elif diff_pct > 1:
        return "minor"    # Small copy or styling changes
    else:
        return "negligible"

from concurrent.futures import ThreadPoolExecutor
 
 
def run_monitoring_cycle(pages: list[dict], base_dir: str = "./screenshots") -> list[dict]:
    """
    Capture screenshots for all pages and detect changes from previous run.
    """
    results = []
 
    def monitor_one(page: dict) -> dict:
        result = {
            "id": page["id"],
            "company": page["company"],
            "url": page["url"],
            "description": page["description"],
            "importance": page["importance"],
            "timestamp": datetime.now().isoformat(),
            "status": "ok",
            "change": None,
            "new_screenshot": None,
            "error": None
        }
 
        try:
            # Get previous screenshot for comparison
            previous = get_latest_screenshot(page["id"], base_dir)
 
            # Capture new screenshot
            png_bytes = capture_screenshot(page["url"])
            new_path = save_screenshot(page["id"], png_bytes, base_dir)
            result["new_screenshot"] = new_path
 
            # Compare if we have a previous one
            if previous and previous != new_path:
                diff = calculate_image_diff(previous, new_path)
                result["change"] = diff
                print(f"  {page['company']} / {page['description']}: "
                      f"{diff['diff_percentage']:.1f}% changed ({diff['change_level']})")
            else:
                print(f"  {page['company']} / {page['description']}: baseline captured")
 
        except Exception as e:
            result["status"] = "error"
            result["error"] = str(e)
            print(f"  ERROR {page['id']}: {e}")
 
        return result
 
    print(f"Running monitoring cycle for {len(pages)} pages...")
 
    with ThreadPoolExecutor(max_workers=3) as executor:
        results = list(executor.map(monitor_one, pages))
 
    return results

from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
 
 
def send_change_alert(results: list[dict], slack_token: str, channel: str = "#competitor-intel"):
    """
    Send a Slack alert for significant changes detected in the monitoring run.
    """
    client = WebClient(token=slack_token)
 
    significant = [
        r for r in results
        if r.get("change") and r["change"]["change_level"] in ("major", "significant")
    ]
 
    if not significant:
        return  # No significant changes, no alert needed
 
    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": f"🔍 Competitor Changes Detected ({len(significant)} pages)"
            }
        },
        {
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": f"Monitored {len(results)} pages • {datetime.now().strftime('%Y-%m-%d %H:%M')}"}]
        }
    ]
 
    for r in significant:
        change = r["change"]
        level_emoji = "🚨" if change["change_level"] == "major" else "⚠️"
 
        blocks.append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": (
                    f"{level_emoji} *{r['company']}* — {r['description']}\n"
                    f"Change: *{change['diff_percentage']}%* pixels modified ({change['change_level']})\n"
                    f"URL: {r['url']}"
                )
            }
        })
 
    try:
        client.chat_postMessage(channel=channel, blocks=blocks)
        print(f"Slack alert sent for {len(significant)} changes")
    except SlackApiError as e:
        print(f"Slack error: {e.response['error']}")

from jinja2 import Template
 
REPORT_TEMPLATE = """
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Competitor Intelligence Report — {{ date }}</title>
    <style>
        body { font-family: system-ui, sans-serif; max-width: 1200px; margin: 0 auto; padding: 2rem; }
        .page-card { border: 1px solid #e0e0e0; border-radius: 8px; margin: 1rem 0; padding: 1rem; }
        .change-major { border-left: 4px solid #ef4444; }
        .change-significant { border-left: 4px solid #f59e0b; }
        .change-minor { border-left: 4px solid #3b82f6; }
        .change-negligible { border-left: 4px solid #22c55e; }
        img { max-width: 100%; border: 1px solid #ccc; margin: 0.5rem 0; }
        .grid { display: grid; grid-template-columns: 1fr 1fr; gap: 1rem; }
    </style>
</head>
<body>
    <h1>Competitor Intelligence Report</h1>
    <p>Generated: {{ date }} | Pages monitored: {{ total }}</p>
 
    {% for r in results %}
    <div class="page-card change-{{ r.change.change_level if r.change else 'negligible' }}">
        <h3>{{ r.company }} — {{ r.description }}</h3>
        <p><a href="{{ r.url }}">{{ r.url }}</a></p>
        {% if r.change %}
            <p><strong>Change detected: {{ r.change.diff_percentage }}% ({{ r.change.change_level }})</strong></p>
            <div class="grid">
                <div>
                    <p>Latest Screenshot:</p>
                    <img src="{{ r.new_screenshot }}" alt="Latest">
                </div>
                <div>
                    <p>Diff View:</p>
                    <img src="{{ r.change.diff_image_path }}" alt="Diff">
                </div>
            </div>
        {% else %}
            <p>No change detected or baseline run.</p>
            <img src="{{ r.new_screenshot }}" alt="Current" style="max-height: 300px; object-fit: cover;">
        {% endif %}
    </div>
    {% endfor %}
</body>
</html>
"""
 
def generate_report(results: list[dict], output_path: str = "./intelligence_report.html"):
    template = Template(REPORT_TEMPLATE)
    html = template.render(
        date=datetime.now().strftime("%Y-%m-%d %H:%M"),
        total=len(results),
        results=results
    )
    Path(output_path).write_text(html, encoding="utf-8")
    print(f"Report generated: {output_path}")

import schedule
import time
 
 
def daily_monitoring_job():
    """The full daily monitoring pipeline."""
    print(f"\n[{datetime.now().isoformat()}] Starting daily competitor monitoring...")
 
    results = run_monitoring_cycle(COMPETITOR_PAGES)
    generate_report(results)
 
    slack_token = os.environ.get("SLACK_TOKEN")
    if slack_token:
        send_change_alert(results, slack_token)
 
    changes = sum(1 for r in results if r.get("change") and r["change"]["change_level"] != "negligible")
    print(f"Cycle complete: {changes}/{len(results)} pages with notable changes")
 
 
# Schedule daily at 8 AM
schedule.every().day.at("08:00").do(daily_monitoring_job)
 
if __name__ == "__main__":
    # Run immediately on start
    daily_monitoring_job()
 
    # Then run on schedule
    while True:
        schedule.run_pending()
        time.sleep(60)