mirror of
https://github.com/VinciGit00/Scrapegraph-ai.git
synced 2026-07-12 21:01:56 +08:00
827f7260ad
Optimize utils modules for better performance and maintainability - Improve HTML cleanup and minification: - Combine regex operations for better performance - Add better error handling for HTML processing - Optimize tag removal and attribute filtering - Enhance deep copy functionality: - Add special case handling for primitive types - Improve type checking and error handling - Optimize recursive copying for collections - Refactor web search functionality: - Add input validation and error handling - Split search logic into separate helper functions - Improve proxy handling and configuration - Add better timeout and error management - Optimize URL filtering and processing Technical improvements: - Better type hints and documentation - More efficient data structures - Improved error handling and validation - Reduced code duplication - Better separation of concerns No breaking changes - all existing functionality maintained
106 lines
4.0 KiB
Python
106 lines
4.0 KiB
Python
"""
|
|
research_web module
|
|
"""
|
|
import re
|
|
from typing import List
|
|
from langchain_community.tools import DuckDuckGoSearchResults
|
|
from googlesearch import search as google_search
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
|
|
def search_on_web(query: str, search_engine: str = "Google",
|
|
max_results: int = 10, port: int = 8080,
|
|
timeout: int = 10, proxy: str | dict = None) -> List[str]:
|
|
"""Search web function with improved error handling and validation"""
|
|
|
|
# Input validation
|
|
if not query or not isinstance(query, str):
|
|
raise ValueError("Query must be a non-empty string")
|
|
|
|
search_engine = search_engine.lower()
|
|
valid_engines = {"google", "duckduckgo", "bing", "searxng"}
|
|
if search_engine not in valid_engines:
|
|
raise ValueError(f"Search engine must be one of: {', '.join(valid_engines)}")
|
|
|
|
# Format proxy once
|
|
formatted_proxy = None
|
|
if proxy:
|
|
formatted_proxy = format_proxy(proxy)
|
|
|
|
try:
|
|
results = []
|
|
if search_engine == "google":
|
|
results = list(google_search(query, num_results=max_results, proxy=formatted_proxy))
|
|
|
|
elif search_engine == "duckduckgo":
|
|
research = DuckDuckGoSearchResults(max_results=max_results)
|
|
res = research.run(query)
|
|
results = re.findall(r'https?://[^\s,\]]+', res)
|
|
|
|
elif search_engine == "bing":
|
|
results = _search_bing(query, max_results, timeout, formatted_proxy)
|
|
|
|
elif search_engine == "searxng":
|
|
results = _search_searxng(query, max_results, port, timeout)
|
|
|
|
return filter_pdf_links(results)
|
|
|
|
except requests.Timeout:
|
|
raise TimeoutError(f"Search request timed out after {timeout} seconds")
|
|
except requests.RequestException as e:
|
|
raise RuntimeError(f"Search request failed: {str(e)}")
|
|
|
|
def _search_bing(query: str, max_results: int, timeout: int, proxy: str = None) -> List[str]:
|
|
"""Helper function for Bing search"""
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
|
|
}
|
|
search_url = f"https://www.bing.com/search?q={query}"
|
|
|
|
proxies = {"http": proxy, "https": proxy} if proxy else None
|
|
response = requests.get(search_url, headers=headers, timeout=timeout, proxies=proxies)
|
|
response.raise_for_status()
|
|
|
|
soup = BeautifulSoup(response.text, "html.parser")
|
|
return [result.find('a')['href'] for result in soup.find_all('li', class_='b_algo', limit=max_results)]
|
|
|
|
def _search_searxng(query: str, max_results: int, port: int, timeout: int) -> List[str]:
|
|
"""Helper function for SearXNG search"""
|
|
url = f"http://localhost:{port}"
|
|
params = {
|
|
"q": query,
|
|
"format": "json",
|
|
"engines": "google,duckduckgo,brave,qwant,bing"
|
|
}
|
|
response = requests.get(url, params=params, timeout=timeout)
|
|
response.raise_for_status()
|
|
return [result['url'] for result in response.json().get("results", [])[:max_results]]
|
|
|
|
def format_proxy(proxy):
|
|
if isinstance(proxy, dict):
|
|
server = proxy.get('server')
|
|
username = proxy.get('username')
|
|
password = proxy.get('password')
|
|
|
|
if all([username, password, server]):
|
|
proxy_url = f"http://{username}:{password}@{server}"
|
|
return proxy_url
|
|
else:
|
|
raise ValueError("Proxy dictionary is missing required fields.")
|
|
elif isinstance(proxy, str):
|
|
return proxy # "https://username:password@ip:port"
|
|
else:
|
|
raise TypeError("Proxy should be a dictionary or a string.")
|
|
|
|
def filter_pdf_links(links: List[str]) -> List[str]:
|
|
"""
|
|
Filters out any links that point to PDF files.
|
|
|
|
Args:
|
|
links (List[str]): A list of URLs as strings.
|
|
|
|
Returns:
|
|
List[str]: A list of URLs excluding any that end with '.pdf'.
|
|
"""
|
|
return [link for link in links if not link.lower().endswith('.pdf')]
|