agent/tools/web_tools.py

"""
NeuraPrompt Agent — Web Tools v8.0 (Multi-Engine Fallback, No API Keys)
"""

import requests
import random
import time
import re
from urllib.parse import quote_plus, unquote
from bs4 import BeautifulSoup
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import logging

log = logging.getLogger("agent.tools.web.v8.0")

# ─────────────────────────────────────────────────────────────
# CONFIG
# ─────────────────────────────────────────────────────────────

TIMEOUT_SEARCH = 12
TIMEOUT_FETCH = 20
MAX_RETRIES = 3
BACKOFF_FACTOR = 1.5
MAX_RESULTS = 6

USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (X11; Linux x86_64; rv:126.0) Gecko/20100101 Firefox/126.0",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Edge/125.0.0.0",
]

SEARXNG_INSTANCES = [
    "https://search.sapti.me",
    "https://searx.be",
    "https://searx.tiekoetter.com",
    "https://searx.prvcy.eu",
]


# ─────────────────────────────────────────────────────────────
# SESSION BUILDER
# ─────────────────────────────────────────────────────────────

def _build_session() -> requests.Session:
    """Create a resilient session with retry logic."""
    session = requests.Session()
    
    retry_strategy = Retry(
        total=MAX_RETRIES,
        backoff_factor=BACKOFF_FACTOR,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["HEAD", "GET", "OPTIONS"],
        raise_on_status=False,
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy, pool_connections=10, pool_maxsize=10)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session


def _headers(referer: str = "") -> dict:
    """Generate realistic browser headers."""
    headers = {
        "User-Agent": random.choice(USER_AGENTS),
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
        "Accept-Language": "en-US,en;q=0.5",
        "Accept-Encoding": "gzip, deflate, br",
        "DNT": "1",
        "Connection": "keep-alive",
        "Upgrade-Insecure-Requests": "1",
        "Sec-Fetch-Dest": "document",
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-Site": "none" if not referer else "same-origin",
        "Sec-Fetch-User": "?1",
        "Cache-Control": "max-age=0",
    }
    if referer:
        headers["Referer"] = referer
    return headers


def _safe_request(session: requests.Session, url: str, **kwargs) -> requests.Response | None:
    """
    Make a request with full error isolation.
    Returns None on any failure so callers can fallback.
    """
    try:
        start = time.time()
        response = session.get(url, **kwargs)
        elapsed = time.time() - start
        
        log.debug(f"[{response.status_code}] {url[:80]}... ({elapsed:.2f}s)")
        
        if response.status_code == 200:
            return response
        elif response.status_code in (403, 429):
            log.warning(f"Blocked [{response.status_code}]: {url[:80]}")
        else:
            log.warning(f"HTTP {response.status_code}: {url[:80]}")
            
    except requests.exceptions.SSLError as e:
        log.warning(f"SSL error for {url[:60]}: {e}")
    except requests.exceptions.ProxyError as e:
        log.warning(f"Proxy error for {url[:60]}: {e}")
    except requests.exceptions.ConnectionError as e:
        log.warning(f"Connection failed for {url[:60]}: {e}")
    except requests.exceptions.Timeout:
        log.warning(f"Timeout for {url[:60]}")
    except requests.exceptions.TooManyRedirects:
        log.warning(f"Redirect loop for {url[:60]}")
    except requests.exceptions.RequestException as e:
        log.warning(f"Request failed for {url[:60]}: {e}")
    except Exception as e:
        log.warning(f"Unexpected error for {url[:60]}: {e}")
    
    return None


# ─────────────────────────────────────────────────────────────
# SEARCH ENGINES
# ─────────────────────────────────────────────────────────────

def _search_ddg_lite(session: requests.Session, query: str) -> list[dict] | None:
    """Engine 1: DuckDuckGo Lite (lowest blocking)."""
    url = f"https://lite.duckduckgo.com/lite/?q={quote_plus(query)}"
    
    response = _safe_request(
        session, url,
        headers=_headers("https://duckduckgo.com/"),
        timeout=TIMEOUT_SEARCH,
        allow_redirects=True
    )
    if not response:
        return None
    
    soup = BeautifulSoup(response.text, "lxml")
    results = []
    
    # DDG Lite uses table.result rows
    for tr in soup.select("table.result")[:MAX_RESULTS]:
        link = tr.select_one("a.result-link")
        if not link:
            continue
            
        title = link.get_text(strip=True)
        href = link.get("href", "")
        # DDG Lite sometimes uses relative or redirect URLs
        if href.startswith("/"):
            href = f"https://lite.duckduckgo.com{href}"
        elif href.startswith("//"):
            href = f"https:{href}"
            
        snippet_el = tr.select_one("td.result-snippet")
        snippet = snippet_el.get_text(strip=True) if snippet_el else ""
        
        if title and href:
            results.append({
                "title": title,
                "url": href,
                "snippet": snippet,
                "engine": "ddg-lite"
            })
    
    return results if results else None


def _search_ddg_api(session: requests.Session, query: str) -> list[dict] | None:
    """Engine 2: DuckDuckGo Instant Answer API (JSON, no scraping)."""
    url = f"https://api.duckduckgo.com/?q={quote_plus(query)}&format=json&no_html=1&skip_disambig=1"
    
    response = _safe_request(
        session, url,
        headers=_headers(),
        timeout=TIMEOUT_SEARCH
    )
    if not response:
        return None
    
    try:
        data = response.json()
    except Exception:
        return None
    
    results = []
    
    # Main abstract answer
    abstract = (data.get("AbstractText") or data.get("Answer") or "").strip()
    abstract_url = data.get("AbstractURL", "")
    if abstract and abstract_url:
        results.append({
            "title": data.get("Heading", "Quick Answer"),
            "url": abstract_url,
            "snippet": abstract,
            "engine": "ddg-api"
        })
    
    # Related topics
    for topic in data.get("RelatedTopics", [])[:MAX_RESULTS - 1]:
        text = topic.get("Text", "").strip()
        first_url = topic.get("FirstURL", "")
        if text and first_url:
            results.append({
                "title": text.split(" - ")[0] if " - " in text else text[:60],
                "url": first_url,
                "snippet": text,
                "engine": "ddg-api"
            })
    
    return results if results else None


def _search_bing(session: requests.Session, query: str) -> list[dict] | None:
    """Engine 3: Bing HTML (moderate blocking, rich results)."""
    url = f"https://www.bing.com/search?q={quote_plus(query)}"
    
    response = _safe_request(
        session, url,
        headers=_headers("https://www.bing.com/"),
        timeout=TIMEOUT_SEARCH + 3,
        allow_redirects=True
    )
    if not response:
        return None
    
    soup = BeautifulSoup(response.text, "lxml")
    results = []
    
    # Bing uses li.b_algo for organic results
    for li in soup.select("li.b_algo")[:MAX_RESULTS]:
        a = li.select_one("a")
        if not a:
            continue
            
        title = a.get_text(strip=True)
        href = a.get("href", "")
        
        # Extract snippet from various Bing structures
        snippet = ""
        for sel in ["p", ".b_caption p", "div.b_attribution+div", ".b_snippet"]:
            el = li.select_one(sel)
            if el:
                snippet = el.get_text(strip=True)
                break
        
        if title and href and href.startswith("http"):
            results.append({
                "title": title,
                "url": href,
                "snippet": snippet,
                "engine": "bing"
            })
    
    return results if results else None


def _search_searxng(session: requests.Session, query: str) -> list[dict] | None:
    """Engine 4: SearXNG public instances (meta-search, JSON output)."""
    random.shuffle(SEARXNG_INSTANCES)
    
    for base in SEARXNG_INSTANCES[:3]:  # Try 3 random instances
        url = f"{base}/search?q={quote_plus(query)}&format=json&language=en"
        
        response = _safe_request(
            session, url,
            headers=_headers(base),
            timeout=TIMEOUT_SEARCH + 5
        )
        if not response:
            continue
        
        try:
            data = response.json()
        except Exception:
            continue
        
        results = []
        for r in data.get("results", [])[:MAX_RESULTS]:
            title = r.get("title", "").strip()
            href = r.get("url", "").strip()
            snippet = r.get("content", r.get("snippet", "")).strip()
            
            if title and href:
                results.append({
                    "title": title,
                    "url": href,
                    "snippet": snippet,
                    "engine": f"searxng-{base.split('//')[1].split('.')[0]}"
                })
        
        if results:
            return results
    
    return None


# ─────────────────────────────────────────────────────────────
# PUBLIC API
# ─────────────────────────────────────────────────────────────

def web_search(query: str) -> str:
    """
    Search the web using multiple no-key engines with automatic fallback.
    
    Priority: DDG Lite → DDG API → Bing → SearXNG
    """
    if not query or not query.strip():
        return "Error: search query cannot be empty."
    
    query = query.strip()
    session = _build_session()
    
    engines = [
        ("DuckDuckGo Lite", _search_ddg_lite),
        ("DuckDuckGo API", _search_ddg_api),
        ("Bing", _search_bing),
        ("SearXNG", _search_searxng),
    ]
    
    all_errors = []
    
    for name, engine_func in engines:
        try:
            log.info(f"Trying {name} for: {query[:50]}...")
            results = engine_func(session, query)
            
            if results:
                formatted = []
                for r in results:
                    line = f"• {r['title']}"
                    if r.get("snippet"):
                        line += f"\n{r['snippet']}"
                    line += f"\n🔗 {r['url']}"
                    formatted.append(line)
                
                footer = f"\n\n[Results via {name} | {len(results)} found]"
                return "\n\n".join(formatted) + footer
                
        except Exception as e:
            log.error(f"Critical error in {name}: {e}")
            all_errors.append(f"{name}: {str(e)}")
            continue
    
    # All engines failed
    error_detail = " | ".join(all_errors) if all_errors else "All engines returned no results."
    return (
        f"Search failed for: '{query}'\n"
        f"All fallback engines exhausted.\n"
        f"Details: {error_detail}\n"
        f"Tip: Check your internet connection or try again later."
    )


def fetch_url(url: str) -> str:
    """
    Fetch a webpage with anti-blocking headers and content extraction.
    """
    if not url or not url.strip():
        return "Error: URL cannot be empty."
    
    url = url.strip()
    if not url.startswith(("http://", "https://")):
        return "Error: URL must start with http:// or https://"
    
    session = _build_session()
    
    response = _safe_request(
        session, url,
        headers=_headers(),
        timeout=TIMEOUT_FETCH,
        allow_redirects=True
    )
    
    if not response:
        return (
            f"Could not fetch URL: {url}\n"
            "The server may be blocking automated requests, or the connection failed.\n"
            "Tip: Try a different URL or check if the site requires JavaScript."
        )
    
    # Handle non-HTML content
    content_type = response.headers.get("content-type", "").lower()
    if "text/html" not in content_type:
        preview = response.text[:8000]
        return f"[Non-HTML content: {content_type}]\n\n{preview}"
    
    # Parse and clean HTML
    soup = BeautifulSoup(response.text, "lxml")
    
    # Remove noise elements
    for tag in soup(["script", "style", "nav", "header", "footer", "aside", 
                     "form", "iframe", "noscript", "svg", "canvas", 
                     "advertisement", ".ad", ".ads", ".cookie-banner"]):
        tag.decompose()
    
    # Strategy 1: Look for semantic content containers
    content_blocks = []
    for selector in ["article", "main", "[role='main']", ".content", ".post", ".entry"]:
        for el in soup.select(selector):
            text = el.get_text(" ", strip=True)
            if len(text) > 300:
                content_blocks.append(text)
    
    # Strategy 2: Fallback to headings + paragraphs
    if not content_blocks:
        for tag in soup.find_all(["h1", "h2", "h3", "h4", "p", "li", "td"]):
            text = tag.get_text(" ", strip=True)
            if len(text) > 30:
                content_blocks.append(text)
    
    # Deduplicate and join
    seen = set()
    final_blocks = []
    for block in content_blocks:
        # Simple dedup: skip if very similar to already-seen
        sig = block[:100].lower()
        if sig not in seen:
            seen.add(sig)
            final_blocks.append(block)
    
    text = "\n\n".join(final_blocks)
    
    if not text:
        return "No readable content found. The page may be JavaScript-rendered or heavily obfuscated."
    
    # Add metadata header
    title = ""
    title_tag = soup.find("title")
    if title_tag:
        title = title_tag.get_text(strip=True)
    
    header = f"Title: {title}\nURL: {url}\n{'='*60}\n\n" if title else f"URL: {url}\n{'='*60}\n\n"
    
    return (header + text[:10000]).strip()


# ─────────────────────────────────────────────────────────────
# TEST / DEBUG
# ─────────────────────────────────────────────────────────────

if __name__ == "__main__":
    # Quick self-test
    print("=" * 60)
    print("TEST: web_search('Python programming language')")
    print("=" * 60)
    print(web_search("Python programming language"))
    print("\n" + "=" * 60)
    print("TEST: fetch_url('https://en.wikipedia.org/wiki/Python_(programming_language)')")
    print("=" * 60)
    print(fetch_url("https://en.wikipedia.org/wiki/Python_(programming_language)")[:1500])