AI competitive intelligence: automatic market monitoring with Claude

import requests
import anthropic
import json
from datetime import date

def get_ahrefs_data(domain: str, api_key: str) -> dict:
    """Gets the main SEO metrics for a domain through Ahrefs API v3.
    The API is available on paid plans. Check required parameters and limits at docs.ahrefs.com:
    the parameter set for endpoints changes."""
    
    headers = {"Authorization": f"Bearer {api_key}"}
    base = "https://api.ahrefs.com/v3"
    
    # Organic metrics
    metrics_resp = requests.get(
        f"{base}/site-explorer/metrics",
        headers=headers,
        params={"target": domain, "mode": "domain", "date": date.today().isoformat()}
    )
    
    # Top pages
    pages_resp = requests.get(
        f"{base}/site-explorer/top-pages",
        headers=headers,
        params={"target": domain, "mode": "domain", "limit": 10}
    )
    
    return {
        "metrics": metrics_resp.json() if metrics_resp.ok else {},
        "top_pages": pages_resp.json() if pages_resp.ok else {}
    }

# Analysis with Claude
client = anthropic.Anthropic()

competitors = ["competitor1.com", "competitor2.com", "competitor3.com"]
# AHREFS_API_KEY = os.getenv("AHREFS_API_KEY")

# For demonstration: the data structure you'll get:
demo_data = {
    "competitor1.com": {
        "organic_traffic": 45200,
        "organic_keywords": 3840,
        "top_pages": [
            {"url": "/blog/ai-automation-guide", "traffic": 8400},
            {"url": "/pricing", "traffic": 4200},
        ]
    }
}

message = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=1500,
    messages=[{
        "role": "user",
        "content": f"""Analyze the competitors' SEO data:

{json.dumps(demo_data, ensure_ascii=False, indent=2)}

My site is [your domain], the niche is AI tools for small businesses.

Answer:
1. Who leads in organic search and why?
2. Which topics/pages generate the most traffic?
3. Which keywords should I go after first?
4. What are competitors missing (content gaps)?"""
    }]
)
print("".join(b.text for b in message.content if b.type == "text"))

pip install playwright
playwright install chromium

import asyncio
from playwright.async_api import async_playwright
import json
from datetime import datetime

async def scrape_pricing_page(url: str) -> dict:
    """Scrapes a competitor's pricing page"""
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        # Identify as a regular browser; we don't bypass the site's protection
        await page.set_extra_http_headers({
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
        })
        
        await page.goto(url, wait_until="networkidle", timeout=30000)
        
        # Extract the page text
        content = await page.inner_text("body")
        
        # Look for prices with JavaScript
        prices = await page.evaluate("""
            () => {
                const priceElements = document.querySelectorAll('[class*="price"], [class*="plan"], [class*="tier"]');
                return Array.from(priceElements).map(el => el.innerText.trim()).filter(t => t.length > 0);
            }
        """)
        
        await browser.close()
        
        return {
            "url": url,
            "scraped_at": datetime.now().isoformat(),
            "page_text": content[:5000],  # first 5000 characters
            "price_elements": prices[:20]
        }

async def monitor_competitors(competitor_urls: list) -> list:
    """Monitors a list of competitors in parallel"""
    tasks = [scrape_pricing_page(url) for url in competitor_urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return [r for r in results if not isinstance(r, Exception)]

# Run
competitors_pricing = [
    "https://competitor1.com/pricing",
    "https://competitor2.com/plans",
]

results = asyncio.run(monitor_competitors(competitors_pricing))

# Save with the date (for differential analysis)
filename = f"competitor_data_{datetime.now().strftime('%Y-%m-%d')}.json"
with open(filename, 'w', encoding='utf-8') as f:
    json.dump(results, f, ensure_ascii=False, indent=2)

print(f"Data saved: {filename}")

import json
from datetime import datetime, timedelta

def find_changes(old_file: str, new_file: str) -> dict:
    """Compares two snapshots of competitor data"""
    
    with open(old_file) as f:
        old_data = json.load(f)
    with open(new_file) as f:
        new_data = json.load(f)
    
    changes = []
    
    old_by_url = {r['url']: r for r in old_data}
    new_by_url = {r['url']: r for r in new_data}
    
    for url in new_by_url:
        if url in old_by_url:
            old_text = old_by_url[url].get('page_text', '')
            new_text = new_by_url[url].get('page_text', '')
            
            # Simple comparison; in production use difflib
            if old_text != new_text:
                changes.append({
                    "url": url,
                    "change_detected": True,
                    "old_length": len(old_text),
                    "new_length": len(new_text),
                    "delta": len(new_text) - len(old_text)
                })
    
    return changes

ChangeDetection.io detected a change
→ Webhook POST to n8n
→ n8n: a request to Claude with the "before" and "after" text
→ Claude analyzes: what changed strategically?
→ A Telegram notification to you with the analysis

[MONDAY 06:00 — automatically]
    │
    ├─ Playwright scraper: collects data on competitors' prices
    ├─ ChangeDetection: checks the sites for changes
    └─ Ahrefs API: exports SEO metrics
    │
    ▼
[Claude analysis]
    Combines all the data
    Compares it with last week
    Writes a report with conclusions
    │
    ▼
[Telegram bot — 08:00]
    📊 Weekly competitor report
    Competitor 1: no changes
    Competitor 2: ⚠️ Price cut from $49 → $39
    Competitor 3: 🆕 New page /ai-assistant
    
    Claude's analysis: [strategic conclusions]

import asyncio
import anthropic
import json
import os
from datetime import datetime
from playwright.async_api import async_playwright
import requests

ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")

COMPETITORS = [
    {
        "name": "Competitor A",
        "pricing_url": "https://competitorA.com/pricing",
        "homepage": "https://competitorA.com"
    },
    {
        "name": "Competitor B",
        "pricing_url": "https://competitorB.com/plans",
        "homepage": "https://competitorB.com"
    },
]

async def scrape_page_text(url: str) -> str:
    """Returns the text content of a page"""
    try:
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            page = await browser.new_page()
            await page.goto(url, wait_until="domcontentloaded", timeout=20000)
            text = await page.inner_text("body")
            await browser.close()
            return text[:8000]
    except Exception as e:
        return f"Scraping error {url}: {e}"

def load_previous_data(filename: str) -> dict:
    try:
        with open(filename) as f:
            return json.load(f)
    except FileNotFoundError:
        return {}

def save_current_data(data: dict, filename: str):
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

def send_telegram(message: str):
    url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
    requests.post(url, json={
        "chat_id": TELEGRAM_CHAT_ID,
        "text": message,
        "parse_mode": "Markdown"
    })

async def generate_weekly_report():
    print(f"Starting data collection: {datetime.now()}")
    
    # Load last week's data
    prev_data = load_previous_data("competitor_snapshot.json")
    current_data = {}
    
    # Scrape all competitors
    for competitor in COMPETITORS:
        print(f"  Scraping {competitor['name']}...")
        pricing_text = await scrape_page_text(competitor['pricing_url'])
        homepage_text = await scrape_page_text(competitor['homepage'])
        
        current_data[competitor['name']] = {
            "scraped_at": datetime.now().isoformat(),
            "pricing": pricing_text,
            "homepage": homepage_text
        }
    
    # Save the current snapshot
    save_current_data(current_data, "competitor_snapshot.json")
    
    # Build the context for Claude
    analysis_input = []
    for name, data in current_data.items():
        prev = prev_data.get(name, {})
        analysis_input.append({
            "competitor": name,
            "current_pricing": data['pricing'][:2000],
            "current_homepage": data['homepage'][:1000],
            "previous_pricing": prev.get('pricing', 'No data')[:2000],
            "change_detected": data['pricing'] != prev.get('pricing', '')
        })
    
    # Analysis with Claude
    client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)
    
    message = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1500,
        messages=[{
            "role": "user",
            "content": f"""This is a weekly competitive intelligence report.

Competitor data (this week vs. last week):
{json.dumps(analysis_input, ensure_ascii=False, indent=2)}

Write a short (up to 500 words) report for Telegram with emoji:
1. 🔍 What changed for each competitor
2. ⚠️ Important signals that call for a response from us
3. 💡 One strategic recommendation for this week

If nothing changed, say so plainly too."""
        }]
    )
    
    report = "".join(b.text for b in message.content if b.type == "text")
    
    # Send to Telegram
    header = f"📊 *Weekly intelligence* — {datetime.now().strftime('%m/%d/%Y')}\n\n"
    send_telegram(header + report)
    
    print("Report sent to Telegram")

# Run
asyncio.run(generate_weekly_report())

# crontab -e
0 6 * * 1 cd /path/to/project && python competitive_monitor.py >> logs/ci_$(date +\%Y\%m\%d).log 2>&1

mkdir competitive-intel && cd competitive-intel
python3 -m venv venv && source venv/bin/activate
pip install playwright anthropic python-dotenv requests
playwright install chromium

# Create a .env file:
# ANTHROPIC_API_KEY=sk-ant-...
# TELEGRAM_TOKEN=...
# TELEGRAM_CHAT_ID=...

python first_snapshot.py

# test_analysis.py
import anthropic
import json

client = anthropic.Anthropic()

# Load the collected data
with open('competitor_snapshot.json') as f:
    data = json.load(f)

# Format it for Claude
summary = ""
for name, info in data.items():
    summary += f"\n### {name}\n"
    summary += f"Pricing page (first 500 characters):\n{info['pricing'][:500]}\n"

message = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=1000,
    messages=[{
        "role": "user",
        "content": f"""Here's data from competitors' websites:

{summary}

Tell me:
1. What pricing model does each one use?
2. Who is the cheapest, who is the most expensive?
3. What looks strongest in their offer?
4. What weak spots do you see?"""
    }]
)

print("".join(b.text for b in message.content if b.type == "text"))

# Open crontab
crontab -e

# Add a line (every Monday at 07:00)
0 7 * * 1 /Users/you/competitive-intel/venv/bin/python /Users/you/competitive-intel/competitive_monitor.py