search機能の実装

app.py

@@ -69,6 +69,8 @@ import numpy as np
 from datetime import datetime
 from ultralytics import YOLO
 from PIL import Image
+from search import WebScraper
+
 app = FastAPI()
 # CORSミドルウェアの追加
 app.add_middleware(
@@ -229,6 +231,25 @@ def create_mask(image, x1, y1, x2, y2):
 
 
 
+async def search_llm():
+    scraper = WebScraper(headless=True) # UIなしで実行
+    
+    # 個人情報流出に関する事例を検索し、上位2件のクリーンなコンテンツを取得
+    personal_breach_docs = await scraper.get_processed_documents(
+        search_query="個人情報流出 事例 SNS", 
+        num_search_results=10
+    )
+    return personal_breach_docs["cleaned_html_content"]
+
+
+
+
+
+
+
+
+
+
 def llm_to_process_image_simple(risk_level, image_path, point1, point2, thresholds=None):
     print(risk_level, image_path, point1, point2, thresholds)
     print('point1,point2', point1, point2)
@@ -262,8 +283,8 @@ def llm_to_process_image_simple(risk_level, image_path, point1, point2, threshol
     return save_dir + debug_image_path
 
 
-
-def llm_to_process_image_simple_auto(risk_level, image_path, point1, point2, thresholds=None):
+import asyncio
+async def llm_to_process_image_simple_auto(risk_level, image_path, point1, point2, thresholds=None):
     print(risk_level, image_path, point1, point2, thresholds)
     print('point1,point2', point1, point2)
     GEMINI_API_KEY=os.getenv('GEMINI_API_KEY')
@@ -273,6 +294,13 @@ def llm_to_process_image_simple_auto(risk_level, image_path, point1, point2, thr
     response=Objectdetector.detect_auto(image_path)
     print(response["objects_to_remove"])
     Objectdetector.prompt_objects=response["objects_to_remove"]
+     # 個人情報流出に関する事例を検索し、上位2件のクリーンなコンテンツを取得
+    scraper = WebScraper(headless=True) 
+    personal_breach_docs = asyncio.run(await scraper.get_processed_documents(
+        search_query="個人情報流出 事例 SNS", 
+        num_search_results=10
+    ))
+    Objectdetector.text=personal_breach_docs["cleaned_html_content"]
     # 画像の読み込みとRGB変換
     print(f"Objectdetector.prompt_objects: {Objectdetector.prompt_objects}")
     image = cv2.imread(image_path)
@@ -1010,13 +1038,6 @@ async def create_mask_sum_auto(image: UploadFile = File(...), risk_level: int =
 
 
 
-
-
-
-
-
-
-
 # カスケードファイルの読み込み (顔検出)
 #face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')
 

requirements.txt

@@ -73,4 +73,4 @@ supervision
 onnxruntime
 google-genai
 python-dotenv
-# moondream # ここを削除
+

search.py

@@ -1,7 +1,7 @@
 import asyncio
-from playwright.async_api import async_playwright, Page, Browser
+from playwright.async_api import async_playwright, Page, Browser, Playwright
 from bs4 import BeautifulSoup
-from bs4.element import Comment # BeautifulSoupのコメント削除用
+from bs4.element import Comment
 from urllib.parse import urlparse, parse_qs
 from typing import List, Dict, Optional
 
@@ -19,23 +19,31 @@ class WebScraper:
         """
         self.headless = headless
         self.default_timeout = default_timeout
-        self._browser: Optional[Browser] = None # Browserインスタンスを保持するため
+        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():
-            self._browser = await async_playwright().chromium.launch(headless=self.headless)
+            if self._playwright_instance is None:
+                self._playwright_instance = await async_playwright().start()
+            self._browser = await self._playwright_instance.chromium.launch(headless=self.headless)
         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()
+        browser = await self._launch_browser() # ブラウザが起動または取得される
         page = await browser.new_page()
         page.set_default_timeout(self.default_timeout)
         return page
@@ -45,37 +53,50 @@ class WebScraper:
         DuckDuckGoで指定されたクエリを検索し、上位N件の検索結果（タイトルとURL）を返します。
         """
         results = []
-        page: Optional[Page] = None # 明示的に型ヒントを追加
+        page: Optional[Page] = None
+        
         try:
             page = await self._get_new_page()
+            """Playwrightのステルス技術を適用し、ボット検出を回避します。"""
+            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' }) });""")
+
             print(f"DuckDuckGoで '{query}' を検索中...")
-            await page.goto("https://duckduckgo.com/")
+            # DuckDuckGoの検索URLは一般的に `?q=` パラメータを使用します
+            await page.goto(f"https://duckduckgo.com/?q={query}")
 
-            await page.fill("#search_form_input_homepage", query)
-            await page.press("#search_form_input_homepage", "Enter")
+            # 検索結果のタイトルリンク要素を特定するセレクタ
+            # DuckDuckGoのHTML構造は変更される可能性があるため、適宜調整が必要
+            # 現在の一般的なセレクタは 'a[data-testid="result-title-link"]'
+            await page.wait_for_selector('h2 > a', timeout=10000)
 
-            await page.wait_for_selector("#links .result__a", timeout=10000)
+            # 検索結果のタイトルリンク要素を取得 (await は不要、Locatorオブジェクトを返す)
+            search_links = page.locator('h2 > a')
 
-            search_elements = await page.query_selector_all("#links .result")
-            
-            for i, element in enumerate(search_elements):
-                if i >= num_results:
-                    break
+            # 取得する結果の数を制限
+            for i in range(min(num_results, await search_links.count())):
+                link_element = search_links.nth(i)
                 
-                title_element = await element.query_selector(".result__a")
-                url_element = await element.query_selector(".result__url")
-
-                title = await title_element.text_content() if title_element else "タイトルなし"
-                url = await url_element.get_attribute("href") if url_element else "URLなし"
+                # タイトルはリンク要素のテキストコンテンツ
+                title = await link_element.text_content()
+                # URLはリンク要素のhref属性
+                url = await link_element.get_attribute("href")
                 
-                # DuckDuckGoのURLのデコードとクリーンアップ
-                if url and url != "URLなし":
+                # DuckDuckGoのリダイレクトURLのデコードとクリーンアップ
+                if url:
                     parsed_url = urlparse(url)
-                    if parsed_url.path == '/l/':
+                    # 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
                     
-                results.append({"title": title.strip(), "url": url.strip()})
+                    # 結果を追加する前に、タイトルとURLが有効か軽くチェック
+                    if title and url and title.strip() != "" and url.strip() != "":
+                        results.append({"title": title.strip(), "url": url.strip()})
+
         except Exception as e:
             print(f"DuckDuckGo検索中にエラーが発生しました: {e}")
         finally:
@@ -91,7 +112,8 @@ class WebScraper:
         try:
             page = await self._get_new_page()
             print(f"  URL: {url} のコンテンツを取得中...")
-            await page.goto(url)
+            # 'domcontentloaded' は 'load' よりも高速な場合が多い
+            await page.goto(url, wait_until='domcontentloaded') 
             return await page.content()
         except Exception as e:
             print(f"  URL: {url} のコンテンツ取得中にエラーが発生しました: {e}")
@@ -121,6 +143,7 @@ class WebScraper:
 
         # 複数の連続する改行を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)
 
@@ -138,11 +161,8 @@ class WebScraper:
         """
         processed_documents = []
         
-        # Playwrightの非同期コンテキストマネージャでブラウザインスタンスを管理
-        async with async_playwright() as p:
-            # ブラウザを一度だけ起動し、インスタンス変数に保持
-            self._browser = await p.chromium.launch(headless=self.headless)
-
+        # クラスのインスタンスでブラウザのライフサイクルを管理
+        try:
             top_results = await self.search_duckduckgo(search_query, num_search_results)
 
             if top_results:
@@ -167,9 +187,27 @@ class WebScraper:
                         print("  クリーンなコンテンツを取得できませんでした。")
             else:
                 print("検索結果が見つからなかったため、処理をスキップします。")
-            
-            await self._close_browser() # 全ての処理後にブラウザを閉じる
+        finally:
+            # すべての処理が完了したらブラウザを閉じる
+            await self._close_browser()
 
         return processed_documents
 
 # クラスの使用例
+async def main():
+    scraper = WebScraper(headless=False) # デバッグのためにheadless=Falseにしても良い
+    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())

test.py

@@ -3,11 +3,13 @@ import os
 from dotenv import load_dotenv
 import numpy as np
 import cv2
+import asyncio
+
 from PIL import Image
 from search import WebScraper
 load_dotenv(dotenv_path='../.env')
 async def main():
-    scraper = WebScraper(headless=True) # UIなしで実行
+    scraper = WebScraper(headless=False) # UIなしで実行
     
     # 個人情報流出に関する事例を検索し、上位2件のクリーンなコンテンツを取得
     personal_breach_docs = await scraper.get_processed_documents(
@@ -24,4 +26,5 @@ async def main():
             print("-" * 30)
     else:
         print("処理されたドキュメントはありませんでした。")
-main()
+if __name__ == "__main__":
+    asyncio.run(main())