mirror of
https://github.com/insistence/RuoYi-Vue3-FastAPI.git
synced 2026-10-03 00:41:16 +00:00
693 lines
25 KiB
Python
693 lines
25 KiB
Python
import ast
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
from plugins.core.discovery.scanner import DiscoveredPlugin
|
||
from plugins.core.environment import PluginRuntimeEnvironmentService
|
||
from plugins.core.manifest.menu_tree import PluginMenuTree
|
||
from plugins.core.manifest.schema import PluginJobManifest, PluginManifest, PluginMenuManifest
|
||
from plugins.core.validation.result import PluginValidationIssue, PluginValidationLevelResolver, ValidationLevel
|
||
from utils.cron_util import CronUtil
|
||
|
||
MAX_PLUGIN_JOB_NAME_LENGTH = 64
|
||
SUPPORTED_SEED_SUFFIXES = {'.py', '.sql'}
|
||
SUPPORTED_MIGRATION_SUFFIXES = {'.py', '.sql'}
|
||
ROUTER_FACTORY_NAMES = {'APIRouter', 'APIRouterPro'}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PluginStructureCheckItem:
|
||
"""
|
||
插件结构检查项。
|
||
"""
|
||
|
||
kind: str
|
||
path: str
|
||
ok: bool
|
||
message: str
|
||
suggestion: str = ''
|
||
|
||
@property
|
||
def level(self) -> ValidationLevel:
|
||
"""
|
||
获取结构检查项等级。
|
||
|
||
:return: 检查等级
|
||
"""
|
||
return PluginValidationLevelResolver.from_ok(self.ok)
|
||
|
||
def to_issue(self) -> PluginValidationIssue:
|
||
"""
|
||
转换为统一校验问题项。
|
||
|
||
:return: 统一校验问题项
|
||
"""
|
||
return PluginValidationIssue(
|
||
level=self.level,
|
||
category='structure',
|
||
kind=self.kind,
|
||
path=self.path,
|
||
message=self.message,
|
||
suggestion=self.suggestion,
|
||
ok=self.ok,
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PluginStructureCheckResult:
|
||
"""
|
||
插件结构检查结果。
|
||
"""
|
||
|
||
plugin_id: str
|
||
items: list[PluginStructureCheckItem]
|
||
|
||
@property
|
||
def ok(self) -> bool:
|
||
"""
|
||
判断结构检查是否整体通过。
|
||
|
||
:return: 是否通过
|
||
"""
|
||
return all(item.ok for item in self.items)
|
||
|
||
@property
|
||
def failed_items(self) -> list[PluginStructureCheckItem]:
|
||
"""
|
||
获取失败检查项列表。
|
||
|
||
:return: 失败检查项列表
|
||
"""
|
||
return [item for item in self.items if not item.ok]
|
||
|
||
@property
|
||
def issues(self) -> list[PluginValidationIssue]:
|
||
"""
|
||
获取统一校验问题项列表。
|
||
|
||
:return: 统一校验问题项列表
|
||
"""
|
||
return [item.to_issue() for item in self.items]
|
||
|
||
|
||
class PluginStructureChecker:
|
||
"""
|
||
插件结构检查器。
|
||
|
||
使用 Checker 模式集中验证 manifest 引用的后端目录、seed 文件和前端页面是否存在。
|
||
"""
|
||
|
||
def __init__(self, backend_root: Path | str, frontend_root: Path | str | None = None) -> None:
|
||
"""
|
||
初始化插件结构检查器。
|
||
|
||
:param backend_root: 后端项目根目录
|
||
:param frontend_root: 前端插件根目录，默认使用运行时环境解析出的 plugins 目录
|
||
"""
|
||
self.backend_root = Path(backend_root)
|
||
self.frontend_root = (
|
||
Path(frontend_root)
|
||
if frontend_root
|
||
else Path(PluginRuntimeEnvironmentService(backend_root=self.backend_root).get_frontend_plugins_dir())
|
||
)
|
||
|
||
def check(
|
||
self,
|
||
discovered_plugin: DiscoveredPlugin,
|
||
*,
|
||
include_frontend: bool = True,
|
||
) -> PluginStructureCheckResult:
|
||
"""
|
||
检查插件结构。
|
||
|
||
:param discovered_plugin: 已发现插件对象
|
||
:param include_frontend: 是否检查前端源码目录和菜单组件
|
||
:return: 插件结构检查结果
|
||
"""
|
||
manifest = discovered_plugin.manifest
|
||
items = [
|
||
self._check_dir('backend_root', discovered_plugin.backend_path),
|
||
self._check_file('manifest', discovered_plugin.manifest_path),
|
||
]
|
||
if manifest.backend.routers.auto_scan:
|
||
controller_dir = discovered_plugin.backend_path / 'controller'
|
||
items.append(self._check_dir('controller_dir', controller_dir))
|
||
items.extend(self._check_controller_route_prefixes(discovered_plugin, controller_dir))
|
||
|
||
entity_dir = discovered_plugin.backend_path / 'entity' / 'do'
|
||
if entity_dir.exists():
|
||
items.append(self._check_dir('entity_do_dir', entity_dir))
|
||
|
||
items.extend(self._check_migration_files(discovered_plugin))
|
||
items.extend(self._check_seed_files(discovered_plugin))
|
||
items.extend(self._check_hooks(discovered_plugin))
|
||
items.extend(self._check_jobs(discovered_plugin))
|
||
if include_frontend:
|
||
items.extend(self._check_frontend(discovered_plugin))
|
||
|
||
return PluginStructureCheckResult(plugin_id=manifest.id, items=items)
|
||
|
||
def _check_controller_route_prefixes(
|
||
self,
|
||
discovered_plugin: DiscoveredPlugin,
|
||
controller_dir: Path,
|
||
) -> list[PluginStructureCheckItem]:
|
||
"""
|
||
检查插件 controller 路由前缀是否位于插件命名空间内。
|
||
|
||
:param discovered_plugin: 已发现插件对象
|
||
:param controller_dir: controller 目录
|
||
:return: 路由前缀检查项列表
|
||
"""
|
||
if not controller_dir.is_dir():
|
||
return []
|
||
|
||
items = []
|
||
for controller_file in sorted(controller_dir.glob('[!_]*.py')):
|
||
items.extend(self.check_controller_file_route_prefixes(discovered_plugin.manifest.id, controller_file))
|
||
|
||
return items
|
||
|
||
def check_controller_file_route_prefixes(
|
||
self,
|
||
plugin_id: str,
|
||
controller_file: Path,
|
||
) -> list[PluginStructureCheckItem]:
|
||
"""
|
||
检查单个 controller 文件内的路由前缀。
|
||
|
||
:param plugin_id: 插件ID
|
||
:param controller_file: controller 文件
|
||
:return: 路由前缀检查项列表
|
||
"""
|
||
try:
|
||
module_ast = ast.parse(controller_file.read_text(encoding='utf-8'), filename=str(controller_file))
|
||
except SyntaxError as exc:
|
||
return [
|
||
PluginStructureCheckItem(
|
||
kind='controller_route_prefix',
|
||
path=str(controller_file),
|
||
ok=False,
|
||
message=f'controller 文件语法错误，无法检查路由前缀：{exc}',
|
||
)
|
||
]
|
||
|
||
items = []
|
||
for node in ast.walk(module_ast):
|
||
if not isinstance(node, ast.Call) or not self._is_router_factory_call(node):
|
||
continue
|
||
|
||
prefix = self._get_router_prefix_literal(node)
|
||
path = f'{controller_file}:{getattr(node, "lineno", 1)}'
|
||
ok = prefix is not None and self._is_plugin_route_prefix(plugin_id, prefix)
|
||
items.append(
|
||
PluginStructureCheckItem(
|
||
kind='controller_route_prefix',
|
||
path=path,
|
||
ok=ok,
|
||
message=(
|
||
f'controller 路由前缀位于插件命名空间内：{prefix}'
|
||
if ok
|
||
else f'controller 路由前缀必须位于插件 {plugin_id} 命名空间内：{prefix or "<未声明>"}'
|
||
),
|
||
suggestion=f'请使用 /{plugin_id} 或 /plugin/{plugin_id} 作为路由前缀根路径',
|
||
)
|
||
)
|
||
|
||
return items
|
||
|
||
@staticmethod
|
||
def _is_router_factory_call(node: ast.Call) -> bool:
|
||
"""
|
||
判断 AST 调用是否为 APIRouter/APIRouterPro 构造。
|
||
|
||
:param node: AST 调用节点
|
||
:return: 是否为路由构造调用
|
||
"""
|
||
if isinstance(node.func, ast.Name):
|
||
return node.func.id in ROUTER_FACTORY_NAMES
|
||
if isinstance(node.func, ast.Attribute):
|
||
return node.func.attr in ROUTER_FACTORY_NAMES
|
||
return False
|
||
|
||
@staticmethod
|
||
def _get_router_prefix_literal(node: ast.Call) -> str | None:
|
||
"""
|
||
提取路由构造调用中的 prefix 字符串字面量。
|
||
|
||
:param node: AST 调用节点
|
||
:return: prefix 字符串，未声明或非字面量时返回 None
|
||
"""
|
||
for keyword in node.keywords:
|
||
if (
|
||
keyword.arg == 'prefix'
|
||
and isinstance(keyword.value, ast.Constant)
|
||
and isinstance(keyword.value.value, str)
|
||
):
|
||
return keyword.value.value
|
||
return None
|
||
|
||
@staticmethod
|
||
def _is_plugin_route_prefix(plugin_id: str, prefix: str) -> bool:
|
||
"""
|
||
判断路由前缀是否属于插件命名空间。
|
||
|
||
:param plugin_id: 插件ID
|
||
:param prefix: 路由前缀
|
||
:return: 是否属于插件命名空间
|
||
"""
|
||
plugin_prefix = f'/{plugin_id}'
|
||
nested_plugin_prefix = f'/plugin/{plugin_id}'
|
||
return prefix in (plugin_prefix, nested_plugin_prefix) or prefix.startswith(
|
||
(f'{plugin_prefix}/', f'{nested_plugin_prefix}/')
|
||
)
|
||
|
||
def _check_hooks(self, discovered_plugin: DiscoveredPlugin) -> list[PluginStructureCheckItem]:
|
||
"""
|
||
检查插件生命周期钩子声明。
|
||
|
||
:param discovered_plugin: 已发现插件对象
|
||
:return: 生命周期钩子检查项列表
|
||
"""
|
||
items = []
|
||
hooks = discovered_plugin.manifest.backend.hooks
|
||
hook_mapping = {
|
||
'on_install': hooks.on_install,
|
||
'on_upgrade': hooks.on_upgrade,
|
||
'on_startup': hooks.on_startup,
|
||
'on_shutdown': hooks.on_shutdown,
|
||
'on_purge': hooks.on_purge,
|
||
}
|
||
for hook_name, hook_path in hook_mapping.items():
|
||
if not hook_path:
|
||
continue
|
||
items.append(self._check_hook_boundary(discovered_plugin.manifest.backend.module, hook_name, hook_path))
|
||
items.append(self._check_hook_callable_importable(discovered_plugin, hook_name, hook_path))
|
||
|
||
return items
|
||
|
||
def _check_hook_callable_importable(
|
||
self,
|
||
discovered_plugin: DiscoveredPlugin,
|
||
hook_name: str,
|
||
hook_path: str,
|
||
) -> PluginStructureCheckItem:
|
||
"""
|
||
检查生命周期钩子 callable 是否可以导入。
|
||
|
||
:param discovered_plugin: 已发现插件对象
|
||
:param hook_name: 生命周期钩子名称
|
||
:param hook_path: 生命周期钩子声明
|
||
:return: 结构检查项
|
||
"""
|
||
try:
|
||
module_path, callable_name = hook_path.split(':', maxsplit=1)
|
||
module_name = self._resolve_hook_module_name(discovered_plugin.manifest.backend.module, module_path)
|
||
module_file = self._resolve_plugin_module_file(discovered_plugin, module_name)
|
||
ok = self._module_file_declares_async_callable(module_file, callable_name)
|
||
except Exception as exc:
|
||
return PluginStructureCheckItem(
|
||
kind='hook_callable',
|
||
path=f'{hook_name}:{hook_path}',
|
||
ok=False,
|
||
message=f'生命周期钩子不可静态解析：{exc}',
|
||
)
|
||
|
||
return PluginStructureCheckItem(
|
||
kind='hook_callable',
|
||
path=f'{hook_name}:{hook_path}',
|
||
ok=ok,
|
||
message=(
|
||
f'生命周期钩子是异步函数：{hook_path}'
|
||
if ok
|
||
else f'生命周期钩子必须是使用 async def 声明的顶层函数：{hook_path}'
|
||
),
|
||
)
|
||
|
||
@staticmethod
|
||
def _check_hook_boundary(backend_module: str, hook_name: str, hook_path: str) -> PluginStructureCheckItem:
|
||
"""
|
||
检查生命周期钩子是否位于当前插件后端模块内。
|
||
|
||
:param backend_module: 插件后端模块路径
|
||
:param hook_name: 生命周期钩子名称
|
||
:param hook_path: 生命周期钩子声明
|
||
:return: 结构检查项
|
||
"""
|
||
module_path = hook_path.split(':', maxsplit=1)[0]
|
||
resolved_module_path = PluginStructureChecker._resolve_hook_module_name(backend_module, module_path)
|
||
ok = resolved_module_path == backend_module or resolved_module_path.startswith(f'{backend_module}.')
|
||
return PluginStructureCheckItem(
|
||
kind='hook_boundary',
|
||
path=f'{hook_name}:{hook_path}',
|
||
ok=ok,
|
||
message=f'生命周期钩子位于插件模块内：{hook_path}' if ok else f'生命周期钩子越界：{hook_path}',
|
||
)
|
||
|
||
@staticmethod
|
||
def _resolve_hook_module_name(backend_module: str, module_path: str) -> str:
|
||
"""
|
||
解析生命周期钩子模块名。
|
||
|
||
:param backend_module: 插件后端模块路径
|
||
:param module_path: manifest 中声明的钩子模块路径
|
||
:return: 完整 Python 模块名
|
||
"""
|
||
if module_path == backend_module or module_path.startswith(f'{backend_module}.'):
|
||
return module_path
|
||
if module_path.startswith('plugins.'):
|
||
return module_path
|
||
|
||
return f'{backend_module}.{module_path}'
|
||
|
||
def _check_jobs(self, discovered_plugin: DiscoveredPlugin) -> list[PluginStructureCheckItem]:
|
||
"""
|
||
检查插件定时任务声明。
|
||
|
||
:param discovered_plugin: 已发现插件对象
|
||
:return: 定时任务检查项列表
|
||
"""
|
||
items = []
|
||
manifest = discovered_plugin.manifest
|
||
for job in manifest.backend.jobs:
|
||
items.append(self._check_job_name_length(manifest.id, job))
|
||
items.append(self._check_job_callable_boundary(manifest.backend.module, job))
|
||
items.append(self._check_job_callable_importable(job))
|
||
items.append(self._check_job_cron_expression(job))
|
||
|
||
return items
|
||
|
||
@staticmethod
|
||
def _check_job_name_length(plugin_id: str, job: PluginJobManifest) -> PluginStructureCheckItem:
|
||
"""
|
||
检查插件任务映射到系统任务表后的名称长度。
|
||
|
||
:param plugin_id: 插件ID
|
||
:param job: 插件定时任务声明
|
||
:return: 结构检查项
|
||
"""
|
||
job_name = f'{plugin_id}:{job.id}'
|
||
ok = len(job_name) <= MAX_PLUGIN_JOB_NAME_LENGTH
|
||
return PluginStructureCheckItem(
|
||
kind='job_name',
|
||
path=job_name,
|
||
ok=ok,
|
||
message=f'任务名称长度有效：{job_name}' if ok else f'任务名称超过 64 字符：{job_name}',
|
||
)
|
||
|
||
@staticmethod
|
||
def _check_job_callable_boundary(
|
||
backend_module: str,
|
||
job: PluginJobManifest,
|
||
) -> PluginStructureCheckItem:
|
||
"""
|
||
检查任务 callable 是否位于当前插件后端模块内。
|
||
|
||
:param backend_module: 插件后端模块路径
|
||
:param job: 插件定时任务声明
|
||
:return: 结构检查项
|
||
"""
|
||
ok = job.callable.startswith(f'{backend_module}.')
|
||
return PluginStructureCheckItem(
|
||
kind='job_callable_boundary',
|
||
path=job.callable,
|
||
ok=ok,
|
||
message=f'任务 callable 位于插件模块内：{job.callable}' if ok else f'任务 callable 越界：{job.callable}',
|
||
)
|
||
|
||
def _check_job_callable_importable(self, job: PluginJobManifest) -> PluginStructureCheckItem:
|
||
"""
|
||
检查任务 callable 是否可通过源码静态解析。
|
||
|
||
:param job: 插件定时任务声明
|
||
:return: 结构检查项
|
||
"""
|
||
try:
|
||
module_path, callable_name = job.callable.rsplit('.', 1)
|
||
module_file = self._resolve_backend_module_file(module_path)
|
||
ok = self._module_file_declares_callable(module_file, callable_name)
|
||
except Exception as exc:
|
||
return PluginStructureCheckItem(
|
||
kind='job_callable',
|
||
path=job.callable,
|
||
ok=False,
|
||
message=f'任务 callable 不可静态解析：{exc}',
|
||
)
|
||
|
||
return PluginStructureCheckItem(
|
||
kind='job_callable',
|
||
path=job.callable,
|
||
ok=ok,
|
||
message=f'任务 callable 可调用：{job.callable}' if ok else f'任务 callable 不是可调用对象：{job.callable}',
|
||
)
|
||
|
||
def _resolve_backend_module_file(self, module_path: str) -> Path:
|
||
"""
|
||
解析后端模块对应的 Python 文件。
|
||
|
||
:param module_path: Python 模块路径
|
||
:return: 模块文件路径
|
||
"""
|
||
module_file = self.backend_root / Path(*module_path.split('.')).with_suffix('.py')
|
||
if module_file.is_file():
|
||
return module_file
|
||
|
||
package_init = self.backend_root / Path(*module_path.split('.')) / '__init__.py'
|
||
if package_init.is_file():
|
||
return package_init
|
||
|
||
raise ImportError(f'无法找到模块文件：{module_path}')
|
||
|
||
def _resolve_plugin_module_file(self, discovered_plugin: DiscoveredPlugin, module_path: str) -> Path:
|
||
"""
|
||
解析插件模块对应的 Python 文件，不导入执行模块。
|
||
|
||
:param discovered_plugin: 已发现插件对象
|
||
:param module_path: Python 模块路径
|
||
:return: 模块文件路径
|
||
"""
|
||
backend_module = discovered_plugin.manifest.backend.module
|
||
if module_path == backend_module:
|
||
module_file = discovered_plugin.backend_path / '__init__.py'
|
||
elif module_path.startswith(f'{backend_module}.'):
|
||
relative_module = module_path.removeprefix(backend_module).lstrip('.')
|
||
module_file = discovered_plugin.backend_path.joinpath(*relative_module.split('.')).with_suffix('.py')
|
||
else:
|
||
module_file = self.backend_root / Path(*module_path.split('.')).with_suffix('.py')
|
||
|
||
if module_file.is_file():
|
||
return module_file
|
||
|
||
raise ImportError(f'无法找到模块文件：{module_path}')
|
||
|
||
@staticmethod
|
||
def _module_file_declares_callable(module_file: Path, callable_name: str) -> bool:
|
||
"""
|
||
通过 AST 判断模块是否声明了顶层可调用符号。
|
||
|
||
:param module_file: 模块文件路径
|
||
:param callable_name: callable 名称
|
||
:return: 是否声明了可调用符号
|
||
"""
|
||
module_ast = ast.parse(module_file.read_text(encoding='utf-8'), filename=str(module_file))
|
||
return any(
|
||
isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef) and node.name == callable_name
|
||
for node in module_ast.body
|
||
)
|
||
|
||
@staticmethod
|
||
def _module_file_declares_async_callable(module_file: Path, callable_name: str) -> bool:
|
||
"""
|
||
通过 AST 判断模块是否声明了顶层异步函数。
|
||
|
||
:param module_file: 模块文件路径
|
||
:param callable_name: callable 名称
|
||
:return: 是否声明了异步函数
|
||
"""
|
||
module_ast = ast.parse(module_file.read_text(encoding='utf-8'), filename=str(module_file))
|
||
return any(isinstance(node, ast.AsyncFunctionDef) and node.name == callable_name for node in module_ast.body)
|
||
|
||
@staticmethod
|
||
def _check_job_cron_expression(job: PluginJobManifest) -> PluginStructureCheckItem:
|
||
"""
|
||
检查任务 cron 表达式。
|
||
|
||
:param job: 插件定时任务声明
|
||
:return: 结构检查项
|
||
"""
|
||
ok = CronUtil.validate_cron_expression(job.cron_expression)
|
||
return PluginStructureCheckItem(
|
||
kind='job_cron',
|
||
path=f'{job.id}:{job.cron_expression}',
|
||
ok=ok,
|
||
message=(
|
||
f'任务 cron 表达式有效：{job.cron_expression}' if ok else f'任务 cron 表达式无效：{job.cron_expression}'
|
||
),
|
||
)
|
||
|
||
def _check_migration_files(self, discovered_plugin: DiscoveredPlugin) -> list[PluginStructureCheckItem]:
|
||
"""
|
||
检查 migration 文件。
|
||
|
||
:param discovered_plugin: 已发现插件对象
|
||
:return: migration 检查项列表
|
||
"""
|
||
items = []
|
||
for migration_path in discovered_plugin.manifest.backend.migrations:
|
||
items.append(
|
||
self._check_plugin_relative_file('migration_file', discovered_plugin.backend_path, migration_path)
|
||
)
|
||
items.append(self._check_migration_type(migration_path))
|
||
|
||
return items
|
||
|
||
@staticmethod
|
||
def _check_migration_type(migration_path: str) -> PluginStructureCheckItem:
|
||
"""
|
||
检查 migration 文件类型。
|
||
|
||
:param migration_path: migration 相对插件根目录路径
|
||
:return: 结构检查项
|
||
"""
|
||
ok = Path(migration_path).suffix in SUPPORTED_MIGRATION_SUFFIXES
|
||
return PluginStructureCheckItem(
|
||
kind='migration_type',
|
||
path=migration_path,
|
||
ok=ok,
|
||
message=f'支持的 migration：{migration_path}' if ok else f'暂不支持的 migration 类型：{migration_path}',
|
||
)
|
||
|
||
def _check_seed_files(self, discovered_plugin: DiscoveredPlugin) -> list[PluginStructureCheckItem]:
|
||
"""
|
||
检查 seed 文件。
|
||
|
||
:param discovered_plugin: 已发现插件对象
|
||
:return: seed 检查项列表
|
||
"""
|
||
items = []
|
||
for seed_path in discovered_plugin.manifest.backend.seeds:
|
||
items.append(self._check_plugin_relative_file('seed_file', discovered_plugin.backend_path, seed_path))
|
||
items.append(self._check_seed_type(seed_path))
|
||
|
||
return items
|
||
|
||
@staticmethod
|
||
def _check_seed_type(seed_path: str) -> PluginStructureCheckItem:
|
||
"""
|
||
检查 seed 文件类型。
|
||
|
||
:param seed_path: seed 相对插件根目录路径
|
||
:return: 结构检查项
|
||
"""
|
||
ok = Path(seed_path).suffix in SUPPORTED_SEED_SUFFIXES
|
||
return PluginStructureCheckItem(
|
||
kind='seed_type',
|
||
path=seed_path,
|
||
ok=ok,
|
||
message=f'支持的 seed：{seed_path}' if ok else f'暂不支持的 seed 类型：{seed_path}',
|
||
)
|
||
|
||
def _check_frontend(self, discovered_plugin: DiscoveredPlugin) -> list[PluginStructureCheckItem]:
|
||
"""
|
||
检查前端插件目录和菜单组件。
|
||
|
||
:param discovered_plugin: 已发现插件对象
|
||
:return: 前端检查项列表
|
||
"""
|
||
manifest = discovered_plugin.manifest
|
||
plugin_view_menus = [
|
||
menu
|
||
for menu in PluginMenuTree.flatten(manifest.frontend.menus)
|
||
if PluginMenuTree.is_plugin_component(menu.component)
|
||
]
|
||
frontend_plugin_root = self.frontend_root / (manifest.frontend.plugin_id or manifest.id)
|
||
if not plugin_view_menus:
|
||
return []
|
||
|
||
items = [self._check_dir('frontend_root', frontend_plugin_root)]
|
||
items.append(self._check_dir('frontend_views_path', frontend_plugin_root / manifest.frontend.views_path))
|
||
items.append(self._check_dir('frontend_api_path', frontend_plugin_root / manifest.frontend.api_path))
|
||
items.extend(self._check_plugin_view(frontend_plugin_root, manifest, menu) for menu in plugin_view_menus)
|
||
|
||
return items
|
||
|
||
def _check_plugin_view(
|
||
self,
|
||
frontend_plugin_root: Path,
|
||
manifest: PluginManifest,
|
||
menu: PluginMenuManifest,
|
||
) -> PluginStructureCheckItem:
|
||
"""
|
||
检查插件菜单组件对应的 Vue 页面。
|
||
|
||
:param frontend_plugin_root: 前端插件根目录
|
||
:param manifest: 插件清单
|
||
:param menu: 菜单声明
|
||
:return: 页面检查项
|
||
"""
|
||
view_path = PluginMenuTree.resolve_plugin_view_path(manifest, menu.component)
|
||
if view_path is None:
|
||
return PluginStructureCheckItem(
|
||
kind='frontend_view',
|
||
path=menu.component,
|
||
ok=False,
|
||
message=f'插件组件路径格式错误：{menu.component}',
|
||
)
|
||
|
||
return self._check_file('frontend_view', frontend_plugin_root / view_path)
|
||
|
||
@staticmethod
|
||
def _check_dir(kind: str, path: Path) -> PluginStructureCheckItem:
|
||
"""
|
||
检查目录是否存在。
|
||
|
||
:param kind: 检查类型
|
||
:param path: 目录路径
|
||
:return: 结构检查项
|
||
"""
|
||
return PluginStructureCheckItem(
|
||
kind=kind,
|
||
path=str(path),
|
||
ok=path.is_dir(),
|
||
message=f'目录存在：{path}' if path.is_dir() else f'目录不存在：{path}',
|
||
)
|
||
|
||
@staticmethod
|
||
def _check_file(kind: str, path: Path) -> PluginStructureCheckItem:
|
||
"""
|
||
检查文件是否存在。
|
||
|
||
:param kind: 检查类型
|
||
:param path: 文件路径
|
||
:return: 结构检查项
|
||
"""
|
||
return PluginStructureCheckItem(
|
||
kind=kind,
|
||
path=str(path),
|
||
ok=path.is_file(),
|
||
message=f'文件存在：{path}' if path.is_file() else f'文件不存在：{path}',
|
||
)
|
||
|
||
@classmethod
|
||
def _check_plugin_relative_file(cls, kind: str, plugin_root: Path, relative_path: str) -> PluginStructureCheckItem:
|
||
"""
|
||
检查插件相对路径文件是否存在且未越过插件根目录。
|
||
|
||
:param kind: 检查类型
|
||
:param plugin_root: 插件后端根目录
|
||
:param relative_path: 相对插件根目录路径
|
||
:return: 结构检查项
|
||
"""
|
||
file_path = (plugin_root / relative_path).resolve()
|
||
resolved_plugin_root = plugin_root.resolve()
|
||
if resolved_plugin_root not in file_path.parents:
|
||
return PluginStructureCheckItem(
|
||
kind=kind,
|
||
path=relative_path,
|
||
ok=False,
|
||
message=f'文件路径不能越过插件根目录：{relative_path}',
|
||
suggestion='请使用插件目录内的相对路径',
|
||
)
|
||
|
||
return cls._check_file(kind, file_path)
|