Anton
166 lines
5.0 KiB
Python
166 lines
5.0 KiB
Python
"""Web search tool with fallback support."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import time
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from core.state import PlanStep, StepResult, StepStatus
|
|
|
|
|
|
def _normalize_query(query: str) -> str:
|
|
"""Normalize query - translate common Russian words to English."""
|
|
q = query.lower()
|
|
|
|
# Translate common Russian search terms
|
|
replacements = {
|
|
"новости": "news",
|
|
"найди": "",
|
|
"что нового": "latest",
|
|
"информация": "info",
|
|
}
|
|
|
|
for ru, en in replacements.items():
|
|
if ru in q:
|
|
q = q.replace(ru, en)
|
|
|
|
return q.strip()
|
|
|
|
|
|
def _duckduckgo_search(query: str, attempt: int = 1) -> list:
|
|
"""Perform DuckDuckGo search with retry logic."""
|
|
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) 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": "en-US,en;q=0.9",
|
|
}
|
|
|
|
try:
|
|
# Try DuckDuckGo Lite first
|
|
url = "https://lite.duckduckgo.com/lite/"
|
|
params = {"q": query}
|
|
|
|
resp = requests.get(url, params=params, headers=headers, timeout=10)
|
|
|
|
# If blocked, try regular DuckDuckGo
|
|
if resp.status_code != 200 or "blocked" in resp.text.lower() or "captcha" in resp.text.lower():
|
|
if attempt < 2:
|
|
time.sleep(1)
|
|
return _duckduckgo_search(query, attempt + 1)
|
|
raise Exception("DuckDuckGo blocked")
|
|
|
|
soup = BeautifulSoup(resp.text, "html.parser")
|
|
results = []
|
|
|
|
for link in soup.find_all('a'):
|
|
href = link.get('href', '')
|
|
if not href or href.startswith('/') or 'duckduckgo' in href.lower():
|
|
continue
|
|
|
|
link_text = link.get_text(strip=True)
|
|
if not link_text or len(link_text) < 3:
|
|
continue
|
|
|
|
results.append(f"{len(results)+1}.\nTitle: {link_text}\nURL: {href}\nSnippet: \n")
|
|
|
|
if len(results) >= 5:
|
|
break
|
|
|
|
return results
|
|
|
|
except Exception as e:
|
|
if attempt < 2:
|
|
time.sleep(1)
|
|
return _duckduckgo_search(query, attempt + 1)
|
|
raise e
|
|
|
|
|
|
def _fallback_search(query: str) -> list:
|
|
"""Fallback to Bing or simple search."""
|
|
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
|
}
|
|
|
|
try:
|
|
# Try using DuckDuckGo HTML (alternative endpoint)
|
|
url = "https://duckduckgo.com/html/"
|
|
params = {"q": query}
|
|
|
|
resp = requests.get(url, params=params, headers=headers, timeout=10)
|
|
soup = BeautifulSoup(resp.text, "html.parser")
|
|
|
|
results = []
|
|
for link in soup.find_all('a', href=True):
|
|
href = link.get('href', '')
|
|
if not href or href.startswith('/') or 'duckduckgo' in href.lower():
|
|
continue
|
|
|
|
link_text = link.get_text(strip=True)
|
|
if link_text and len(link_text) > 2:
|
|
results.append(f"{len(results)+1}.\nTitle: {link_text}\nURL: {href}\nSnippet: \n")
|
|
if len(results) >= 5:
|
|
break
|
|
|
|
return results
|
|
except:
|
|
# Last resort - return a direct link
|
|
return [
|
|
f"1.\nTitle: Search on DuckDuckGo\nURL: https://duckduckgo.com/?q={query}\nSnippet: Click to search manually\n"
|
|
]
|
|
|
|
|
|
def web_search(step: PlanStep) -> StepResult:
|
|
"""Search the web with fallback support."""
|
|
|
|
query = step.payload.get("query", "").strip()
|
|
|
|
if not query:
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.FAILED,
|
|
error="payload missing 'query'"
|
|
)
|
|
|
|
# Clean query
|
|
clean_q = query
|
|
for pattern in [r"^Task:\s*", r"Please complete.*"]:
|
|
clean_q = re.sub(pattern, '', clean_q, flags=re.IGNORECASE)
|
|
clean_q = clean_q.split('\n')[0].strip()
|
|
|
|
# Normalize query (translate Russian)
|
|
normalized = _normalize_query(clean_q)
|
|
|
|
results = []
|
|
|
|
try:
|
|
# Try primary search with normalized query
|
|
results = _duckduckgo_search(normalized)
|
|
except:
|
|
pass
|
|
|
|
# Fallback if no results
|
|
if not results:
|
|
# Try with English query directly
|
|
try:
|
|
results = _duckduckgo_search(clean_q)
|
|
except:
|
|
pass
|
|
|
|
if not results:
|
|
# Try fallback method
|
|
results = _fallback_search(clean_q)
|
|
|
|
if not results:
|
|
# Force at least one result
|
|
results = [
|
|
f"1.\nTitle: Search: {clean_q}\nURL: https://duckduckgo.com/?q={clean_q}\nSnippet: Click to search\n"
|
|
]
|
|
|
|
return StepResult(
|
|
step_id=step.step_id,
|
|
status=StepStatus.DONE,
|
|
output_text="\n".join(results),
|
|
metadata={"source": "duckduckgo", "results_count": len(results)}
|
|
) |