mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-19 18:47:41 +08:00
fix: support dynamic plugin web api routes
This commit is contained in:
@@ -72,6 +72,9 @@ class StarMetadata:
|
||||
i18n: dict[str, dict] = field(default_factory=dict)
|
||||
"""插件自带的国际化文案,按 locale 分组。"""
|
||||
|
||||
pages: list[dict] = field(default_factory=list)
|
||||
"""插件注册的 Pages 元数据。"""
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"Plugin {self.name} ({self.version}) by {self.author}: {self.desc}"
|
||||
|
||||
|
||||
@@ -519,6 +519,9 @@ class PluginManager:
|
||||
if isinstance(metadata.get("astrbot_version"), str)
|
||||
else None
|
||||
),
|
||||
pages=metadata["pages"]
|
||||
if isinstance(metadata.get("pages"), list)
|
||||
else [],
|
||||
i18n=PluginManager._load_plugin_i18n(plugin_path),
|
||||
)
|
||||
|
||||
|
||||
@@ -14,6 +14,8 @@ from hypercorn.asyncio import serve
|
||||
from hypercorn.config import Config as HyperConfig
|
||||
from quart import Quart, g, jsonify, request
|
||||
from quart.logging import default_handler
|
||||
from werkzeug.exceptions import MethodNotAllowed, NotFound
|
||||
from werkzeug.routing import Map, Rule
|
||||
|
||||
from astrbot.core import logger
|
||||
from astrbot.core.config.default import VERSION
|
||||
@@ -46,6 +48,43 @@ class _AddrWithPort(Protocol):
|
||||
APP: Quart
|
||||
|
||||
|
||||
def _normalize_plugin_api_route(route: str) -> str:
|
||||
route = route.strip()
|
||||
if not route.startswith("/"):
|
||||
route = f"/{route}"
|
||||
return route
|
||||
|
||||
|
||||
def _match_registered_web_api(registered_web_apis, subpath: str, method: str):
|
||||
request_path = f"/{subpath.lstrip('/')}"
|
||||
request_method = method.upper()
|
||||
|
||||
for route, view_handler, methods, _ in registered_web_apis:
|
||||
allowed_methods = [item.upper() for item in methods]
|
||||
if request_method not in allowed_methods:
|
||||
continue
|
||||
|
||||
url_map = Map(
|
||||
[
|
||||
Rule(
|
||||
_normalize_plugin_api_route(route),
|
||||
endpoint="plugin_api",
|
||||
methods=allowed_methods,
|
||||
),
|
||||
]
|
||||
)
|
||||
try:
|
||||
_, path_values = url_map.bind("").match(
|
||||
request_path,
|
||||
method=request_method,
|
||||
)
|
||||
except (MethodNotAllowed, NotFound):
|
||||
continue
|
||||
return view_handler, path_values
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _parse_env_bool(value: str | None, default: bool) -> bool:
|
||||
if value is None:
|
||||
return default
|
||||
@@ -155,10 +194,14 @@ class AstrBotDashboard:
|
||||
async def srv_plug_route(self, subpath, *args, **kwargs):
|
||||
"""插件路由"""
|
||||
registered_web_apis = self.core_lifecycle.star_context.registered_web_apis
|
||||
for api in registered_web_apis:
|
||||
route, view_handler, methods, _ = api
|
||||
if route == f"/{subpath}" and request.method in methods:
|
||||
return await view_handler(*args, **kwargs)
|
||||
matched_api = _match_registered_web_api(
|
||||
registered_web_apis,
|
||||
subpath,
|
||||
request.method,
|
||||
)
|
||||
if matched_api:
|
||||
view_handler, path_values = matched_api
|
||||
return await view_handler(*args, **{**kwargs, **path_values})
|
||||
return jsonify(Response().error("未找到该路由").__dict__)
|
||||
|
||||
async def auth_middleware(self):
|
||||
|
||||
@@ -14,7 +14,7 @@ from urllib.parse import parse_qs, urlsplit, urlunsplit
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from quart import Quart
|
||||
from quart import Quart, jsonify
|
||||
from werkzeug.datastructures import FileStorage
|
||||
|
||||
from astrbot.core import LogBroker
|
||||
@@ -44,8 +44,7 @@ def _strip_query(url: str) -> str:
|
||||
@pytest.fixture
|
||||
def registered_plugin_page(core_lifecycle_td: AstrBotCoreLifecycle, monkeypatch):
|
||||
plugin_root = (
|
||||
Path(core_lifecycle_td.plugin_manager.plugin_store_path)
|
||||
/ PLUGIN_PAGE_DEMO_NAME
|
||||
Path(core_lifecycle_td.plugin_manager.plugin_store_path) / PLUGIN_PAGE_DEMO_NAME
|
||||
)
|
||||
page_root = plugin_root / "pages" / PLUGIN_PAGE_DEMO_PAGE_NAME
|
||||
shared_root = page_root / "shared"
|
||||
@@ -230,6 +229,44 @@ async def test_auth_login_secure_cookie_override(
|
||||
assert "SameSite=Strict" in jwt_cookie_header
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_plugin_web_api_supports_dynamic_route(
|
||||
app: Quart,
|
||||
core_lifecycle_td: AstrBotCoreLifecycle,
|
||||
authenticated_header: dict[str, str],
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
):
|
||||
calls = []
|
||||
|
||||
async def group_detail(name: str):
|
||||
calls.append(name)
|
||||
return jsonify({"name": name})
|
||||
|
||||
monkeypatch.setattr(
|
||||
core_lifecycle_td.star_context,
|
||||
"registered_web_apis",
|
||||
[
|
||||
(
|
||||
f"/{PLUGIN_PAGE_DEMO_NAME}/groups/<name>",
|
||||
group_detail,
|
||||
["GET"],
|
||||
"Group detail",
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
test_client = app.test_client()
|
||||
response = await test_client.get(
|
||||
f"/api/plug/{PLUGIN_PAGE_DEMO_NAME}/groups/example",
|
||||
headers=authenticated_header,
|
||||
)
|
||||
data = await response.get_json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert data == {"name": "example"}
|
||||
assert calls == ["example"]
|
||||
|
||||
|
||||
def test_plugin_page_content_path_escapes_plugin_name():
|
||||
assert (
|
||||
PluginRoute._build_plugin_page_content_path("plugin with space", "main page")
|
||||
|
||||
@@ -106,9 +106,24 @@ def test_load_plugin_metadata_includes_i18n(tmp_path: Path):
|
||||
|
||||
assert metadata is not None
|
||||
assert metadata.short_desc == "Local test short description"
|
||||
assert metadata.pages == []
|
||||
assert metadata.i18n == {"zh-CN": {"metadata": {"display_name": "你好世界"}}}
|
||||
|
||||
|
||||
def test_load_plugin_metadata_includes_pages(tmp_path: Path):
|
||||
plugin_path = tmp_path / "helloworld"
|
||||
_write_local_test_plugin(plugin_path, TEST_PLUGIN_REPO)
|
||||
metadata_path = plugin_path / "metadata.yaml"
|
||||
metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
|
||||
metadata["pages"] = [{"name": "dashboard", "title": "Dashboard"}]
|
||||
metadata_path.write_text(yaml.dump(metadata), encoding="utf-8")
|
||||
|
||||
loaded_metadata = PluginManager._load_plugin_metadata(str(plugin_path))
|
||||
|
||||
assert loaded_metadata is not None
|
||||
assert loaded_metadata.pages == [{"name": "dashboard", "title": "Dashboard"}]
|
||||
|
||||
|
||||
def test_loaded_metadata_can_copy_i18n_into_existing_star_metadata(tmp_path: Path):
|
||||
plugin_path = tmp_path / "helloworld"
|
||||
_write_local_test_plugin(plugin_path, TEST_PLUGIN_REPO)
|
||||
|
||||
Reference in New Issue
Block a user