tools/web_tools.py

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)