Python Webスクレイピング 2025 第1部：Playwright入門と環境構築

# Python環境管理（推奨）
# pyenvとPoetryの組み合わせを使用する理由：
# - pyenv: 複数のPythonバージョンを管理
# - Poetry: 依存関係とパッケージ管理を効率化

# pyenvのインストール（Pythonバージョン管理）
curl -L https://github.com/pyenv/pyenv-installer/raw/master/bin/pyenv-installer | bash

# Poetryのインストール（パッケージ管理）
curl -sSL https://install.python-poetry.org | python3 -

# Python 3.11以上を推奨（型ヒントとパフォーマンス改善のため）
pyenv install 3.11.7
pyenv global 3.11.7

# プロジェクト初期化
# pyproject.tomlファイルが作成され、依存関係が管理される
poetry init

# 必要なパッケージのインストール
# playwright: ブラウザ自動化
# aiohttp: 非同期HTTP通信
# beautifulsoup4: HTML解析
# pandas: データ操作
poetry add playwright aiohttp asyncio beautifulsoup4 pandas

# フレームワーク使用時（オプション）
# scrapy: 大規模スクレイピングフレームワーク
# scrapy-playwright: ScrapyとPlaywrightの統合
poetry add scrapy scrapy-playwright

# Playwrightブラウザのインストール
# 実際のブラウザバイナリをダウンロード
poetry run playwright install

# 2025年標準構成 - 各インポートの役割を解説

# 非同期処理の核となるライブラリ
import asyncio  # 非同期プログラミングのためのライブラリ

# HTTP通信用（非同期対応）
import aiohttp  # 高速な非同期HTTP通信ライブラリ

# Playwright関連
from playwright.async_api import async_playwright  # 非同期版のPlaywright API

# HTML解析
from bs4 import BeautifulSoup  # HTMLパーサー（要素の抽出に使用）

# データ処理
import pandas as pd  # データフレーム操作（CSV出力、統計処理等）
import json         # JSON形式のデータ処理
import time         # 時間関連の処理（遅延、計測等）

# 日時処理
from datetime import datetime, timedelta  # 日付・時刻の操作

# 型ヒント（コードの可読性と安全性向上）
from typing import List, Dict, Optional, Union

# ログ出力
import logging      # エラーログや処理状況の記録

# データクラス（構造化データの定義）
from dataclasses import dataclass, field

# ファイルパス操作
from pathlib import Path  # モダンなファイル操作

# ログ設定の詳細解説
logging.basicConfig(
    level=logging.INFO,  # INFO以上のレベルを出力
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    # フォーマット: 時刻 - ロガー名 - レベル - メッセージ
)
logger = logging.getLogger(__name__)  # モジュール固有のロガーを作成

@dataclass
class ScrapingConfig:
    """
    スクレイピング設定クラス
    
    @dataclassデコレータを使用する理由：
    - 自動的に__init__, __repr__, __eq__メソッドを生成
    - 型ヒント対応
    - デフォルト値の簡潔な記述
    """
    headless: bool = True  # ブラウザをヘッドレスモードで実行（画面表示なし）
    timeout: int = 30000   # タイムアウト時間（ミリ秒）
    user_agent: str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"  # User-Agent設定
    # viewportのデフォルト値をfactory関数で設定（辞書の参照問題を回避）
    viewport: Dict[str, int] = field(default_factory=lambda: {"width": 1920, "height": 1080})
    proxy: Optional[Dict[str, str]] = None  # プロキシ設定（任意）

class ModernWebScraper:
    """
    2025年版Webスクレイピングクラス
    
    非同期コンテキストマネージャーとして実装する理由：
    - リソースの確実な開放
    - with文による直感的な使用
    - 例外処理時の安全なクリーンアップ
    """
    
    def __init__(self, config: ScrapingConfig = None):
        """
        初期化メソッド
        
        Args:
            config: スクレイピング設定（Noneの場合デフォルト設定を使用）
        """
        self.config = config or ScrapingConfig()  # 設定が渡されなければデフォルトを使用
        self.browser = None      # ブラウザインスタンス
        self.context = None      # ブラウザコンテキスト（タブ群の管理）
        self.session_data = {}   # セッションデータ（Cookie等の保存）
    
    async def __aenter__(self):
        """
        非同期コンテキストマネージャー開始
        
        async with文で呼び出される際に実行される
        ブラウザの起動とコンテキストの作成を行う
        """
        # Playwrightインスタンスを起動
        self.playwright = await async_playwright().start()
        
        # ブラウザ起動（Chromiumを使用）
        # headless=Trueで非表示モード、Falseで画面表示
        self.browser = await self.playwright.chromium.launch(
            headless=self.config.headless,
            proxy=self.config.proxy  # プロキシ設定があれば適用
        )
        
        # ブラウザコンテキスト作成
        # コンテキスト = 独立したブラウザセッション（Cookie、ローカルストレージ等を管理）
        self.context = await self.browser.new_context(
            user_agent=self.config.user_agent,  # User-Agent設定
            viewport=self.config.viewport        # 画面解像度設定
        )
        
        # デフォルトタイムアウト設定（すべての操作に適用）
        self.context.set_default_timeout(self.config.timeout)
        
        logger.info("Playwright scraper initialized")
        return self  # selfを返すことでwith文の変数に代入される
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """
        非同期コンテキストマネージャー終了
        
        with文を抜ける際に自動実行される
        リソースの適切な解放を行う
        
        Args:
            exc_type: 例外の型
            exc_val: 例外の値
            exc_tb: トレースバック
        """
        # 順番に終了処理を実行（重要：逆順で閉じる）
        if self.context:
            await self.context.close()  # コンテキスト終了
        if self.browser:
            await self.browser.close()  # ブラウザ終了
        if self.playwright:
            await self.playwright.stop()  # Playwright終了
        logger.info("Playwright scraper closed")
    
    async def scrape_page(self, url: str, wait_selector: str = None) -> Dict:
        """
        単一ページのスクレイピング
        
        Args:
            url: スクレイピング対象のURL
            wait_selector: 待機対象のCSSセレクター（動的コンテンツ対応）
            
        Returns:
            スクレイピング結果の辞書
        """
        # 新しいページ（タブ）を作成
        page = await self.context.new_page()
        
        try:
            # ページ読み込み開始
            logger.info(f"Loading page: {url}")
            
            # wait_until="domcontentloaded"の説明：
            # - "load": すべてのリソース（画像等）の読み込み完了まで待機
            # - "domcontentloaded": HTMLの解析完了まで待機（推奨）
            # - "networkidle": ネットワーク通信が停止するまで待機
            await page.goto(url, wait_until="domcontentloaded")
            
            # 動的コンテンツの読み込み待機
            if wait_selector:
                # 指定されたCSSセレクターの要素が表示されるまで待機
                # JavaScript実行後に表示される要素に対応
                await page.wait_for_selector(wait_selector, timeout=10000)
            
            # ページの基本情報を取得
            title = await page.title()           # <title>タグの内容
            html_content = await page.content()  # ページのHTML全体
            
            # JavaScriptの実行結果を取得
            # page.evaluate()でブラウザ内でJavaScriptコードを実行
            js_data = await page.evaluate("""
                () => {
                    // ブラウザ内で実行されるJavaScriptコード
                    return {
                        url: window.location.href,        // 現在のURL
                        userAgent: navigator.userAgent,   // User-Agent
                        timestamp: Date.now(),             // タイムスタンプ
                        viewport: {
                            width: window.innerWidth,      // ビューポート幅
                            height: window.innerHeight     // ビューポート高さ
                        }
                    }
                }
            """)
            
            # 成功時の戻り値
            return {
                'url': url,
                'title': title,
                'html_content': html_content,
                'js_data': js_data,
                'timestamp': datetime.now(),
                'status': 'success'
            }
            
        except Exception as e:
            # エラー時のログ出力と戻り値
            logger.error(f"Error scraping {url}: {e}")
            return {
                'url': url,
                'error': str(e),
                'timestamp': datetime.now(),
                'status': 'error'
            }
        finally:
            # 必ずページを閉じる（メモリリーク防止）
            await page.close()

# 使用例の詳細解説
async def demo_basic_scraping():
    """
    基本的なスクレイピングのデモ
    
    この関数はasyncで定義されている理由：
    - Playwrightが非同期APIを使用
    - 複数のページを効率的に処理可能
    - ブロッキング処理を回避
    """
    
    # 設定オブジェクトの作成
    config = ScrapingConfig(
        headless=False,  # デバッグ時はブラウザを表示
        timeout=20000    # 20秒のタイムアウト
    )
    
    # 非同期コンテキストマネージャーの使用
    # with文により、自動的にリソースが解放される
    async with ModernWebScraper(config) as scraper:
        # テストサイトの配列
        # httpbin.orgは開発者向けHTTPテストサービス
        test_urls = [
            "https://httpbin.org/html",  # HTML形式のレスポンス
            "https://httpbin.org/json",  # JSON形式のレスポンス
            "https://httpbin.org/xml"    # XML形式のレスポンス
        ]
        
        results = []  # 結果を格納するリスト
        
        # 各URLを順次処理
        for url in test_urls:
            print(f"スクレイピング中: {url}")
            
            # ページをスクレイピング
            result = await scraper.scrape_page(url)
            
            # 結果の判定と処理
            if result['status'] == 'success':
                print(f"✅ 成功: {result['title']}")
                results.append(result)
            else:
                print(f"❌ エラー: {result['error']}")
            
            # 適切な間隔を空ける（サーバー負荷軽減）
            # asyncio.sleep()は非ブロッキング待機
            await asyncio.sleep(1)
        
        return results

# 実行方法の解説
# Jupyter Notebookの場合：
# results = await demo_basic_scraping()

# 通常のPythonスクリプトの場合：
# import asyncio
# results = asyncio.run(demo_basic_scraping())

class BasicDataExtractor:
    """
    基本的なデータ抽出クラス
    
    BeautifulSoupを使用してHTML要素を抽出する
    正規表現パターンも事前定義して効率化
    """
    
    def __init__(self):
        """
        初期化メソッド
        よく使用される正規表現パターンを事前定義
        """
        self.extraction_patterns = {
            # メールアドレス抽出用の正規表現
            'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
            # 電話番号抽出用（日本・米国形式対応）
            'phone': r'(\+?\d{1,3}[-.]?)?\(?\d{3}\)?[-.]?\d{3}[-.]?\d{4}',
            # 価格抽出用（通貨記号対応）
            'price': r'[¥$€£]\s*[\d,]+\.?\d*|[\d,]+\.?\d*\s*[¥$€£]',
            # 日付抽出用（複数形式対応）
            'date': r'\d{4}[-/]\d{1,2}[-/]\d{1,2}|\d{1,2}[-/]\d{1,2}[-/]\d{4}',
            # URL抽出用
            'url': r'https?://(?:[-\w.])+(?:[:\d]+)?(?:/(?:[\w/_.])*(?:\?(?:[\w&=%.])*)?(?:#(?:[\w.])*)?)?'
        }
    
    def extract_basic_info(self, html_content: str) -> Dict:
        """
        基本情報の抽出（包括的な情報取得）
        
        Args:
            html_content: HTML文字列
            
        Returns:
            抽出された情報の辞書
        """
        # BeautifulSoupでHTMLを解析
        # 'html.parser'は標準ライブラリのパーサー（追加インストール不要）
        soup = BeautifulSoup(html_content, 'html.parser')
        
        return {
            'title': self.get_title(soup),           # ページタイトル
            'headings': self.get_headings(soup),     # 見出し一覧
            'links': self.get_links(soup),           # リンク一覧
            'images': self.get_images(soup),         # 画像一覧
            'meta_data': self.get_meta_data(soup)    # メタデータ
        }
    
    def get_title(self, soup: BeautifulSoup) -> str:
        """
        タイトル取得（フォールバック機能付き）
        
        Args:
            soup: BeautifulSoupオブジェクト
            
        Returns:
            ページタイトル
        """
        # まず<title>タグを探す
        title_tag = soup.find('title')
        if title_tag:
            return title_tag.get_text(strip=True)  # strip=Trueで前後の空白を除去
        
        # <title>がない場合は<h1>タグをフォールバック
        h1_tag = soup.find('h1')
        return h1_tag.get_text(strip=True) if h1_tag else None
    
    def get_headings(self, soup: BeautifulSoup) -> List[Dict]:
        """
        見出し取得（階層情報付き）
        
        Args:
            soup: BeautifulSoupオブジェクト
            
        Returns:
            見出し情報のリスト
        """
        headings = []
        
        # h1からh6まですべての見出しを取得
        for level in range(1, 7):  # h1-h6
            # find_all()で該当するすべての要素を取得
            for heading in soup.find_all(f'h{level}'):
                headings.append({
                    'level': level,                         # 見出しレベル
                    'text': heading.get_text(strip=True),   # テキスト内容
                    'id': heading.get('id'),                # id属性
                    'class': heading.get('class')           # class属性（リスト形式）
                })
        
        return headings
    
    def get_links(self, soup: BeautifulSoup) -> List[Dict]:
        """
        リンク取得（属性情報も含む）
        
        Args:
            soup: BeautifulSoupオブジェクト
            
        Returns:
            リンク情報のリスト
        """
        links = []
        
        # href属性を持つ<a>タグをすべて取得
        # href=Trueで「href属性が存在する」条件を指定
        for link in soup.find_all('a', href=True):
            links.append({
                'url': link['href'],                     # リンクURL
                'text': link.get_text(strip=True),       # リンクテキスト
                'title': link.get('title'),              # title属性
                'target': link.get('target')             # target属性（_blank等）
            })
        
        return links
    
    def get_images(self, soup: BeautifulSoup) -> List[Dict]:
        """
        画像取得（属性情報も含む）
        
        Args:
            soup: BeautifulSoupオブジェクト
            
        Returns:
            画像情報のリスト
        """
        images = []
        
        # src属性を持つ<img>タグをすべて取得
        for img in soup.find_all('img', src=True):
            images.append({
                'src': img['src'],              # 画像URL
                'alt': img.get('alt'),          # alt属性（代替テキスト）
                'title': img.get('title'),      # title属性
                'width': img.get('width'),      # width属性
                'height': img.get('height')     # height属性
            })
        
        return images
    
    def get_meta_data(self, soup: BeautifulSoup) -> Dict:
        """
        メタデータ取得（SEO情報等）
        
        Args:
            soup: BeautifulSoupオブジェクト
            
        Returns:
            メタデータの辞書
        """
        meta_data = {}
        
        # すべての<meta>タグを処理
        for meta in soup.find_all('meta'):
            # name属性またはproperty属性を取得
            # name: 一般的なメタタグ（description, keywords等）
            # property: Open Graphメタタグ（og:title等）
            name = meta.get('name') or meta.get('property')
            content = meta.get('content')
            
            # 両方の属性が存在する場合のみ辞書に追加
            if name and content:
                meta_data[name] = content
        
        return meta_data
    
    def extract_with_regex(self, text: str, pattern_name: str) -> List[str]:
        """
        正規表現による抽出
        
        Args:
            text: 検索対象のテキスト
            pattern_name: パターン名（self.extraction_patternsのキー）
            
        Returns:
            マッチした文字列のリスト
        """
        import re
        
        # 事前定義されたパターンが存在するかチェック
        if pattern_name in self.extraction_patterns:
            pattern = self.extraction_patterns[pattern_name]
            # re.findall()で全てのマッチを取得
            return re.findall(pattern, text)
        
        # パターンが存在しない場合は空のリストを返す
        return []

# 使用例の詳細解説
async def demo_data_extraction():
    """
    データ抽出のデモ
    実際のHTMLからの情報抽出を体験
    """
    
    # サンプルHTMLの作成
    # 実際のWebページの構造を模倣
    sample_html = """
    <html>
        <head>
            <!-- ページのメタ情報 -->
            <title>サンプルページ</title>
            <meta name="description" content="これはテストページです">
            <meta property="og:title" content="OGタイトル">
        </head>
        <body>
            <!-- 見出し階層 -->
            <h1>メインタイトル</h1>
            <h2>サブタイトル</h2>
            
            <!-- コンテンツ（抽出対象の情報を含む） -->
            <p>お問い合わせ: contact@example.com</p>
            <p>電話: 03-1234-5678</p>
            
            <!-- リンク -->
            <a href="https://example.com" title="外部リンク">リンク</a>
            
            <!-- 画像 -->
            <img src="image.jpg" alt="サンプル画像">
        </body>
    </html>
    """
    
    # データ抽出器のインスタンス作成
    extractor = BasicDataExtractor()
    
    # 基本情報の抽出実行
    basic_info = extractor.extract_basic_info(sample_html)
    
    # 結果の表示
    print("=== 抽出結果 ===")
    print(f"タイトル: {basic_info['title']}")
    print(f"見出し数: {len(basic_info['headings'])}")
    print(f"リンク数: {len(basic_info['links'])}")
    print(f"画像数: {len(basic_info['images'])}")
    print(f"メタデータ: {basic_info['meta_data']}")
    
    # 正規表現による詳細抽出
    all_text = sample_html
    emails = extractor.extract_with_regex(all_text, 'email')
    phones = extractor.extract_with_regex(all_text, 'phone')
    
    print(f"\n=== 正規表現抽出結果 ===")
    print(f"メールアドレス: {emails}")
    print(f"電話番号: {phones}")
    
    return basic_info

# 実行例
# extraction_result = await demo_data_extraction()

class ResponsibleScraper(ModernWebScraper):
    """
    責任あるスクレイピングクラス
    
    継承を使用してModernWebScraperを拡張
    マナーを守ったスクレイピングを実現
    """
    
    def __init__(self, config: ScrapingConfig = None):
        """
        初期化メソッド
        
        親クラスの初期化に加えて、
        リクエスト管理用の属性を追加
        """
        super().__init__(config)  # 親クラスの初期化を実行
        
        self.request_history = []        # リクエスト履歴を記録
        self.min_delay = 1.0            # 最小遅延時間（秒）
        self.last_request_time = {}     # ドメイン別の最終リクエスト時刻
    
    async def polite_scrape(self, url: str, custom_delay: float = None) -> Dict:
        """
        マナーを守ったスクレイピング
        
        ドメインごとに適切な間隔を空けてリクエストを送信
        サーバー負荷軽減とアクセス制限回避を両立
        
        Args:
            url: スクレイピング対象のURL
            custom_delay: カスタム遅延時間（指定がなければデフォルト使用）
            
        Returns:
            スクレイピング結果の辞書
        """
        from urllib.parse import urlparse
        
        # URLからドメイン名を抽出
        # 例: "https://example.com/page" → "example.com"
        domain = urlparse(url).netloc
        current_time = time.time()
        
        # ドメインごとの遅延管理
        if domain in self.last_request_time:
            # 前回のリクエストからの経過時間を計算
            elapsed = current_time - self.last_request_time[domain]
            
            # 使用する遅延時間を決定
            delay = custom_delay or self.min_delay
            
            # まだ十分な時間が経過していない場合は待機
            if elapsed < delay:
                wait_time = delay - elapsed
                logger.info(f"Waiting {wait_time:.1f}s for {domain}")
                
                # 非ブロッキング待機（他の処理を妨げない）
                await asyncio.sleep(wait_time)
        
        # 実際のリクエスト実行
        result = await self.scrape_page(url)
        
        # 履歴の更新
        self.last_request_time[domain] = time.time()  # 最終リクエスト時刻を更新
        self.request_history.append({
            'url': url,
            'timestamp': datetime.now(),
            'status': result['status']
        })
        
        return result
    
    def get_request_stats(self) -> Dict:
        """
        リクエスト統計取得
        
        スクレイピングの実行状況を分析
        成功率やドメイン分布を把握
        
        Returns:
            統計情報の辞書
        """
        if not self.request_history:
            return {}
        
        total_requests = len(self.request_history)
        
        # 成功・失敗の集計
        successful = sum(1 for req in self.request_history if req['status'] == 'success')
        failed = total_requests - successful
        
        # ドメイン別統計の計算
        domain_counts = {}
        for req in self.request_history:
            from urllib.parse import urlparse
            domain = urlparse(req['url']).netloc
            
            # 辞書のget()メソッドで安全にカウント
            domain_counts[domain] = domain_counts.get(domain, 0) + 1
        
        return {
            'total_requests': total_requests,
            'successful': successful,
            'failed': failed,
            'success_rate': successful / total_requests * 100,  # 成功率（%）
            'domain_distribution': domain_counts
        }

# 使用例の詳細解説
async def demo_responsible_scraping():
    """
    責任あるスクレイピングのデモ
    
    適切な間隔を空けながら複数のページを処理
    統計情報の取得方法も示す
    """
    
    # 設定作成（ヘッドレスモードで高速実行）
    config = ScrapingConfig(headless=True)
    
    # 責任あるスクレイパーを使用
    async with ResponsibleScraper(config) as scraper:
        # テスト用URL配列
        test_urls = [
            "https://httpbin.org/delay/1",  # 1秒の遅延付きレスポンス
            "https://httpbin.org/html",     # HTML形式のレスポンス
            "https://httpbin.org/json"      # JSON形式のレスポンス
        ]
        
        results = []
        
        # 各URLを適切な間隔で処理
        for url in test_urls:
            print(f"Polite scraping: {url}")
            
            # カスタム遅延（2秒）を指定してスクレイピング
            result = await scraper.polite_scrape(url, custom_delay=2.0)
            results.append(result)
        
        # 統計情報の取得と表示
        stats = scraper.get_request_stats()
        print(f"\n=== リクエスト統計 ===")
        print(f"総リクエスト数: {stats['total_requests']}")
        print(f"成功率: {stats['success_rate']:.1f}%")
        print(f"ドメイン分布: {stats['domain_distribution']}")
        
        return results, stats

# 実行例
# responsible_results = await demo_responsible_scraping()