How to Fix Hermes SearXNG Blocking with API Fallback

import os, requests, time, json
from datetime import datetime

SEARXNG_URL = os.environ.get('SEARXNG_URL', 'http://localhost:8888')
API_KEY = os.environ['SCAVIO_API_KEY']
SH = {'x-api-key': API_KEY, 'Content-Type': 'application/json'}

def check_searxng_health():
    """Check if SearXNG is responsive and returning results."""
    try:
        resp = requests.get(f'{SEARXNG_URL}/search',
            params={'q': 'test', 'format': 'json'}, timeout=5)
        if resp.status_code != 200:
            return {'healthy': False, 'reason': f'HTTP {resp.status_code}'}
        data = resp.json()
        if not data.get('results'):
            return {'healthy': False, 'reason': 'No results (likely blocked)'}
        return {'healthy': True, 'results': len(data['results'])}
    except requests.exceptions.ConnectionError:
        return {'healthy': False, 'reason': 'Connection refused (SearXNG down)'}
    except requests.exceptions.Timeout:
        return {'healthy': False, 'reason': 'Timeout (SearXNG overloaded)'}
    except Exception as e:
        return {'healthy': False, 'reason': str(e)[:50]}

def search_api_fallback(query, num_results=5):
    """Scavio API fallback when SearXNG fails."""
    resp = requests.post('https://api.scavio.dev/api/v1/search',
        headers=SH, json={'query': query, 'country_code': 'us'}, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    return [{'title': r.get('title', ''), 'url': r.get('link', ''),
             'content': r.get('snippet', '')} for r in data.get('organic_results', [])[:num_results]]

health = check_searxng_health()
print(f'SearXNG status: {"healthy" if health["healthy"] else "UNHEALTHY"}')
if not health['healthy']:
    print(f'  Reason: {health["reason"]}')
    print(f'  Fallback: Scavio API ready ($0.005/query)')

class SmartSearchRouter:
    def __init__(self):
        self.searxng_failures = 0
        self.last_check = 0
        self.check_interval = 300  # Re-check SearXNG every 5 min
        self.api_calls = 0
    
    def search(self, query, num_results=5):
        # Try SearXNG first if it has been healthy recently
        if self.searxng_failures < 3 or time.time() - self.last_check > self.check_interval:
            try:
                resp = requests.get(f'{SEARXNG_URL}/search',
                    params={'q': query, 'format': 'json', 'engines': 'google'},
                    timeout=5)
                if resp.status_code == 200:
                    data = resp.json()
                    results = data.get('results', [])
                    if results:
                        self.searxng_failures = 0
                        return {'provider': 'searxng', 'results': [
                            {'title': r.get('title', ''), 'url': r.get('url', ''),
                             'content': r.get('content', '')}
                            for r in results[:num_results]
                        ]}
                self.searxng_failures += 1
                self.last_check = time.time()
            except Exception:
                self.searxng_failures += 1
                self.last_check = time.time()
        # Fallback to API
        results = search_api_fallback(query, num_results)
        self.api_calls += 1
        return {'provider': 'scavio_api', 'results': results}

router = SmartSearchRouter()

# Simulate searches
for q in ['python async tutorial', 'FastAPI deployment', 'database migration tools']:
    result = router.search(q)
    print(f'  [{result["provider"]:12}] {q:30} | {len(result["results"])} results')

print(f'\nAPI fallback calls: {router.api_calls}')
print(f'API cost so far: ${router.api_calls * 0.005:.3f}')

def patch_hermes_search():
    """Replace Hermes SearXNG search with smart router."""
    router = SmartSearchRouter()
    
    def hermes_search_tool(query, num_results=5):
        """Drop-in replacement for Hermes search tool."""
        result = router.search(query, num_results)
        # Format in SearXNG response style for compatibility
        return {
            'results': result['results'],
            'number_of_results': len(result['results']),
            '_provider': result['provider'],
        }
    
    return hermes_search_tool

# Apply the patch
search = patch_hermes_search()

# Test with simulated SearXNG failure
print('Testing with SearXNG down...')
result = search('how to fix SearXNG blocking')
print(f'  Provider: {result["_provider"]}')
print(f'  Results: {result["number_of_results"]}')
for r in result['results'][:3]:
    print(f'    {r["title"][:50]}')

print(f'\n  SearXNG blocking fix:')
print(f'    Problem: SearXNG gets blocked by Google, returns 0 results')
print(f'    Solution: Smart router falls back to Scavio API')
print(f'    Cost: $0.005/query (only when SearXNG is down)')
print(f'    Latency: ~300ms API vs 2-5s SearXNG')