AI駆動開発１（google検索アプリ）

ブラウザ活用MCPサーバー作成用に、内部で動くアプリを作ってます。

目的

• 最終的にＡＩエージェントやDifyから活用します

今回のアプリ

• google検索 => 検索結果すべてにアクセスしてbodyを取得するアプリ
• プロンプトとアプリのコードをこちらでシェアします
• 以下のプロンプトでほぼ完成しましたが、少し改造して最後のソースコードをシェアしてます。

特徴

• playwrightでgoogle検索をしても、ロボットチェックに引っかからない仕組みを導入しました。（どのようなプロンプトでできるかは以下を参照ください。）
• 検索結果のurlにアクセスする際は、同時に最大５ページまでアクセスします。（見てると楽しいですよ。）（同時実行するために、非同期で作成）

プロンプト

ＡＩへの指示１

Python playwright でアプリを作ります。 質問があれば、事前に聞いてください

非同期モード作成します。
google検索するので、ユーザーエージェントなどの偽装により、ロボットチェックを回避します。
ステルスモードを活用して、ロボットチェックを回避します。

googleで検索をして、検索結果が並んだら、全てのURLに最大５並列でアクセスして、アクセス先のデータを人間が見れる形で取得します。
全てのURLアクセスについては、networkidleで待ってください。

全てのURLにアクセスするには、以下のセレクターを少し改造します。
改造方法は、以下のセレクターから、:nth-child(5)　と、　> を削除します。

改造前
#rso > div:nth-child(5) > div > div > div.kb0PBd.A9Y9g.jGGQ5e > div > div > span > a

改造前
#rso div div div div.kb0PBd.A9Y9g.jGGQ5e div div span a

ＡＩからの回答１

はい、承知いたしました！Python Playwright を使った非同期モードのGoogle検索結果スクレイピングアプリ開発ですね。 ステルスモードやユーザーエージェント偽装によるロボットチェック回避、検索結果URLへの並列アクセスとコンテンツ取得、そして `networkidle` での待機、セレクタの調整、これらの要件を理解しました。

開発を進めるにあたり、事前にいくつか質問させてください。これらの点が明確になることで、よりご要望に沿ったアプリを作成できるかと思います。

質問事項:

1. Google検索の具体的な入力:

   • 検索キーワードはどのようにアプリに渡されますか？ (コマンドライン引数、設定ファイル、関数への引数など)
   • 検索対象のGoogleドメインは指定がありますか？ (例: google.com, google.co.jp など。デフォルトは google.com で良いでしょうか？)

2. 「検索結果が並んだら」の判断基準:

   • 検索結果ページが読み込まれたことを判断する具体的なセレクタや条件はありますか？ (例: #rso が表示されたら、特定の数の検索結果アイテムが表示されたらなど)
   • あるいは、単純に page.wait_for_load_state("domcontentloaded") や page.wait_for_load_state("networkidle") で判断しますか？

3. 検索結果URLの取得セレクタについて:

   • 提示いただいたセレクタは、特定の構造の検索結果（オーガニック検索など）を指していると思われます。広告や特殊な検索結果（ナレッジパネル、動画など）もURL取得の対象としますか？
   • 「改造後」として提示いただいたセレクタ #rso div div div div.kb0PBd.A9Y9g.jGGQ5e div div span a について、これはスペース区切りなのでCSSの子孫セレクタとして解釈されます。このセレクタでGoogle検索結果の全ての主要なリンクが取得できるという認識でよろしいでしょうか？（GoogleのHTML構造は頻繁に変わるため、このセレクタの堅牢性が気になります。）
   • 取得するURLの最大数はありますか？ (例: 最初の10件まで、1ページ目全てなど)

4. 並列アクセスについて:

   • 「最大5並列」とは、同時に最大5つのタブ（またはコンテキスト）を開いて各URLにアクセスするという理解でよろしいでしょうか？
   • Playwrightの非同期処理 (asyncio と async/await) を活用してこれを実装します。

5. 「人間が見れる形」のデータ取得:

   • 各アクセス先ページから具体的にどのような情報を取得したいですか？
     • ページ全体のテキスト (page.inner_text('body') など)
     • 特定の要素内のテキスト (例:記事本文 <article>...</article>)
     • ページのタイトル (page.title())
     • HTML全体 (page.content())
   • 取得したデータの出力形式はどのようにしますか？ (標準出力に表示、テキストファイルに保存、JSONファイルに保存など)

6. エラーハンドリング:

   • Google検索自体に失敗した場合（ネットワークエラー、ロボットチェックによるブロックなど）。
   • 検索結果からURLが取得できなかった場合。
   • 個別のURLへのアクセスに失敗した場合（タイムアウト、404エラーなど）。
   • これらの場合に、アプリはどのように振る舞うべきですか？ (エラーをログに出力して続行、処理を中断など)

7. ステルスモードの具体的な内容:

   • Playwrightには直接的な「ステルスモード」という機能はありませんが、一般的には以下のような手法を組み合わせて検出を回避します。
     • 適切なユーザーエージェントの設定 (これは要件にありますね)。
     • navigator.webdriver = false のようなJavaScriptプロパティの上書き。
     • その他、playwright-stealth のような外部ライブラリの利用も考えられますが、まずはPlaywright標準機能でどこまでできるか試しますか？それとも特定のライブラリ利用のご希望がありますか？
     • (参考: playwright-extra とそのプラグイン playwright-extra-plugin-stealth などがあります)

8. 非同期処理のタイムアウト:

   • Google検索ページへの遷移タイムアウト、検索結果URLへのアクセスごとのタイムアウト（networkidle 待機を含む）は、どの程度の値を想定していますか？

9. 依存ライブラリ:

   • playwright 以外に、導入を検討すべき特定のライブラリはありますか？ (例: 上述のステルス用ライブラリなど)

これらの点について、可能な範囲でお答えいただけると幸いです。
特に、検索結果URLを取得するセレクタの正確性と、アクセス先のページから具体的に何を取得したいかは、アプリの核となる部分ですので、詳細をいただけると助かります。

お手数ですが、よろしくお願いいたします。

---

ＡＩへの指示２

Google検索の具体的な入力: 検索キーワードはどのようにアプリに渡されますか？ (コマンドライン引数、設定ファイル、関数への引数など) ＝＞　コマンドライン引数です

検索対象のGoogleドメインは指定がありますか？ (例: google.com, google.co.jp など。デフォルトは google.com で良いでしょうか？)
＝＞　google.co.jp

「検索結果が並んだら」の判断基準:
検索結果ページが読み込まれたことを判断する具体的なセレクタや条件はありますか？ (例: #rso が表示されたら、特定の数の検索結果アイテムが表示されたらなど)
あるいは、単純に page.wait_for_load_state("domcontentloaded") や page.wait_for_load_state("networkidle") で判断しますか？
＝＞　page.wait_for_load_state("networkidle")　でお願い

検索結果URLの取得セレクタについて:
提示いただいたセレクタは、特定の構造の検索結果（オーガニック検索など）を指していると思われます。広告や特殊な検索結果（ナレッジパネル、動画など）もURL取得の対象としますか？
＝＞　しません

「改造後」として提示いただいたセレクタ #rso div div div div.kb0PBd.A9Y9g.jGGQ5e div div span a について、これはスペース区切りなのでCSSの子孫セレクタとして解釈されます。このセレクタでGoogle検索結果の全ての主要なリンクが取得できるという認識でよろしいでしょうか？（GoogleのHTML構造は頻繁に変わるため、このセレクタの堅牢性が気になります。）
＝＞　変わったらエラーになるのでその際に対応します。　永遠に変更しなくて良いセレクターがあれば教えて

取得するURLの最大数はありますか？ (例: 最初の10件まで、1ページ目全てなど)
＝＞　あるだけ全部です。多分１ページ目全部になると思うが、２ページ目とか可能なら、どんな感じが知りたい

並列アクセスについて:
「最大5並列」とは、同時に最大5つのタブ（またはコンテキスト）を開いて各URLにアクセスするという理解でよろしいでしょうか？
Playwrightの非同期処理 (asyncio と async/await) を活用してこれを実装します。
＝＞　その認識です

「人間が見れる形」のデータ取得:
各アクセス先ページから具体的にどのような情報を取得したいですか？
ページ全体のテキスト (page.inner_text('body') など)
特定の要素内のテキスト (例:記事本文 ... )
ページのタイトル (page.title())
HTML全体 (page.content())
＝＞　page.inner_text('body')　で、改行なども取れますよね？

取得したデータの出力形式はどのようにしますか？ (標準出力に表示、テキストファイルに保存、JSONファイルに保存など)
＝＞　標準出力とJSONにお願いします。
JSONでは、何件目、URL、内容が分かるといいです。（これをAIに投げるかもしれないので、AIに理解しやすい形にして）

エラーハンドリング:
Google検索自体に失敗した場合（ネットワークエラー、ロボットチェックによるブロックなど）。
検索結果からURLが取得できなかった場合。
個別のURLへのアクセスに失敗した場合（タイムアウト、404エラーなど）。
これらの場合に、アプリはどのように振る舞うべきですか？ (エラーをログに出力して続行、処理を中断など)
＝＞　処理中断でよいです

ステルスモードの具体的な内容:
Playwrightには直接的な「ステルスモード」という機能はありませんが、一般的には以下のような手法を組み合わせて検出を回避します。
適切なユーザーエージェントの設定 (これは要件にありますね)。
navigator.webdriver = false のようなJavaScriptプロパティの上書き。
その他、playwright-stealth のような外部ライブラリの利用も考えられますが、まずはPlaywright標準機能でどこまでできるか試しますか？それとも特定のライブラリ利用のご希望がありますか？
(参考: playwright-extra とそのプラグイン playwright-extra-plugin-stealth などがあります)
＝＞　stealthとつく、playwrightのライブラリを活用したい

非同期処理のタイムアウト:
Google検索ページへの遷移タイムアウト、検索結果URLへのアクセスごとのタイムアウト（networkidle 待機を含む）は、どの程度の値を想定していますか？
＝＞　30秒

依存ライブラリ:
playwright 以外に、導入を検討すべき特定のライブラリはありますか？ (例: 上述のステルス用ライブラリなど)
＝＞　stealthとつく、playwrightのライブラリを活用したい

ＡＩからの回答２

詳細なご回答ありがとうございます！非常に明確になりました。 これらの情報に基づいて、アプリの設計と実装を進めていきます。

整理された要件:

1. 入力:
   • コマンドライン引数で検索キーワードを受け取る。
   • 検索対象ドメイン: google.co.jp
2. Google検索ページ:
   • 読み込み完了判断: page.wait_for_load_state("networkidle")
3. 検索結果URL取得:
   • 対象: オーガニック検索結果のみ（広告、ナレッジパネル、動画などは除く）。
   • セレクタ: #rso div div div div.kb0PBd.A9Y9g.jGGQ5e div div span a を使用。Googleの構造変更時は都度対応。
     • 補足: より堅牢なセレクタとしては、特定の data- 属性や、より汎用的なクラス名（例: 検索結果の各アイテムを囲むコンテナ要素のクラス名を探し、その中の a タグ）を利用することが考えられますが、まずはご指定のセレクタで実装します。
   • 取得数: 取得できる限り全て。
     • 2ページ目以降の対応について:
       • Google検索結果は通常、下にスクロールすると次の結果が読み込まれるか、「次へ」のページネーションリンクがあります。
       • 「あるだけ全部」となると、これらのページ遷移処理を実装する必要があります。
       • まずは1ページ目の結果を取得するロジックを実装し、その上でページネーション対応をオプションとして検討する形が良いかと思います。今回はまず1ページ目のみを対象としますか？それともページネーション処理の試みも初期実装に含めますか？ （今回はまず1ページ目のみで進めます。ご要望があれば後で拡張しましょう。）
4. 並列アクセス:
   • 最大5並列で、取得した各URLに非同期アクセス。
5. アクセス先ページのデータ取得:
   • page.inner_text('body') でページ全体のテキストを取得（改行も含む）。
   • 各ページ読み込み完了判断: networkidle で待機。
6. 出力:
   • 標準出力: 取得したテキスト（整形して表示）。
   • JSONファイル:

     [
       {
         "index": 1, // 何件目か
         "url": "アクセスしたURL",
         "title": "ページのタイトル (取得できれば)", // 追加提案
         "content": "取得したページ全体のテキスト (inner_text('body'))"
       },
       // ... 他のURLの結果
     ]
     

     （AIに理解しやすいように、とのことなので、各エントリが検索結果1件に対応するようにします。タイトルも追加すると文脈が分かりやすくなるかもしれません。）
7. エラーハンドリング:
   • いずれかの段階でエラーが発生した場合（Google検索失敗、URL取得失敗、個別URLアクセス失敗）、処理を中断し、エラーメッセージを出力。
8. ステルス機能:
   • playwright-extra と playwright-extra-plugin-stealth ライブラリの利用を検討・実装。
9. タイムアウト:
   • Google検索ページ遷移、各検索結果URLへのアクセス（networkidle 待機含む）共に30秒。
10. 依存ライブラリ:
    • playwright
    • playwright-extra
    • playwright-extra-plugin-stealth
    • asyncio (Python標準ライブラリ)
    • argparse (Python標準ライブラリ)

1点だけ追加で確認させてください（上記3の補足）：

• 検索結果URLの取得数について: 「あるだけ全部」とのことですが、Googleの1ページに表示される検索結果の数は変動することがあります。今回は、ご指定のセレクタで取得できる1ページ目の全オーガニック検索結果のURLを対象とする、という理解で進めてよろしいでしょうか？ 2ページ目以降の対応は、別途ご相談させてください。

上記内容でアプリの実装を進めます。
もし、上記の認識で相違点や追加のご要望があればお知らせください。

特に問題なければ、これより具体的なコードの作成に入ります。

---

ＡＩへの指示３

検索結果URLの取得数について: 「あるだけ全部」とのことですが、Googleの1ページに表示される検索結果の数は変動することがあります。今回は、ご指定のセレクタで取得できる1ページ目の全オーガニック検索結果のURLを対象とする、という理解で進めてよろしいでしょうか？ 2ページ目以降の対応は、別途ご相談させてください。 ＝＞　はい！　よろしくお願いします

ＡＩからの回答３

はい、承知いたしました！

検索結果URLの取得数については、ご指定のセレクタで取得できる1ページ目の全オーガニック検索結果のURLを対象とする、という方針で実装を進めます。
2ページ目以降の対応は、今回の初期実装が完了した後、必要に応じて拡張機能として検討しましょう。

それでは、これらの要件に基づいて、Python Playwrightを使った非同期Google検索スクレイピングアプリのコードを作成します。
少しお時間をください。コードが完成次第、提示いたします。

---

ロジック

import asyncio
import argparse
import json
import logging
import os
from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeoutError, Error as PlaywrightError
from playwright_stealth import stealth_async

# --- ロギング設定 ---
log_file_name = "google_scraper.log"
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(module)s - %(funcName)s - %(message)s')

# コンソールハンドラー
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)

# ファイルハンドラー
try:
    file_handler = logging.FileHandler(log_file_name, mode='a', encoding='utf-8')
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)
    print(f"ログは {log_file_name} にも出力されます。")
except Exception as e:
    print(f"ログファイル '{log_file_name}' の設定に失敗しました: {e}")
    print("ログはコンソールにのみ出力されます。")


async def fetch_page_content(context, url: str, semaphore: asyncio.Semaphore, timeout_ms: int):
    """指定されたURLにアクセスし、ページのタイトルと本文テキストを取得する"""
    async with semaphore: # 同時実行数を制限
        page = None
        try:
            page = await context.new_page()
            # await stealth_async(page) # playwright-extra: 各ページにもstealthを適用 (コンテキスト単位で十分な場合もある)
            logger.info(f"Navigating to: {url}")
            await page.goto(url, timeout=timeout_ms, wait_until="networkidle")
            
            page_title = await page.title()
            body_text = await page.inner_text('body')
            
            logger.info(f"Successfully fetched content from: {url} (Title: {page_title})")
            return {"url": url, "title": page_title, "content": body_text}
        except PlaywrightTimeoutError:
            logger.error(f"Timeout error when accessing {url}")
            return {"url": url, "title": "Error: Timeout", "content": "Failed to load page due to timeout."}
        except PlaywrightError as e:
            logger.error(f"Playwright error when accessing {url}: {e}")
            return {"url": url, "title": f"Error: Playwright ({type(e).__name__})", "content": f"An error occurred: {str(e)}"}
        except Exception as e:
            logger.error(f"Unexpected error when accessing {url}: {e}")
            return {"url": url, "title": f"Error: Unexpected ({type(e).__name__})", "content": f"An unexpected error occurred: {str(e)}"}
        finally:
            if page:
                await page.close()

async def search_google_and_scrape_results(keyword: str, google_domain: str, results_selector: str, max_parallel_tasks: int, timeout_ms: int):
    """Googleで検索し、結果のURLに並列アクセスしてコンテンツを取得する"""
    async with async_playwright() as p:
        try:
            browser = await p.chromium.launch(headless=False) # ヘッドレスモードで起動
            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",
                java_script_enabled=True,
                # viewport={'width': 1920, 'height': 1080} # 必要に応じて
            )
            page = await context.new_page()
            await stealth_async(page) # playwright-extra: stealth機能を適用

            search_url = f"https://www.{google_domain}/search?q={keyword.replace(' ', '+')}&hl=ja" # 日本語指定
            logger.info(f"Navigating to Google search: {search_url}")
            await page.goto(search_url, timeout=timeout_ms, wait_until="networkidle")
            logger.info("Google search page loaded.")

            # 検索結果のURLを取得
            logger.info(f"Extracting result URLs using selector: {results_selector}")
            
            # セレクタに一致する全てのaタグのhref属性を取得
            link_elements = await page.locator(results_selector).all()
            urls_to_scrape = []
            for link_element in link_elements:
                href = await link_element.get_attribute("href")
                if href and href.startswith("http"): # 有効なURLのみ
                    urls_to_scrape.append(href)
            
            if not urls_to_scrape:
                logger.warning("No search result URLs found with the given selector.")
                await browser.close()
                return []

            logger.info(f"Found {len(urls_to_scrape)} URLs to scrape (1st page):")
            for i, url in enumerate(urls_to_scrape):
                 logger.info(f"  {i+1}. {url}")


            # 取得したURLに並列アクセス
            semaphore = asyncio.Semaphore(max_parallel_tasks)
            tasks = []
            for url in urls_to_scrape:
                # 新しいコンテキストで各ページを開くことで、Cookieやストレージの分離を図る
                # (ただし、パフォーマンスへの影響も考慮。同じコンテキストでタブを開く方が速い場合もある)
                # new_page_context = await browser.new_context(user_agent=...) # もし各ページでUAを変えたいなど
                # task = fetch_page_content(new_page_context, url, semaphore, timeout_ms)
                task = fetch_page_content(context, url, semaphore, timeout_ms) # 同じコンテキストで実行
                tasks.append(task)
            
            scraped_data = await asyncio.gather(*tasks)
            
            await browser.close()
            return scraped_data

        except PlaywrightTimeoutError:
            logger.critical(f"Timeout error during Google search or initial page load.")
            if 'browser' in locals() and browser.is_connected(): await browser.close()
            raise # エラーを再送出してmainでキャッチ
        except PlaywrightError as e:
            logger.critical(f"Playwright error during Google search: {e}")
            if 'browser' in locals() and browser.is_connected(): await browser.close()
            raise
        except Exception as e:
            logger.critical(f"Unexpected error during Google search: {e}")
            if 'browser' in locals() and browser.is_connected(): await browser.close()
            raise


async def main():
    parser = argparse.ArgumentParser(description="Google検索結果のURLにアクセスし、内容を取得します。")
    parser.add_argument("keyword", help="Googleで検索するキーワード")
    parser.add_argument("--domain", default="google.co.jp", help="検索対象のGoogleドメイン (例: google.com, google.co.jp)")
    parser.add_argument(
        "--selector", 
        default="#rso div div div div.kb0PBd.A9Y9g.jGGQ5e div div span a", 
        help="検索結果のURLを取得するためのCSSセレクタ"
    )
    parser.add_argument("--parallel", type=int, default=5, help="最大並列アクセス数")
    parser.add_argument("--timeout", type=int, default=30000, help="各操作のタイムアウト時間 (ミリ秒)")
    parser.add_argument("--output_json", default="scraped_results.json", help="結果を保存するJSONファイル名")

    args = parser.parse_args()

    try:
        logger.info(f"Starting script with keyword: '{args.keyword}'")
        results = await search_google_and_scrape_results(
            args.keyword,
            args.domain,
            args.selector,
            args.parallel,
            args.timeout
        )

        if results:
            logger.info(f"Successfully scraped {len(results)} pages.")
            
            # 標準出力
            print("\n--- Scraped Results ---")
            for i, data in enumerate(results):
                if data: # エラーでない結果のみ表示を試みる
                    print(f"\n[{i+1}] URL: {data.get('url')}")
                    print(f"    Title: {data.get('title', 'N/A')}")
                    print(f"    Content (first 200 chars): {data.get('content', '')[:200]}...")
            
            # JSONファイルに出力
            output_data = []
            for i, data in enumerate(results):
                if data: # エラーでない結果
                    output_data.append({
                        "index": i + 1,
                        "url": data.get("url"),
                        "title": data.get("title", "N/A"),
                        "content": data.get("content")
                    })
                else: # エラーの場合も記録は残す
                     output_data.append({
                        "index": i + 1,
                        "url": "Unknown or failed URL",
                        "title": "Error",
                        "content": "Failed to retrieve content for this entry."
                    })
                    
            import datetime
            # 現在の日時を取得
            now = datetime.datetime.now()

            # 日時を yyyymmdd-hhmmss 形式の文字列に変換
            timestamp = now.strftime("%Y%m%d-%H%M%S")

            # 出力ファイル名
            output_filename = f"{timestamp}--"
                    
            output_folder = "output"

            # output フォルダが存在するか確認
            if not os.path.exists(output_folder):
                # 存在しない場合は作成
                os.makedirs(output_folder)
            
            args.output_json = output_filename + args.output_json
            # JSONファイルに出力

            with open(os.path.join(output_folder, args.output_json), 'w', encoding='utf-8') as f:
                json.dump(output_data, f, ensure_ascii=False, indent=2)
            logger.info(f"Results saved to {args.output_json}")
            print(f"\nResults saved to {args.output_json}")

        else:
            logger.info("No results were scraped.")
            print("No results were scraped.")

    except Exception as e:
        logger.critical(f"An critical error occurred in main: {e}", exc_info=True)
        print(f"処理中にクリティカルなエラーが発生しました: {e}")
    finally:
        logger.info("Script finished.")


if __name__ == "__main__":
    asyncio.run(main())