ニュースサイトのスクレイピングと分析
NewsAPIやスクレイピングでニュース記事を収集し、トレンド分析を実装します。
この記事で学べること
- NewsAPIでニュース収集
- BeautifulSoupでWebページ解析
- テキストマイニングとキーワード抽出
- トレンドの可視化
1. 準備：必要なライブラリ
# インストール
pip install newsapi-python
pip install beautifulsoup4
pip install requests
pip install pandas
pip install matplotlib
pip install wordcloud
pip install janome
pip install python-dotenvrequirements.txt
newsapi-python==0.2.7
beautifulsoup4==4.12.2
requests==2.31.0
pandas==2.1.0
matplotlib==3.8.0
wordcloud==1.9.2
janome==0.5.0
python-dotenv==1.0.02. NewsAPI キーの取得
アカウント作成手順
1. NewsAPI にアクセス
https://newsapi.org/
2. 「Get API Key」をクリック
- メールアドレス登録
- パスワード設定
3. APIキー確認
ダッシュボードでキーをコピー
4. 無料プラン制限
- 1日: 100リクエスト
- 過去1ヶ月のニュースのみ
- 商用利用不可.envファイルに保存
# .env
NEWS_API_KEY=your_api_key_here3. NewsAPIでニュース取得
news_scraper.py
#!/usr/bin/env python3
"""ニューススクレイパー（NewsAPI）"""
import os
from newsapi import NewsApiClient
from dotenv import load_dotenv
import pandas as pd
from datetime import datetime, timedelta
load_dotenv()
class NewsScraper:
def __init__(self):
api_key = os.getenv('NEWS_API_KEY')
self.newsapi = NewsApiClient(api_key=api_key)
def get_top_headlines(self, country='jp', category=None, q=None):
"""トップヘッドラインを取得
Parameters:
-----------
country : str
国コード ('jp', 'us', 'gb' など)
category : str
カテゴリ ('business', 'entertainment', 'general',
'health', 'science', 'sports', 'technology')
q : str
検索キーワード
"""
try:
response = self.newsapi.get_top_headlines(
country=country,
category=category,
q=q,
page_size=100
)
if response['status'] != 'ok':
print(f"❌ エラー: {response.get('message', '不明')}")
return None
articles = self._parse_articles(response['articles'])
return pd.DataFrame(articles)
except Exception as e:
print(f"❌ エラー: {e}")
return None
def search_news(self, q, language='jp', sort_by='publishedAt', days=7):
"""ニュースを検索
Parameters:
-----------
q : str
検索キーワード
language : str
言語コード ('jp', 'en' など)
sort_by : str
ソート順 ('relevancy', 'popularity', 'publishedAt')
days : int
遡る日数
"""
try:
from_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
response = self.newsapi.get_everything(
q=q,
language=language,
sort_by=sort_by,
from_param=from_date,
page_size=100
)
if response['status'] != 'ok':
print(f"❌ エラー: {response.get('message', '不明')}")
return None
articles = self._parse_articles(response['articles'])
return pd.DataFrame(articles)
except Exception as e:
print(f"❌ エラー: {e}")
return None
def get_sources(self, country='jp', category=None, language='jp'):
"""ニュースソース一覧を取得"""
try:
response = self.newsapi.get_sources(
country=country,
category=category,
language=language
)
sources = []
for source in response['sources']:
sources.append({
'id': source['id'],
'name': source['name'],
'description': source.get('description', ''),
'url': source['url'],
'category': source['category']
})
return pd.DataFrame(sources)
except Exception as e:
print(f"❌ エラー: {e}")
return None
def _parse_articles(self, articles):
"""記事データをパース"""
parsed = []
for article in articles:
parsed.append({
'title': article.get('title', ''),
'description': article.get('description', ''),
'content': article.get('content', ''),
'author': article.get('author', '不明'),
'source': article['source']['name'],
'url': article.get('url', ''),
'published_at': article.get('publishedAt', ''),
'image_url': article.get('urlToImage', '')
})
return parsed
# 使用例
if __name__ == '__main__':
scraper = NewsScraper()
# 日本のトップヘッドライン
print("=== 日本のトップニュース ===")
df = scraper.get_top_headlines(country='jp')
if df is not None:
print(f"取得記事数: {len(df)}\n")
print(df[['title', 'source']].head(10))
# キーワード検索
print("\n=== 'AI' 関連ニュース ===")
df_ai = scraper.search_news('AI OR 人工知能', days=7)
if df_ai is not None:
print(f"取得記事数: {len(df_ai)}\n")
print(df_ai[['title', 'published_at']].head(5))実行結果
=== 日本のトップニュース ===
取得記事数: 87
title source
0 日本経済、3四半期連続でプラス成長 日本経済新聞
1 新型コロナワクチン、追加接種開始 NHK NEWS WEB
2 東京株式市場、日経平均が年初来高値更新 Bloomberg
3 政府、デジタル化推進で新予算案発表 朝日新聞
4 スポーツ界、来年の国際大会開催地決定 スポーツ報知
=== 'AI' 関連ニュース ===
取得記事数: 42
title published_at
0 ChatGPT新機能、マルチモーダル対応へ 2025-10-26T10:30:00Z
1 Google、AI検索機能を大幅強化 2025-10-26T08:15:00Z
2 日本企業、生成AI導入が加速 2025-10-25T14:22:00Z4. BeautifulSoupで個別サイトをスクレイピング
web_scraper.py
#!/usr/bin/env python3
"""Webスクレイパー（BeautifulSoup）"""
import requests
from bs4 import BeautifulSoup
import pandas as pd
from datetime import datetime
import time
class WebNewsScraper:
def __init__(self):
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
def scrape_yahoo_news(self, max_articles=20):
"""Yahoo!ニューストピックスをスクレイピング
注意: 実際の使用前に利用規約を確認すること
"""
try:
url = 'https://news.yahoo.co.jp/topics/top-picks'
response = requests.get(url, headers=self.headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
articles = []
# 記事要素を探す（実際の構造に合わせて調整必要）
article_elements = soup.find_all('article', limit=max_articles)
for article in article_elements:
# タイトル取得
title_elem = article.find('a')
if title_elem:
title = title_elem.get_text(strip=True)
link = title_elem.get('href', '')
articles.append({
'title': title,
'url': link,
'source': 'Yahoo!ニュース',
'scraped_at': datetime.now()
})
return pd.DataFrame(articles)
except Exception as e:
print(f"❌ スクレイピングエラー: {e}")
return None
def scrape_nhk_news(self, max_articles=20):
"""NHKニュースをスクレイピング
注意: robots.txtとサービス利用規約を遵守
"""
try:
url = 'https://www3.nhk.or.jp/news/'
response = requests.get(url, headers=self.headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
articles = []
# 記事リンクを探す
links = soup.find_all('a', class_='content--summary', limit=max_articles)
for link in links:
title = link.get_text(strip=True)
href = link.get('href', '')
if href.startswith('/'):
href = 'https://www3.nhk.or.jp' + href
articles.append({
'title': title,
'url': href,
'source': 'NHK NEWS WEB',
'scraped_at': datetime.now()
})
# 負荷軽減のため待機
time.sleep(0.5)
return pd.DataFrame(articles)
except Exception as e:
print(f"❌ スクレイピングエラー: {e}")
return None
def scrape_article_content(self, url):
"""記事本文を取得"""
try:
response = requests.get(url, headers=self.headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
# 本文を抽出（サイトによって構造が異なる）
# article, mainタグやclass="content"などを探す
content = ''
# 一般的なパターン
for tag in soup.find_all(['article', 'main']):
paragraphs = tag.find_all('p')
content = '\n'.join([p.get_text(strip=True) for p in paragraphs])
if content:
break
return content
except Exception as e:
print(f"❌ 本文取得エラー: {e}")
return None
# 使用例
if __name__ == '__main__':
scraper = WebNewsScraper()
# Yahoo!ニューススクレイピング
df = scraper.scrape_yahoo_news(max_articles=10)
if df is not None:
print("=== Yahoo!ニュース ===")
print(df[['title', 'source']].head())5. テキストマイニング・キーワード抽出
text_analyzer.py
#!/usr/bin/env python3
"""ニューステキスト分析"""
import pandas as pd
from janome.tokenizer import Tokenizer
from collections import Counter
import re
class NewsTextAnalyzer:
def __init__(self):
self.tokenizer = Tokenizer()
# ストップワード（除外する一般的な語）
self.stop_words = set([
'こと', 'もの', 'ため', 'よう', 'それ', 'ここ',
'さん', 'これ', 'そこ', 'あれ', 'どこ', 'なる',
'ある', 'する', 'いる', 'れる', 'られる', 'ない'
])
def extract_keywords(self, text, top_n=20):
"""キーワードを抽出"""
# 形態素解析
tokens = self.tokenizer.tokenize(text)
# 名詞のみ抽出
nouns = []
for token in tokens:
parts = token.part_of_speech.split(',')
if parts[0] == '名詞' and parts[1] == '一般':
surface = token.surface
# 1文字の単語とストップワードを除外
if len(surface) > 1 and surface not in self.stop_words:
nouns.append(surface)
# 頻度カウント
counter = Counter(nouns)
return counter.most_common(top_n)
def analyze_news_trends(self, df, text_column='title'):
"""ニューストレンドを分析"""
# 全テキストを結合
all_text = ' '.join(df[text_column].dropna().astype(str))
# キーワード抽出
keywords = self.extract_keywords(all_text, top_n=30)
# DataFrameに変換
keywords_df = pd.DataFrame(keywords, columns=['キーワード', '出現回数'])
return keywords_df
def extract_entities(self, text):
"""固有表現抽出（簡易版）"""
entities = []
tokens = self.tokenizer.tokenize(text)
for token in tokens:
parts = token.part_of_speech.split(',')
# 固有名詞
if parts[0] == '名詞' and parts[1] == '固有名詞':
entities.append(token.surface)
return Counter(entities)
def get_sentiment_words(self, text):
"""感情語を抽出（簡易版）"""
# ポジティブワード
positive_words = ['成功', '向上', '増加', '達成', '改善', '発展', '期待']
# ネガティブワード
negative_words = ['失敗', '減少', '低下', '悪化', '問題', '懸念', '危機']
pos_count = sum(1 for word in positive_words if word in text)
neg_count = sum(1 for word in negative_words if word in text)
return {
'positive': pos_count,
'negative': neg_count,
'sentiment': 'ポジティブ' if pos_count > neg_count else
'ネガティブ' if neg_count > pos_count else '中立'
}
# 使用例
from news_scraper import NewsScraper
scraper = NewsScraper()
df = scraper.get_top_headlines(country='jp')
if df is not None:
analyzer = NewsTextAnalyzer()
# トレンドキーワード
trends = analyzer.analyze_news_trends(df)
print("=== トレンドキーワード ===")
print(trends.head(15))
# 感情分析
sentiments = []
for title in df['title']:
sent = analyzer.get_sentiment_words(str(title))
sentiments.append(sent['sentiment'])
df['sentiment'] = sentiments
print("\n=== 感情分布 ===")
print(df['sentiment'].value_counts())6. データの可視化
news_visualizer.py
#!/usr/bin/env python3
"""ニュースデータ可視化"""
import matplotlib.pyplot as plt
import seaborn as sns
from wordcloud import WordCloud
import pandas as pd
sns.set_style('whitegrid')
plt.rcParams['font.sans-serif'] = ['MS Gothic', 'Yu Gothic', 'Arial']
plt.rcParams['axes.unicode_minus'] = False
class NewsVisualizer:
def __init__(self, df):
self.df = df.copy()
def plot_keyword_ranking(self, keywords_df, top_n=20):
"""キーワードランキンググラフ"""
top_keywords = keywords_df.head(top_n)
plt.figure(figsize=(12, 8))
bars = plt.barh(
range(len(top_keywords)),
top_keywords['出現回数'],
color='#2196F3'
)
plt.yticks(
range(len(top_keywords)),
top_keywords['キーワード']
)
plt.xlabel('出現回数', fontsize=12)
plt.title(f'ニュース トップ{top_n}キーワード', fontsize=14, fontweight='bold')
plt.gca().invert_yaxis()
# 値を表示
for i, (bar, count) in enumerate(zip(bars, top_keywords['出現回数'])):
plt.text(
count,
i,
f' {count}',
va='center',
fontsize=10
)
plt.tight_layout()
plt.savefig('news_keywords.png', dpi=300, bbox_inches='tight')
plt.show()
def create_wordcloud(self, text_column='title'):
"""ワードクラウド作成"""
from janome.tokenizer import Tokenizer
# テキスト結合
all_text = ' '.join(self.df[text_column].dropna().astype(str))
# 形態素解析
tokenizer = Tokenizer()
tokens = tokenizer.tokenize(all_text, wakati=True)
wakati_text = ' '.join(tokens)
# ワードクラウド生成
wordcloud = WordCloud(
width=1200,
height=600,
background_color='white',
font_path='C:/Windows/Fonts/msgothic.ttc', # 日本語フォント
colormap='viridis',
max_words=100,
relative_scaling=0.5
).generate(wakati_text)
plt.figure(figsize=(15, 8))
plt.imshow(wordcloud, interpolation='bilinear')
plt.axis('off')
plt.title('ニュース ワードクラウド', fontsize=16, fontweight='bold', pad=20)
plt.tight_layout()
plt.savefig('news_wordcloud.png', dpi=300, bbox_inches='tight')
plt.show()
def plot_sources_distribution(self):
"""ニュースソース別分布"""
source_counts = self.df['source'].value_counts().head(15)
plt.figure(figsize=(12, 6))
bars = plt.bar(
range(len(source_counts)),
source_counts.values,
color='#FF9800'
)
plt.xticks(
range(len(source_counts)),
source_counts.index,
rotation=45,
ha='right'
)
plt.ylabel('記事数', fontsize=12)
plt.title('ニュースソース別記事数', fontsize=14, fontweight='bold')
# 値を表示
for bar in bars:
height = bar.get_height()
plt.text(
bar.get_x() + bar.get_width()/2.,
height,
f'{int(height)}',
ha='center',
va='bottom',
fontsize=10
)
plt.tight_layout()
plt.savefig('news_sources.png', dpi=300, bbox_inches='tight')
plt.show()
def plot_timeline(self):
"""時系列グラフ"""
if 'published_at' not in self.df.columns:
print("published_at カラムがありません")
return
# 日時に変換
self.df['published_date'] = pd.to_datetime(
self.df['published_at'],
errors='coerce'
)
# 日別集計
daily_counts = self.df.groupby(
self.df['published_date'].dt.date
).size()
plt.figure(figsize=(14, 6))
plt.plot(
daily_counts.index,
daily_counts.values,
marker='o',
linewidth=2,
markersize=8,
color='#4CAF50'
)
plt.xlabel('日付', fontsize=12)
plt.ylabel('記事数', fontsize=12)
plt.title('日別ニュース記事数', fontsize=14, fontweight='bold')
plt.grid(True, alpha=0.3)
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig('news_timeline.png', dpi=300, bbox_inches='tight')
plt.show()7. 完全なニュース分析システム
#!/usr/bin/env python3
"""完全なニュース分析システム"""
from news_scraper import NewsScraper
from text_analyzer import NewsTextAnalyzer
from news_visualizer import NewsVisualizer
import pandas as pd
def full_news_analysis(keyword=None, country='jp', days=7):
"""完全なニュース分析"""
print(f"\n{'='*60}")
print(f" ニュース分析レポート")
if keyword:
print(f" キーワード: {keyword}")
print(f"{'='*60}\n")
scraper = NewsScraper()
# 1. データ収集
print("[1/5] ニュースデータ収集中...")
if keyword:
df = scraper.search_news(keyword, days=days)
else:
df = scraper.get_top_headlines(country=country)
if df is None or len(df) == 0:
print("❌ データ取得失敗")
return None
print(f"✅ {len(df)}件のニュース記事を取得\n")
# 2. 基本統計
print("[2/5] 基本統計...")
print(f"期間: {df['published_at'].min()} ～ {df['published_at'].max()}")
print(f"ユニークソース数: {df['source'].nunique()}")
print(f"\nトップソース:")
print(df['source'].value_counts().head(5))
print()
# 3. テキスト分析
print("[3/5] テキスト分析中...")
analyzer = NewsTextAnalyzer()
keywords_df = analyzer.analyze_news_trends(df)
print("\n=== トップキーワード ===")
print(keywords_df.head(10))
print()
# 4. 可視化
print("[4/5] グラフ作成中...")
visualizer = NewsVisualizer(df)
visualizer.plot_keyword_ranking(keywords_df, top_n=20)
visualizer.create_wordcloud()
visualizer.plot_sources_distribution()
visualizer.plot_timeline()
print("✅ グラフを保存しました\n")
# 5. CSV保存
print("[5/5] データ保存中...")
from datetime import datetime
filename = f'news_analysis_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv'
df.to_csv(filename, index=False, encoding='utf-8-sig')
print(f"✅ データ保存: {filename}\n")
print(f"{'='*60}")
print(" 分析完了")
print(f"{'='*60}\n")
return df, keywords_df
if __name__ == '__main__':
# 日本のニュース分析
df, keywords = full_news_analysis(country='jp')
# または特定キーワードで分析
# df, keywords = full_news_analysis(keyword='AI OR 人工知能', days=7)8. 定期実行とアラート
import schedule
import time
from datetime import datetime
def monitor_news_keywords(keywords, interval_minutes=60):
"""特定キーワードのニュースを監視"""
scraper = NewsScraper()
def check_keywords():
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] ニュース監視実行")
for keyword in keywords:
df = scraper.search_news(keyword, days=1)
if df is not None and len(df) > 0:
new_articles = len(df)
print(f"📰 '{keyword}': {new_articles}件の新着記事")
# アラート条件
if new_articles >= 10:
print(f" 🚨 注目: 多数の記事が検出されました！")
# メール通知などを実装可能
# 初回実行
check_keywords()
# 定期実行
schedule.every(interval_minutes).minutes.do(check_keywords)
while True:
schedule.run_pending()
time.sleep(60)
# 使用例
monitor_keywords = ['AI', '株価', '気候変動']
# monitor_news_keywords(monitor_keywords, interval_minutes=30)まとめ
NewsAPIとスクレイピングでニュースを収集・分析できます！
重要ポイント
- ✅ NewsAPIは無料で簡単
- ✅ BeautifulSoupでカスタムスクレイピング
- ✅ Janomeで日本語テキスト分析
- ✅ トレンドキーワードを可視化
- ✅ 定期監視でリアルタイム追跡
応用例
- メディアモニタリング
- 競合分析・市場調査
- 危機管理（炎上検知）
- 投資判断支援
- コンテンツ企画
注意事項
- ⚠️ 必ず利用規約とrobots.txtを確認
- ⚠️ スクレイピング頻度に注意
- ⚠️ APIレート制限を遵守
- ⚠️ 著作権に配慮
- ⚠️ 個人情報の取り扱いに注意