29e77f97b2
Both plugins called urllib.request.urlopen directly, bypassing the proxy. Switch to derp.http.urlopen and update the SearXNG endpoint to the public domain (searx.mymx.me). Update test mocks to match. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
96 lines
2.7 KiB
Python
96 lines
2.7 KiB
Python
"""Plugin: SearXNG web search."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
from derp.http import urlopen as _urlopen
|
|
from derp.plugin import command
|
|
|
|
# -- Constants ---------------------------------------------------------------
|
|
|
|
_SEARX_URL = "https://searx.mymx.me/search"
|
|
_FETCH_TIMEOUT = 10
|
|
_MAX_RESULTS = 3
|
|
_MAX_TITLE_LEN = 80
|
|
_MAX_QUERY_LEN = 200
|
|
|
|
|
|
# -- Pure helpers ------------------------------------------------------------
|
|
|
|
def _truncate(text: str, max_len: int = _MAX_TITLE_LEN) -> str:
|
|
"""Truncate text with ellipsis if needed."""
|
|
if len(text) <= max_len:
|
|
return text
|
|
return text[: max_len - 3].rstrip() + "..."
|
|
|
|
|
|
# -- SearXNG search (blocking) ----------------------------------------------
|
|
|
|
def _search(query: str) -> list[dict]:
|
|
"""Search SearXNG. Blocking.
|
|
|
|
Returns list of dicts with keys: title, url, snippet.
|
|
Raises on HTTP or parse errors.
|
|
"""
|
|
params = urllib.parse.urlencode({"q": query, "format": "json"})
|
|
url = f"{_SEARX_URL}?{params}"
|
|
|
|
req = urllib.request.Request(url, method="GET")
|
|
resp = _urlopen(req, timeout=_FETCH_TIMEOUT)
|
|
raw = resp.read()
|
|
resp.close()
|
|
|
|
data = json.loads(raw)
|
|
results: list[dict] = []
|
|
for item in data.get("results", []):
|
|
results.append({
|
|
"title": item.get("title", ""),
|
|
"url": item.get("url", ""),
|
|
"snippet": item.get("content", item.get("snippet", "")),
|
|
})
|
|
return results
|
|
|
|
|
|
# -- Command handler ---------------------------------------------------------
|
|
|
|
@command("searx", help="Search: !searx <query>")
|
|
async def cmd_searx(bot, message):
|
|
"""Search SearXNG and show top results.
|
|
|
|
Usage: !searx <query...>
|
|
"""
|
|
if not message.is_channel:
|
|
await bot.reply(message, "Use this command in a channel")
|
|
return
|
|
|
|
parts = message.text.split(None, 1)
|
|
if len(parts) < 2 or not parts[1].strip():
|
|
await bot.reply(message, "Usage: !searx <query>")
|
|
return
|
|
|
|
query = parts[1].strip()
|
|
if len(query) > _MAX_QUERY_LEN:
|
|
await bot.reply(message, f"Query too long (max {_MAX_QUERY_LEN} chars)")
|
|
return
|
|
|
|
import asyncio
|
|
loop = asyncio.get_running_loop()
|
|
|
|
try:
|
|
results = await loop.run_in_executor(None, _search, query)
|
|
except Exception as exc:
|
|
await bot.reply(message, f"Search failed: {exc}")
|
|
return
|
|
|
|
if not results:
|
|
await bot.reply(message, f"No results for: {query}")
|
|
return
|
|
|
|
for item in results[:_MAX_RESULTS]:
|
|
title = _truncate(item["title"]) if item["title"] else "(no title)"
|
|
url = item["url"]
|
|
await bot.reply(message, f"{title} -- {url}")
|