97ced1b5e6
Key features implemented: - Updated package metadata and dependencies in PKG-INFO, setup files - Added main.py entry point for backward compatibility with README launch method - Enhanced CLI with config options, system info command, and proper signal handling - Implemented minimal PluginManager loading only plugin-loader core plugin - Refactored PluginLoader to follow minimal core design, removed sandbox/isolation complexity - Updated auto-dependency plugin with safer PL injection mechanism and disabled pl_injection - Removed legacy plugin files (firewall, frp_proxy, ftp_server, multi_lang_deploy, ops_toolbox, security_gateway) as functionality moved to core plugin system - Improved gitignore with comprehensive ignore patterns The changes implement a minimal core framework design where only the plugin-loader is directly loaded by the core, with all other plugins managed through the PL injection mechanism, significantly simplifying the architecture.
324 lines
13 KiB
Python
324 lines
13 KiB
Python
"""引用检查器 - 检测导入错误、变量错误等"""
|
|
import ast
|
|
import sys
|
|
import os
|
|
from pathlib import Path
|
|
|
|
|
|
class ReferenceChecker:
|
|
"""引用检查器"""
|
|
|
|
# Python 标准库模块列表
|
|
STD_MODULES = {
|
|
'os', 'sys', 'json', 're', 'time', 'datetime', 'pathlib',
|
|
'typing', 'collections', 'functools', 'itertools', 'io',
|
|
'string', 'math', 'random', 'hashlib', 'hmac', 'secrets',
|
|
'urllib', 'http', 'email', 'html', 'xml', 'csv', 'configparser',
|
|
'logging', 'warnings', 'traceback', 'inspect', 'importlib',
|
|
'threading', 'multiprocessing', 'subprocess', 'socket',
|
|
'asyncio', 'concurrent', 'queue', 'contextlib', 'abc',
|
|
'enum', 'dataclasses', 'copy', 'pprint', 'textwrap',
|
|
'struct', 'codecs', 'locale', 'gettext', 'argparse',
|
|
'unittest', 'doctest', 'pdb', 'profile', 'timeit',
|
|
'tempfile', 'glob', 'fnmatch', 'stat', 'fileinput',
|
|
'shutil', 'pickle', 'shelve', 'sqlite3', 'dbm',
|
|
'gzip', 'bz2', 'lzma', 'zipfile', 'tarfile',
|
|
'base64', 'binascii', 'quopri', 'uu',
|
|
}
|
|
|
|
# Python 内置函数和类型（不应报告为未定义）
|
|
BUILTINS = {
|
|
'print', 'len', 'str', 'int', 'float', 'bool', 'list', 'dict',
|
|
'set', 'tuple', 'range', 'enumerate', 'zip', 'map', 'filter',
|
|
'sorted', 'reversed', 'min', 'max', 'sum', 'abs', 'round',
|
|
'isinstance', 'issubclass', 'type', 'id', 'hash', 'repr',
|
|
'True', 'False', 'None', 'Exception', 'ValueError', 'TypeError',
|
|
'KeyError', 'AttributeError', 'ImportError', 'FileNotFoundError',
|
|
'IndexError', 'RuntimeError', 'StopIteration', 'GeneratorExit',
|
|
'staticmethod', 'classmethod', 'property', 'super',
|
|
'open', 'input', 'format', 'hex', 'oct', 'bin', 'chr', 'ord',
|
|
'dir', 'vars', 'locals', 'globals', 'callable', 'getattr',
|
|
'setattr', 'hasattr', 'delattr', 'exec', 'eval', 'compile',
|
|
'any', 'all', 'slice', 'frozenset', 'bytearray', 'bytes',
|
|
'memoryview', 'complex', 'divmod', 'pow', 'object',
|
|
'dict', 'list', 'str', 'int', 'float', 'bool', 'set',
|
|
'tuple', 'Exception', 'ValueError', 'TypeError', 'KeyError',
|
|
'self', 'cls', 'args', 'kwargs',
|
|
}
|
|
|
|
def __init__(self, project_root: str = "."):
|
|
self.project_root = Path(project_root)
|
|
self._available_modules = set(self.STD_MODULES)
|
|
self._scan_project_modules()
|
|
|
|
def _scan_project_modules(self):
|
|
"""扫描项目中的可用模块"""
|
|
# 扫描 oss 目录（框架核心）
|
|
oss_dir = self.project_root / "oss"
|
|
if oss_dir.exists():
|
|
self._available_modules.add("oss")
|
|
self._scan_module_dir(oss_dir, "oss")
|
|
|
|
# 扫描 store 目录下的所有插件
|
|
store_dir = self.project_root / "store"
|
|
if store_dir.exists():
|
|
for author_dir in store_dir.iterdir():
|
|
if not author_dir.is_dir():
|
|
continue
|
|
for plugin_dir in author_dir.iterdir():
|
|
if not plugin_dir.is_dir():
|
|
continue
|
|
plugin_name = plugin_dir.name
|
|
# 添加插件名作为可用模块
|
|
self._available_modules.add(plugin_name)
|
|
# 扫描插件内部的子模块
|
|
self._scan_plugin_modules(plugin_dir, plugin_name)
|
|
|
|
def _scan_module_dir(self, dir_path: Path, base_name: str):
|
|
"""扫描模块目录"""
|
|
if dir_path.exists():
|
|
for item in dir_path.iterdir():
|
|
if item.is_file() and item.name.endswith(".py") and item.name != "__init__.py":
|
|
module_name = item.name[:-3]
|
|
full_name = f"{base_name}.{module_name}"
|
|
self._available_modules.add(full_name)
|
|
elif item.is_dir() and (item / "__init__.py").exists():
|
|
full_name = f"{base_name}.{item.name}"
|
|
self._available_modules.add(full_name)
|
|
self._scan_module_dir(item, full_name)
|
|
|
|
def _scan_plugin_modules(self, plugin_dir: Path, base_name: str):
|
|
"""扫描插件内部的子模块"""
|
|
for item in plugin_dir.iterdir():
|
|
if item.is_dir() and (item / "__init__.py").exists():
|
|
full_name = f"{base_name}.{item.name}"
|
|
self._available_modules.add(full_name)
|
|
self._scan_module_dir(item, full_name)
|
|
elif item.is_file() and item.name.endswith(".py") and item.name != "__init__.py":
|
|
module_name = item.name[:-3]
|
|
full_name = f"{base_name}.{module_name}"
|
|
self._available_modules.add(full_name)
|
|
|
|
def _add_module_from_dir(self, dir_path: Path, base_name: str):
|
|
"""从目录添加模块"""
|
|
if dir_path.exists():
|
|
for item in dir_path.iterdir():
|
|
if item.is_file() and item.name.endswith(".py") and item.name != "__init__.py":
|
|
module_name = item.name[:-3]
|
|
self._available_modules.add(f"{base_name}.{module_name}")
|
|
elif item.is_dir() and (item / "__init__.py").exists():
|
|
self._add_module_from_dir(item, f"{base_name}.{item.name}")
|
|
|
|
def check(self, filepath: str, content: str) -> list:
|
|
"""执行引用检查"""
|
|
issues = []
|
|
|
|
try:
|
|
tree = ast.parse(content)
|
|
except SyntaxError as e:
|
|
return [{
|
|
"file": filepath,
|
|
"line": e.lineno or 0,
|
|
"severity": "critical",
|
|
"type": "syntax_error",
|
|
"message": f"语法错误: {e.msg}"
|
|
}]
|
|
|
|
# 检查导入语句（跳过相对导入）
|
|
issues.extend(self._check_imports(filepath, tree))
|
|
|
|
# 检查属性访问错误
|
|
issues.extend(self._check_attribute_access(filepath, tree, content))
|
|
|
|
# 检查函数调用错误
|
|
issues.extend(self._check_function_calls(filepath, tree, content))
|
|
|
|
return issues
|
|
|
|
def _check_imports(self, filepath: str, tree: ast.AST) -> list:
|
|
"""检查导入语句"""
|
|
issues = []
|
|
file_path = Path(filepath)
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
# 跳过 oss 框架模块（运行时可用）
|
|
if alias.name.startswith('oss.') or alias.name == 'oss':
|
|
continue
|
|
# 跳过 websockets 等第三方库
|
|
if alias.name in ('websockets', 'yaml', 'click'):
|
|
continue
|
|
if not self._is_module_available(alias.name, file_path):
|
|
issues.append({
|
|
"file": filepath,
|
|
"line": node.lineno,
|
|
"severity": "critical",
|
|
"type": "import_error",
|
|
"message": f"无法导入模块: {alias.name}"
|
|
})
|
|
|
|
elif isinstance(node, ast.ImportFrom):
|
|
# 跳过相对导入（以 . 开头）
|
|
if node.level and node.level > 0:
|
|
continue
|
|
|
|
# 跳过 oss 框架模块
|
|
if node.module and (node.module.startswith('oss.') or node.module == 'oss'):
|
|
continue
|
|
|
|
# 跳过第三方库
|
|
if node.module and node.module.split('.')[0] in ('websockets', 'yaml', 'click'):
|
|
continue
|
|
|
|
if node.module:
|
|
if not self._is_module_available(node.module, file_path):
|
|
issues.append({
|
|
"file": filepath,
|
|
"line": node.lineno,
|
|
"severity": "critical",
|
|
"type": "import_error",
|
|
"message": f"无法导入模块: {node.module}"
|
|
})
|
|
|
|
return issues
|
|
|
|
def _check_variable_references(self, filepath: str, tree: ast.AST, content: str) -> list:
|
|
"""检查变量引用"""
|
|
issues = []
|
|
lines = content.split('\n')
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load):
|
|
# 检查是否引用了未定义的变量
|
|
if not self._is_name_defined(node.id, tree, node.lineno):
|
|
if node.id not in ('True', 'False', 'None', 'self', 'cls'):
|
|
issues.append({
|
|
"file": filepath,
|
|
"line": node.lineno,
|
|
"severity": "warning",
|
|
"type": "undefined_variable",
|
|
"message": f"使用了未定义的变量: {node.id}"
|
|
})
|
|
|
|
return issues
|
|
|
|
def _check_attribute_access(self, filepath: str, tree: ast.AST, content: str) -> list:
|
|
"""检查属性访问"""
|
|
issues = []
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Attribute):
|
|
# 检查可能的属性错误
|
|
if isinstance(node.value, ast.Name):
|
|
var_name = node.value.id
|
|
if var_name in ('None', 'True', 'False'):
|
|
issues.append({
|
|
"file": filepath,
|
|
"line": node.lineno,
|
|
"severity": "critical",
|
|
"type": "attribute_error",
|
|
"message": f"尝试访问 {var_name} 的属性: {node.attr}"
|
|
})
|
|
|
|
return issues
|
|
|
|
def _check_function_calls(self, filepath: str, tree: ast.AST, content: str) -> list:
|
|
"""检查函数调用"""
|
|
issues = []
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Call):
|
|
# 检查调用不存在的方法
|
|
if isinstance(node.func, ast.Attribute):
|
|
if isinstance(node.func.value, ast.Constant) and node.func.value.value is None:
|
|
issues.append({
|
|
"file": filepath,
|
|
"line": node.lineno,
|
|
"severity": "critical",
|
|
"type": "method_call_on_none",
|
|
"message": f"在 None 上调用方法: {node.func.attr}"
|
|
})
|
|
|
|
return issues
|
|
|
|
def _is_module_available(self, module_name: str, file_path: Path = None) -> bool:
|
|
"""检查模块是否可用"""
|
|
# 检查是否在已扫描的模块中
|
|
if module_name in self._available_modules:
|
|
return True
|
|
|
|
# 检查标准库
|
|
base_module = module_name.split('.')[0]
|
|
if base_module in self.STD_MODULES:
|
|
return True
|
|
|
|
# 检查是否是 oss 框架模块
|
|
if module_name.startswith('oss.') or module_name == 'oss':
|
|
return True
|
|
|
|
# 检查是否是常见第三方库
|
|
third_party = {'websockets', 'yaml', 'click', 'requests', 'flask', 'django', 'numpy', 'pandas'}
|
|
if module_name.split('.')[0] in third_party:
|
|
return True
|
|
|
|
# 检查是否是当前文件的同目录模块（相对导入的情况）
|
|
if file_path:
|
|
file_dir = file_path.parent
|
|
# 检查同级 .py 文件
|
|
sibling_module = file_dir / f"{module_name}.py"
|
|
if sibling_module.exists():
|
|
return True
|
|
# 检查同级包
|
|
sibling_pkg = file_dir / module_name
|
|
if sibling_pkg.is_dir() and (sibling_pkg / "__init__.py").exists():
|
|
return True
|
|
# 检查 store 目录下的插件
|
|
store_dir = self.project_root / "store"
|
|
if store_dir.exists():
|
|
for author_dir in store_dir.iterdir():
|
|
if author_dir.is_dir():
|
|
for plugin_dir in author_dir.iterdir():
|
|
if plugin_dir.is_dir() and plugin_dir.name == module_name.split('.')[0]:
|
|
return True
|
|
|
|
return False
|
|
|
|
def _is_name_defined(self, name: str, tree: ast.AST, line: int) -> bool:
|
|
"""检查名称是否已定义"""
|
|
# 检查是否是内置函数/类型
|
|
if name in self.BUILTINS:
|
|
return True
|
|
|
|
# 检查是否是函数参数
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.FunctionDef):
|
|
for arg in node.args.args:
|
|
if arg.arg == name:
|
|
return True
|
|
|
|
# 检查是否是赋值目标
|
|
elif isinstance(node, ast.Assign):
|
|
for target in node.targets:
|
|
if isinstance(target, ast.Name) and target.id == name:
|
|
return True
|
|
|
|
# 检查是否是循环变量
|
|
elif isinstance(node, ast.For):
|
|
if isinstance(node.target, ast.Name) and node.target.id == name:
|
|
return True
|
|
|
|
# 检查是否是导入
|
|
elif isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
if alias.asname == name or alias.name == name:
|
|
return True
|
|
|
|
elif isinstance(node, ast.ImportFrom):
|
|
if node.module:
|
|
for alias in node.names:
|
|
if alias.asname == name or alias.name == name:
|
|
return True
|
|
|
|
return False
|