HitonabiandClaude Opus 5.5
292a9d2b6b
Ampel / ampel (push) Failing after 21s
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>
68 lines
2.6 KiB
Python
68 lines
2.6 KiB
Python
"""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()
|