Spaces:
Running
Running
File size: 3,284 Bytes
a31f556 | 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 | # backend/tasks/tools/web_search_tools.py
import asyncio
import json
import urllib.request
import urllib.parse
import os
async def search_web(query: str) -> dict:
"""Search web using Brave Search API (fallback to DuckDuckGo if no key)"""
brave_api_key = os.environ.get("BRAVE_API_KEY")
if not brave_api_key:
# Fallback to DuckDuckGo
def _duck_search():
url = f"https://api.duckduckgo.com/?q={urllib.parse.quote(query)}&format=json"
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
try:
with urllib.request.urlopen(req, timeout=10) as response:
data = json.loads(response.read().decode('utf-8'))
return {"results": [{"title": "Abstract", "snippet": data.get('AbstractText', ''), "url": data.get('AbstractURL', '')}]}
except Exception as e:
return {"error": str(e)}
return await asyncio.to_thread(_duck_search)
def _brave_search():
url = f"https://api.search.brave.com/res/v1/web/search?q={urllib.parse.quote(query)}"
req = urllib.request.Request(url, headers={
'Accept': 'application/json',
'Accept-Encoding': 'gzip',
'X-Subscription-Token': brave_api_key
})
try:
with urllib.request.urlopen(req, timeout=10) as response:
if response.info().get('Content-Encoding') == 'gzip':
import gzip
data = gzip.decompress(response.read()).decode('utf-8')
else:
data = response.read().decode('utf-8')
parsed = json.loads(data)
results = []
for item in parsed.get('web', {}).get('results', [])[:5]:
results.append({
"title": item.get("title"),
"snippet": item.get("description"),
"url": item.get("url")
})
return {"results": results}
except Exception as e:
return {"error": str(e)}
return await asyncio.to_thread(_brave_search)
async def fetch_page(url: str) -> dict:
"""Fetch page via Playwright, bypassing CAPTCHAs if encountered"""
from backend.tools.browser_tools import get_ephemeral_context, stealth_async
from backend.tools.captcha_solver import solve_captcha_if_present
try:
async with get_ephemeral_context() as context:
page = await context.new_page()
if stealth_async:
await stealth_async(page)
await page.goto(url, wait_until="domcontentloaded", timeout=15000)
await page.wait_for_timeout(2000)
# Auto-solve CAPTCHA if Cloudflare/Turnstile/hCaptcha hard-blocks the article
await solve_captcha_if_present(page)
# Extract main text
content = await page.evaluate("document.body.innerText")
await page.close()
return {"status": "success", "content": content[:5000]} # Limit to 5000 chars for context limits
except Exception as e:
return {"error": str(e)}
|