Claudeで作ってみた!ホームページの公式ニュースをLINEに自動通知するアプリ

tech

どうも、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

コメント

タイトルとURLをコピーしました