AIでWebサイト監視 — 変更検知・ダウン検知・価格追跡
約5分で読めます
AIでWebサイト監視 — 変更検知・ダウン検知・価格追跡
「競合サイトの価格が変わったらすぐ知りたい」「取引先のサイトがダウンしたら即座に対応したい」「政府系サイトの入札情報が更新されたら通知してほしい」。こうしたニーズに応えるのがAI駆動のWeb監視システムだ。定期スクレイピングで変更を検知し、AIで変更内容を分析して、重要度に応じてDiscordやSlackに通知する。Pythonで構築する実践的なパイプラインを解説する。
Web監視システムの全体アーキテクチャ
パイプライン構成
[スケジューラ] -> [スクレイパー] -> [差分検知] -> [AI分析] -> [通知]
cron requests/ difflib Claude API Discord/
schedule Playwright Slack/Email監視タイプと用途
| 監視タイプ | 用途 | チェック頻度 |
|---|---|---|
| ダウン検知 | サイトの死活監視、レスポンス時間 | 5分ごと |
| コンテンツ変更 | ニュース、ブログ、求人情報の更新 | 1〜24時間ごと |
| 価格追跡 | ECサイトの価格変動、セール検知 | 1〜6時間ごと |
| 入札・公募監視 | 官公庁の入札情報、補助金情報 | 6〜12時間ごと |
| SEO監視 | 検索順位、メタ情報の変更 | 24時間ごと |
基本的なWeb監視スクリプト
HTTPステータスとレスポンス時間の監視
import requests
import time
import json
from datetime import datetime
from pathlib import Path
class SiteMonitor:
"""Webサイトの死活監視とレスポンス時間を追跡する"""
def __init__(self, config_path: str = "monitor_config.json"):
self.config = self._load_config(config_path)
self.results_dir = Path("monitor_results")
self.results_dir.mkdir(exist_ok=True)
def _load_config(self, path: str) -> dict:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def check_site(self, url: str, timeout: int = 10) -> dict:
"""単一サイトのヘルスチェック"""
try:
start = time.time()
response = requests.get(url, timeout=timeout, headers={
"User-Agent": "SiteMonitor/1.0"
})
elapsed = round(time.time() - start, 3)
return {
"url": url,
"status_code": response.status_code,
"response_time": elapsed,
"is_up": response.status_code == 200,
"content_length": len(response.text),
"checked_at": datetime.now().isoformat()
}
except requests.RequestException as e:
return {
"url": url, "status_code": 0, "response_time": -1,
"is_up": False, "error": str(e),
"checked_at": datetime.now().isoformat()
}
def check_all(self) -> list[dict]:
"""設定ファイルの全サイトをチェック"""
results = []
for site in self.config.get("sites", []):
result = self.check_site(site["url"])
result["name"] = site.get("name", site["url"])
results.append(result)
if not result["is_up"]:
print(f"[ALERT] {result['name']} is DOWN!")
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output = self.results_dir / f"check_{timestamp}.json"
output.write_text(json.dumps(results, ensure_ascii=False, indent=2))
return resultsHTML差分検知とAI分析
コンテンツの変更検知
import difflib
import hashlib
import anthropic
from bs4 import BeautifulSoup
client = anthropic.Anthropic()
class ContentChangeDetector:
"""Webページのコンテンツ変更を検知しAIで分析する"""
def __init__(self, storage_dir: str = "page_snapshots"):
self.storage_dir = Path(storage_dir)
self.storage_dir.mkdir(exist_ok=True)
def _extract_text(self, html: str) -> str:
"""HTMLからテキストコンテンツを抽出"""
soup = BeautifulSoup(html, "html.parser")
for tag in soup(["script", "style", "nav", "footer", "header"]):
tag.decompose()
return soup.get_text(separator="\n", strip=True)
def _get_snapshot_path(self, url: str) -> Path:
url_hash = hashlib.md5(url.encode()).hexdigest()
return self.storage_dir / f"{url_hash}.txt"
def check_for_changes(self, url: str) -> dict:
"""ページの変更を検知する"""
response = requests.get(url, headers={"User-Agent": "ContentMonitor/1.0"})
current_text = self._extract_text(response.text)
snapshot_path = self._get_snapshot_path(url)
if not snapshot_path.exists():
snapshot_path.write_text(current_text, encoding="utf-8")
return {"url": url, "changed": False, "reason": "初回スナップショット保存"}
previous_text = snapshot_path.read_text(encoding="utf-8")
if current_text == previous_text:
return {"url": url, "changed": False}
diff = list(difflib.unified_diff(
previous_text.splitlines(), current_text.splitlines(), lineterm=""
))
snapshot_path.write_text(current_text, encoding="utf-8")
return {"url": url, "changed": True, "diff_lines": len(diff),
"diff": "\n".join(diff[:100])}
def analyze_change_with_ai(self, change_result: dict) -> dict:
"""AIで変更内容を分析する"""
if not change_result.get("changed"):
return {"analysis": "変更なし", "importance": "NONE"}
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1000,
messages=[{
"role": "user",
"content": f"""Webページの変更差分を分析してください。
URL: {change_result['url']}
差分: {change_result['diff']}
判定: 変更種類、重要度(HIGH/MEDIUM/LOW)、要約、推奨アクション（JSON形式）"""
}]
)
return json.loads(response.content[0].text)価格追跡システム
ECサイトの価格変動監視
class PriceTracker:
"""ECサイトの価格を追跡し変動をアラートする"""
def __init__(self, db_path: str = "price_history.json"):
self.db_path = Path(db_path)
self.history = self._load_history()
def _load_history(self) -> dict:
if self.db_path.exists():
return json.loads(self.db_path.read_text(encoding="utf-8"))
return {}
def _save_history(self):
self.db_path.write_text(
json.dumps(self.history, ensure_ascii=False, indent=2), encoding="utf-8"
)
def track_price(self, product_id: str, url: str, css_selector: str) -> dict:
"""指定商品の価格を取得し履歴に記録する"""
response = requests.get(url, headers={"User-Agent": "PriceTracker/1.0"})
soup = BeautifulSoup(response.text, "html.parser")
price_element = soup.select_one(css_selector)
if not price_element:
return {"error": "価格要素が見つかりません", "product_id": product_id}
price_text = price_element.get_text(strip=True)
price = int("".join(c for c in price_text if c.isdigit()))
if product_id not in self.history:
self.history[product_id] = {"url": url, "prices": []}
entry = {"price": price, "timestamp": datetime.now().isoformat()}
self.history[product_id]["prices"].append(entry)
self._save_history()
prices = self.history[product_id]["prices"]
if len(prices) >= 2:
prev_price = prices[-2]["price"]
change_pct = ((price - prev_price) / prev_price) * 100
return {
"product_id": product_id, "current_price": price,
"previous_price": prev_price,
"change_percent": round(change_pct, 1),
"alert": abs(change_pct) >= 5
}
return {"product_id": product_id, "current_price": price, "alert": False}Discord/Slack通知の実装
Discord Webhookによるアラート通知
def send_discord_alert(webhook_url: str, alert_data: dict):
"""Discord Webhookでアラートを送信する"""
colors = {"HIGH": 0xFF0000, "MEDIUM": 0xFFA500, "LOW": 0x00FF00}
color = colors.get(alert_data.get("importance", "LOW"), 0x808080)
embed = {
"title": f"Web監視アラート: {alert_data.get('type', '変更検知')}",
"description": alert_data.get("summary", ""),
"color": color,
"fields": [
{"name": "URL", "value": alert_data.get("url", "N/A"), "inline": False},
{"name": "重要度", "value": alert_data.get("importance", "N/A"), "inline": True},
{"name": "検知時刻", "value": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "inline": True},
],
"footer": {"text": "AI Web Monitor v1.0"}
}
if alert_data.get("recommended_action"):
embed["fields"].append({
"name": "推奨アクション", "value": alert_data["recommended_action"], "inline": False
})
requests.post(webhook_url, json={"embeds": [embed]})定期実行とコスト
スケジュール設定
import schedule
def run_monitoring():
"""全監視タスクを実行する"""
monitor = SiteMonitor()
detector = ContentChangeDetector()
health_results = monitor.check_all()
down_sites = [r for r in health_results if not r["is_up"]]
for site in monitor.config.get("content_watch", []):
change = detector.check_for_changes(site["url"])
if change.get("changed"):
analysis = detector.analyze_change_with_ai(change)
if analysis.get("importance") in ["HIGH", "MEDIUM"]:
send_discord_alert(WEBHOOK_URL, {
"type": "コンテンツ変更",
"url": site["url"],
"summary": analysis.get("summary", ""),
"importance": analysis["importance"],
"recommended_action": analysis.get("action", "")
})
schedule.every(5).minutes.do(lambda: SiteMonitor().check_all())
schedule.every(6).hours.do(run_monitoring)月間コスト試算
| 項目 | コスト |
|---|---|
| Claude API（1日10回の分析） | 約500〜1,000円/月 |
| サーバー（VPS/Cloud Run） | 0〜1,000円/月 |
| 通知（Discord/Slack） | 無料 |
| 合計 | 500〜2,000円/月 |
監視サービスとの比較
| サービス | 月額 | 監視サイト数 | AI分析 |
|---|---|---|---|
| UptimeRobot（Pro） | $7〜 | 50サイト | なし |
| Datadog | $15〜/ホスト | 無制限 | あり |
| 自作（本記事の方法） | 500〜2,000円 | 無制限 | あり |
自作の利点は、カスタマイズの自由度とAI分析の統合だ。特に「変更内容の意味を理解して通知する」機能は、既存サービスにはない強みである。
まとめ — 今日から始めるWeb監視
- Step 1: 監視したいサイトを3つ選ぶ（自社サイト、競合、取引先）
- Step 2: 基本的な死活監視スクリプトを動かす
- Step 3: コンテンツ変更検知とAI分析を追加する
- Step 4: Discord通知とcron定期実行で自動化を完成させる
Web監視は「問題が起きてから気づく」を「問題が起きた瞬間に知る」に変える。この差が、ビジネスにおける対応速度の決定的な差になる。
関連記事
A
Agentive 編集部
AIエージェントを実際に使い倒す個人開発者。サイト制作の自動化を実践しながら、その知見を発信しています。