Blame for plugins/search_engine.py

"""
plugins/search_engine.py — Web search via DuckDuckGo (free, no API key).
Used to give Phantom real-time information beyond training cutoff.
"""
import logging
import httpx

log = logging.getLogger(__name__)

_DDG_URL = "https://api.duckduckgo.com/"
_HEADERS = {"User-Agent": "ZyverixBot/1.0"}


async def search(query: str, max_results: int = 5) -> list[dict]:
    """
    Search DuckDuckGo and return top results.
    Returns list of {title, body, url}.
    Falls back to empty list on any error.
    """
    try:
        params = {
            "q": query, "format": "json",
            "no_html": "1", "skip_disambig": "1",
        }
        async with httpx.AsyncClient(timeout=8, headers=_HEADERS) as client:
            r = await client.get(_DDG_URL, params=params)
            data = r.json()

        results: list[dict] = []

        # Abstract (main result)
        if data.get("AbstractText"):
            results.append({
                "title": data.get("Heading", ""),
                "body":  data["AbstractText"],
                "url":   data.get("AbstractURL", ""),
            })

        # Related topics
        for item in data.get("RelatedTopics", []):
            if len(results) >= max_results:
                break
            if isinstance(item, dict) and item.get("Text"):
                results.append({
                    "title": item.get("Text", "")[:80],
                    "body":  item.get("Text", ""),
                    "url":   item.get("FirstURL", ""),
                })

        return results[:max_results]

    except (httpx.ConnectError, httpx.TimeoutException) as e:
        log.warning("DDG search network error: %s", e)
        return []
    except Exception as e:
        log.warning("DDG search failed: %s", e)
        return []


def format_results(results: list[dict]) -> str:
    """Format search results for injection into AI context."""
    if not results:
        return ""
    lines = ["[نتائج بحث الويب]"]
    for i, r in enumerate(results, 1):
        lines.append(f"{i}. {r['title']}: {r['body'][:200]}")
    return "\n".join(lines)


# Trigger words that indicate the user needs real-time info
SEARCH_TRIGGERS = {
    "ar": ["الآن", "اليوم", "أمس", "هذا الأسبوع", "أحدث", "آخر", "جديد",
           "أخبار", "سعر", "حالياً", "متى", "2025", "2026"],
    "en": ["now", "today", "latest", "current", "price", "news",
           "when", "recent", "2025", "2026"],
}


def needs_search(text: str) -> bool:
    """Detect if a message likely needs real-time web data."""
    text_lower = text.lower()
    for lang, triggers in SEARCH_TRIGGERS.items():
        if any(t in text_lower for t in triggers):
            return True
    return False