Create websearch.py

tools/websearch.py

@@ -0,0 +1,72 @@
+import requests
+from bs4 import BeautifulSoup, Comment
+import logging
+import re
+from duckduckgo_search import DDGS
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_TIMEOUT = 10
+MAX_CONTENT_LENGTH_PER_URL = 3000
+MAX_TOTAL_SCRAPED_CONTENT = 9000
+
+def scrape_url(url_to_scrape):
+    try:
+        headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36'}
+        response = requests.get(url_to_scrape, headers=headers, timeout=DEFAULT_TIMEOUT, allow_redirects=True)
+        response.raise_for_status()
+        if 'html' not in response.headers.get('content-type', '').lower():
+            return {"url": url_to_scrape, "error": "Non-HTML content"}
+
+        soup = BeautifulSoup(response.content, 'html.parser')
+        for el_type in ["script", "style", "nav", "footer", "aside", "form", "header", "menu"]:
+            for el in soup.select(el_type): el.decompose()
+        for comment in soup.find_all(string=lambda text: isinstance(text, Comment)):
+            comment.extract()
+        
+        main_selectors = ['main', 'article', '.main-content', '.entry-content', 'div[role="main"]']
+        content_area = next((soup.select_one(s) for s in main_selectors if soup.select_one(s)), soup.body or soup)
+        
+        text_parts = [el.get_text(separator=' ', strip=True) for el in content_area.find_all(['p', 'h1', 'h2', 'h3', 'li', 'td'])]
+        full_text = "\n\n".join(filter(None, text_parts))
+        if not full_text: full_text = content_area.get_text(separator='\n', strip=True)
+        
+        full_text = re.sub(r'\s{3,}', ' ', full_text).strip()
+        full_text = re.sub(r'(\n\s*){3,}', '\n\n', full_text)
+        title = soup.find('title').string.strip() if soup.find('title') and soup.find('title').string else url_to_scrape
+        
+        return {"url": url_to_scrape, "title": title, "content": full_text[:MAX_CONTENT_LENGTH_PER_URL]}
+    except requests.exceptions.RequestException as e:
+        logger.error(f"WEBSEARCH: Request failed for {url_to_scrape}: {e}")
+        return {"url": url_to_scrape, "error": f"Request failed: {e}"}
+    except Exception as e:
+        logger.error(f"WEBSEARCH: Scraping error for {url_to_scrape}: {e}", exc_info=True)
+        return {"url": url_to_scrape, "error": f"Scraping error: {e}"}
+
+def search_and_scrape_duckduckgo(search_query, num_results=3):
+    scraped_data = []
+    total_content_len = 0
+    try:
+        with DDGS() as ddgs:
+            results = list(ddgs.text(search_query, max_results=num_results + 2, region='wt-wt'))
+        if not results: return [{"query": search_query, "error": "No search results."}]
+        
+        for res in results[:num_results]:
+            if total_content_len >= MAX_TOTAL_SCRAPED_CONTENT: break
+            if not res.get('href'): continue
+            
+            scraped_info = scrape_url(res['href'])
+            if not scraped_info.get("title") or scraped_info.get("title") == res['href']:
+                scraped_info["title"] = res.get('title', res['href'])
+            
+            current_content = scraped_info.get("content", "")
+            if res.get('body') and (not current_content or len(current_content) < 150 or scraped_info.get("error")):
+                scraped_info["content"] = f"Search snippet: {res['body']}\n\n(Content from page):\n{current_content}"
+            
+            scraped_data.append(scraped_info)
+            if scraped_info.get("content") and not scraped_info.get("error"):
+                total_content_len += len(scraped_info["content"])
+        return scraped_data
+    except Exception as e:
+        logger.error(f"WEBSEARCH (DDG): Error for '{search_query}': {e}", exc_info=True)
+        return [{"query": search_query, "error": f"DDG search failed: {e}"}]