|
|
@@ -0,0 +1,841 @@
|
|
|
+import requests
|
|
|
+from bs4 import BeautifulSoup
|
|
|
+from urllib.parse import quote_plus, urljoin
|
|
|
+import time
|
|
|
+import re
|
|
|
+import html
|
|
|
+
|
|
|
+class BrowserTool:
|
|
|
+ name = "browser_search"
|
|
|
+ description = "执行网页搜索（支持多种搜索引擎和内容提取）"
|
|
|
+
|
|
|
+ def get_parameters(self):
|
|
|
+ return {
|
|
|
+ "input": {"type": "str", "description": "搜索关键词", "required": True}
|
|
|
+ }
|
|
|
+
|
|
|
+ def _is_valid_result(self, title, url):
|
|
|
+ """验证搜索结果的有效性"""
|
|
|
+ if not title or len(title.strip()) < 3:
|
|
|
+ return False
|
|
|
+
|
|
|
+ # 过滤导航链接和无意义内容
|
|
|
+ skip_keywords = [
|
|
|
+ "next", "previous", "more", "about", "help", "settings",
|
|
|
+ "privacy", "terms", "feedback", "donate", "install",
|
|
|
+ "download", "login", "register", "sign in", "sign up"
|
|
|
+ ]
|
|
|
+
|
|
|
+ title_lower = title.lower()
|
|
|
+ if any(keyword in title_lower for keyword in skip_keywords):
|
|
|
+ return False
|
|
|
+
|
|
|
+ # 过滤广告和推广链接
|
|
|
+ ad_indicators = ["ad", "sponsored", "promotion", "广告", "推广"]
|
|
|
+ if any(indicator in title_lower for indicator in ad_indicators):
|
|
|
+ return False
|
|
|
+
|
|
|
+ return True
|
|
|
+
|
|
|
+ def _clean_text(self, text):
|
|
|
+ """清理文本内容"""
|
|
|
+ if not text:
|
|
|
+ return ""
|
|
|
+
|
|
|
+ # 移除多余空白字符
|
|
|
+ text = re.sub(r'\s+', ' ', text.strip())
|
|
|
+
|
|
|
+ # 移除特殊字符
|
|
|
+ text = re.sub(r'[^\w\s\u4e00-\u9fff.,!?;:()[\]{}"\'-]', '', text)
|
|
|
+
|
|
|
+ return text[:200] # 限制长度
|
|
|
+
|
|
|
+ def _search_searx(self, query, limit=5):
|
|
|
+ """使用多个搜索引擎实例 - 稳定版，优先支持中文搜索"""
|
|
|
+ # 精选多个稳定的搜索引擎，优先支持中文
|
|
|
+ search_instances = [
|
|
|
+ {
|
|
|
+ "name": "Searx.xyz",
|
|
|
+ "url": "https://searx.xyz/search",
|
|
|
+ "timeout": 10,
|
|
|
+ "type": "searx"
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "name": "Searx.be",
|
|
|
+ "url": "https://searx.be/search",
|
|
|
+ "timeout": 10,
|
|
|
+ "type": "searx"
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "name": "Brave搜索",
|
|
|
+ "url": "https://search.brave.com/search",
|
|
|
+ "timeout": 8,
|
|
|
+ "type": "brave"
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "name": "Ecosia",
|
|
|
+ "url": "https://www.ecosia.org/search",
|
|
|
+ "timeout": 8,
|
|
|
+ "type": "ecosia"
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "name": "Qwant",
|
|
|
+ "url": "https://www.qwant.com",
|
|
|
+ "timeout": 8,
|
|
|
+ "type": "qwant"
|
|
|
+ }
|
|
|
+ ]
|
|
|
+
|
|
|
+ for instance in search_instances:
|
|
|
+ try:
|
|
|
+ print(f"🔍 尝试 {instance['name']}...")
|
|
|
+ result = self._try_search_instance(instance, query, limit)
|
|
|
+ if result and len(result) > 0:
|
|
|
+ print(f"✅ {instance['name']} 搜索成功，找到 {len(result)} 个结果")
|
|
|
+ return result, True
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ print(f"⚠️ {instance['name']} 失败: {str(e)[:50]}")
|
|
|
+ continue # 静默失败，快速切换
|
|
|
+
|
|
|
+ # 快速降级到搜索建议
|
|
|
+ print("🔗 所有搜索引擎失败，提供搜索建议")
|
|
|
+ return self._get_search_suggestions(query), True
|
|
|
+
|
|
|
+ def _try_search_instance(self, instance, query, limit):
|
|
|
+ """尝试单个搜索引擎实例"""
|
|
|
+ if instance['type'] == 'searx':
|
|
|
+ return self._try_searx_instance(instance, query, limit)
|
|
|
+ elif instance['type'] == 'duckduckgo':
|
|
|
+ return self._try_duckduckgo_instance(instance, query, limit)
|
|
|
+ elif instance['type'] == 'startpage':
|
|
|
+ return self._try_startpage_instance(instance, query, limit)
|
|
|
+ elif instance['type'] == 'qwant':
|
|
|
+ return self._try_qwant_instance(instance, query, limit)
|
|
|
+ elif instance['type'] == 'brave':
|
|
|
+ return self._try_brave_instance(instance, query, limit)
|
|
|
+ elif instance['type'] == 'ecosia':
|
|
|
+ return self._try_ecosia_instance(instance, query, limit)
|
|
|
+ else:
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _try_searx_instance(self, instance, query, limit):
|
|
|
+ """尝试Searx实例 - 优化中文搜索支持"""
|
|
|
+ # 检测是否为中文查询
|
|
|
+ is_chinese = any('\u4e00' <= char <= '\u9fff' for char in query)
|
|
|
+
|
|
|
+ params = {
|
|
|
+ 'q': query,
|
|
|
+ 'format': 'json',
|
|
|
+ 'engines': 'google,bing,duckduckgo,yandex' if not is_chinese else 'google,bing,yandex,baidu',
|
|
|
+ 'language': 'zh-CN' if is_chinese else 'auto'
|
|
|
+ }
|
|
|
+
|
|
|
+ headers = {
|
|
|
+ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
|
|
+ "Accept": "application/json, text/plain, */*",
|
|
|
+ "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8" if is_chinese else "en-US,en;q=0.9"
|
|
|
+ }
|
|
|
+
|
|
|
+ try:
|
|
|
+ response = requests.get(
|
|
|
+ instance['url'],
|
|
|
+ params=params,
|
|
|
+ headers=headers,
|
|
|
+ timeout=instance['timeout']
|
|
|
+ )
|
|
|
+
|
|
|
+ if response.status_code == 200:
|
|
|
+ try:
|
|
|
+ data = response.json()
|
|
|
+ results = []
|
|
|
+
|
|
|
+ for item in data.get('results', [])[:limit]:
|
|
|
+ title = self._clean_text(item.get('title', ''))
|
|
|
+ url = item.get('url', '')
|
|
|
+ content = item.get('content', '')
|
|
|
+
|
|
|
+ if self._is_valid_result(title, url):
|
|
|
+ results.append({
|
|
|
+ 'title': title,
|
|
|
+ 'url': url,
|
|
|
+ 'snippet': self._clean_text(content)[:200],
|
|
|
+ 'source': f"{instance['name']}/{item.get('engine', 'unknown')}"
|
|
|
+ })
|
|
|
+
|
|
|
+ return results if results else None
|
|
|
+ except Exception as e:
|
|
|
+ print(f"⚠️ 解析Searx响应失败: {str(e)[:50]}")
|
|
|
+ return None
|
|
|
+ else:
|
|
|
+ print(f"⚠️ Searx返回状态码: {response.status_code}")
|
|
|
+ return None
|
|
|
+ except requests.Timeout:
|
|
|
+ print(f"⚠️ {instance['name']} 请求超时")
|
|
|
+ return None
|
|
|
+ except Exception as e:
|
|
|
+ print(f"⚠️ {instance['name']} 请求异常: {str(e)[:50]}")
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _try_duckduckgo_instance(self, instance, query, limit):
|
|
|
+ """尝试DuckDuckGo实例"""
|
|
|
+ params = {
|
|
|
+ 'q': query,
|
|
|
+ 'kl': 'cn-zh'
|
|
|
+ }
|
|
|
+
|
|
|
+ headers = {
|
|
|
+ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
|
|
|
+ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
|
|
|
+ }
|
|
|
+
|
|
|
+ response = requests.get(
|
|
|
+ instance['url'],
|
|
|
+ params=params,
|
|
|
+ headers=headers,
|
|
|
+ timeout=instance['timeout']
|
|
|
+ )
|
|
|
+
|
|
|
+ if response.status_code == 200:
|
|
|
+ soup = BeautifulSoup(response.text, 'html.parser')
|
|
|
+ return self._extract_duckduckgo_results_from_soup(soup, limit)
|
|
|
+
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _try_startpage_instance(self, instance, query, limit):
|
|
|
+ """尝试Startpage实例"""
|
|
|
+ params = {
|
|
|
+ 'query': query,
|
|
|
+ 'cat': 'web',
|
|
|
+ 'pl': 'ext-ff',
|
|
|
+ 'extVersion': '1.3.0'
|
|
|
+ }
|
|
|
+
|
|
|
+ headers = {
|
|
|
+ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
|
|
|
+ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
|
|
|
+ }
|
|
|
+
|
|
|
+ response = requests.get(
|
|
|
+ instance['url'],
|
|
|
+ params=params,
|
|
|
+ headers=headers,
|
|
|
+ timeout=instance['timeout']
|
|
|
+ )
|
|
|
+
|
|
|
+ if response.status_code == 200:
|
|
|
+ soup = BeautifulSoup(response.text, 'html.parser')
|
|
|
+ return self._extract_startpage_results(soup, limit)
|
|
|
+
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _try_qwant_instance(self, instance, query, limit):
|
|
|
+ """尝试Qwant实例"""
|
|
|
+ params = {
|
|
|
+ 'q': query,
|
|
|
+ 't': 'web',
|
|
|
+ 'locale': 'zh_CN'
|
|
|
+ }
|
|
|
+
|
|
|
+ headers = {
|
|
|
+ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
|
|
|
+ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
|
|
|
+ }
|
|
|
+
|
|
|
+ response = requests.get(
|
|
|
+ instance['url'],
|
|
|
+ params=params,
|
|
|
+ headers=headers,
|
|
|
+ timeout=instance['timeout']
|
|
|
+ )
|
|
|
+
|
|
|
+ if response.status_code == 200:
|
|
|
+ soup = BeautifulSoup(response.text, 'html.parser')
|
|
|
+ return self._extract_qwant_results(soup, limit)
|
|
|
+
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _try_brave_instance(self, instance, query, limit):
|
|
|
+ """尝试Brave搜索实例"""
|
|
|
+ params = {
|
|
|
+ 'q': query,
|
|
|
+ 'source': 'web'
|
|
|
+ }
|
|
|
+
|
|
|
+ headers = {
|
|
|
+ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
|
|
+ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
|
+ "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
|
|
|
+ }
|
|
|
+
|
|
|
+ try:
|
|
|
+ response = requests.get(
|
|
|
+ instance['url'],
|
|
|
+ params=params,
|
|
|
+ headers=headers,
|
|
|
+ timeout=instance['timeout']
|
|
|
+ )
|
|
|
+
|
|
|
+ if response.status_code == 200:
|
|
|
+ soup = BeautifulSoup(response.text, 'html.parser')
|
|
|
+ # Brave搜索结果提取（需要根据实际HTML结构调整）
|
|
|
+ results = []
|
|
|
+ result_divs = soup.find_all('div', class_=['result', 'web-result'])
|
|
|
+
|
|
|
+ for div in result_divs[:limit]:
|
|
|
+ title_elem = div.find('a') or div.find('h2')
|
|
|
+ snippet_elem = div.find('p') or div.find('span', class_='snippet')
|
|
|
+
|
|
|
+ if title_elem:
|
|
|
+ title = self._clean_text(title_elem.get_text())
|
|
|
+ url = title_elem.get('href', '')
|
|
|
+ snippet = self._clean_text(snippet_elem.get_text()) if snippet_elem else ''
|
|
|
+
|
|
|
+ if self._is_valid_result(title, url):
|
|
|
+ results.append({
|
|
|
+ 'title': title,
|
|
|
+ 'url': url,
|
|
|
+ 'snippet': snippet[:200],
|
|
|
+ 'source': 'Brave'
|
|
|
+ })
|
|
|
+
|
|
|
+ return results if results else None
|
|
|
+ except Exception:
|
|
|
+ return None
|
|
|
+
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _try_ecosia_instance(self, instance, query, limit):
|
|
|
+ """尝试Ecosia搜索实例"""
|
|
|
+ params = {
|
|
|
+ 'q': query
|
|
|
+ }
|
|
|
+
|
|
|
+ headers = {
|
|
|
+ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
|
|
+ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
|
+ "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
|
|
|
+ }
|
|
|
+
|
|
|
+ try:
|
|
|
+ response = requests.get(
|
|
|
+ instance['url'],
|
|
|
+ params=params,
|
|
|
+ headers=headers,
|
|
|
+ timeout=instance['timeout']
|
|
|
+ )
|
|
|
+
|
|
|
+ if response.status_code == 200:
|
|
|
+ soup = BeautifulSoup(response.text, 'html.parser')
|
|
|
+ # Ecosia搜索结果提取（需要根据实际HTML结构调整）
|
|
|
+ results = []
|
|
|
+ result_divs = soup.find_all('div', class_=['result', 'web-result', 'result__body'])
|
|
|
+
|
|
|
+ for div in result_divs[:limit]:
|
|
|
+ title_elem = div.find('a') or div.find('h2')
|
|
|
+ snippet_elem = div.find('p') or div.find('span', class_='result__snippet')
|
|
|
+
|
|
|
+ if title_elem:
|
|
|
+ title = self._clean_text(title_elem.get_text())
|
|
|
+ url = title_elem.get('href', '')
|
|
|
+ snippet = self._clean_text(snippet_elem.get_text()) if snippet_elem else ''
|
|
|
+
|
|
|
+ if self._is_valid_result(title, url):
|
|
|
+ results.append({
|
|
|
+ 'title': title,
|
|
|
+ 'url': url,
|
|
|
+ 'snippet': snippet[:200],
|
|
|
+ 'source': 'Ecosia'
|
|
|
+ })
|
|
|
+
|
|
|
+ return results if results else None
|
|
|
+ except Exception:
|
|
|
+ return None
|
|
|
+
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _extract_duckduckgo_results_from_soup(self, soup, limit):
|
|
|
+ """从DuckDuckGo HTML中提取结果"""
|
|
|
+ results = []
|
|
|
+
|
|
|
+ # 查找搜索结果
|
|
|
+ result_divs = soup.find_all('div', class_='result')
|
|
|
+
|
|
|
+ for div in result_divs[:limit]:
|
|
|
+ title_elem = div.find('a', class_='result__a')
|
|
|
+ snippet_elem = div.find('a', class_='result__snippet')
|
|
|
+
|
|
|
+ if title_elem:
|
|
|
+ title = self._clean_text(title_elem.get_text())
|
|
|
+ url = title_elem.get('href', '')
|
|
|
+ snippet = self._clean_text(snippet_elem.get_text()) if snippet_elem else ''
|
|
|
+
|
|
|
+ if self._is_valid_result(title, url):
|
|
|
+ results.append({
|
|
|
+ 'title': title,
|
|
|
+ 'url': url,
|
|
|
+ 'snippet': snippet[:200],
|
|
|
+ 'source': 'DuckDuckGo'
|
|
|
+ })
|
|
|
+
|
|
|
+ return results
|
|
|
+
|
|
|
+ def _extract_startpage_results(self, soup, limit):
|
|
|
+ """从Startpage HTML中提取结果"""
|
|
|
+ results = []
|
|
|
+
|
|
|
+ # 查找搜索结果
|
|
|
+ result_divs = soup.find_all('div', class_='w-gl__result')
|
|
|
+
|
|
|
+ for div in result_divs[:limit]:
|
|
|
+ title_elem = div.find('h3')
|
|
|
+ link_elem = title_elem.find('a') if title_elem else None
|
|
|
+ snippet_elem = div.find('p', class_='w-gl__description')
|
|
|
+
|
|
|
+ if link_elem:
|
|
|
+ title = self._clean_text(link_elem.get_text())
|
|
|
+ url = link_elem.get('href', '')
|
|
|
+ snippet = self._clean_text(snippet_elem.get_text()) if snippet_elem else ''
|
|
|
+
|
|
|
+ if self._is_valid_result(title, url):
|
|
|
+ results.append({
|
|
|
+ 'title': title,
|
|
|
+ 'url': url,
|
|
|
+ 'snippet': snippet[:200],
|
|
|
+ 'source': 'Startpage'
|
|
|
+ })
|
|
|
+
|
|
|
+ return results
|
|
|
+
|
|
|
+ def _extract_qwant_results(self, soup, limit):
|
|
|
+ """从Qwant HTML中提取结果"""
|
|
|
+ results = []
|
|
|
+
|
|
|
+ # 查找搜索结果
|
|
|
+ result_divs = soup.find_all('div', class_='result')
|
|
|
+
|
|
|
+ for div in result_divs[:limit]:
|
|
|
+ title_elem = div.find('a', class_='result--web')
|
|
|
+ snippet_elem = div.find('p', class_='result__desc')
|
|
|
+
|
|
|
+ if title_elem:
|
|
|
+ title = self._clean_text(title_elem.get_text())
|
|
|
+ url = title_elem.get('href', '')
|
|
|
+ snippet = self._clean_text(snippet_elem.get_text()) if snippet_elem else ''
|
|
|
+
|
|
|
+ if self._is_valid_result(title, url):
|
|
|
+ results.append({
|
|
|
+ 'title': title,
|
|
|
+ 'url': url,
|
|
|
+ 'snippet': snippet[:200],
|
|
|
+ 'source': 'Qwant'
|
|
|
+ })
|
|
|
+
|
|
|
+ return results
|
|
|
+
|
|
|
+ def _extract_duckduckgo_results(self, soup, limit=5):
|
|
|
+ """提取DuckDuckGo搜索结果"""
|
|
|
+ results = []
|
|
|
+
|
|
|
+ # DuckDuckGo现在返回202状态码，需要JavaScript渲染
|
|
|
+ # 我们尝试从HTML中提取任何有用的信息
|
|
|
+
|
|
|
+ # 方法1：查找所有外部链接
|
|
|
+ all_links = soup.find_all('a', href=True)
|
|
|
+ external_links = []
|
|
|
+
|
|
|
+ for link in all_links:
|
|
|
+ href = link.get('href', '')
|
|
|
+ title = self._clean_text(link.get_text(strip=True))
|
|
|
+
|
|
|
+ # 过滤外部链接（非DuckDuckGo内部链接）
|
|
|
+ if (href and
|
|
|
+ not href.startswith('javascript:') and
|
|
|
+ not href.startswith('#') and
|
|
|
+ 'duckduckgo.com' not in href and
|
|
|
+ len(title) > 3 and
|
|
|
+ self._is_valid_result(title, href)):
|
|
|
+
|
|
|
+ external_links.append({
|
|
|
+ 'title': title,
|
|
|
+ 'url': href,
|
|
|
+ 'snippet': '',
|
|
|
+ 'link_element': link
|
|
|
+ })
|
|
|
+
|
|
|
+ # 方法2：如果外部链接不够，尝试从页面文本中提取信息
|
|
|
+ if len(external_links) < 2:
|
|
|
+ print("⚠️ 外部链接较少，尝试文本提取")
|
|
|
+
|
|
|
+ # 查找页面中的主要文本内容
|
|
|
+ text_content = soup.get_text()
|
|
|
+
|
|
|
+ # 尝试提取URL模式
|
|
|
+ import re
|
|
|
+ url_pattern = r'https?://[^\s<>"\'()]+'
|
|
|
+ urls = re.findall(url_pattern, text_content)
|
|
|
+
|
|
|
+ for url in urls[:limit]:
|
|
|
+ # 从URL中提取可能的标题
|
|
|
+ domain = url.split('/')[2] if '/' in url else url
|
|
|
+ title = domain.replace('www.', '').title()
|
|
|
+
|
|
|
+ if self._is_valid_result(title, url):
|
|
|
+ external_links.append({
|
|
|
+ 'title': title,
|
|
|
+ 'url': url,
|
|
|
+ 'snippet': f'来自 {domain}',
|
|
|
+ 'link_element': None
|
|
|
+ })
|
|
|
+
|
|
|
+ # 方法3：如果还是没有足够结果，提供搜索建议
|
|
|
+ if len(external_links) < 2:
|
|
|
+ print("⚠️ 搜索结果有限，提供搜索建议")
|
|
|
+
|
|
|
+ suggestions = [
|
|
|
+ {
|
|
|
+ 'title': f'在Google搜索 "{self.last_query}"',
|
|
|
+ 'url': f'https://www.google.com/search?q={self.last_query}',
|
|
|
+ 'snippet': '使用Google搜索引擎',
|
|
|
+ 'link_element': None
|
|
|
+ },
|
|
|
+ {
|
|
|
+ 'title': f'在Bing搜索 "{self.last_query}"',
|
|
|
+ 'url': f'https://www.bing.com/search?q={self.last_query}',
|
|
|
+ 'snippet': '使用Bing搜索引擎',
|
|
|
+ 'link_element': None
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ external_links.extend(suggestions)
|
|
|
+
|
|
|
+ # 去重并限制结果数量
|
|
|
+ seen_urls = set()
|
|
|
+ unique_results = []
|
|
|
+
|
|
|
+ for result in external_links:
|
|
|
+ if result['url'] and result['url'] not in seen_urls:
|
|
|
+ seen_urls.add(result['url'])
|
|
|
+ unique_results.append(result)
|
|
|
+ if len(unique_results) >= limit:
|
|
|
+ break
|
|
|
+
|
|
|
+ return unique_results
|
|
|
+
|
|
|
+ def _extract_content_from_url(self, url, max_length=300):
|
|
|
+ """从URL提取主要内容"""
|
|
|
+ try:
|
|
|
+ headers = {
|
|
|
+ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
|
|
+ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
|
+ "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
|
|
|
+ }
|
|
|
+
|
|
|
+ response = requests.get(url, headers=headers, timeout=10)
|
|
|
+ if response.status_code != 200:
|
|
|
+ return "内容获取失败"
|
|
|
+
|
|
|
+ soup = BeautifulSoup(response.text, 'html.parser')
|
|
|
+
|
|
|
+ # 移除脚本和样式标签
|
|
|
+ for script in soup(["script", "style", "nav", "footer", "header", "aside", "advertisement"]):
|
|
|
+ script.decompose()
|
|
|
+
|
|
|
+ # 智能内容提取策略
|
|
|
+ content = self._extract_main_content(soup)
|
|
|
+
|
|
|
+ if not content:
|
|
|
+ content = soup.get_text(strip=True)
|
|
|
+
|
|
|
+ # 清理和优化内容
|
|
|
+ content = self._clean_and_format_content(content)
|
|
|
+
|
|
|
+ return content[:max_length] + "..." if len(content) > max_length else content
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ return f"内容提取失败: {str(e)[:50]}"
|
|
|
+
|
|
|
+ def _extract_main_content(self, soup):
|
|
|
+ """智能提取页面主要内容"""
|
|
|
+ # 优先级策略：从最具体到最通用
|
|
|
+ extraction_strategies = [
|
|
|
+ # 1. 文章相关标签
|
|
|
+ ['article', 'main article', '.article-content', '.post-content'],
|
|
|
+ # 2. 主要内容区域
|
|
|
+ ['main', '.main', '.content', '.main-content'],
|
|
|
+ # 3. 常见内容类名
|
|
|
+ ['.entry-content', '.post-body', '.article-body', '.content-area'],
|
|
|
+ # 4. 通用容器
|
|
|
+ ['.container', '.wrapper', '.page-content'],
|
|
|
+ # 5. 最后尝试body
|
|
|
+ ['body']
|
|
|
+ ]
|
|
|
+
|
|
|
+ for strategy in extraction_strategies:
|
|
|
+ for selector in strategy:
|
|
|
+ element = soup.select_one(selector)
|
|
|
+ if element:
|
|
|
+ content = element.get_text(strip=True)
|
|
|
+ # 验证内容质量
|
|
|
+ if self._is_quality_content(content):
|
|
|
+ return content
|
|
|
+
|
|
|
+ return ""
|
|
|
+
|
|
|
+ def _is_quality_content(self, content):
|
|
|
+ """验证内容质量"""
|
|
|
+ if not content or len(content) < 50:
|
|
|
+ return False
|
|
|
+
|
|
|
+ # 过滤导航和菜单内容
|
|
|
+ nav_keywords = ['导航', '菜单', '首页', '登录', '注册', '搜索', '联系', '关于', 'privacy', 'terms', 'home', 'login', 'register', 'contact', 'about']
|
|
|
+ content_lower = content.lower()
|
|
|
+
|
|
|
+ for keyword in nav_keywords:
|
|
|
+ if keyword in content_lower:
|
|
|
+ return False
|
|
|
+
|
|
|
+ # 检查是否包含有意义的句子
|
|
|
+ sentences = content.split('。')
|
|
|
+ meaningful_sentences = [s.strip() for s in sentences if len(s.strip()) > 10]
|
|
|
+
|
|
|
+ return len(meaningful_sentences) >= 2
|
|
|
+
|
|
|
+ def _clean_and_format_content(self, content):
|
|
|
+ """清理和格式化内容"""
|
|
|
+ if not content:
|
|
|
+ return ""
|
|
|
+
|
|
|
+ # 移除多余空白
|
|
|
+ content = re.sub(r'\s+', ' ', content.strip())
|
|
|
+
|
|
|
+ # 移除特殊字符，保留中文标点
|
|
|
+ content = re.sub(r'[^\w\s\u4e00-\u9fff.,!?;:()[\]{}"\'。，！？：；（）【】""''-]', '', content)
|
|
|
+
|
|
|
+ # 移除重复的换行和空格
|
|
|
+ content = re.sub(r'\n\s*\n', '\n', content)
|
|
|
+ content = re.sub(r' {2,}', ' ', content)
|
|
|
+
|
|
|
+ # 提取前几个有意义的句子
|
|
|
+ sentences = re.split(r'[。！？.!?]', content)
|
|
|
+ meaningful_sentences = []
|
|
|
+
|
|
|
+ for sentence in sentences:
|
|
|
+ sentence = sentence.strip()
|
|
|
+ if len(sentence) > 10 and len(sentence) < 100: # 合理的句子长度
|
|
|
+ meaningful_sentences.append(sentence)
|
|
|
+ if len(meaningful_sentences) >= 3: # 最多3个句子
|
|
|
+ break
|
|
|
+
|
|
|
+ return '。'.join(meaningful_sentences)
|
|
|
+
|
|
|
+ def _enhance_search_results(self, results, limit=3):
|
|
|
+ """增强搜索结果，提取内容预览"""
|
|
|
+ enhanced_results = []
|
|
|
+
|
|
|
+ for i, result in enumerate(results):
|
|
|
+ if i >= limit: # 只增强前几个结果
|
|
|
+ break
|
|
|
+
|
|
|
+ if result['url'] and result['url'].startswith('http'):
|
|
|
+ print(f"📄 提取内容: {result['title'][:30]}...")
|
|
|
+ content = self._extract_content_from_url(result['url'])
|
|
|
+ result['snippet'] = content
|
|
|
+ result['enhanced'] = True
|
|
|
+ else:
|
|
|
+ result['enhanced'] = False
|
|
|
+
|
|
|
+ enhanced_results.append(result)
|
|
|
+
|
|
|
+ # 添加未增强的结果
|
|
|
+ enhanced_results.extend(results[limit:])
|
|
|
+
|
|
|
+ return enhanced_results
|
|
|
+
|
|
|
+ def _fallback_extraction(self, soup, limit=5):
|
|
|
+ """备用结果提取方法"""
|
|
|
+ results = []
|
|
|
+
|
|
|
+ # 方法1：提取标题元素
|
|
|
+ for tag in ["h1", "h2", "h3", "h4"]:
|
|
|
+ elements = soup.find_all(tag)
|
|
|
+ for elem in elements:
|
|
|
+ if len(results) >= limit:
|
|
|
+ break
|
|
|
+
|
|
|
+ title = self._clean_text(elem.get_text(strip=True))
|
|
|
+ if self._is_valid_result(title, ""):
|
|
|
+ results.append({
|
|
|
+ "title": title,
|
|
|
+ "url": "",
|
|
|
+ "snippet": ""
|
|
|
+ })
|
|
|
+
|
|
|
+ # 方法2：提取文本块
|
|
|
+ if not results:
|
|
|
+ text_blocks = soup.get_text().split('\n')
|
|
|
+ for block in text_blocks:
|
|
|
+ if len(results) >= limit:
|
|
|
+ break
|
|
|
+
|
|
|
+ block = self._clean_text(block)
|
|
|
+ if len(block) > 20 and len(block) < 150:
|
|
|
+ results.append({
|
|
|
+ "title": block,
|
|
|
+ "url": "",
|
|
|
+ "snippet": ""
|
|
|
+ })
|
|
|
+
|
|
|
+ return results
|
|
|
+
|
|
|
+ def run(self, parameters):
|
|
|
+ # 确保参数处理的安全性
|
|
|
+ if isinstance(parameters, dict):
|
|
|
+ query = parameters.get("input", "")
|
|
|
+ else:
|
|
|
+ query = str(parameters) if parameters else ""
|
|
|
+
|
|
|
+ # 参数验证
|
|
|
+ if not query or not query.strip():
|
|
|
+ return "错误：搜索关键词不能为空"
|
|
|
+
|
|
|
+ query = query.strip()
|
|
|
+ self.last_query = query # 保存查询用于建议
|
|
|
+ limit = 5 # 增加结果数量
|
|
|
+
|
|
|
+ # URL 编码查询参数
|
|
|
+ encoded_query = quote_plus(query)
|
|
|
+ url = f"https://duckduckgo.com/html/?q={encoded_query}"
|
|
|
+
|
|
|
+ # 使用更真实的User-Agent
|
|
|
+ headers = {
|
|
|
+ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
|
|
+ "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
|
|
|
+ "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
|
|
|
+ "Accept-Encoding": "gzip, deflate, br",
|
|
|
+ "DNT": "1",
|
|
|
+ "Connection": "keep-alive",
|
|
|
+ "Upgrade-Insecure-Requests": "1"
|
|
|
+ }
|
|
|
+
|
|
|
+ # 检测是否为中文查询
|
|
|
+ is_chinese = any('\u4e00' <= char <= '\u9fff' for char in query)
|
|
|
+
|
|
|
+ # 对于中文搜索，直接使用Searx搜索引擎，跳过DuckDuckGo（避免202问题）
|
|
|
+ if is_chinese:
|
|
|
+ print(f"🌐 检测到中文查询，使用多引擎搜索策略...")
|
|
|
+ searx_results, searx_success = self._search_searx(query, limit)
|
|
|
+
|
|
|
+ if searx_success and searx_results:
|
|
|
+ results = searx_results
|
|
|
+ search_engine = "Searx多引擎"
|
|
|
+ print(f"✅ 中文搜索成功，找到 {len(results)} 个结果")
|
|
|
+ else:
|
|
|
+ # 如果Searx失败，提供搜索建议
|
|
|
+ print("⚠️ 所有搜索引擎失败，提供搜索建议")
|
|
|
+ results = self._get_search_suggestions(query)
|
|
|
+ search_engine = "搜索建议"
|
|
|
+ else:
|
|
|
+ # 英文搜索：先尝试DuckDuckGo，失败后使用Searx
|
|
|
+ max_retries = 2 # 减少重试次数，快速切换到Searx
|
|
|
+ duckduckgo_success = False
|
|
|
+
|
|
|
+ for attempt in range(max_retries):
|
|
|
+ try:
|
|
|
+ print(f"🔍 尝试DuckDuckGo搜索: {query} (尝试 {attempt + 1}/{max_retries})")
|
|
|
+
|
|
|
+ response = requests.get(url, headers=headers, timeout=10)
|
|
|
+
|
|
|
+ # DuckDuckGo经常返回202，直接跳过
|
|
|
+ if response.status_code == 202:
|
|
|
+ print("⚠️ DuckDuckGo返回202（需要JavaScript），切换到Searx...")
|
|
|
+ break
|
|
|
+
|
|
|
+ if response.status_code != 200:
|
|
|
+ if attempt < max_retries - 1:
|
|
|
+ time.sleep(1)
|
|
|
+ continue
|
|
|
+ break
|
|
|
+
|
|
|
+ # 检查响应内容
|
|
|
+ if len(response.text) < 1000:
|
|
|
+ if attempt < max_retries - 1:
|
|
|
+ time.sleep(1)
|
|
|
+ continue
|
|
|
+ break
|
|
|
+
|
|
|
+ soup = BeautifulSoup(response.text, "html.parser")
|
|
|
+ results = self._extract_duckduckgo_results(soup, limit)
|
|
|
+
|
|
|
+ if results and len(results) > 0:
|
|
|
+ duckduckgo_success = True
|
|
|
+ search_engine = "DuckDuckGo"
|
|
|
+ print(f"✅ DuckDuckGo搜索成功，找到 {len(results)} 个结果")
|
|
|
+ break
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ print(f"⚠️ DuckDuckGo尝试失败: {str(e)[:50]}")
|
|
|
+ if attempt < max_retries - 1:
|
|
|
+ time.sleep(1)
|
|
|
+ continue
|
|
|
+ break
|
|
|
+
|
|
|
+ # 如果DuckDuckGo失败，使用Searx
|
|
|
+ if not duckduckgo_success:
|
|
|
+ print("🌐 DuckDuckGo失败，切换到Searx搜索引擎...")
|
|
|
+ searx_results, searx_success = self._search_searx(query, limit)
|
|
|
+
|
|
|
+ if searx_success and searx_results:
|
|
|
+ results = searx_results
|
|
|
+ search_engine = "Searx多引擎"
|
|
|
+ print(f"✅ Searx搜索成功，找到 {len(results)} 个结果")
|
|
|
+ else:
|
|
|
+ print("⚠️ 所有搜索引擎失败，提供搜索建议")
|
|
|
+ results = self._get_search_suggestions(query)
|
|
|
+ search_engine = "搜索建议"
|
|
|
+
|
|
|
+ # 增强搜索结果（提取内容预览）
|
|
|
+ if results:
|
|
|
+ print("🚀 增强搜索结果，提取内容预览...")
|
|
|
+ enhanced_results = self._enhance_search_results(results, limit=3)
|
|
|
+ results = enhanced_results
|
|
|
+
|
|
|
+ # 格式化输出结果
|
|
|
+ if results:
|
|
|
+ formatted_results = []
|
|
|
+ for i, result in enumerate(results, 1):
|
|
|
+ result_text = f"{i}. {result['title']}"
|
|
|
+
|
|
|
+ if result['url']:
|
|
|
+ result_text += f"\n 🔗 {result['url']}"
|
|
|
+
|
|
|
+ if result['snippet']:
|
|
|
+ # 如果是增强的结果，显示内容预览
|
|
|
+ if result.get('enhanced'):
|
|
|
+ result_text += f"\n 📄 内容预览: {result['snippet']}"
|
|
|
+ else:
|
|
|
+ result_text += f"\n 📝 {result['snippet']}"
|
|
|
+
|
|
|
+ formatted_results.append(result_text)
|
|
|
+
|
|
|
+ return "\n\n".join(formatted_results)
|
|
|
+ else:
|
|
|
+ return f"未找到关于 '{query}' 的搜索结果。请尝试使用不同的关键词。"
|
|
|
+
|
|
|
+ def _get_search_suggestions(self, query):
|
|
|
+ """快速提供搜索建议"""
|
|
|
+ return [
|
|
|
+ {
|
|
|
+ 'title': f'Google搜索: {query}',
|
|
|
+ 'url': f'https://www.google.com/search?q={query}',
|
|
|
+ 'snippet': '使用Google搜索引擎',
|
|
|
+ 'source': 'Google'
|
|
|
+ },
|
|
|
+ {
|
|
|
+ 'title': f'Bing搜索: {query}',
|
|
|
+ 'url': f'https://www.bing.com/search?q={query}',
|
|
|
+ 'snippet': '使用Bing搜索引擎',
|
|
|
+ 'source': 'Bing'
|
|
|
+ }
|
|
|
+ ]
|
|
|
+
|
|
|
+ return "搜索失败，已多次重试。请稍后再试。"
|
haoye2