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
51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
"""PluginWebMixin — plugin web panel SDK"""
|
|
import logging, json, asyncio
|
|
from aiohttp import web
|
|
from typing import Dict, Callable
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class PluginWebMixin:
|
|
def __init__(self):
|
|
self._web_pages: Dict[str, dict] = {}
|
|
self._api_routes: list = []
|
|
self._sse_clients: list = []
|
|
|
|
def register_web_page(self, path: str, title: str, html_content: str, icon: str = "P"):
|
|
self._web_pages[path] = {"title": title, "icon": icon, "html": html_content}
|
|
logger.info(f"Web page registered: {path}")
|
|
|
|
def register_api_route(self, method: str, path: str, handler: Callable):
|
|
self._api_routes.append((method, path, handler))
|
|
|
|
def get_web_pages(self) -> dict:
|
|
return self._web_pages
|
|
|
|
def get_api_routes(self) -> list:
|
|
return self._api_routes
|
|
|
|
async def push_sse_event(self, event_type: str, data: dict):
|
|
payload = json.dumps(data)
|
|
dead = []
|
|
for client in self._sse_clients:
|
|
try:
|
|
await client.send(f"event: {event_type}\\ndata: {payload}\\n\\n")
|
|
except:
|
|
dead.append(client)
|
|
for d in dead:
|
|
self._sse_clients.remove(d)
|
|
|
|
async def handle_sse(self, request):
|
|
resp = web.StreamResponse()
|
|
resp.headers["Content-Type"] = "text/event-stream"
|
|
resp.headers["Cache-Control"] = "no-cache"
|
|
await resp.prepare(request)
|
|
self._sse_clients.append(resp)
|
|
try:
|
|
while True:
|
|
await asyncio.sleep(30)
|
|
await resp.write(b": keepalive\\n\\n")
|
|
except:
|
|
self._sse_clients.remove(resp)
|
|
return resp
|