qinglong
45 lines
1.9 KiB
Python
45 lines
1.9 KiB
Python
from aiohttp import web
|
|
import json, logging
|
|
from ..utils.auth import panel_auth
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def setup_plugin_web_routes(app, service_manager, base_path=""):
|
|
prefix = base_path.rstrip("/")
|
|
|
|
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:
|
|
resp = web.Response(text=pages[name]["html"], content_type="text/html")
|
|
resp.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
|
|
return resp
|
|
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(f"{prefix}/plugin/{{name}}", panel_auth(plugin_page))
|
|
app.router.add_get(f"{prefix}/plugin/{{name}}/sse", panel_auth(plugin_sse))
|
|
app.router.add_post(f"{prefix}/plugin/{{name}}/event", panel_auth(plugin_event))
|
|
logger.info(f"Plugin web routes registered at {prefix}/plugin/* (已加认证)")
|