【実践】Pythonでスクレイピング｜playwrightで攻略

# ターミナルまたはコマンドプロンプトで実行
pip install playwright
# Chromium, Firefox, WebKitの3つまとめてインストール
playwright install

# Chromiumだけをインストール
playwright install chromium

# Firefoxだけをインストール
playwright install firefox

# WebKit (Safariのエンジン) だけをインストール
playwright install webkit

# Directions for robots.  See this URL:
# http://www.robotstxt.org/robotstxt.html
# for a description of the file format.

User-agent: HTTrack
User-agent: puf
User-agent: MSIECrawler
Disallow: /

# The Krugle web crawler (though based on Nutch) is OK.
User-agent: Krugle
Allow: /
Disallow: /~guido/orlijn/
Disallow: /webstats/

# No one should be crawling us with Nutch.
User-agent: Nutch
Disallow: /

# Hide old versions of the documentation and various large sets of files.
User-agent: *
Disallow: /~guido/orlijn/
Disallow: /webstats/

# ==============================================================================
# セクション1: 環境構築
# ==============================================================================
import logging
import os
import sys

for handler in logging.root.handlers[:]:
    logging.root.removeHandler(handler)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

IS_COLAB = "google.colab" in sys.modules

try:
    if IS_COLAB:
        logging.info(">>> Google Colab環境を検出しました。Playwrightの環境構築を開始します。")
        get_ipython().system('pip install playwright pandas beautifulsoup4 lxml -q')
        get_ipython().system('playwright install chromium')
        logging.info(">>> 環境構築が正常に完了しました。")
    else:
        logging.info(">>> ローカル環境とみなし、環境構築をスキップします。")
except Exception as e:
    logging.error(f"--- 環境構築中にエラーが発生しました: {e} ---")
    raise SystemExit("環境構築に失敗したため、処理を中断します。")


# ==============================================================================
# セクション2: ライブラリのインポート
# ==============================================================================
import time
import asyncio
import pandas as pd
from bs4 import BeautifulSoup
from datetime import datetime
from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeoutError

if IS_COLAB:
    from google.colab import files


# ==============================================================================
# セクション3: メイン処理
# ==============================================================================

async def main():
    async with async_playwright() as p:
        browser = None
        page = None
        try:
            browser = await p.chromium.launch(headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"])
            context = await browser.new_context(
                user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.102 Safari/537.36',
                viewport={'width': 1920, 'height': 1080}
            )
            page = await context.new_page()
            logging.info("Playwrightのセットアップが完了しました。")

            await page.goto("https://www.python.org/", timeout=60000)
            logging.info(f"'{await page.title()}' にアクセスしました。")

            await page.locator("#news > a").click(force=True)
            await page.wait_for_url("**/blogs/")
            logging.info(f"ニュース一覧ページ ({page.url}) に遷移しました。")

            await page.locator('xpath=//*[@id="content"]/div/section/div/div[1]/div/p/a').click()
            await page.wait_for_url(lambda url: "blog.python.org" in url or "pythoninsider.blogspot.com" in url)
            logging.info(f"ブログサイト ({page.url}) に正常に遷移しました。")

            is_mobile_site = 'pythoninsider.blogspot.com' in page.url
            if is_mobile_site:
                logging.info(">>> モバイル版ブログサイトのスクレイピングを開始します。")
            else:
                logging.info(">>> PC版ブログサイトのスクレイピングを開始します。")

            news_data = []
            MAX_PAGES = 10

            for i in range(MAX_PAGES):
                logging.info(f"--- ページ {i + 1} のデータを取得します ---")
                await page.wait_for_selector(".blog-posts")
                html = await page.content()
                soup = BeautifulSoup(html, 'lxml')

                page_articles_found = 0

                def format_date(date_str):
                    try:
                        date_obj = datetime.strptime(date_str, "%A, %B %d, %Y")
                        return date_obj.strftime("%Y-%m-%d")
                    except ValueError:
                        return date_str

                post_containers = soup.select(".blog-posts > div.date-outer")
                for container in post_containers:
                    date_tag = container.select_one("h2.date-header span")
                    date_str = date_tag.get_text(strip=True) if date_tag else "日付不明"
                    formatted_date = format_date(date_str)
                    articles = container.select("div.post-outer")
                    for article in articles:
                        title_tag = article.select_one("h3.post-title a")
                        if title_tag:
                            title = title_tag.get_text(strip=True)
                            if title and not any(d['タイトル'] == title for d in news_data):
                                # logging.info(f"  >> 発見: [投稿日: {formatted_date}] [タイトル: {title}]")
                                news_data.append({'投稿日': formatted_date, 'タイトル': title})
                                page_articles_found += 1
                
                logging.info(f"このページから新たに {page_articles_found} 件の記事を取得しました。")
                if page_articles_found == 0 and i > 0:
                    logging.info("新たな記事がなかったため、最終ページと判断し、処理を終了します。")
                    break

                if i < MAX_PAGES - 1:
                    try:
                        await page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
                        await page.wait_for_timeout(1000)

                        next_button_locator_str = "#blog-pager-older-link" if is_mobile_site else "#Blog1_blog-pager-older-link"
                        next_button = page.locator(next_button_locator_str)

                        if await next_button.count() > 0:
                            await next_button.click()
                            logging.info(f"次のページ({i + 2}ページ目)へ遷移します...")
                            await page.wait_for_load_state('domcontentloaded')
                        else:
                            logging.info("「Older Posts」ボタンがDOM内に見つかりませんでした。ここで取得を終了します。")
                            break
                    except PlaywrightTimeoutError:
                        logging.info("「Older Posts」ボタンの操作でタイムアウトしました。ここで取得を終了します。")
                        break
            
            logging.info(f"合計 {len(news_data)} 件のニュースを取得しました。")
            if news_data:
                df = pd.DataFrame(news_data, columns=['投稿日', 'タイトル'])
                csv_filename = 'python_blog_news_playwright.csv'
                df.to_csv(csv_filename, index=False, encoding='utf-8-sig')
                logging.info(f"取得したデータを'{csv_filename}'に保存しました。")
                if IS_COLAB:
                    files.download(csv_filename)
            else:
                logging.warning("取得できたニュースがなかったため、CSVファイルは作成されませんでした。")

        except Exception as e:
            logging.error(f"処理中に予期せぬエラーが発生しました: {e}", exc_info=True)
            if page:
                error_filename = 'error_screenshot_playwright.png'
                await page.screenshot(path=error_filename)
                if IS_COLAB:
                    files.download(error_filename)
        finally:
            logging.info("処理が終了しました。ブラウザは自動的に閉じられます。")

# asyncioを使ってmain関数を実行
import nest_asyncio
nest_asyncio.apply()
asyncio.run(main())

# ==============================================================================
# セクション1: 環境構築
# ==============================================================================
import logging
import os
import sys

# 既存のログハンドラをクリア
for handler in logging.root.handlers[:]:
    logging.root.removeHandler(handler)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

IS_COLAB = "google.colab" in sys.modules

try:
    if IS_COLAB:
        logging.info(">>> Google Colab環境を検出しました。Playwrightの環境構築を開始します。")
        get_ipython().system('apt-get update -qq && apt-get install -y fonts-ipafont-gothic -qq')
        get_ipython().system('pip install playwright pandas -q')
        get_ipython().system('playwright install chromium')
        logging.info(">>> 環境構築が正常に完了しました。")
    else:
        logging.info(">>> ローカル環境とみなし、環境構築をスキップします。")
except Exception as e:
    logging.error(f"--- 環境構築中にエラーが発生しました: {e} ---")
    raise SystemExit("環境構築に失敗したため、処理を中断します。")


# ==============================================================================
# セクション2: ライブラリのインポート
# ==============================================================================
import asyncio
import pandas as pd
from playwright.async_api import async_playwright
from urllib.parse import quote_plus

if IS_COLAB:
    from google.colab import files

# ★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★
# ★ 設定項目 ★
SEARCH_KEYWORD = "ここに検索したいキーワード"
MAX_PAGES_TO_SCRAPE = 5 # 取得したいページ数
# ★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★★


# ==============================================================================
# セクション3: メイン処理
# ==============================================================================

async def main():
    async with async_playwright() as p:
        browser = None
        page = None
        try:
            browser_args = ["--no-sandbox", "--disable-dev-shm-usage", "--disable-blink-features=AutomationControlled"]
            browser = await p.chromium.launch(headless=True, args=browser_args)
            context = await browser.new_context(
                user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36',
                viewport={'width': 1920, 'height': 1080},
                locale='ja-JP'
            )
            page = await context.new_page()
            logging.info("Playwrightのセットアップが完了しました。")

            encoded_keyword = quote_plus(SEARCH_KEYWORD)
            start_url = f"https://search.yahoo.co.jp/search?p={encoded_keyword}"
            logging.info(f"検索結果ページに直接アクセスします: {start_url}")
            await page.goto(start_url, wait_until="domcontentloaded", timeout=60000)

            logging.info("ポップアップや同意画面が表示されていないか確認し、処理します...")
            try:
                later_button = page.locator('button:has-text("後で")')
                if await later_button.is_visible(timeout=5000):
                    await later_button.click()
                    logging.info("「アドレスバーを設定」ポップアップの「後で」ボタンをクリックしました。")
                    await later_button.wait_for(state="hidden", timeout=5000)
                    logging.info("「アドレスバーを設定」ポップアップが閉じられたことを確認しました。")

                agree_button = page.locator('button:has-text("同意して閉じる")')
                if await agree_button.is_visible(timeout=3000):
                    await agree_button.click()
                    logging.info("クッキー同意ボタンをクリックしました。")
                    await agree_button.wait_for(state="hidden", timeout=5000)
                    logging.info("クッキー同意画面が閉じられたことを確認しました。")

            except Exception as e:
                logging.warning(f"ポップアップ処理中に軽微なエラー（またはポップアップ無）が発生しましたが、処理を続行します: {e}")

            all_search_results = []
            
            for page_num in range(1, MAX_PAGES_TO_SCRAPE + 1):
                logging.info(f"--- {page_num} ページ目のデータを取得します ---")

                await page.wait_for_selector("#contents", timeout=30000)
                
                #  page.title()が不安定な場合があるため、エラーが発生しても処理を続行する
                try:
                    page_title = await page.title()
                    logging.info(f"検索結果ページ '{page_title}' の読み込み完了。")
                except Exception as e:
                    logging.warning(f"ページのタイトル取得に失敗しましたが、処理を続行します: {e}")
                
                await page.wait_for_timeout(1000) # 動的コンテンツの読み込みを少し待つ

                result_items = await page.locator("div.sw-Card.Algo").all()

                # このページに検索結果がない場合はループを抜ける
                if not result_items:
                    logging.info("このページに検索結果が見つかりませんでした。取得を終了します。")
                    break

                logging.info(f"ページ内の検索結果(div.sw-Card.Algo)を{len(result_items)}件見つけました。")

                rank_offset = len(all_search_results)
                for i, item in enumerate(result_items, 1):
                    rank = rank_offset + i
                    link_loc = item.locator("a:has(h3)").first
                    title_loc = link_loc.locator("h3")
                    desc_loc = item.locator("p.sw-Card__summary")
                    
                    if await title_loc.count() == 0:
                        continue

                    title = await title_loc.inner_text()
                    url = await link_loc.get_attribute("href")
                    description = await desc_loc.inner_text() if await desc_loc.count() > 0 else ""
                    
                    logging.info(f"  [順位 {rank}] {title}")
                    all_search_results.append({"順位": rank, "タイトル": title, "URL": url, "ディスクリプション": description})

                logging.info(f"--- {page_num} ページ目から {len(result_items)} 件の記事を取得しました ---")

                # 次のページへ遷移するか、ループを終了するかを判断
                if page_num < MAX_PAGES_TO_SCRAPE:
                    next_button = page.locator('.Pagenation__next a:has-text("次へ")')
                    if await next_button.count() > 0:
                        logging.info("「次へ」ボタンをクリックして、次のページへ遷移します...")
                        await next_button.click()
                        await page.wait_for_load_state("domcontentloaded", timeout=60000)
                    else:
                        # 「次へ」ボタンがない場合は最終ページなのでループを抜ける
                        logging.info("「次へ」ボタンが見つかりませんでした。最終ページと判断し、取得を終了します。")
                        break
                else:
                    logging.info(f"指定された {MAX_PAGES_TO_SCRAPE} ページの取得が完了したため、ループを終了します。")
            
            logging.info(f"--- 合計 {len(all_search_results)} 件の検索結果を取得しました ---")
            
            related_keywords = []
            paa_questions = []
            try:
                logging.info("--- 関連キーワードとPAAの取得を試みます ---")
                # 関連検索ワードのセレクタ
                related_keywords_loc = page.locator('div.Unit--south li.SouthUnitItem a') 
                paa_loc = page.locator('div.AnswerRelatedQuestions .sw-Accordion__title')
                
                if await related_keywords_loc.count() > 0:
                    related_keywords = [await loc.inner_text() for loc in await related_keywords_loc.all()]
                    logging.info(f"関連キーワード: {related_keywords}")
                else:
                    logging.warning("関連キーワードのセクションが見つかりませんでした。")
                
                if await paa_loc.count() > 0:
                    paa_questions = await paa_loc.all_inner_texts()
                    logging.info(f"PAA: {paa_questions}")
                else:
                    logging.warning("PAAのセクションが見つかりませんでした。")
            except Exception as e:
                logging.warning(f"関連情報（キーワード/PAA）の取得中に軽微なエラーが発生しました: {e}")

            if all_search_results:
                df_results = pd.DataFrame(all_search_results)
                filename_results = f"yahoo_search_results_{SEARCH_KEYWORD.replace(' ', '_')}.csv"
                df_results.to_csv(filename_results, index=False, encoding='utf-8-sig')
                logging.info(f"検索結果を'{filename_results}'に保存しました。")
                if IS_COLAB: files.download(filename_results)
            
            if related_keywords:
                df_related = pd.DataFrame(related_keywords, columns=["関連キーワード"])
                filename_related = f"yahoo_related_keywords_{SEARCH_KEYWORD.replace(' ', '_')}.csv"
                df_related.to_csv(filename_related, index=False, encoding='utf-8-sig')
                logging.info(f"関連キーワードを'{filename_related}'に保存しました。")
                if IS_COLAB: files.download(filename_related)

            if paa_questions:
                df_paa = pd.DataFrame(paa_questions, columns=["他の人はこちらも質問"])
                filename_paa = f"yahoo_paa_{SEARCH_KEYWORD.replace(' ', '_')}.csv"
                df_paa.to_csv(filename_paa, index=False, encoding='utf-8-sig')
                logging.info(f"PAAを'{filename_paa}'に保存しました。")
                if IS_COLAB: files.download(filename_paa)

        except Exception as e:
            logging.error(f"処理中に予期せぬエラーが発生しました: {e}", exc_info=True)
            if page:
                error_filename = f"error_screenshot_{SEARCH_KEYWORD.replace(' ', '_')}.png"
                await page.screenshot(path=error_filename, full_page=True)
                logging.info(f"エラー発生時のスクリーンショットを '{error_filename}' として保存しました。")
                if IS_COLAB:
                    try:
                        files.download(error_filename)
                        html_filename = f"error_page_{SEARCH_KEYWORD.replace(' ', '_')}.html"
                        with open(html_filename, 'w', encoding='utf-8') as f:
                            f.write(await page.content())
                        logging.info(f"エラー発生時のHTMLを '{html_filename}' として保存しました。")
                        files.download(html_filename)
                    except Exception as download_error:
                        logging.error(f"エラーファイルのダウンロード中に問題が発生しました: {download_error}")
        finally:
            if browser: await browser.close()
            logging.info("処理が終了しました。")

# asyncioを使ってmain関数を実行
import nest_asyncio
nest_asyncio.apply()
asyncio.run(main())