AskaEth
e52f7f3566
- New: sdk/plugin_web.py (PluginWebMixin: web pages, API routes, SSE) - New: services/web_panel/routes/plugin_web.py (3 endpoints) - New: plugins/example_plugin/dashboard.html (counter + SSE demo) - Updated: example_plugin uses PluginWebMixin - Tests: 28/28 passing
39 lines
1.6 KiB
Python
39 lines
1.6 KiB
Python
from aiohttp import web
|
|
import json, logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def setup_plugin_web_routes(app, service_manager):
|
|
async def plugin_page(request):
|
|
name = request.match_info.get("name","")
|
|
ps = service_manager.get_service("plugin")
|
|
plugin = ps.plugins.get(name)
|
|
if not plugin or not hasattr(plugin, "get_web_pages"):
|
|
return web.Response(text=f"Plugin {name} not found", status=404)
|
|
pages = plugin.get_web_pages()
|
|
if name in pages:
|
|
return web.Response(text=pages[name]["html"], content_type="text/html")
|
|
return web.json_response({"error": "no web page"})
|
|
|
|
async def plugin_sse(request):
|
|
name = request.match_info.get("name","")
|
|
ps = service_manager.get_service("plugin")
|
|
plugin = ps.plugins.get(name)
|
|
if plugin and hasattr(plugin, "handle_sse"):
|
|
return await plugin.handle_sse(request)
|
|
return web.json_response({"error": "SSE not supported"}, status=404)
|
|
|
|
async def plugin_event(request):
|
|
name = request.match_info.get("name","")
|
|
data = await request.json()
|
|
ps = service_manager.get_service("plugin")
|
|
plugin = ps.plugins.get(name)
|
|
if plugin and hasattr(plugin, "push_sse_event"):
|
|
await plugin.push_sse_event(data.get("type","event"), data)
|
|
return web.json_response({"ok": True})
|
|
return web.json_response({"ok": False}, status=404)
|
|
|
|
app.router.add_get("/plugin/{name}", plugin_page)
|
|
app.router.add_get("/plugin/{name}/sse", plugin_sse)
|
|
app.router.add_post("/plugin/{name}/event", plugin_event)
|
|
logger.info("Plugin web routes registered")
|