backend/tools/web_search_tools.py

# 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)}