Spaces:
Runtime error
Runtime error
| """ | |
| 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 |