297 lines
12 KiB
Python
Executable File
297 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
宠小它 本体—代码漂移审计脚本。
|
||
|
||
扫描 backend/src/main/java/com/petstore/controller/*.java，提取所有 HTTP 端点
|
||
（类级 @RequestMapping + 方法级 @GetMapping/@PostMapping/@PutMapping/@DeleteMapping），
|
||
对比 docs/graph/ontology.jsonl 中 Action 的 entrypoints，报告：
|
||
|
||
1. code-only：代码里有但本体未录入的端点（drift：应补录本体或确认废弃）
|
||
2. ontology-only：本体里有但代码里找不到的端点（drift：可能是已删除接口或路径写错）
|
||
3. entity-coverage：扫描 backend/src/main/java/com/petstore/entity/*.java，对比 ontology.jsonl 的 Entity，
|
||
报告未录入的实体类。
|
||
|
||
用法：
|
||
python3 docs/graph/audit_drift.py [repo_root]
|
||
|
||
默认 repo_root = . （当前工作目录，应能访问 backend/ 和 docs/）。
|
||
退出码：0 = 无漂移；1 = 有漂移（warning）；2 = 解析错误。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from collections import defaultdict
|
||
|
||
VALID_TYPES = {"Entity", "Action"}
|
||
|
||
|
||
def find_controller_files(repo_root):
|
||
ctrl_dir = os.path.join(repo_root, "backend", "src", "main", "java", "com", "petstore", "controller")
|
||
if not os.path.isdir(ctrl_dir):
|
||
return []
|
||
return sorted(
|
||
os.path.join(ctrl_dir, f) for f in os.listdir(ctrl_dir) if f.endswith(".java")
|
||
)
|
||
|
||
|
||
def find_entity_files(repo_root):
|
||
ent_dir = os.path.join(repo_root, "backend", "src", "main", "java", "com", "petstore", "entity")
|
||
if not os.path.isdir(ent_dir):
|
||
return []
|
||
return sorted(
|
||
os.path.join(ent_dir, f) for f in os.listdir(ent_dir) if f.endswith(".java")
|
||
)
|
||
|
||
|
||
# 类级 @RequestMapping("/api/xxx")
|
||
CLASS_MAPPING_RE = re.compile(r'@RequestMapping\s*\(\s*(?:value\s*=\s*)?"([^"]+)"', re.MULTILINE)
|
||
# 方法级 mapping 注解
|
||
METHOD_MAPPING_RE = re.compile(
|
||
r'@(GetMapping|PostMapping|PutMapping|DeleteMapping)\s*\(\s*(?:value\s*=\s*)?"([^"]*)"',
|
||
re.MULTILINE,
|
||
)
|
||
# 方法签名（紧跟 mapping 注解后的 public 方法）
|
||
METHOD_SIG_RE = re.compile(
|
||
r'public\s+\S+\s+(\w+)\s*\(',
|
||
re.MULTILINE,
|
||
)
|
||
# 类名
|
||
CLASS_NAME_RE = re.compile(r'class\s+(\w+)\b')
|
||
|
||
|
||
def extract_endpoints(java_path):
|
||
"""从单个 controller 文件提取 HTTP 端点：返回 [(http_method, full_path, java_method_name), ...]"""
|
||
with open(java_path, "r", encoding="utf-8") as f:
|
||
content = f.read()
|
||
|
||
class_name_match = CLASS_NAME_RE.search(content)
|
||
class_name = class_name_match.group(1) if class_name_match else os.path.basename(java_path)
|
||
|
||
# 类级基础路径
|
||
class_path = ""
|
||
cm = CLASS_MAPPING_RE.search(content)
|
||
if cm:
|
||
class_path = cm.group(1)
|
||
|
||
endpoints = []
|
||
|
||
# 按方法级 mapping 注解切分
|
||
# 找所有方法级 mapping 注解的位置
|
||
for m in METHOD_MAPPING_RE.finditer(content):
|
||
annotation = m.group(1)
|
||
method_path = m.group(2)
|
||
http_method = {
|
||
"GetMapping": "GET",
|
||
"PostMapping": "POST",
|
||
"PutMapping": "PUT",
|
||
"DeleteMapping": "DELETE",
|
||
}[annotation]
|
||
|
||
# 在该注解之后找最近的方法签名
|
||
rest = content[m.end():]
|
||
sig = METHOD_SIG_RE.search(rest)
|
||
java_method = sig.group(1) if sig else "?"
|
||
|
||
# 拼接完整路径
|
||
full = class_path.rstrip("/") + "/" + method_path.lstrip("/") if method_path else class_path
|
||
# 去掉多余斜杠
|
||
full = re.sub(r"/+", "/", full)
|
||
# 处理通配符路径（如 /api/upload/image/**）——保留通配符用于匹配
|
||
endpoints.append((http_method, full, java_method, class_name))
|
||
|
||
return endpoints
|
||
|
||
|
||
def extract_entity_classes(java_path):
|
||
"""从实体文件提取类名"""
|
||
with open(java_path, "r", encoding="utf-8") as f:
|
||
content = f.read()
|
||
m = re.search(r'@Entity\b', content)
|
||
if not m:
|
||
return None
|
||
cm = CLASS_NAME_RE.search(content)
|
||
return cm.group(1) if cm else None
|
||
|
||
|
||
def load_ontology(repo_root):
|
||
"""加载 ontology.jsonl，返回 actions 与 entities 的端点/类名集合"""
|
||
jsonl_path = os.path.join(repo_root, "docs", "graph", "ontology.jsonl")
|
||
if not os.path.exists(jsonl_path):
|
||
print(f"ERROR: {jsonl_path} not found")
|
||
sys.exit(2)
|
||
|
||
action_endpoints = set() # (HTTP_METHOD, path_pattern)
|
||
action_endpoints_by_id = defaultdict(list)
|
||
entity_class_names = set() # PascalCase 类名
|
||
entity_ids = {} # class_name -> entity_id
|
||
non_jpa_entity_class_names = set() # 本体里标 jpa:false 的（不要求代码有 @Entity 类）
|
||
|
||
with open(jsonl_path, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
obj = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if obj.get("type") == "Action":
|
||
entrypoints = obj.get("entrypoints", [])
|
||
eid = obj.get("id", "?")
|
||
for ep in entrypoints:
|
||
if isinstance(ep, str) and (" " in ep or ep.startswith(("GET", "POST", "PUT", "DELETE"))):
|
||
# 格式 "POST /api/report/create" 或 "ReportController#create"
|
||
parts = ep.split(" ", 1)
|
||
if len(parts) == 2 and parts[0] in ("GET", "POST", "PUT", "DELETE"):
|
||
action_endpoints.add((parts[0], parts[1]))
|
||
action_endpoints_by_id[eid].append((parts[0], parts[1]))
|
||
elif obj.get("type") == "Entity":
|
||
eid = obj.get("id", "")
|
||
# 从 entity:store 推 PascalCase Store
|
||
if eid.startswith("entity:"):
|
||
class_name = "".join(p.capitalize() for p in eid[7:].split("_"))
|
||
# jpa:false 表示本体内建模为领域实体但代码里不是 JPA @Entity（如字段承载或 record）
|
||
if obj.get("jpa") is False:
|
||
non_jpa_entity_class_names.add(class_name)
|
||
else:
|
||
entity_class_names.add(class_name)
|
||
entity_ids[class_name] = eid
|
||
|
||
return action_endpoints, action_endpoints_by_id, entity_class_names, entity_ids, non_jpa_entity_class_names
|
||
|
||
|
||
def path_matches_ontology_pattern(code_path, ontology_path):
|
||
"""代码路径是否匹配本体里的路径模式（支持 ** 通配符）"""
|
||
if code_path == ontology_path:
|
||
return True
|
||
# 处理 ** 通配符
|
||
if "**" in ontology_path:
|
||
# 转换为正则：** → .*, * → [^/]*
|
||
pattern = re.escape(ontology_path).replace(r"\*\*", ".*").replace(r"\*", r"[^/]*")
|
||
return re.fullmatch(pattern, code_path) is not None
|
||
# 处理单 * 通配符（如 /api/report/*/suggestion）
|
||
if "*" in ontology_path:
|
||
pattern = re.escape(ontology_path).replace(r"\*", r"[^/]*")
|
||
return re.fullmatch(pattern, code_path) is not None
|
||
return False
|
||
|
||
|
||
def main():
|
||
repo_root = sys.argv[1] if len(sys.argv) > 1 else "."
|
||
repo_root = os.path.abspath(repo_root)
|
||
|
||
action_endpoints, action_endpoints_by_id, entity_class_names, entity_ids, non_jpa_entity_class_names = load_ontology(repo_root)
|
||
|
||
# === 1. Controller 端点漂移 ===
|
||
controller_files = find_controller_files(repo_root)
|
||
code_endpoints = [] # (http_method, path, java_method, class_name, file)
|
||
for cf in controller_files:
|
||
for http_method, path, java_method, class_name in extract_endpoints(cf):
|
||
code_endpoints.append((http_method, path, java_method, class_name, cf))
|
||
|
||
# 对比
|
||
code_only = [] # 代码里有，本体没有
|
||
for http_method, path, java_method, class_name, cf in code_endpoints:
|
||
found = False
|
||
for ont_method, ont_path in action_endpoints:
|
||
if ont_method == http_method and path_matches_ontology_pattern(path, ont_path):
|
||
found = True
|
||
break
|
||
if not found:
|
||
code_only.append((http_method, path, java_method, class_name, cf))
|
||
|
||
ontology_only = [] # 本体里有，代码没有
|
||
for ont_method, ont_path in action_endpoints:
|
||
found = False
|
||
for http_method, path, java_method, class_name, cf in code_endpoints:
|
||
if http_method == ont_method and path_matches_ontology_pattern(path, ont_path):
|
||
found = True
|
||
break
|
||
if not found:
|
||
ontology_only.append((ont_method, ont_path))
|
||
|
||
# === 2. Entity 类覆盖 ===
|
||
entity_files = find_entity_files(repo_root)
|
||
code_entity_classes = set()
|
||
for ef in entity_files:
|
||
cn = extract_entity_classes(ef)
|
||
if cn:
|
||
code_entity_classes.add(cn)
|
||
|
||
entities_missing_in_ontology = code_entity_classes - entity_class_names
|
||
entities_missing_in_code = entity_class_names - code_entity_classes
|
||
|
||
# non-jpa 本体实体（如 HighlightVideo 字段承载、SessionToken record）不算 extra drift
|
||
# 它们在本体里是领域实体，但代码里不是 @Entity 类，单独列出作为信息项
|
||
non_jpa_in_ontology = non_jpa_entity_class_names
|
||
|
||
# === 输出报告 ===
|
||
print("=" * 60)
|
||
print("宠小它 本体—代码漂移审计")
|
||
print("=" * 60)
|
||
|
||
print(f"\n扫描 controller 文件: {len(controller_files)}")
|
||
print(f"代码 HTTP 端点数: {len(code_endpoints)}")
|
||
print(f"本体 Action 端点数: {len(action_endpoints)}")
|
||
print(f"扫描 entity 文件: {len(entity_files)}")
|
||
print(f"代码 Entity 类数: {len(code_entity_classes)}")
|
||
print(f"本体 Entity 数（JPA）: {len(entity_class_names)}")
|
||
print(f"本体 Entity 数（非 JPA/领域实体）: {len(non_jpa_in_ontology)}")
|
||
|
||
print("\n--- Controller 端点漂移 ---")
|
||
if code_only:
|
||
print(f"\n⚠️ code-only（代码里有，本体未录入）: {len(code_only)} 个")
|
||
for http_method, path, java_method, class_name, cf in sorted(code_only):
|
||
rel = os.path.relpath(cf, repo_root)
|
||
print(f" {http_method:6s} {path:40s} {class_name}#{java_method} [{rel}]")
|
||
else:
|
||
print("\n✅ 无 code-only 漂移（所有代码端点都已录入本体）")
|
||
|
||
if ontology_only:
|
||
print(f"\n⚠️ ontology-only（本体里有，代码找不到）: {len(ontology_only)} 个")
|
||
for ont_method, ont_path in sorted(ontology_only):
|
||
print(f" {ont_method:6s} {ont_path}")
|
||
else:
|
||
print("\n✅ 无 ontology-only 漂移（所有本体端点都有对应代码）")
|
||
|
||
print("\n--- Entity 类覆盖 ---")
|
||
if entities_missing_in_ontology:
|
||
print(f"\n⚠️ 代码里有 Entity 类但本体未录入: {len(entities_missing_in_ontology)} 个")
|
||
for cn in sorted(entities_missing_in_ontology):
|
||
print(f" {cn}")
|
||
else:
|
||
print("\n✅ 所有代码 Entity 类都已录入本体")
|
||
|
||
if entities_missing_in_code:
|
||
print(f"\n⚠️ 本体里有 Entity 但代码找不到: {len(entities_missing_in_code)} 个")
|
||
for cn in sorted(entities_missing_in_code):
|
||
print(f" {cn} (本体 ID: {entity_ids.get(cn, '?')})")
|
||
else:
|
||
print("✅ 所有本体 Entity 都有对应代码类")
|
||
|
||
if non_jpa_in_ontology:
|
||
print(f"\nℹ️ 本体非 JPA 领域实体（代码里不是 @Entity 类，正常）: {len(non_jpa_in_ontology)} 个")
|
||
for cn in sorted(non_jpa_in_ontology):
|
||
print(f" {cn}")
|
||
|
||
# === 退出码 ===
|
||
has_drift = bool(code_only or ontology_only or entities_missing_in_ontology or entities_missing_in_code)
|
||
if has_drift:
|
||
print("\n" + "=" * 60)
|
||
print("漂移汇总: code-only={} ontology-only={} entity-missing={} entity-extra={}".format(
|
||
len(code_only), len(ontology_only), len(entities_missing_in_ontology), len(entities_missing_in_code)
|
||
))
|
||
print("=" * 60)
|
||
sys.exit(1)
|
||
else:
|
||
print("\n" + "=" * 60)
|
||
print("✅ 无漂移：本体与代码完全对齐")
|
||
print("=" * 60)
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|