b973635445
SearXNG instance at 192.168.122.119 is reachable via grokbox static route -- no need to tunnel through SOCKS5. Reverts searx and alert plugins to stdlib urlopen for SearXNG queries. YouTube and Twitch in alert.py still use the proxy. Also removes cprofile flag from docker-compose command. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
95 lines
2.6 KiB
Python
95 lines
2.6 KiB
Python
"""Plugin: SearXNG web search."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
from derp.plugin import command
|
|
|
|
# -- Constants ---------------------------------------------------------------
|
|
|
|
_SEARX_URL = "http://192.168.122.119:3000/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 = urllib.request.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}")
|