File size: 2,926 Bytes
6c511d6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | from urllib.parse import parse_qs, unquote, urlparse
import requests
from bs4 import BeautifulSoup
from config import REQUEST_HEADERS
from tools.common import truncate_text
def clean_html_text(raw_html: str) -> str:
soup = BeautifulSoup(raw_html, "html.parser")
for tag in soup(["script", "style", "noscript", "svg", "nav", "footer"]):
tag.decompose()
title = soup.title.string.strip() if soup.title and soup.title.string else ""
text = soup.get_text("\n")
lines = [line.strip() for line in text.splitlines() if line.strip()]
compact_text = "\n".join(lines)
if title:
return f"标题：{title}\n正文：\n{compact_text}"
return compact_text
def fetch_url_text(url: str, limit: int = 8000) -> str:
try:
response = requests.get(url, headers=REQUEST_HEADERS, timeout=25)
response.raise_for_status()
content_type = response.headers.get("content-type", "")
if "text/html" in content_type or "<html" in response.text[:500].lower():
return truncate_text(clean_html_text(response.text), limit)
return truncate_text(response.text, limit)
except Exception as exc:
return f"无法读取网页 {url}：{exc}"
def decode_duckduckgo_href(href: str) -> str:
parsed = urlparse(href)
query = parse_qs(parsed.query)
if "uddg" in query:
return unquote(query["uddg"][0])
return href
def search_web(query: str, max_results: int = 5) -> list[dict[str, str]]:
try:
response = requests.get(
"https://duckduckgo.com/html/",
params={"q": query},
headers=REQUEST_HEADERS,
timeout=25,
)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
results = []
for anchor in soup.select("a.result__a"):
title = anchor.get_text(" ", strip=True)
href = decode_duckduckgo_href(anchor.get("href", ""))
if title and href.startswith("http"):
results.append({"title": title, "url": href})
if len(results) >= max_results:
break
return results
except Exception as exc:
print(f"搜索失败：{exc}")
return []
def collect_web_evidence(question: str, limit: int = 10000) -> str:
search_results = search_web(question)
if not search_results:
return "网页搜索没有返回可用结果。"
evidence_parts = ["网页搜索结果："]
for index, result in enumerate(search_results, start=1):
evidence_parts.append(f"{index}. {result['title']} - {result['url']}")
for result in search_results[:3]:
page_text = fetch_url_text(result["url"], limit=3500)
evidence_parts.append(
f"\n--- 网页内容：{result['title']} ({result['url']}) ---\n{page_text}"
)
return truncate_text("\n".join(evidence_parts), limit)
|