ページ監視の方法論

ウェブページの変更追跡、コンテンツ削除の検出、重要なページの消失前の保護のためのパターン。

監視サービスの比較

無料層の制限と保持期間は毎年変わります — 特定の数値に依存する前に、各サービスの価格ページで確認してください。以下の表は 2026 年時点のものです。

サービス	無料プラン	最適な用途	履歴	アラート速度
Visualping	1日数回のチェック（最近は無料プランが厳格化）	視覚的な変更	標準	数分
ChangeTower	あり（現在の制限を確認）	コンプライアンス、アーカイブ	有料プランで複数年	数分
Distill.io	約5個のモニター、7日間の履歴	要素レベルの追跡	無料層では制限あり	数秒
Wachete	制限あり	ログイン保護ページ	12ヶ月	数分
UptimeRobot	50個のモニター、5分間隔（無料SMS廃止）	アップタイム監視のみ	60日	5分チェック
changedetection.io	セルフホスト、無料	プライバシー / DIY	ディスク容量次第	設定可能
urlwatch	セルフホスト、無料	Cron駆動CLI	設定可能	設定可能

クイックスタート: ページの監視

Distill.io 要素監視

// Distill.io は正確な監視用に CSS/XPath セレクターを使用できます
// 一般的なユースケース用のセレクター例:

// ニュース記事の見出しを監視
const newsSelector = '.article-headline, h1.title, .story-title';

// 価格変更を監視
const priceSelector = '.price, .product-price, [data-price]';

// 在庫/可用性を監視
const availabilitySelector = '.in-stock, .availability, .stock-status';

// 特定の段落またはセクションを監視
const sectionSelector = '#main-content p:first-child';

// テーブルデータを監視
const tableSelector = 'table.data-table tbody tr';

Python 監視スクリプト

import requests
import hashlib
import json
import smtplib
from email.mime.text import MIMEText
from datetime import datetime
from pathlib import Path
from typing import Optional
from bs4 import BeautifulSoup

class PageMonitor:
    """ローカルストレージを備えたシンプルなページ変更監視。"""

    def __init__(self, storage_dir: Path):
        self.storage_dir = storage_dir
        self.storage_dir.mkdir(parents=True, exist_ok=True)
        self.state_file = storage_dir / 'monitor_state.json'
        self.state = self._load_state()

    def _load_state(self) -> dict:
        if self.state_file.exists():
            return json.loads(self.state_file.read_text())
        return {'pages': {}}

    def _save_state(self):
        self.state_file.write_text(json.dumps(self.state, indent=2))

    def _get_page_hash(self, url: str, selector: Optional[str] = None) -> tuple[str, str]:
        """ページまたは要素のコンテンツハッシュとコンテンツを取得。"""

        response = requests.get(url, timeout=30, headers={
            'User-Agent': 'Mozilla/5.0 (PageMonitor/1.0)'
        })
        response.raise_for_status()

        if selector:
            soup = BeautifulSoup(response.text, 'html.parser')
            element = soup.select_one(selector)
            content = element.get_text(strip=True) if element else ''
        else:
            content = response.text

        content_hash = hashlib.sha256(content.encode()).hexdigest()
        return content_hash, content

    def add_page(self, url: str, name: str, selector: Optional[str] = None):
        """監視対象にページを追加。"""

        content_hash, content = self._get_page_hash(url, selector)

        self.state['pages'][url] = {
            'name': name,
            'selector': selector,
            'last_hash': content_hash,
            'last_check': datetime.now().isoformat(),
            'last_content': content[:1000],  # プレビューを保存
            'change_count': 0
        }

        self._save_state()
        print(f"追加: {name} ({url})")

    def check_page(self, url: str) -> Optional[dict]:
        """単一ページの変更を確認。"""

        if url not in self.state['pages']:
            return None

        page = self.state['pages'][url]
        selector = page.get('selector')

        try:
            new_hash, new_content = self._get_page_hash(url, selector)
        except Exception as e:
            return {
                'url': url,
                'name': page['name'],
                'status': 'error',
                'error': str(e)
            }

        changed = new_hash != page['last_hash']

        result = {
            'url': url,
            'name': page['name'],
            'status': 'changed' if changed else 'unchanged',
            'previous_content': page['last_content'],
            'new_content': new_content[:1000] if changed else None
        }

        if changed:
            page['last_hash'] = new_hash
            page['last_content'] = new_content[:1000]
            page['change_count'] += 1

            # 変更をアーカイブ
            archive_file = self.storage_dir / f"{hashlib.md5(url.encode()).hexdigest()}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
            archive_file.write_text(new_content)

        page['last_check'] = datetime.now().isoformat()
        self._save_state()

        return result

    def check_all(self) -> list[dict]:
        """監視対象のすべてのページをチェック。"""
        results = []
        for url in self.state['pages']:
            result = self.check_page(url)
            if result:
                results.append(result)
        return results

# 使用例
monitor = PageMonitor(Path('./page_monitor_data'))

# 監視対象ページを追加
monitor.add_page(
    'https://example.com/important-page',
    'Important Page',
    selector='.main-content'  # オプション: 特定の要素のみ監視
)

# 変更を確認
results = monitor.check_all()
for result in results:
    if result['status'] == 'changed':
        print(f"変更: {result['name']}")
        print(f"  前: {result['previous_content'][:100]}...")
        print(f"  新: {result['new_content'][:100]}...")

アップタイム監視

UptimeRobot API 統合

import requests
from typing import List, Optional

class UptimeRobotClient:
    """UptimeRobot API クライアント、ページ可用性監視用。"""

    def __init__(self, api_key: str):
        self.api_key = api_key
        # v2 は 2026 年時点でも動作していますが保守モード中です; v3 は
        # 現在の REST API で https://api.uptimerobot.com/v3 にあり、
        # 異なるリクエスト形式（Bearer認証、JSONボディ）を使用します。
        self.base_url = "https://api.uptimerobot.com/v2"

    def _request(self, endpoint: str, params: dict = None) -> dict:
        data = {'api_key': self.api_key}
        if params:
            data.update(params)

        response = requests.post(f"{self.base_url}/{endpoint}", data=data)
        return response.json()

    def get_monitors(self) -> List[dict]:
        """すべてのモニターを取得。"""
        result = self._request('getMonitors')
        return result.get('monitors', [])

    def create_monitor(self, friendly_name: str, url: str,
                       monitor_type: int = 1) -> dict:
        """新しいモニターを作成。

        型: 1=HTTP(s), 2=Keyword, 3=Ping, 4=Port
        """
        return self._request('newMonitor', {
            'friendly_name': friendly_name,
            'url': url,
            'type': monitor_type
        })

    def get_monitor_uptime(self, monitor_id: int,
                           custom_uptime_ratios: str = "7-30-90") -> dict:
        """モニターのアップタイム統計を取得。"""
        return self._request('getMonitors', {
            'monitors': monitor_id,
            'custom_uptime_ratios': custom_uptime_ratios
        })

    def pause_monitor(self, monitor_id: int) -> dict:
        """モニターを一時停止。"""
        return self._request('editMonitor', {
            'id': monitor_id,
            'status': 0
        })

    def resume_monitor(self, monitor_id: int) -> dict:
        """モニターを再開。"""
        return self._request('editMonitor', {
            'id': monitor_id,
            'status': 1
        })

# 使用例
client = UptimeRobotClient('your-api-key')

# 重要なページ用のモニターを作成
client.create_monitor('News Homepage', 'https://example-news.com')
client.create_monitor('API Status', 'https://api.example.com/health')

# すべてのモニターを確認
for monitor in client.get_monitors():
    status = 'UP' if monitor['status'] == 2 else 'DOWN'
    print(f"{monitor['friendly_name']}: {status}")

RSS フィード生成

RSS なしのページから RSS を生成

import requests
from bs4 import BeautifulSoup
from feedgen.feed import FeedGenerator
from datetime import datetime
import hashlib

class RSSGenerator:
    """ウェブページから RSS フィードを生成。"""

    def __init__(self, feed_id: str, title: str, link: str):
        self.fg = FeedGenerator()
        self.fg.id(feed_id)
        self.fg.title(title)
        self.fg.link(href=link)
        self.fg.description(f'{title} の自動生成フィード')

    def add_from_page(self, url: str, item_selector: str,
                      title_selector: str, link_selector: str,
                      description_selector: Optional[str] = None):
        """ページを解析してフィードに項目を追加。

        引数:
            url: 解析するページの URL
            item_selector: 各項目コンテナの CSS セレクター
            title_selector: タイトルの CSS セレクター（項目からの相対位置）
            link_selector: リンクの CSS セレクター（項目からの相対位置）
            description_selector: オプションの説明 CSS セレクター
        """

        response = requests.get(url, timeout=30)
        soup = BeautifulSoup(response.text, 'html.parser')

        items = soup.select(item_selector)

        for item in items[:20]:  # 最大20項目に制限
            title_elem = item.select_one(title_selector)
            link_elem = item.select_one(link_selector)

            if not title_elem or not link_elem:
                continue

            title = title_elem.get_text(strip=True)
            link = link_elem.get('href', '')

            # 相対 URL の場合は絶対 URL に変換
            if link.startswith('/'):
                from urllib.parse import urljoin
                link = urljoin(url, link)

            fe = self.fg.add_entry()
            fe.id(hashlib.md5(link.encode()).hexdigest())
            fe.title(title)
            fe.link(href=link)

            if description_selector:
                desc_elem = item.select_one(description_selector)
                if desc_elem:
                    fe.description(desc_elem.get_text(strip=True))

            fe.published(datetime.now())

    def generate_rss(self) -> str:
        """RSS XML 文字列を生成。"""
        return self.fg.rss_str(pretty=True).decode()

    def save_rss(self, filepath: str):
        """RSS フィードをファイルに保存。"""
        self.fg.rss_file(filepath)

# 例: RSS なしのニュースサイト用にフィードを生成
rss = RSSGenerator(
    'https://example.com/news',
    'Example News Feed',
    'https://example.com/news'
)

rss.add_from_page(
    'https://example.com/news',
    item_selector='.news-item',
    title_selector='h2 a',
    link_selector='h2 a',
    description_selector='.summary'
)

# フィードを保存
rss.save_rss('example_feed.xml')

RSS-Bridge の使用（セルフホスト）

# RSS-Bridge はフィードなしのサイト用フィードを生成
# Twitter、Instagram、YouTube など多くのサイトをサポート

# Docker インストール
docker pull rssbridge/rss-bridge
docker run -d -p 3000:80 rssbridge/rss-bridge

# http://localhost:3000 でアクセス
# ブリッジを選択し、パラメーターを入力、RSS フィード URL を取得

ソーシャルメディア監視

Twarc を使用した Twitter/X アーカイブ

# Twarc には X (Twitter) API 認証情報が必要です。
#
# 重要 (2023年以降): X は無料の Twitter API 層と無料のアカデミック
# リサーチアクセスプログラムを廃止しました。現在の X API 価格は
# 2023年の変更以降複数回変わっています — Basic / Pro / Enterprise
# のティア名と従量課金制クレジットモデルは改訂されています;
# コスト推定前に https://developer.x.com/en/products/x-api で現在の
# 価格ページを確認してください。単一アーカイブの場合、snscrape /
# nitter / web-scraping パスがより費用効果的かもしれません —
# web-scraping スキルを参照してください。

# インストール
# pip install twarc

# 設定（インタラクティブ — API キー + ベアラートークンを提供）
# twarc2 configure

import subprocess
import json
from pathlib import Path

class TwitterArchiver:
    """Twitter の検索とタイムラインをアーカイブ。"""

    def __init__(self, output_dir: Path):
        self.output_dir = output_dir
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def search(self, query: str, max_results: int = 100) -> Path:
        """ツイートを検索してファイルに保存。"""

        output_file = self.output_dir / f"search_{query.replace(' ', '_')}.jsonl"

        subprocess.run([
            'twarc2', 'search',
            '--max-results', str(max_results),
            query,
            str(output_file)
        ], check=True)

        return output_file

    def get_timeline(self, username: str, max_results: int = 100) -> Path:
        """ユーザーのタイムラインを取得。"""

        output_file = self.output_dir / f"timeline_{username}.jsonl"

        subprocess.run([
            'twarc2', 'timeline',
            '--max-results', str(max_results),
            username,
            str(output_file)
        ], check=True)

        return output_file

    def parse_archive(self, filepath: Path) -> list[dict]:
        """アーカイブされたツイートを解析。"""

        tweets = []
        with open(filepath) as f:
            for line in f:
                data = json.loads(line)
                if 'data' in data:
                    tweets.extend(data['data'])

        return tweets

ウェブフック通知

変更時にアラートを送信

import requests
from datetime import datetime
from typing import Optional

class AlertManager:
    """監視対象ページが変更されたときにアラートを送信。"""

    def __init__(self, slack_webhook: str = None,
                 discord_webhook: str = None,
                 email_config: dict = None):
        self.slack_webhook = slack_webhook
        self.discord_webhook = discord_webhook
        self.email_config = email_config

    def send_slack(self, message: str, channel: str = None):
        """Slack 通知を送信。"""
        if not self.slack_webhook:
            return

        payload = {'text': message}
        if channel:
            payload['channel'] = channel

        requests.post(self.slack_webhook, json=payload)

    def send_discord(self, message: str):
        """Discord 通知を送信。"""
        if not self.discord_webhook:
            return

        requests.post(self.discord_webhook, json={'content': message})

    def send_email(self, subject: str, body: str, to: str):
        """メール通知を送信。"""
        if not self.email_config:
            return

        import smtplib
        from email.mime.text import MIMEText

        msg = MIMEText(body)
        msg['Subject'] = subject
        msg['From'] = self.email_config['from']
        msg['To'] = to

        with smtplib.SMTP(self.email_config['smtp_host'],
                         self.email_config['smtp_port']) as server:
            server.starttls()
            server.login(self.email_config['username'],
                        self.email_config['password'])
            server.send_message(msg)

    def alert_change(self, page_name: str, url: str,
                     old_content: str, new_content: str):
        """変更アラートを設定されたすべてのチャネルに送信。"""

        message = f"""
ページが変更されました: {page_name}
URL: {url}
時刻: {datetime.now().isoformat()}

前のコンテンツ（プレビュー）:
{old_content[:200]}...

新しいコンテンツ（プレビュー）:
{new_content[:200]}...
"""

        if self.slack_webhook:
            self.send_slack(message)

        if self.discord_webhook:
            self.send_discord(message)

Cron を使用したスケジュール監視

継続的監視用の Cron セットアップ

# crontab を編集
crontab -e

# 15分ごとにページをチェック
*/15 * * * * /usr/bin/python3 /path/to/monitor_script.py >> /var/log/monitor.log 2>&1

# 重要ページを5分ごとにチェック
*/5 * * * * /usr/bin/python3 /path/to/critical_monitor.py >> /var/log/critical.log 2>&1

# 毎日午前8時に日次レポートを作成
0 8 * * * /usr/bin/python3 /path/to/daily_report.py

監視スクリプトテンプレート

#!/usr/bin/env python3
"""Cron 実行用のページ監視スクリプト。"""

import sys
from pathlib import Path
from datetime import datetime

# プロジェクトをパスに追加
sys.path.insert(0, str(Path(__file__).parent))

from monitor import PageMonitor
from alerts import AlertManager

def main():
    # 初期化
    monitor = PageMonitor(Path('./data'))
    alerts = AlertManager(
        slack_webhook='https://hooks.slack.com/services/...',
        discord_webhook='https://discord.com/api/webhooks/...'
    )

    # すべてのページをチェック
    results = monitor.check_all()

    # 結果を処理
    changes = [r for r in results if r['status'] == 'changed']
    errors = [r for r in results if r['status'] == 'error']

    # 変更時にアラート
    for change in changes:
        alerts.alert_change(
            change['name'],
            change['url'],
            change['previous_content'],
            change['new_content']
        )
        print(f"[{datetime.now()}] 変更: {change['name']}")

    # エラー時にアラート
    for error in errors:
        alerts.send_slack(f"{error['name']} の監視エラー: {error['error']}")
        print(f"[{datetime.now()}] エラー: {error['name']} - {error['error']}")

    # サマリー
    print(f"[{datetime.now()}] {len(results)} ページをチェック、"
          f"{len(changes)} 件の変更、{len(errors)} 件のエラー")

if __name__ == '__main__':
    main()

変更時のアーカイブ

変更検出時の自動アーカイブ

# `MultiArchiver` は兄妹の web-archiving スキル
# (research-toolkit/skills/web-archiving/ を参照) からの
# カスケードアーカイブヘルパーです。
# このインポートを独自のマルチサービスアーカイバーに置き換えるか、
# そのスキルから MultiArchiver クラスをポーティングしてください。
from multiarchiver import MultiArchiver

class ArchivingMonitor(PageMonitor):
    """変更検出時にコンテンツをアーカイブするページモニター。"""

    def __init__(self, storage_dir: Path):
        super().__init__(storage_dir)
        self.archiver = MultiArchiver()

    def check_page(self, url: str) -> dict:
        """ページをチェックして変更があればアーカイブ。"""

        result = super().check_page(url)

        if result and result['status'] == 'changed':
            # 複数のサービスにアーカイブ
            archive_results = self.archiver.archive_url(url)

            successful_archives = [
                r.archived_url for r in archive_results
                if r.success
            ]

            result['archives'] = successful_archives

            # アーカイブ URL をログ
            print(f"{url} をアーカイブしました:")
            for archive_url in successful_archives:
                print(f"  - {archive_url}")

        return result

ユースケース別の監視戦略

ニュース監視

## ニュース / 現在のイベント監視

### 監視対象ページ:
- 速報ニュースセクション
- プレスリリースページ
- 政府発表ページ
- 企業ニュースルーム

### 監視頻度:
- 速報: 5分ごと
- プレスリリース: 15～30分ごと
- 一般ニュース: 1時間ごと

### アーカイブ戦略:
- 検出直後にアーカイブ
- Wayback Machine と Archive.today の両方を使用
- タイムスタンプ付きでローカルコピーを保存

研究監視

## 学術 / 研究監視

### 監視対象ページ:
- プレプリントサーバー (arXiv, SSRN)
- ジャーナル目次
- 学会議論録
- 研究者プロフィール

### 監視頻度:
- アクティブなトピック: 毎日
- 一般監視: 毎週

### 推奨ツール:
- Google Scholar アラート（無料、組み込み）
- Semantic Scholar アラート
- RSS フィード（利用可能な場合）
- 特定ページ用のカスタムモニター

競合インテリジェンス

## 競合監視

### 監視対象ページ:
- 価格ページ
- 製品ページ
- 求人情報
- プレスリリース
- 経営陣の経歴

### 監視頻度:
- 価格: 毎日
- 製品: 毎日
- 採用: 毎週
- プレス: 毎日

### 法的配慮:
- サービス利用規約に違反しない
- アクセス制御を回避しない
- 公開ページのみ
- 高頻度でスクレイプしない

ベストプラクティス

監視チェックリスト

## ページ監視を開始する前に:

- [ ] ページは公開アクセス可能か?
- [ ] robots.txt を尊重しているか?
- [ ] 監視頻度は適切か?
- [ ] 正当な目的があるか?
- [ ] データを安全に保管しているか?
- [ ] アラートが設定されているか?
- [ ] 重要ページのアーカイブが設定されているか?

## メンテナンス:

- [ ] 月1回モニターをレビュー
- [ ] 使用されていないモニターを削除
- [ ] ページ変更時にセレクターを更新
- [ ] アラート配信を確認
- [ ] アーカイブが正常に動作しているか確認

レート制限

import time
from functools import wraps

def rate_limit(min_interval: float = 1.0):
    """関数呼び出しをレート制限するデコレーター。"""
    last_call = [0.0]

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_call[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            last_call[0] = time.time()
            return func(*args, **kwargs)
        return wrapper
    return decorator

# 使用例
@rate_limit(min_interval=2.0)  # 最大2秒ごと
def check_page(url: str):
    return requests.get(url)