Coverage for src\genesis\impact\code

1"""CodeParser：多语言源码解析 → CodeStructure（Impact Agent）。

3通过语言适配器注册表（LANGUAGE_PARSERS）分发：每种开发语言一个适配器

4（BaseLanguageParser 子类），定义文件扩展名、类名识别、角色（控制器/服务/实体）

5与端点提取。当前内置 Java 适配器（Spring Boot / JPA / MyBatis-Plus）。

7新增语言 = 实现一个适配器类并调用 register_language_parser 注册，

8下游（ExistingSystemExplorer / ImpactAgent / Writer）零改动。

9"""

10from __future__ import annotations

12import re

13from pathlib import Path

15from genesis.data_models import (

16 CodeStructure,

17 ControllerInfo,

18 EndpointInfo,

19 EntityInfo,

20 ServiceInfo,

21)

24class CodeParseError(Exception):

25 """既有系统解析失败（路径无效或无可识别源码）。"""

28def _relative(path: Path, root: Path) -> str:

29 return path.relative_to(root).as_posix()

32def _read(path: Path) -> str:

33 return path.read_text(encoding="utf-8", errors="ignore")

36class BaseLanguageParser:

37 """语言适配器基类：定义扩展名与单文件解析契约。

39 子类需设置 language / extensions，并实现 parse_file：

40 返回 dict（keys: imports / classes / controllers / services / entities / endpoints）。

41 """

43 language: str = ""

44 extensions: tuple[str, ...] = ()

46 def source_files(self, root: Path) -> list[Path]:

47 """返回本语言适配器覆盖的源文件（排序 + 去重，保证确定性）。"""

48 files: list[Path] = []

49 for ext in self.extensions:

50 files.extend(p for p in root.rglob(f"*{ext}") if p.is_file())

51 return sorted(set(files))

53 def parse_file(self, text: str, rel: str) -> dict:

54 """解析单个源文件，返回该文件对 CodeStructure 各分层的贡献。"""

55 raise NotImplementedError

58class JavaLanguageParser(BaseLanguageParser):

59 """Java（Spring Boot / JPA / MyBatis-Plus）适配器。"""

61 language = "java"

62 extensions = (".java",)

64 _CLASS_RE = re.compile(

66 )

67 _TABLE_RE = re.compile(r"@(?:Table|TableName)\s*\(\s*(?:name\s*=\s*)?[\"']([^\"']+)[\"']")

68 _CLASS_MAPPING_RE = re.compile(r"@RequestMapping\s*\(\s*[\"']([^\"']+)[\"']")

69 _METHOD_MAPPING_RE = re.compile(

71 )

72 _METHOD_DECL_RE = re.compile(

74 r"[\w<>\[\],.?]+\s+(\w+)\s*\("

75 )

76 _FIELD_DECL_RE = re.compile(

77 r"(?:private|public|protected)\s+[\w<>\[\],]+\s+(\w+)\s*;"

78 )

79 _IMPORT_RE = re.compile(r"^import\s+([\w.]+);", re.MULTILINE)

81 def parse_file(self, text: str, rel: str) -> dict:

82 class_name = self._class_name(text)

83 imports = self._IMPORT_RE.findall(text)

84 out: dict = {

85 "imports": imports,

86 "classes": [],

87 "controllers": [],

88 "services": [],

89 "entities": [],

90 "endpoints": [],

91 }

92 if not class_name:

93 # package-info.java 等无类声明文件：仅登记 imports，不参与要素提取

94 return out

96 out["classes"].append({"class_name": class_name, "path": rel})

98 is_controller = "@RestController" in text or "@Controller" in text

99 is_service = "@Service" in text

100 # 既有系统实体可能用 JPA @Entity 或 MyBatis-Plus @TableName 标注

101 is_entity = "@Entity" in text or "@TableName" in text

102

103 if is_controller:

104 ctrl, endpoints = self._parse_controller(text, rel, class_name)

105 out["controllers"].append(ctrl)

106 out["endpoints"] = endpoints

107 elif is_service:

108 out["services"].append(self._parse_service(text, rel, class_name))

109 elif is_entity:

110 out["entities"].append(self._parse_entity(text, rel, class_name))

111 return out

112

113 @staticmethod

114 def _class_name(text: str) -> str | None:

115 m = JavaLanguageParser._CLASS_RE.search(text)

116 return m.group(1) if m else None

117

118 @staticmethod

119 def _parse_controller(

120 text: str, rel: str, class_name: str

121 ) -> tuple[ControllerInfo, list[EndpointInfo]]:

122 base_path = ""

123 m = JavaLanguageParser._CLASS_MAPPING_RE.search(text)

124 if m:

125 base_path = m.group(1)

126

127 ctrl_endpoints: list[str] = []

128 endpoints: list[EndpointInfo] = []

129 for m in JavaLanguageParser._METHOD_MAPPING_RE.finditer(text):

130 verb, sub = m.group(1).upper(), m.group(2) or ""

131 if verb == "REQUEST":

132 verb = "ANY"

133 full = f"{base_path.rstrip('/')}/{sub.lstrip('/')}".rstrip("/") or base_path

134 ctrl_endpoints.append(full)

135 endpoints.append(

136 EndpointInfo(

137 method=verb,

138 path=full,

139 controller=class_name,

140 description="",

141 source_uri=rel,

142 )

143 )

144

145 return (

146 ControllerInfo(

147 name=class_name,

148 class_name=class_name,

149 path=rel,

150 base_path=base_path,

151 endpoints=ctrl_endpoints,

152 source_uri=rel,

153 ),

154 endpoints,

155 )

156

157 @staticmethod

158 def _parse_service(text: str, rel: str, class_name: str) -> ServiceInfo:

159 methods = list(dict.fromkeys(JavaLanguageParser._METHOD_DECL_RE.findall(text)))

160 return ServiceInfo(

161 name=class_name,

162 class_name=class_name,

163 path=rel,

164 methods=methods,

165 source_uri=rel,

166 )

167

168 @staticmethod

169 def _parse_entity(text: str, rel: str, class_name: str) -> EntityInfo:

170 table_name = None

171 m = JavaLanguageParser._TABLE_RE.search(text)

172 if m: 172 ↛ 174line 172 didn't jump to line 174 because the condition on line 172 was always true

173 table_name = m.group(1)

174 fields = list(dict.fromkeys(JavaLanguageParser._FIELD_DECL_RE.findall(text)))

175 return EntityInfo(

176 name=class_name,

177 class_name=class_name,

178 path=rel,

179 table_name=table_name,

180 fields=fields,

181 source_uri=rel,

182 )

183

184

185# ---------- 语言适配器注册表 ----------

186

187LANGUAGE_PARSERS: dict[str, type[BaseLanguageParser]] = {}

188

189

190def register_language_parser(name: str, parser_cls: type[BaseLanguageParser]) -> None:

191 """注册语言适配器（未来扩展点：新增语言只需实现并注册，下游零改动）。"""

192 LANGUAGE_PARSERS[name] = parser_cls

193

194

195register_language_parser(JavaLanguageParser.language, JavaLanguageParser)

196

197

198class CodeParser:

199 """按语言分发解析既有系统目录，输出 CodeStructure。"""

200

201 def parse(self, root_path: str | Path, language: str | None = None) -> CodeStructure:

202 root = Path(root_path)

203 if not root.is_dir():

204 raise CodeParseError(f"既有系统路径无效或不存在: {root_path}")

205

206 if language is not None:

207 cls = LANGUAGE_PARSERS.get(language)

208 if cls is None:

209 raise CodeParseError(

210 f"不支持的源码语言: {language}（支持: {', '.join(sorted(LANGUAGE_PARSERS))}）"

211 )

212 parser_cls = [cls]

213 else:

214 # 自动探测：仅保留在根目录下确有源文件的语言

215 present = [

216 name for name in sorted(LANGUAGE_PARSERS)

217 if LANGUAGE_PARSERS[name]().source_files(root)

218 ]

219 if not present:

220 raise CodeParseError(

221 f"未找到可识别的源码（支持: {', '.join(sorted(LANGUAGE_PARSERS))}）: {root_path}"

222 )

223 parser_cls = [LANGUAGE_PARSERS[name] for name in present]

224

225 controllers: list[ControllerInfo] = []

226 services: list[ServiceInfo] = []

227 entities: list[EntityInfo] = []

228 endpoints: list[EndpointInfo] = []

229 classes: list[dict] = []

230 raw_imports: list[dict] = []

231 all_files: list[Path] = []

232

233 for cls in parser_cls:

234 parser = cls()

235 files = parser.source_files(root)

236 if not files:

237 # 自动探测的 present 过滤已保证有文件；此处仅显式 language 时可达

238 raise CodeParseError(f"未找到 {parser.language} 源码: {root_path}")

239 all_files.extend(files)

240 for path in files:

241 text = _read(path)

242 rel = _relative(path, root)

243 out = parser.parse_file(text, rel)

244 raw_imports.append({"path": rel, "imports": out["imports"]})

245 controllers.extend(out["controllers"])

246 services.extend(out["services"])

247 entities.extend(out["entities"])

248 endpoints.extend(out["endpoints"])

249 classes.extend(out["classes"])

250

251 modules = self._modules(root, all_files)

252 langs = [cls.language for cls in parser_cls]

253 return CodeStructure(

254 root_path=str(root),

255 language=",".join(langs) if len(langs) > 1 else langs[0],

256 modules=modules,

257 classes=classes,

258 controllers=controllers,

259 services=services,

260 entities=entities,

261 endpoints=endpoints,

262 raw_imports=raw_imports,

263 )

264

265 @staticmethod

266 def _modules(root: Path, files: list[Path]) -> list[str]:

267 """顶层目录中凡包含源码者视为一个模块（按名排序，保证确定性）。"""

268 mods = {

269 p.relative_to(root).parts[0]

270 for p in files

271 if len(p.relative_to(root).parts) > 1

272 }

273 return sorted(mods)

Coverage for src\genesis\impact\code_parser.py: 99%

128 statements