Spaces:
Runtime error
Runtime error
| 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()) |