どうも、ukimaruです。
サッカーオタクなので最新の情報を誰よりも早く入手したいと思ってます。
そこで、今回は生成AI「Claude」を使って、Jリーグクラブチームの公式サイトのニュース更新を自動でLINEに通知するアプリを開発してみました。
その名も「ニュース監視アプリ」☀️
背景:情報の取りこぼしが悔しい
今回は我がレイソルで。
公式サイト(reysol.co.jp)って、ニュース更新の頻度が高いんですが通知ができないんですよね。
でもRSSもないし、公式アプリも通知が不十分…。ファンとして「気づいたら終わってた」イベントやグッズ情報、地味に悔しい経験あります。
そこで自動監視して、LINEに通知させよう!という発想に至りました。
Claudeの使い方:要件整理からスクリプト生成までお任せ
生成AI「Claude」に以下のように投げました:
「柏レイソルの公式サイトを定期スクレイピングして、新着ニュースがあればLINEに通知するPythonアプリを作って。環境変数でトークン設定できるようにして、ログ出力やキャッシュ制御も欲しい。」
すると…かなり完成度の高いベースコードを即座に提案してくれました。
Claudeに依頼したポイント:
- ニュース判定ロジック(タイトル+URLのハッシュ管理)
- LINE通知処理(Messaging API利用)
- エラーハンドリング
- systemdやDocker対応のガイド
これらをベースに、必要に応じてチューニング&テストしました。
アプリの構成
📁 reysol-news-monitor/
├── reysol_news_monitor.py # メイン処理
├── config.json # 設定ファイル(or 環境変数)
├── news_cache.json # 通知済みニュースのキャッシュ
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
実装のハイライト
✅ スクレイピング(BeautifulSoup + requests)
HTML構造が結構クセあり。Claudeは lxml
を使って効率よく処理するよう提案してくれました。
soup = BeautifulSoup(response.content, "lxml")
✅ LINE通知(Messaging API)
公式Botを作成し、ユーザーIDとアクセストークンを取得して送信。
headers = {"Authorization": f"Bearer {LINE_ACCESS_TOKEN}"}
body = {"to": LINE_USER_ID, "messages": [{"type": "text", "text": message}]}
requests.post(LINE_API_URL, headers=headers, json=body)
✅ キャッシュで重複通知を防止
news_cache.json
に過去のニュースをハッシュ保存して、既読管理しています。
実行方法
ローカル実行
python reysol_news_monitor.py
バックグラウンド(Linux)
nohup python reysol_news_monitor.py > monitor.log 2>&1 &
systemdで常駐化
ExecStart=/usr/bin/python3 /path/to/reysol_news_monitor.py
Restart=always
Docker化もOK
docker-compose up -d
生成AIを使って感じたこと
◎ 良かった点
- 面倒な構文や仕様調査をせず、構造化されたコードが即出てくる
- Claudeの出力が読みやすく丁寧。Python初学者でも追いやすい
- ドキュメント(セットアップ手順やエラー対応)も自然に生成
△ 惜しかった点
- HTML構造が変わるとエラーになる → セレクタの保守は必要
- Webhook設定やLINE側の仕様変更に注意
コード
import requests
import json
import time
import hashlib
from datetime import datetime
from bs4 import BeautifulSoup
import logging
import os
from typing import List, Dict, Set
# ログ設定
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('reysol_monitor.log', encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class ReysalNewsMonitor:
def __init__(self, line_access_token: str, line_user_id: str):
"""
柏レイソルのニュース監視クラス
Args:
line_access_token: LINE Messaging APIのアクセストークン
line_user_id: 通知を送信するLINE USER ID
"""
self.reysol_url = "https://www.reysol.co.jp/"
self.line_access_token = line_access_token
self.line_user_id = line_user_id
self.line_api_url = "https://api.line.me/v2/bot/message/push"
self.cache_file = "news_cache.json"
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
# キャッシュされたニュースを読み込み
self.cached_news = self.load_cached_news()
def load_cached_news(self) -> Set[str]:
"""キャッシュされたニュースのハッシュを読み込み"""
try:
if os.path.exists(self.cache_file):
with open(self.cache_file, 'r', encoding='utf-8') as f:
data = json.load(f)
return set(data.get('news_hashes', []))
return set()
except Exception as e:
logger.error(f"キャッシュファイル読み込みエラー: {e}")
return set()
def save_cached_news(self, news_hashes: Set[str]):
"""ニュースのハッシュをキャッシュファイルに保存"""
try:
data = {
'news_hashes': list(news_hashes),
'last_update': datetime.now().isoformat()
}
with open(self.cache_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
except Exception as e:
logger.error(f"キャッシュファイル保存エラー: {e}")
def fetch_news(self) -> List[Dict[str, str]]:
"""柏レイソルサイトからニュースを取得"""
try:
response = requests.get(self.reysol_url, headers=self.headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
news_items = []
# ニュースセクションを検索(サイト構造に基づいて調整)
# メインのニュースセクションを探す
news_sections = soup.find_all('div', class_=lambda x: x and 'news' in x.lower())
# より汎用的なアプローチでニュースリンクを探す
news_links = soup.find_all('a', href=True)
for link in news_links:
href = link.get('href', '')
text = link.get_text(strip=True)
# ニュース記事らしいリンクを判定
if (text and len(text) > 10 and
any(keyword in text for keyword in ['選手', '試合', 'チケット', 'イベント', 'お知らせ', '募集']) and
(href.startswith('/') or 'reysol.co.jp' in href)):
# 日付を検索(周辺のテキストから)
date_text = ""
parent = link.parent
if parent:
parent_text = parent.get_text()
# 日付パターンを検索 (YYYY年MM月DD日 形式)
import re
date_match = re.search(r'(\d{4})年(\d{1,2})月(\d{1,2})日', parent_text)
if date_match:
date_text = f"{date_match.group(1)}-{date_match.group(2):0>2}-{date_match.group(3):0>2}"
news_item = {
'title': text,
'url': href if href.startswith('http') else f"https://www.reysol.co.jp{href}",
'date': date_text or datetime.now().strftime('%Y-%m-%d'),
'hash': hashlib.md5((text + href).encode('utf-8')).hexdigest()
}
news_items.append(news_item)
# 重複を除去し、最新のニュースを優先
unique_news = {}
for item in news_items:
if item['hash'] not in unique_news:
unique_news[item['hash']] = item
# 日付でソート(新しい順)
sorted_news = sorted(unique_news.values(),
key=lambda x: x['date'], reverse=True)[:20] # 最新20件
logger.info(f"取得したニュース数: {len(sorted_news)}")
return sorted_news
except Exception as e:
logger.error(f"ニュース取得エラー: {e}")
return []
def send_line_notification(self, news_item: Dict[str, str]):
"""LINEに通知を送信"""
try:
message = f"""🏆 柏レイソル 新着ニュース
📰 {news_item['title']}
📅 {news_item['date']}
🔗 {news_item['url']}
#柏レイソル #REYSOL"""
payload = {
"to": self.line_user_id,
"messages": [
{
"type": "text",
"text": message
}
]
}
headers = {
"Authorization": f"Bearer {self.line_access_token}",
"Content-Type": "application/json"
}
response = requests.post(
self.line_api_url,
data=json.dumps(payload),
headers=headers,
timeout=10
)
if response.status_code == 200:
logger.info(f"LINE通知送信成功: {news_item['title']}")
return True
else:
logger.error(f"LINE通知送信失敗: {response.status_code} - {response.text}")
return False
except Exception as e:
logger.error(f"LINE通知送信エラー: {e}")
return False
def check_for_new_news(self):
"""新しいニュースをチェックして通知"""
try:
current_news = self.fetch_news()
new_news_found = False
current_hashes = set()
for news_item in current_news:
news_hash = news_item['hash']
current_hashes.add(news_hash)
# 新しいニュースかチェック
if news_hash not in self.cached_news:
logger.info(f"新しいニュース発見: {news_item['title']}")
# LINE通知送信
if self.send_line_notification(news_item):
new_news_found = True
# 少し間隔を空けて送信
time.sleep(2)
# レート制限を考慮して最大5件まで
if len([h for h in current_hashes if h not in self.cached_news]) >= 5:
logger.info("一度に送信するニュース数の上限に達しました")
break
# キャッシュを更新
self.cached_news = current_hashes
self.save_cached_news(current_hashes)
if not new_news_found:
logger.info("新しいニュースはありませんでした")
return new_news_found
except Exception as e:
logger.error(f"ニュースチェックエラー: {e}")
return False
def run_continuous_monitoring(self, interval_minutes: int = 30):
"""継続的な監視を実行"""
logger.info(f"柏レイソルニュース監視を開始します(間隔: {interval_minutes}分)")
while True:
try:
logger.info("ニュースチェック開始...")
self.check_for_new_news()
logger.info(f"次回チェックまで {interval_minutes} 分待機...")
time.sleep(interval_minutes * 60)
except KeyboardInterrupt:
logger.info("監視を停止します")
break
except Exception as e:
logger.error(f"監視中にエラーが発生: {e}")
logger.info("5分後に再試行します...")
time.sleep(300) # 5分待機
def main():
"""メイン実行関数"""
# 環境変数または直接設定
LINE_ACCESS_TOKEN = os.getenv('LINE_ACCESS_TOKEN', 'YOUR_LINE_ACCESS_TOKEN_HERE')
LINE_USER_ID = os.getenv('LINE_USER_ID', 'YOUR_LINE_USER_ID_HERE')
if LINE_ACCESS_TOKEN == 'YOUR_LINE_ACCESS_TOKEN_HERE' or LINE_USER_ID == 'YOUR_LINE_USER_ID_HERE':
print("LINE_ACCESS_TOKEN と LINE_USER_ID を設定してください")
print("環境変数または直接コード内で設定可能です")
return
# 監視インスタンス作成
monitor = ReysalNewsMonitor(LINE_ACCESS_TOKEN, LINE_USER_ID)
# 初回テスト実行
print("初回ニュースチェックを実行します...")
monitor.check_for_new_news()
# 継続監視開始の確認
response = input("継続監視を開始しますか? (y/n): ")
if response.lower() == 'y':
interval = input("チェック間隔を分で入力してください (デフォルト: 30): ")
try:
interval = int(interval) if interval else 30
except ValueError:
interval = 30
monitor.run_continuous_monitoring(interval)
else:
print("監視を終了します")
if __name__ == "__main__":
main()
今後やりたいこと
- Web UI追加して、通知ログをブラウザで確認できるようにする
- 複数サイト(Jリーグ公式・サポーターサイトなど)への対応
- ユーザーごとの通知カスタマイズ(カテゴリ別通知)
おわりに
今回は、「Claude × Python」でまあまあ実用的なアプリを短時間で構築できました。
ちょっとしたアイデアも、生成AIの力を借りることで形にできます。
exit 0
コメント