"""
title: Web Search using SearXNG and Scrape first N Pages
author: constLiakos with enhancements by justinh-rahb and ther3zz
funding_url: https://github.com/EntropyYue/web_search
version: 0.4.4
license: MIT
"""
import requests
import json
from bs4 import BeautifulSoup
import concurrent.futures
from urllib.parse import urlparse
import re
import unicodedata
from pydantic import BaseModel, Field
from typing import Callable, Any
class HelpFunctions:
def __init__(self):
pass
def get_base_url(self, url):
parsed_url = urlparse(url)
base_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
return base_url
def generate_excerpt(self, content, max_length=200):
return content[:max_length] + "..." if len(content) > max_length else content
def format_text(self, original_text, valves):
soup = BeautifulSoup(original_text, "html.parser")
formatted_text = soup.get_text(separator=" ", strip=True)
formatted_text = unicodedata.normalize("NFKC", formatted_text)
formatted_text = re.sub(r"\s+", " ", formatted_text)
formatted_text = formatted_text.strip()
formatted_text = self.remove_emojis(formatted_text)
if valves.REMOVE_LINKS:
formatted_text = self.replace_urls_with_text(formatted_text)
return formatted_text
def remove_emojis(self, text):
return "".join(c for c in text if not unicodedata.category(c).startswith("So"))
def replace_urls_with_text(self, text, replacement="(links)"):
pattern = r"\(https?://[^\s]+\)"
return re.sub(pattern, replacement, text)
def process_search_result(self, result, valves):
title_site = self.remove_emojis(result["title"])
url_site = result["url"]
snippet = result.get("content", "")
# Check if the website is in the ignored list, but only if IGNORED_WEBSITES is not empty
if valves.IGNORED_WEBSITES:
base_url = self.get_base_url(url_site)
if any(
ignored_site.strip() in base_url
for ignored_site in valves.IGNORED_WEBSITES.split(",")
):
return None
try:
response_site = requests.get(
valves.JINA_READER_BASE_URL + url_site, timeout=20
)
response_site.raise_for_status()
html_content = response_site.text
soup = BeautifulSoup(html_content, "html.parser")
content_site = self.format_text(
soup.get_text(separator=" ", strip=True), valves
)
truncated_content = self.truncate_to_n_words(
content_site, valves.PAGE_CONTENT_WORDS_LIMIT
)
return {
"title": title_site,
"url": url_site,
"content": truncated_content,
"snippet": self.remove_emojis(snippet),
}
except requests.exceptions.RequestException:
return None
def truncate_to_n_words(self, text, token_limit):
tokens = text.split()
truncated_tokens = tokens[:token_limit]
return " ".join(truncated_tokens)
class EventEmitter:
def __init__(self, event_emitter: Callable[[dict], Any] = None):
self.event_emitter = event_emitter
async def emit(
self,
description="未知状态",
status="in_progress",
done=False,
action="",
urls=[],
):
if self.event_emitter:
await self.event_emitter(
{
"type": "status",
"data": {
"status": status,
"description": description,
"done": done,
"action": action,
"urls": urls,
},
}
)
class Tools:
class Valves(BaseModel):
SEARXNG_ENGINE_API_BASE_URL: str = Field(
default="https://example.com/search",
description="搜索引擎的基础URL",
)
IGNORED_WEBSITES: str = Field(
default="",
description="以逗号分隔的要忽略的网站列表",
)
RETURNED_SCRAPPED_PAGES_NO: int = Field(
default=3,
description="要分析的搜索引擎结果数",
)
SCRAPPED_PAGES_NO: int = Field(
default=5,
description="已分页的总页数。理想情况下，大于返回的页面之一",
)
PAGE_CONTENT_WORDS_LIMIT: int = Field(
default=5000,
description="限制每页的字数",
)
CITATION_LINKS: bool = Field(
default=False,
description="如果为True，则发送带有链接的自定义引用",
)
JINA_READER_BASE_URL: str = Field(
default="https://r.jina.ai/",
description="Jina Reader的基础URL",
)
REMOVE_LINKS: bool = Field(
default=True,
description="检索中的返回是否移除链接",
)
def __init__(self):
self.valves = self.Valves()
self.headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"
}
async def search_web(
self,
query: str,
__event_emitter__: Callable[[dict], Any] = None,
) -> str:
"""
搜索网络并获取相关页面的内容，搜索未知知识、新闻、信息、公共联系信息、天气等
:params query: 搜索中使用的关键词
:return: The content of the pages in json format.
"""
functions = HelpFunctions()
emitter = EventEmitter(__event_emitter__)
await emitter.emit(f"正在搜索: {query}")
search_engine_url = self.valves.SEARXNG_ENGINE_API_BASE_URL
# Ensure RETURNED_SCRAPPED_PAGES_NO does not exceed SCRAPPED_PAGES_NO
if self.valves.RETURNED_SCRAPPED_PAGES_NO > self.valves.SCRAPPED_PAGES_NO:
self.valves.RETURNED_SCRAPPED_PAGES_NO = self.valves.SCRAPPED_PAGES_NO
params = {
"q": query,
"format": "json",
"number_of_results": self.valves.RETURNED_SCRAPPED_PAGES_NO,
}
try:
await emitter.emit("正在向搜索引擎发送请求")
resp = requests.get(
search_engine_url, params=params, headers=self.headers, timeout=120
)
resp.raise_for_status()
data = resp.json()
results = data.get("results", [])
limited_results = results[: self.valves.SCRAPPED_PAGES_NO]
await emitter.emit(f"返回了 {len(limited_results)} 个搜索结果")
except requests.exceptions.RequestException as e:
await emitter.emit(
status="error",
description=f"搜索时出错: {str(e)}",
done=True,
)
return json.dumps({"error": str(e)})
results_json = []
if limited_results:
await emitter.emit("正在处理搜索结果")
try:
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = [
executor.submit(
functions.process_search_result, result, self.valves
)
for result in limited_results
]
processed_count = 0
for future in concurrent.futures.as_completed(futures):
result_json = future.result()
if result_json:
try:
results_json.append(result_json)
processed_count += 1
await emitter.emit(
f"处理页面 {processed_count}/{len(limited_results)}",
)
except (TypeError, ValueError, Exception) as e:
print(f"处理时出错: {str(e)}")
continue
if len(results_json) >= self.valves.RETURNED_SCRAPPED_PAGES_NO:
break
except BaseException as e:
await emitter.emit(
status="error",
description=f"处理时出错: {str(e)}",
done=True,
)
results_json = results_json[: self.valves.RETURNED_SCRAPPED_PAGES_NO]
if self.valves.CITATION_LINKS and __event_emitter__:
if len(results_json):
for result in results_json:
await __event_emitter__(
{
"type": "citation",
"data": {
"document": [result["content"]],
"metadata": [{"source": result["url"]}],
"source": {"name": result["title"]},
},
}
)
urls = []
for result in results_json:
urls.append(result["url"])
await emitter.emit(
status="complete",
description=f"网络搜索已完成,将从 {len(results_json)} 个页面检索内容",
done=True,
action="web_search",
urls=urls,
)
return json.dumps(results_json, indent=4, ensure_ascii=False)
async def get_website(
self, url: str, __event_emitter__: Callable[[dict], Any] = None
) -> str:
"""
打开输入的网站并获取其内容
:params url: 需要打开的网站
:return: The content of the website in json format.
"""
functions = HelpFunctions()
emitter = EventEmitter(__event_emitter__)
await emitter.emit(f"正在从URL获取内容: {url}")
results_json = []
try:
response_site = requests.get(
self.valves.JINA_READER_BASE_URL + url,
headers=self.headers,
timeout=120,
)
response_site.raise_for_status()
html_content = response_site.text
soup = BeautifulSoup(html_content, "html.parser")
page_title = soup.title.string if soup.title else "No title found"
page_title = unicodedata.normalize("NFKC", page_title.strip())
page_title = functions.remove_emojis(page_title)
title_site = page_title
url_site = url
content_site = functions.format_text(
soup.get_text(separator=" ", strip=True), self.valves
)
truncated_content = functions.truncate_to_n_words(
content_site, self.valves.PAGE_CONTENT_WORDS_LIMIT
)
result_site = {
"title": title_site,
"url": url_site,
"content": truncated_content,
"excerpt": functions.generate_excerpt(content_site),
}
results_json.append(result_site)
if self.valves.CITATION_LINKS and __event_emitter__:
await __event_emitter__(
{
"type": "citation",
"data": {
"document": [truncated_content],
"metadata": [{"source": url_site}],
"source": {"name": title_site},
},
}
)
await emitter.emit(
status="complete",
description="已成功检索和处理网站内容",
done=True,
)
except requests.exceptions.RequestException as e:
results_json.append(
{
"url": url,
"content": f"检索页面失败,错误: {str(e)}",
}
)
await emitter.emit(
status="error",
description=f"获取网站内容时出错: {str(e)}",
done=True,
)
return json.dumps(results_json, indent=4, ensure_ascii=False)