search.py

import asyncio
from playwright.async_api import async_playwright, Page, Browser, Playwright
from bs4 import BeautifulSoup
from bs4.element import Comment
from urllib.parse import urlparse, parse_qs
from typing import List, Dict, Optional
import random # randomモジュールを追加

class WebScraper:
    """
    DuckDuckGoでの検索、URLからのコンテンツ取得、HTMLクリーンアップを行うクラス。
    """
    # User-Agentのリストをクラス変数として定義
    USER_AGENTS = [
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36",
        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/108.0",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:109.0) Gecko/20100101 Firefox/108.0",
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", # 最新版に近いChrome
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0.3 Safari/605.1.15", # Safari
        "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 OPR/106.0.0.0", # Opera
    ]

    def __init__(self, headless: bool = True, default_timeout: int = 30000):
        """
        WebScraperのインスタンスを初期化します。

        Args:
            headless (bool): Playwrightをヘッドレスモードで実行するかどうか (デフォルト: True)。
            default_timeout (int): ページのロードタイムアウト (ミリ秒、デフォルト: 30000 = 30秒)。
        """
        self.headless = headless
        self.default_timeout = default_timeout
        self._browser: Optional[Browser] = None
        self._playwright_instance: Optional[Playwright] = None # Playwrightインスタンスを保持

    async def _launch_browser(self) -> Browser:
        """Playwrightを起動し、ブラウザを立ち上げます。
        既にブラウザが起動していればそれを再利用します。
        """
        if not self._browser or not self._browser.is_connected():
            if self._playwright_instance is None:
                self._playwright_instance = await async_playwright().start()
            # ヘッドレスモードでの検出を避けるための引数を追加
            self._browser = await self._playwright_instance.chromium.launch(
                headless=self.headless,
                args=[
                    '--no-sandbox',
                    '--disable-setuid-sandbox',
                    '--disable-infobars',
                    '--window-size=1280,720', # 一般的なデスクトップサイズに設定
                    '--disable-blink-features=AutomationControlled' # ヘッドレス検出を回避
                ]
            )
        return self._browser

    async def _close_browser(self):
        """ブラウザを閉じ、Playwrightインスタンスも停止します。"""
        if self._browser and self._browser.is_connected():
            await self._browser.close()
        self._browser = None
        if self._playwright_instance:
            await self._playwright_instance.stop()
            self._playwright_instance = None

    async def _get_new_page(self) -> Page:
        """新しいページ（タブ）を作成します。"""
        browser = await self._launch_browser() # ブラウザが起動または取得される
        page = await browser.new_page()
        page.set_default_timeout(self.default_timeout)
        
        # User-Agentをランダムに選択して設定
        await page.set_extra_http_headers({
            "User-Agent": random.choice(self.USER_AGENTS)
        })

        # より包括的なステルス対策をページに適用
        await page.evaluate("""Object.defineProperty(navigator, 'webdriver', { get: () => false });""")
        await page.evaluate("""Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });""")
        await page.evaluate("""Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });""")
        await page.evaluate("""window.chrome = { runtime: {}, loadTimes: function() {}, csi: function() {}, app: {} };""")
        await page.evaluate("""Object.defineProperty(navigator.permissions, 'query', { enumerable: true, configurable: true, writable: true, value: async (parameters) => ({ state: 'prompt' }) });""")
        
        # ページロード後の追加待機
        await asyncio.sleep(random.uniform(2, 5)) 

        return page

    async def search_duckduckgo(self, query: str, num_results: int = 3) -> List[Dict[str, str]]:
        """
        DuckDuckGoで指定されたクエリを検索し、上位N件の検索結果（タイトルとURL）を返します。
        """
        results = []
        page: Optional[Page] = None
        
        try:
            page = await self._get_new_page()
            
            print(f"Bingで '{query}' を検索中...")
            # networkidle でより安定したページロードを待機
            await page.goto(f"https://www.bing.com/search?q=={query}&setlang=ja", wait_until='networkidle') 
            
            # デバッグのためにページのスクリーンショットを保存
            await page.screenshot(path="./duckduckgo_search_results.png")
            
            # 検索結果のタイトルリンク要素を特定するセレクタ
            # DuckDuckGoのHTML構造は変更される可能性があるため、適宜調整が必要です。
            # 現在の一般的なセレクタは 'a[data-testid="result-title-link"]' もしくは 'h2 > a' ですが、
            # ページ構造が変わった場合は、開発者ツールで適切なセレクタを見つけてください。
            
            # 要素が見つかるまで、より長く待機するか、別のセレクタを試す
            try:
                await page.wait_for_selector('h2 > a', timeout=20000) # タイムアウトを長くする
            except Exception as e:
                print(f"セレクタ 'h2 > a' の待機中にタイムアウトしました: {e}")
                # ここで代替のセレクタを試すか、処理を終了する
                return []

            search_links = page.locator('h2 > a')

            for i in range(min(num_results, await search_links.count())):
                link_element = search_links.nth(i)
                
                title = await link_element.text_content()
                url = await link_element.get_attribute("href")
                
                if url:
                    parsed_url = urlparse(url)
                    # DuckDuckGoのリダイレクトURLのデコードとクリーンアップ
                    if parsed_url.netloc == 'duckduckgo.com' and parsed_url.path == '/l/':
                        decoded_url = parse_qs(parsed_url.query).get('uddg', [''])[0]
                        url = decoded_url
                    
                    # 結果を追加する前に、タイトルとURLが有効か軽くチェック
                    if title and url and title.strip() != "" and url.strip() != "":
                        results.append({"title": title.strip(), "url": url.strip()})
            
            # 検索結果が一つも見つからなかった場合もスクリーンショットを保存
            if not results:
                print(f"検索結果が見つかりませんでした。ページのスクリーンショットを './duckduckgo_no_results.png' に保存します。")
                await page.screenshot(path="./duckduckgo_no_results.png")

        except Exception as e:
            print(f"DuckDuckGo検索中にエラーが発生しました: {e}")
        finally:
            if page:
                await page.close() # ページを閉じる
        
        print(f"検索が完了しました。{len(results)} 件の結果が見つかりました。")
        return results

    async def _get_raw_html_content(self, url: str) -> Optional[str]:
        """指定されたURLから生のHTMLコンテンツを取得します。"""
        page: Optional[Page] = None
        try:
            page = await self._get_new_page()
            print(f"   URL: {url} のコンテンツを取得中...")
            # networkidle でより安定したページロードを待機
            await page.goto(url) 
            return await page.content()
        except Exception as e:
            print(f"   URL: {url} のコンテンツ取得中にエラーが発生しました: {e}")
            return None
        finally:
            if page:
                await page.close()

    def _clean_html_to_text(self, html_content: str) -> str:
        """
        HTMLコンテンツからJavaScript、スタイル、不要なリンクなどを除去し、整形されたテキストを返します。
        """
        soup = BeautifulSoup(html_content, 'html.parser')

        # スクリプトタグとスタイルタグを削除
        for script_or_style in soup(["script", "style"]):
            script_or_style.decompose()

        # headタグ内のリンクタグ（CSSなど）を削除
        if soup.head:
            for link_tag in soup.head.find_all('link'):
                link_tag.decompose()
        
        # HTMLコメントを削除
        for comment in soup.find_all(string=lambda text: isinstance(text, Comment)):
            comment.extract()

        # 複数の連続する改行を1つに減らす
        cleaned_text = soup.get_text(separator='\n', strip=True)
        # 空行を削除し、各行をトリム
        cleaned_text_lines = [line.strip() for line in cleaned_text.splitlines() if line.strip()]
        return '\n'.join(cleaned_text_lines)

    async def get_processed_documents(self, search_query: str, num_search_results: int = 2) -> List[Dict[str, str]]:
        """
        DuckDuckGoで検索し、上位N件の検索結果のURLからクリーンなHTMLコンテンツを取得します。

        Args:
            search_query (str): 検索クエリ。
            num_search_results (int): 取得する検索結果の数。

        Returns:
            List[Dict[str, str]]: 処理されたドキュメントのリスト。
                                    各ドキュメントは 'title', 'original_url', 'cleaned_html_content' を含む。
        """
        processed_documents = []
        
        # クラスのインスタンスでブラウザのライフサイクルを管理
        try:
            top_results = await self.search_duckduckgo(search_query, num_search_results)

            if top_results:
                for i, result in enumerate(top_results):
                    print(f"\n--- 処理中の記事 {i+1} ---")
                    print(f"タイトル: {result['title']}")
                    print(f"元URL: {result['url']}")
                    
                    # 個別のURLのコンテンツを取得・クリーンアップ
                    raw_html = await self._get_raw_html_content(result['url'])
                    
                    if raw_html:
                        cleaned_content = self._clean_html_to_text(raw_html)
                        processed_documents.append({
                            "title": result['title'],
                            "original_url": result['url'],
                            "cleaned_html_content": cleaned_content
                        })
                        print(f"   クリーンなコンテンツの長さ: {len(cleaned_content)} 文字")
                        print(f"   クリーンなコンテンツ（一部）:\n{cleaned_content[:500]}...")
                    else:
                        print("   クリーンなコンテンツを取得できませんでした。")
            else:
                print("検索結果が見つからなかったため、処理をスキップします。")
        finally:
            # すべての処理が完了したらブラウザを閉じる
            await self._close_browser()

        return processed_documents

# クラスの使用例
async def main():
    scraper = WebScraper(headless=False) # まずはheadless=Trueで試してください
    query = "個人情報流出 事例"
    documents = await scraper.get_processed_documents(query, num_search_results=2)

    if documents:
        print("\n--- 処理されたドキュメント ---")
        for doc in documents:
            print(f"タイトル: {doc['title']}")
            print(f"URL: {doc['original_url']}")
            # print(f"コンテンツの長さ: {len(doc['cleaned_html_content'])} 文字")
            # print(f"コンテンツの一部: {doc['cleaned_html_content'][:200]}...\n")
    else:
        print("処理されたドキュメントはありませんでした。")

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