競合分析の自動化——スクレイピングで月次レポートを作る

競合分析の自動化——なぜ必要か

競合他社のWebサイト・ECサイトを月1回手動でチェックするのは現実的ではありません。特に価格・サービスラインナップ・コンテンツ戦略は週単位・日単位で変化することがあります。

スクレイピングとレポート生成を自動化することで、常に最新の競合情報を把握し、ビジネス判断を迅速に行えます。

収集する情報の設計

月次競合レポートに含める情報例：

REPORT_SECTIONS = {
    'pricing': {
        'description': '価格比較',
        'metrics': ['最安値', '最高値', '平均価格', '価格変動率'],
    },
    'content': {
        'description': 'コンテンツ分析',
        'metrics': ['新規記事数', '更新頻度', 'キーワード傾向'],
    },
    'product_lineup': {
        'description': '商品ラインナップ',
        'metrics': ['総商品数', '新着商品', '廃止商品'],
    },
}

スクレイピングパイプラインの構築

import asyncio
import aiohttp
from bs4 import BeautifulSoup
from dataclasses import dataclass, field
from datetime import datetime
import pandas as pd

@dataclass
class CompetitorSnapshot:
    """競合サイトの1時点のスナップショット"""
    competitor_id:  str
    url:            str
    scraped_at:     datetime
    products:       list[dict] = field(default_factory=list)
    blog_posts:     list[dict] = field(default_factory=list)
    price_summary:  dict       = field(default_factory=dict)

async def scrape_competitor(session: aiohttp.ClientSession, url: str) -> str:
    """単一ページを非同期で取得"""
    headers = {
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) '
                      'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'
    }
    async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=15)) as resp:
        return await resp.text()

async def collect_all_competitors(competitors: list[dict]) -> list[CompetitorSnapshot]:
    """全競合サイトを並列でスクレイピング"""
    snapshots = []

    async with aiohttp.ClientSession() as session:
        tasks = []
        for comp in competitors:
            task = asyncio.create_task(
                scrape_single_competitor(session, comp)
            )
            tasks.append(task)

        results = await asyncio.gather(*tasks, return_exceptions=True)

        for comp, result in zip(competitors, results):
            if isinstance(result, Exception):
                print(f"エラー [{comp['name']}]: {result}")
            else:
                snapshots.append(result)
                await asyncio.sleep(1)  # レート制限対策

    return snapshots

async def scrape_single_competitor(session, competitor: dict) -> CompetitorSnapshot:
    snapshot = CompetitorSnapshot(
        competitor_id = competitor['id'],
        url           = competitor['url'],
        scraped_at    = datetime.now(),
    )

    # 商品ページの取得
    html = await scrape_competitor(session, f"{competitor['url']}/products")
    soup = BeautifulSoup(html, 'lxml')

    snapshot.products = [
        {
            'name':  el.select_one('.product-name').get_text(strip=True),
            'price': int(el.select_one('.price').get_text(strip=True).replace(',', '').replace('¥', '')),
        }
        for el in soup.select('.product-item')
        if el.select_one('.product-name') and el.select_one('.price')
    ]

    if snapshot.products:
        prices = [p['price'] for p in snapshot.products]
        snapshot.price_summary = {
            'min':   min(prices),
            'max':   max(prices),
            'avg':   sum(prices) / len(prices),
            'count': len(prices),
        }

    return snapshot

データの保存と比較

import json
from pathlib import Path

def save_snapshot(snapshot: CompetitorSnapshot, data_dir: Path) -> None:
    """スナップショットをJSONで保存"""
    month_dir = data_dir / snapshot.scraped_at.strftime('%Y-%m')
    month_dir.mkdir(parents=True, exist_ok=True)

    filename = f"{snapshot.competitor_id}_{snapshot.scraped_at.strftime('%Y%m%d_%H%M')}.json"
    filepath = month_dir / filename

    with open(filepath, 'w', encoding='utf-8') as f:
        json.dump({
            'competitor_id': snapshot.competitor_id,
            'url':          snapshot.url,
            'scraped_at':   snapshot.scraped_at.isoformat(),
            'products':     snapshot.products,
            'price_summary': snapshot.price_summary,
        }, f, ensure_ascii=False, indent=2)


def compare_with_previous(current: CompetitorSnapshot, data_dir: Path) -> dict:
    """前月のデータと比較"""
    prev_month = current.scraped_at.replace(day=1) - pd.DateOffset(months=1)
    prev_dir = data_dir / prev_month.strftime('%Y-%m')

    if not prev_dir.exists():
        return {'error': '前月データなし'}

    prev_files = sorted(prev_dir.glob(f"{current.competitor_id}_*.json"))
    if not prev_files:
        return {'error': '前月データなし'}

    with open(prev_files[-1], encoding='utf-8') as f:
        prev_data = json.load(f)

    prev_summary = prev_data.get('price_summary', {})
    curr_summary = current.price_summary

    return {
        'product_count_change': curr_summary.get('count', 0) - prev_summary.get('count', 0),
        'avg_price_change':     curr_summary.get('avg', 0) - prev_summary.get('avg', 0),
        'min_price_change':     curr_summary.get('min', 0) - prev_summary.get('min', 0),
    }

HTMLレポートの生成

from jinja2 import Environment, FileSystemLoader

def generate_html_report(snapshots: list[CompetitorSnapshot], comparisons: dict) -> str:
    env = Environment(loader=FileSystemLoader('./templates'))
    template = env.get_template('competitor_report.html')

    return template.render(
        report_date = datetime.now().strftime('%Y年%m月'),
        snapshots   = snapshots,
        comparisons = comparisons,
    )

メール送信（Python smtplib）

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_report_email(html_content: str, recipients: list[str]) -> None:
    msg = MIMEMultipart('alternative')
    msg['Subject'] = f"競合分析月次レポート {datetime.now().strftime('%Y年%m月')}"
    msg['From']    = 'report@example.com'
    msg['To']      = ', '.join(recipients)

    msg.attach(MIMEText(html_content, 'html', 'utf-8'))

    with smtplib.SMTP_SSL('smtp.gmail.com', 465) as server:
        server.login(os.environ['SMTP_USER'], os.environ['SMTP_PASS'])
        server.send_message(msg)