The AI Competitive Intelligence Analyst: Daily Briefing on Autopilot

[
  {
    "name": "Linear",
    "pages": [
      {"url": "https://linear.app", "label": "homepage"},
      {"url": "https://linear.app/pricing", "label": "pricing"},
      {"url": "https://linear.app/changelog", "label": "changelog"},
      {"url": "https://linear.app/blog", "label": "blog"}
    ],
    "extract": {
      "pricing": ["tier names", "prices", "feature lists"],
      "homepage": ["hero headline", "CTA text", "customer logos"]
    }
  },
  {
    "name": "Notion",
    "pages": [
      {"url": "https://notion.so", "label": "homepage"},
      {"url": "https://notion.so/pricing", "label": "pricing"},
      {"url": "https://notion.so/releases", "label": "releases"}
    ],
    "extract": {
      "pricing": ["tier names", "prices", "feature lists", "AI add-on pricing"],
      "homepage": ["hero headline", "social proof numbers"]
    }
  }
]

import anthropic, base64

client = anthropic.Anthropic()

async def compare_pages(
    yesterday_screenshot: bytes,
    today_screenshot: bytes,
    competitor: str,
    page_label: str,
) -> dict:
    """Compare two screenshots and describe changes."""
    b64_old = base64.b64encode(yesterday_screenshot).decode()
    b64_new = base64.b64encode(today_screenshot).decode()

    response = client.messages.create(
        model="claude-sonnet-4-6-20250514",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": f"""Compare these two screenshots of {competitor}'s {page_label} page.
The first image is yesterday. The second is today.

Describe any meaningful changes:
- Pricing changes (amounts, tier names, feature additions/removals)
- Visual changes (layout shifts, new badges, redesigned cards)
- Copy changes (headlines, CTAs, value propositions)
- New content (logos, testimonials, banners, announcements)

If the pages look identical, respond with exactly: NO_CHANGES
Otherwise, describe each change concisely. Be specific about what changed."""},
                {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": b64_old}},
                {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": b64_new}},
            ],
        }],
    )

    text = response.content[0].text.strip()
    has_changes = text != "NO_CHANGES"

    return {
        "competitor": competitor,
        "page": page_label,
        "has_changes": has_changes,
        "description": text if has_changes else None,
    }

import asyncio, json, time, os, httpx
from datetime import datetime
from pathlib import Path
from playwright.async_api import async_playwright

SANDBOX_URL = os.environ["SANDBOX_URL"]
API_KEY = os.environ["SANDBOX_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
TODAY = datetime.utcnow().strftime("%Y-%m-%d")
ARCHIVE = Path("screenshots")


async def monitor_competitor(client, competitor: dict) -> list:
    """Visit all pages for one competitor, screenshot each."""
    name = competitor["name"]

    # Create browser
    browser_data = (await client.post(
        f"{SANDBOX_URL}/api/browsers", headers=HEADERS,
        json={"browser_type": "chrome", "enable_debug_port": True, "idle_timeout_minutes": 10}
    )).json()
    cid = browser_data["container_id"]

    try:
        # Wait for browser
        for _ in range(40):
            s = (await client.get(f"{SANDBOX_URL}/api/containers/{cid}/status", headers=HEADERS)).json()
            if s["state"] == "assigned": break
            await asyncio.sleep(3)

        results = []
        async with async_playwright() as p:
            browser = await p.chromium.connect_over_cdp(s["cdp_url"])
            page = await browser.contexts[0].new_page()
            await page.set_viewport_size({"width": 1280, "height": 720})

            for pg in competitor["pages"]:
                try:
                    await page.goto(pg["url"], wait_until="networkidle", timeout=20000)
                    await page.wait_for_timeout(2000)

                    # Dismiss cookie banners
                    for sel in ["button:has-text('Accept')", "[class*='cookie'] button"]:
                        try:
                            btn = page.locator(sel).first
                            if await btn.is_visible(timeout=1000):
                                await btn.click()
                                break
                        except: pass

                    # Screenshot
                    img = await page.screenshot(full_page=True)
                    save_path = ARCHIVE / name / TODAY / f"{pg['label']}.png"
                    save_path.parent.mkdir(parents=True, exist_ok=True)
                    save_path.write_bytes(img)

                    # Extract key text
                    text = await page.evaluate("() => document.body.innerText")

                    results.append({
                        "competitor": name, "page": pg["label"],
                        "url": pg["url"], "screenshot_path": str(save_path),
                        "text_excerpt": text[:3000], "status": "ok",
                    })
                except Exception as e:
                    results.append({"competitor": name, "page": pg["label"], "status": "error", "error": str(e)})

        return results

    finally:
        await client.post(f"{SANDBOX_URL}/api/containers/{cid}/release", headers=HEADERS)


async def main():
    with open("competitors.json") as f:
        competitors = json.load(f)

    print(f"Monitoring {len(competitors)} competitors...")

    # Phase 1: Screenshot all competitors in parallel
    async with httpx.AsyncClient(timeout=300) as client:
        tasks = [monitor_competitor(client, c) for c in competitors]
        all_results = await asyncio.gather(*tasks, return_exceptions=True)

    # Flatten results
    pages = [r for batch in all_results if isinstance(batch, list) for r in batch]

    # Phase 2: Compare against yesterday
    changes = []
    for page in pages:
        if page["status"] != "ok": continue
        yesterday_path = page["screenshot_path"].replace(TODAY, "YESTERDAY_DATE")
        # In production, compute yesterday's date and check if file exists
        # If it does, compare using compare_pages() from compare.py
        # Collect changes

    # Phase 3: Build and post Slack briefing
    briefing = build_slack_briefing(changes, pages)
    post_to_slack(briefing)

    print(f"Done. {len(changes)} changes detected across {len(pages)} pages.")

asyncio.run(main())

# For competitors that require login
{
  "name": "CompetitorX",
  "auth": {
    "login_url": "https://app.competitorx.com/login",
    "username": "${COMPETITORX_USER}",
    "password": "${COMPETITORX_PASS}"
  },
  "pages": [
    {"url": "https://app.competitorx.com/changelog", "label": "changelog"},
    {"url": "https://app.competitorx.com/integrations", "label": "integrations"}
  ]
}

def build_slack_briefing(changes: list, all_pages: list) -> str:
    date = datetime.utcnow().strftime("%B %d, %Y")
    total = len(all_pages)
    changed = len([c for c in changes if c["has_changes"]])

    msg = f"""*Competitive Intelligence Briefing — {date}*
Monitored: {total} pages across {len(set(c['competitor'] for c in changes))} competitors
Changes detected: {changed}
"""

    if changed == 0:
        msg += "\n_No changes detected. All quiet._"
    else:
        msg += "\n---\n"
        for c in changes:
            if c["has_changes"]:
                msg += f"\n*{c['competitor']}* — {c['page']}\n{c['description']}\n"

    msg += f"\n_Full screenshots archived. Run cost: ~$0.05._"
    return msg