AI Конкурентная разведка — автоматический мониторинг рынка

import requests
import anthropic
import json
from datetime import date

def get_ahrefs_data(domain: str, api_key: str) -> dict:
    """Получает основные SEO-метрики домена через Ahrefs API v3.
    API доступен на платных тарифах. Обязательные параметры и лимиты сверяй с docs.ahrefs.com:
    набор параметров у эндпоинтов меняется."""
    
    headers = {"Authorization": f"Bearer {api_key}"}
    base = "https://api.ahrefs.com/v3"
    
    # Органические метрики
    metrics_resp = requests.get(
        f"{base}/site-explorer/metrics",
        headers=headers,
        params={"target": domain, "mode": "domain", "date": date.today().isoformat()}
    )
    
    # Топ страницы
    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 {}
    }

# Анализ через Claude
client = anthropic.Anthropic()

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

# Для демонстрации — структура данных которую ты получишь:
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"""Проанализируй SEO-данные конкурентов:

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

Мой сайт — [твой домен], ниша — AI-инструменты для малого бизнеса.

Ответь:
1. Кто лидирует в органике и почему?
2. Какие темы/страницы создают больше всего трафика?
3. Какие ключевые слова мне стоит атаковать первыми?
4. Что конкуренты упускают (контентные дыры)?"""
    }]
)
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:
    """Скрейпит страницу цен конкурента"""
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        # Представляемся обычным браузером; защиту сайта не обходим
        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)
        
        # Извлекаем текст страницы
        content = await page.inner_text("body")
        
        # Ищем цены через 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],  # первые 5000 символов
            "price_elements": prices[:20]
        }

async def monitor_competitors(competitor_urls: list) -> list:
    """Мониторит список конкурентов параллельно"""
    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)]

# Запуск
competitors_pricing = [
    "https://competitor1.com/pricing",
    "https://competitor2.com/plans",
]

results = asyncio.run(monitor_competitors(competitors_pricing))

# Сохраняем с датой (для дифференциального анализа)
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"Данные сохранены: {filename}")

import json
from datetime import datetime, timedelta

def find_changes(old_file: str, new_file: str) -> dict:
    """Сравнивает два снимка данных конкурентов"""
    
    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', '')
            
            # Простое сравнение — в продакшне используй 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 обнаружил изменение
→ Webhook POST на n8n
→ n8n: запрос к Claude с текстом "было" и "стало"
→ Claude анализирует: что изменилось стратегически?
→ Telegram-уведомление тебе с анализом

[ПОНЕДЕЛЬНИК 06:00 — автоматически]
    │
    ├─ Playwright scraper: собирает данные о ценах конкурентов
    ├─ ChangeDetection: проверяет изменения на сайтах
    └─ Ahrefs API: выгружает SEO-метрики
    │
    ▼
[Claude анализ]
    Объединяет все данные
    Сравнивает с прошлой неделей
    Составляет отчёт с выводами
    │
    ▼
[Telegram-бот — 08:00]
    📊 Еженедельный отчёт о конкурентах
    Конкурент 1: изменений нет
    Конкурент 2: ⚠️ Цена снижена с $49 → $39
    Конкурент 3: 🆕 Новая страница /ai-assistant
    
    Анализ Claude: [стратегические выводы]

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": "Конкурент A",
        "pricing_url": "https://competitorA.com/pricing",
        "homepage": "https://competitorA.com"
    },
    {
        "name": "Конкурент B",
        "pricing_url": "https://competitorB.com/plans",
        "homepage": "https://competitorB.com"
    },
]

async def scrape_page_text(url: str) -> str:
    """Возвращает текстовое содержимое страницы"""
    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"Ошибка скрейпинга {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"Начинаю сбор данных: {datetime.now()}")
    
    # Загружаем прошлые данные
    prev_data = load_previous_data("competitor_snapshot.json")
    current_data = {}
    
    # Скрейпим всех конкурентов
    for competitor in COMPETITORS:
        print(f"  Скрейплю {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_current_data(current_data, "competitor_snapshot.json")
    
    # Строим контекст для 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', 'Нет данных')[:2000],
            "change_detected": data['pricing'] != prev.get('pricing', '')
        })
    
    # Анализ через 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"""Это еженедельный отчёт конкурентной разведки.

Данные по конкурентам (текущая неделя vs прошлая):
{json.dumps(analysis_input, ensure_ascii=False, indent=2)}

Составь краткий (до 500 слов) отчёт для Telegram с эмодзи:
1. 🔍 Что изменилось у каждого конкурента
2. ⚠️ Важные сигналы которые требуют нашей реакции
3. 💡 Одна стратегическая рекомендация на эту неделю

Если изменений нет — тоже скажи об этом прямо."""
        }]
    )
    
    report = "".join(b.text for b in message.content if b.type == "text")
    
    # Отправляем в Telegram
    header = f"📊 *Еженедельная разведка* — {datetime.now().strftime('%d.%m.%Y')}\n\n"
    send_telegram(header + report)
    
    print("Отчёт отправлен в Telegram")

# Запуск
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

# Создай .env файл:
# ANTHROPIC_API_KEY=sk-ant-...
# TELEGRAM_TOKEN=...
# TELEGRAM_CHAT_ID=...

python first_snapshot.py

# test_analysis.py
import anthropic
import json

client = anthropic.Anthropic()

# Загружаем снятые данные
with open('competitor_snapshot.json') as f:
    data = json.load(f)

# Форматируем для Claude
summary = ""
for name, info in data.items():
    summary += f"\n### {name}\n"
    summary += f"Страница цен (первые 500 символов):\n{info['pricing'][:500]}\n"

message = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=1000,
    messages=[{
        "role": "user",
        "content": f"""Вот данные с сайтов конкурентов:

{summary}

Скажи:
1. Какую модель ценообразования использует каждый?
2. Кто самый дешёвый, кто самый дорогой?
3. Что в их предложении кажется самым сильным?
4. Какие слабые места видишь?"""
    }]
)

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

# Открываем crontab
crontab -e

# Добавляем строку (каждый понедельник в 07:00)
0 7 * * 1 /Users/тебя/competitive-intel/venv/bin/python /Users/тебя/competitive-intel/competitive_monitor.py