hermes: web_extract liest Seiten lokal (Plugin mc2-web-lesen) statt jeden Morgen zu scheitern

Die Box hat als Web-Anbieter nur DuckDuckGo, der kann nicht lesen. Hermes bot web_extract trotzdem an, jeder Aufruf scheiterte, und der Waechter meldete den Nachrichten-Job jeden Morgen gelb. Das Plugin nutzt Hermes' offizielle Anbieter-Schnittstelle (kein Eingriff in den Hermes-Code) und liest wie MC2s fetch_url mit httpx und trafilatura, ohne fremden Dienst. deploy.sh kopiert Hermes-Plugins aus dem Repo; aktiviert wird einmalig von Hand. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-24 12:41:52 +02:00
parent 8d46adfd86
commit 292a9d2b6b
4 changed files with 185 additions and 0 deletions
@@ -0,0 +1,67 @@
 """Hermes-Plugin mc2-web-lesen (deploy/hermes-plugins): liest Seiten lokal für web_extract.
 Das Plugin läuft in Hermes' eigener Umgebung. Hier wird die Hermes-Schnittstelle durch eine
 Attrappe ersetzt und das Netz durch httpx.MockTransport — ohne Hermes, ohne Internet.
 """
 import abc
 import asyncio
 import importlib.util
 import sys
 import types
 from pathlib import Path
 import httpx
 import pytest
 PLUGIN = Path(__file__).resolve().parents[2] / "deploy" / "hermes-plugins" / "mc2-web-lesen" / "__init__.py"
@pytest.fixture
 def plugin(monkeypatch):
    class WebSearchProvider(abc.ABC):   # Attrappe der Hermes-Schnittstelle
        @property
        @abc.abstractmethod
        def name(self) -> str: ...
    agent = types.ModuleType("agent")
    schnittstelle = types.ModuleType("agent.web_search_provider")
    schnittstelle.WebSearchProvider = WebSearchProvider
    monkeypatch.setitem(sys.modules, "agent", agent)
    monkeypatch.setitem(sys.modules, "agent.web_search_provider", schnittstelle)
    spec = importlib.util.spec_from_file_location("mc2_web_lesen", PLUGIN)
    modul = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(modul)
    return modul
 SEITE = ("<html><head><title>Box-News</title><script>var x = 1;</script></head>"
         "<body><article><h1>Neues Modell</h1><p>Die Box testet nachts selbst.</p></article></body></html>")
 def _lies(plugin, antwort: httpx.Response, url: str = "https://example.org/a") -> dict:
    async def lauf():
        async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: antwort)) as client:
            return await plugin.lies(client, url)
    return asyncio.run(lauf())
 def test_liest_titel_und_haupttext(plugin):
    ergebnis = _lies(plugin, httpx.Response(200, headers={"content-type": "text/html; charset=utf-8"}, text=SEITE))
    assert ergebnis["url"] == "https://example.org/a" and ergebnis["title"] == "Box-News"
    assert "testet nachts selbst" in ergebnis["content"]
    assert "var x" not in ergebnis["content"] and "error" not in ergebnis
 def test_fehler_werden_je_seite_gemeldet_statt_geworfen(plugin):
    assert _lies(plugin, httpx.Response(404))["error"] == "HTTP 404"
    pdf = _lies(plugin, httpx.Response(200, headers={"content-type": "application/pdf"}, content=b"%PDF"))
    assert "Kein Text-Inhalt (application/pdf)" in pdf["error"]
 def test_anbieter_liest_nur_und_meldet_sich_an(plugin):
    angemeldet = []
    plugin.register(types.SimpleNamespace(register_web_search_provider=angemeldet.append))
    anbieter = angemeldet[0]
    assert anbieter.name == "mc2-lesen" and anbieter.is_available()
    assert anbieter.supports_extract() and not anbieter.supports_search()
@@ -66,6 +66,14 @@ main() {
      fi
  done
  # 4.6 Hermes-Plugins aus dem Repo (09/2026: mc2-web-lesen, Seiten lokal lesen für web_extract).
  # Nur kopieren: Aktiviert wird ein Plugin einmalig von Hand (hermes plugins enable …), und neuer
  # Plugin-Code greift erst beim nächsten Neustart des Hermes-Gateways — den macht deploy.sh bewusst nicht.
  mkdir -p ~/.hermes/plugins
  for plugin_dir in deploy/hermes-plugins/*; do
      [ -d "$plugin_dir" ] && cp -r "$plugin_dir" ~/.hermes/plugins/
  done
  # 5. Dienste neu starten
  # ‼️ mc2-steward gehoert hier dazu: er laeuft aus DEMSELBEN Checkout (backend/steward.py).
  # Ohne ihn lief der Waechter-Prozess nach jedem Deploy mit dem alten Code weiter — der
@@ -0,0 +1,103 @@
 """MC2: Webseiten lokal lesen — Anbieter für Hermes' Werkzeug web_extract (Box-Wart, 09/2026).
 Warum: Die Box hat als Web-Anbieter nur DuckDuckGo (ddgs). Der kann suchen, aber keine Seiten lesen.
 Hermes bot web_extract trotzdem an (die Prüfung fragt nur „gibt es irgendeinen Web-Anbieter?“), jeder
 Aufruf scheiterte mit „ddgs is a search-only backend“, und der Nachrichten-Job warf deshalb jeden Morgen
 einen Werkzeugfehler. Dieser Anbieter liest Seiten selbst: httpx lädt, trafilatura zieht den Haupttext
 heraus — dasselbe wie MC2s MCP-Werkzeug fetch_url. Kein fremder Dienst, kein Schlüssel, die Box bleibt lokal.
 Hermes-Quellcode bleibt unberührt: das hier ist ein Plugin über die offizielle Anbieter-Schnittstelle.
 Einbau (deploy.sh kopiert den Ordner nach ~/.hermes/plugins/, der Rest einmalig):
    hermes plugins enable mc2-web-lesen
    hermes config set web.extract_backend mc2-lesen
    systemctl --user restart hermes-gateway
 Rückweg: web.extract_backend wieder auf ddgs setzen und `hermes plugins disable mc2-web-lesen`.
 """
 from __future__ import annotations
 import re
 from typing import Any
 import httpx
 from agent.web_search_provider import WebSearchProvider
 KOPFZEILEN = {
    "User-Agent": ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
                   "Chrome/120.0.0.0 Safari/537.36"),
    "Accept-Language": "de,en;q=0.8",
 }
 ZEITLIMIT_S = 20
 MAX_ZEICHEN = 3_000_000       # rohes HTML; Hermes kürzt das Ergebnis danach selbst (web.extract_char_limit)
 def _fehler(url: str, grund: str) -> dict:
    return {"url": url, "title": "", "content": "", "error": grund}
 def _titel(html: str) -> str:
    m = re.search(r"<title[^>]*>(.*?)</title>", html, re.IGNORECASE | re.DOTALL)
    return re.sub(r"\s+", " ", m.group(1)).strip()[:200] if m else ""
 def haupttext(html: str) -> str:
    """Haupttext einer Seite: trafilatura, sonst grob ohne Skripte, Stile und Tags."""
    try:
        import trafilatura
        text = trafilatura.extract(html, include_comments=False, include_tables=True, no_fallback=False) or ""
    except Exception:   # trafilatura fehlt oder stolpert über die Seite → grober Rückfall
        text = ""
    if not text:
        ohne = re.sub(r"<(script|style|noscript)[^>]*>.*?</\1>", " ", html, flags=re.IGNORECASE | re.DOTALL)
        text = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", ohne)).strip()
    return text
 async def lies(client: httpx.AsyncClient, url: str) -> dict:
    try:
        antwort = await client.get(url)
        antwort.raise_for_status()
    except httpx.TimeoutException:
        return _fehler(url, f"Zeitüberschreitung nach {ZEITLIMIT_S} s")
    except httpx.HTTPStatusError as e:
        return _fehler(url, f"HTTP {e.response.status_code}")
    except httpx.HTTPError as e:
        return _fehler(url, f"Seite nicht erreichbar: {e}")
    art = antwort.headers.get("content-type", "").lower()
    if art and not any(t in art for t in ("html", "xml", "text/")):
        return _fehler(url, f"Kein Text-Inhalt ({art.split(';')[0]}), lokal nicht lesbar")
    html = antwort.text[:MAX_ZEICHEN]
    text = haupttext(html)
    if not text:
        return _fehler(url, "Kein Text auf der Seite gefunden")
    return {"url": url, "title": _titel(html), "content": text, "raw_content": text,
            "metadata": {"quelle": "mc2-lesen", "endadresse": str(antwort.url)}}
 class LokalLesen(WebSearchProvider):
    """Nur Lesen (extract), keine Suche — die Suche bleibt bei ddgs (web.search_backend)."""
    @property
    def name(self) -> str:
        return "mc2-lesen"
    @property
    def display_name(self) -> str:
        return "MC2 · Seiten lokal lesen"
    def is_available(self) -> bool:
        return True
    def supports_search(self) -> bool:
        return False
    def supports_extract(self) -> bool:
        return True
    async def extract(self, urls: list[str], **kwargs: Any) -> list[dict]:
        async with httpx.AsyncClient(headers=KOPFZEILEN, timeout=ZEITLIMIT_S, follow_redirects=True) as client:
            return [await lies(client, u) for u in urls]
 def register(ctx) -> None:
    ctx.register_web_search_provider(LokalLesen())
@@ -0,0 +1,7 @@
 name: mc2-web-lesen
 version: 1.0.0
 description: "Liest Webseiten lokal für web_extract (httpx + trafilatura) — ohne fremden Dienst und ohne API-Schlüssel."
 author: MC2 Box-Wart
 kind: backend
 provides_web_providers:
  - mc2-lesen