221 lines
7.4 KiB
Python
221 lines
7.4 KiB
Python
"""task-124 @Transactional 方法审计清单工具。
|
||
|
||
对 backend-java modules 下的 Java 源码做机械化扫描：定位全部方法级
|
||
@Transactional 注解，按方法体内容分类事务内代码段——必须事务内（落库写/状态/锁）
|
||
与可移出（纯计算/DTO 组装/日志/临时文件清理），标记 Python 回调相关方法，
|
||
按可移出收益排序，产出审计清单供 plan 07 的任务 125-131 事务收缩参考。
|
||
本工具只读，不改任何代码。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
|
||
from n1_scan import _match_span, _strip_comments
|
||
|
||
MODULES_ROOT = (
|
||
Path(__file__).resolve().parent.parent / "src/main/java/com/nanri/aiimage/modules"
|
||
)
|
||
|
||
TX_RE = re.compile(r"@Transactional\b")
|
||
METHOD_RE = re.compile(
|
||
r"\s*(?:(?:public|protected|private)\s+)?"
|
||
r"(?:[\w<>,.?\[\] ]+[\s])?(\w+)\s*\("
|
||
)
|
||
PYTHON_FACING_RE = re.compile(r"(?i)(result|submit|upload|chunk|done|ack|report)")
|
||
|
||
WRITE_RE = re.compile(
|
||
r"\.(insert|update|deleteById|deleteBatchIds|removeById|removeByIds|remove\b|"
|
||
r"save|saveBatch|saveOrUpdate|updateById|insertOrUpdate|updateBatchById)\s*\("
|
||
)
|
||
LOCK_RE = re.compile(r"acquireTaskLock|\.tryLock\(|LockHandle|taskLockHandle")
|
||
CLEANUP_RE = re.compile(
|
||
r"Files\.(delete|deleteIfExists|move)|deleteIfExists\(|"
|
||
r"tempDir|tempFile|temp-dir|cleanupTemp|deleteTemp|临时文件"
|
||
)
|
||
COMPUTE_RE = re.compile(
|
||
r"\.stream\(\)|\.map\(|\.collect\(|\.filter\(|\.reduce\(|\.sorted\(|"
|
||
r"Collectors\.|computeIfAbsent|\.distinct\(|\.flatMap\(|\.peek\("
|
||
)
|
||
ASSEMBLY_RE = re.compile(
|
||
r"build\w*(Vo|VO|Dto|DTO)|toVo\(|toEntity\(|convert\w*\(|assemble\w*\(|"
|
||
r"new \w+(Vo|VO|Dto|DTO)\b"
|
||
)
|
||
LOG_RE = re.compile(r"log\.(info|warn|debug|error|trace)\s*\(")
|
||
|
||
|
||
def _match_body_span(text: str, open_pos: int) -> int | None:
|
||
"""从 { 出发，返回匹配的 } 偏移（含）。
|
||
|
||
圆括号深度同步跟踪：lambda/方法调用内的 { } 不计入方法体层级，
|
||
方法体的闭合 } 总是在圆括号外层（paren == 0）首次归零时命中。
|
||
"""
|
||
depth = 0
|
||
paren = 0
|
||
for i in range(open_pos, len(text)):
|
||
c = text[i]
|
||
if c == "(":
|
||
paren += 1
|
||
elif c == ")":
|
||
paren -= 1
|
||
elif c == "{" and paren == 0:
|
||
depth += 1
|
||
elif c == "}" and paren == 0:
|
||
depth -= 1
|
||
if depth == 0:
|
||
return i
|
||
return None
|
||
|
||
|
||
def _count_hits(pattern: re.Pattern[str], text: str) -> int:
|
||
return len(pattern.findall(text))
|
||
|
||
|
||
def scan_text(text: str, filename: str = "fixture.java") -> list[dict]:
|
||
"""扫描单份 Java 源码，返回 @Transactional 方法审计条目（按收益降序）。"""
|
||
stripped = _strip_comments(text)
|
||
findings: list[dict] = []
|
||
pos = 0
|
||
while True:
|
||
tx_match = TX_RE.search(stripped, pos)
|
||
if not tx_match:
|
||
break
|
||
anno_end = tx_match.end()
|
||
if stripped[anno_end : anno_end + 1] == "(":
|
||
close = _match_span(stripped, anno_end)
|
||
if close is None:
|
||
pos = tx_match.end()
|
||
continue
|
||
anno_end = close + 1
|
||
method_match = METHOD_RE.search(stripped, anno_end)
|
||
if not method_match:
|
||
pos = tx_match.end()
|
||
continue
|
||
method_name = method_match.group(1)
|
||
paren_pos = method_match.end() - 1
|
||
close_paren = _match_span(stripped, paren_pos)
|
||
if close_paren is None:
|
||
pos = method_match.end()
|
||
continue
|
||
body_start = close_paren + 1
|
||
while body_start < len(stripped) and stripped[body_start].isspace():
|
||
body_start += 1
|
||
if body_start >= len(stripped) or stripped[body_start] != "{":
|
||
# 抽象方法/接口声明无方法体，跳过
|
||
pos = method_match.end()
|
||
continue
|
||
body_end = _match_body_span(stripped, body_start)
|
||
if body_end is None:
|
||
pos = method_match.end()
|
||
continue
|
||
body = stripped[body_start + 1 : body_end]
|
||
|
||
must_stay: list[str] = []
|
||
counts: dict[str, int] = {}
|
||
if _count_hits(WRITE_RE, body) > 0:
|
||
must_stay.append("write")
|
||
counts["write"] = _count_hits(WRITE_RE, body)
|
||
if _count_hits(LOCK_RE, body) > 0:
|
||
must_stay.append("lock")
|
||
counts["lock"] = _count_hits(LOCK_RE, body)
|
||
|
||
movable: list[str] = []
|
||
for key, pattern in (
|
||
("compute", COMPUTE_RE),
|
||
("assembly", ASSEMBLY_RE),
|
||
("cleanup", CLEANUP_RE),
|
||
("log", LOG_RE),
|
||
):
|
||
hits = _count_hits(pattern, body)
|
||
counts[key] = hits
|
||
if hits > 0:
|
||
movable.append(key)
|
||
|
||
priority = sum(
|
||
counts.get(key, 0) for key in ("compute", "assembly", "cleanup", "log")
|
||
)
|
||
findings.append(
|
||
{
|
||
"file": filename,
|
||
"line": _line_of(text, tx_match.start()),
|
||
"method": method_name,
|
||
"signature": " ".join(
|
||
stripped[method_match.start() : close_paren].split()
|
||
)[:100],
|
||
"class_level": False,
|
||
"python_facing": bool(PYTHON_FACING_RE.search(method_name)),
|
||
"must_stay": must_stay,
|
||
"movable": movable,
|
||
"counts": counts,
|
||
"priority": priority,
|
||
}
|
||
)
|
||
pos = body_end + 1
|
||
|
||
findings.sort(key=lambda f: (-f["priority"], f["line"]))
|
||
return findings
|
||
|
||
|
||
def _line_of(text: str, offset: int) -> int:
|
||
return text.count("\n", 0, offset) + 1
|
||
|
||
|
||
@dataclass
|
||
class Report:
|
||
scanned_files: int = 0
|
||
modules: list[str] = field(default_factory=list)
|
||
findings: list[dict] = field(default_factory=list)
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"scanned_files": self.scanned_files,
|
||
"modules": self.modules,
|
||
"findings": self.findings,
|
||
}
|
||
|
||
|
||
def scan_modules(root: Path) -> Report:
|
||
"""扫描 modules 根目录下全部 Java 文件，产出 @Transactional 审计报告。"""
|
||
report = Report()
|
||
module_set: set[str] = set()
|
||
for java_file in sorted(root.rglob("*.java")):
|
||
report.scanned_files += 1
|
||
rel = java_file.relative_to(root)
|
||
module_set.add(rel.parts[0])
|
||
for finding in scan_text(java_file.read_text(encoding="utf-8"), str(rel)):
|
||
finding["module"] = rel.parts[0]
|
||
report.findings.append(finding)
|
||
report.findings.sort(key=lambda f: (-f["priority"], f["file"], f["line"]))
|
||
report.modules = sorted(module_set)
|
||
return report
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="@Transactional 方法审计")
|
||
parser.add_argument("--json", help="输出 JSON 报告路径（默认 stdout）")
|
||
args = parser.parse_args()
|
||
|
||
report = scan_modules(MODULES_ROOT)
|
||
payload = report.to_dict()
|
||
if args.json:
|
||
Path(args.json).write_text(
|
||
json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8"
|
||
)
|
||
else:
|
||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
||
print(
|
||
f"scanned {report.scanned_files} files, "
|
||
f"{len(report.findings)} @Transactional methods "
|
||
f"across {len(report.modules)} modules",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|