Coverage for src/symphra_modules/loader/directory.py: 49.26%

1"""目录加载器实现.""" 

2 

3import importlib 

4import importlib.util 

5import sys 

6from pathlib import Path 

7from typing import Any 

8 

9from symphra_modules.abc import ModuleInterface 

10from symphra_modules.exceptions import ModuleLoadError 

11from symphra_modules.loader.base import ModuleLoader 

12from symphra_modules.utils import get_logger 

13 

14logger = get_logger() 

15 

16 

17class DirectoryLoader(ModuleLoader): 

18 """从目录加载模块.""" 

19 

20 def __init__(self, base_path: Path | None = None) -> None: 

21 """初始化目录加载器. 

22 

23 Args: 

24 base_path: 基础路径，默认为当前工作目录 

25 """ 

26 self.base_path = base_path or Path.cwd() 

27 

28 def _to_module_name(self, path: Path) -> str | None: 

29 """将文件或包路径转换为模块名（相对 base_path）. 

30 

31 Args: 

32 path: 文件或包路径 

33 

34 Returns: 

35 模块名，如果无法转换则返回 None 

36 """ 

37 try: 

38 relative = path.relative_to(self.base_path) 

39 except ValueError: 

40 # 无法转换为相对路径时返回 None 

41 return None 

42 

43 parts = list(relative.parts) 

44 if parts and parts[-1].endswith(".py"): 

45 parts[-1] = parts[-1][:-3] 

46 

47 return ".".join(parts) 

48 

49 def load(self, source: str) -> dict[str, type[ModuleInterface]]: 

50 """从指定目录加载所有模块. 

51 

52 Args: 

53 source: 目录路径（相对于 base_path） 

54 

55 Returns: 

56 模块名到模块类的映射字典 

57 

58 Raises: 

59 ModuleLoadError: 目录不存在时抛出 

60 """ 

61 modules: dict[str, type[ModuleInterface]] = {} 

62 module_dir = Path(self.base_path) / source 

63 

64 if not module_dir.exists(): 

65 raise ModuleLoadError(f"模块目录不存在: {module_dir}") 

66 

67 # 查找所有 Python 文件 

68 for py_file in module_dir.glob("*.py"): 

69 if py_file.name.startswith("_"): 69 ↛ 72line 69 didn't jump to line 72 because the condition on line 69 was always true

70 continue 

71 

72 try: 

73 module_classes = self._load_from_file(py_file) 

74 modules.update(module_classes) 

75 except Exception as e: 

76 logger.warning(f"无法从 {py_file} 加载模块: {e}") 

77 

78 # 查找包 

79 for package_dir in module_dir.iterdir(): 

80 if package_dir.is_dir() and (package_dir / "__init__.py").exists(): 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true

81 try: 

82 module_classes = self._load_from_package(package_dir) 

83 modules.update(module_classes) 

84 except Exception as e: 

85 logger.error(f"无法从 {package_dir} 加载模块: {e}") 

86 

87 return modules 

88 

89 def discover(self, source: str) -> list[str]: 

90 """发现目录中的模块. 

91 

92 Args: 

93 source: 目录路径 

94 

95 Returns: 

96 模块名称列表 

97 """ 

98 module_names: list[str] = [] 

99 module_dir = Path(self.base_path) / source 

100 

101 if not module_dir.exists(): 

102 return module_names 

103 

104 # Python 文件 

105 for py_file in module_dir.glob("*.py"): 

106 if not py_file.name.startswith("_"): 

107 module_names.append(py_file.stem) 

108 

109 # 包 

110 for package_dir in module_dir.iterdir(): 

111 if package_dir.is_dir() and (package_dir / "__init__.py").exists(): 

112 module_names.append(package_dir.name) 

113 

114 return module_names 

115 

116 def _load_from_file(self, file_path: Path, package: str | None = None) -> dict[str, type[ModuleInterface]]: 

117 """从 Python 文件加载模块. 

118 

119 Args: 

120 file_path: Python 文件路径 

121 package: 包名（可选） 

122 

123 Returns: 

124 模块类字典 

125 

126 Raises: 

127 ModuleLoadError: 加载失败时抛出 

128 """ 

129 module_name: str 

130 if package is not None: 

131 module_name = f"{package}.{file_path.stem}" 

132 else: 

133 inferred = self._to_module_name(file_path) 

134 module_name = inferred or file_path.stem 

135 

136 module: Any = None 

137 if "." in module_name: 

138 try: 

139 module = importlib.import_module(module_name) 

140 except ImportError: 

141 module = None 

142 

143 if module is None: 

144 spec = importlib.util.spec_from_file_location(module_name, file_path) 

145 if not spec or not spec.loader: 

146 raise ModuleLoadError(f"无法从 {file_path} 创建模块规范") 

147 

148 module = importlib.util.module_from_spec(spec) 

149 sys.modules[module_name] = module 

150 spec.loader.exec_module(module) 

151 

152 return self._find_module_classes(module) 

153 

154 def _load_from_package(self, package_dir: Path) -> dict[str, type[ModuleInterface]]: 

155 """从包加载模块. 

156 

157 Args: 

158 package_dir: 包目录路径 

159 

160 Returns: 

161 模块类字典 

162 

163 Raises: 

164 ModuleLoadError: 加载失败时抛出 

165 """ 

166 inferred_package = self._to_module_name(package_dir) 

167 package_name = inferred_package or package_dir.name 

168 

169 # 将包路径添加到 sys.path，便于定位依赖 

170 parent_path = str(package_dir.parent) 

171 if parent_path not in sys.path: 

172 sys.path.insert(0, parent_path) 

173 

174 modules: dict[str, type[ModuleInterface]] = {} 

175 

176 # 优先加载显式的 module.py，再加载其他非私有 Python 文件 

177 candidate_files: list[Path] = [] 

178 module_file = package_dir / "module.py" 

179 if module_file.exists(): 

180 candidate_files.append(module_file) 

181 

182 if not candidate_files: 

183 # 回退：若没有独立文件，则尝试加载包本身 

184 try: 

185 module = importlib.import_module(package_name) 

186 return self._find_module_classes(module) 

187 except ImportError as e: 

188 raise ModuleLoadError(f"无法从 {package_name} 加载模块: {e}") from e 

189 

190 for py_file in candidate_files: 

191 try: 

192 module_classes = self._load_from_file(py_file, package=package_name) 

193 modules.update(module_classes) 

194 except Exception as e: 

195 logger.warning(f"无法从 {package_name} 的 {py_file.name} 加载模块: {e}") 

196 

197 return modules