期間内のコミット取得・変更ファイル一覧の抽出
Git Bash
cd /c/path/to/repo
git fetch --all --prune
SINCE="2026-07-01"
UNTIL="2026-08-01" # 7/31 まで含む
BR="origin/develop"
# ① 期間内コミット一覧
git -c core.quotepath=false --no-pager log "$BR" \
--since="$SINCE" --until="$UNTIL" --no-merges \
--date=iso-strict --pretty=format:'%h%x09%ad%x09%an%x09%s'
# ② 変更ファイル(コミット単位・重複除去なし)
git -c core.quotepath=false --no-pager log "$BR" \
--since="$SINCE" --until="$UNTIL" --no-merges \
-M --name-status --date=short \
--pretty=format:'@@@%x09%h%x09%ad%x09%an%x09%s'
# ②' 変更ファイル(ユニーク・変更回数つき=優先度の材料)
git -c core.quotepath=false --no-pager log "$BR" \
--since="$SINCE" --until="$UNTIL" --no-merges --name-only --pretty=format: \
| grep -v '^$' | sort | uniq -c | sort -rn
# ②'' 累積差分(テスト対象の確定リスト)
BASE=$(git rev-list -n1 --before="$SINCE" "$BR")
HEAD_C=$(git rev-list -n1 --before="$UNTIL" "$BR")
git -c core.quotepath=false --no-pager diff --name-status -M "$BASE" "$HEAD_C" \
-- ':(exclude)*.md' ':(exclude)docs/**'
Python
python git_change_extract.py -r C:\work\myapp -b origin/develop --since 2026-07-01 --until 2026-08-01 -o .\out
python git_change_extract.py -r C:\work\myapp --range v1.2.0..v1.3.0 -o .\out
git_change_extract.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
git_change_extract.py
Step 1: 期間内のコミット取得
Step 2: 変更ファイル一覧の抽出
出力 (--out ディレクトリ):
01_commits.csv … コミット1件 = 1行
02_commit_files.csv … コミット x 変更ファイル = 1行(追跡用)
03_files_summary.csv … ファイル1件 = 1行(変更回数・分類つき/③④の作業台)
04_net_diff.csv … 期間の累積差分(テスト対象の確定リスト)
使用例:
python git_change_extract.py -r C:/work/myapp -b origin/develop ^
--since 2026-07-01 --until 2026-08-01 -o ./out
python git_change_extract.py -r C:/work/myapp --range v1.2.0..v1.3.0 -o ./out
"""
import argparse
import csv
import subprocess
import sys
from collections import OrderedDict
from pathlib import Path
EMPTY_TREE = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"
SENTINEL = "@@C@@"
DEFAULT_EXCLUDES = [
":(exclude)*.md",
":(exclude)docs/**",
":(exclude).gitignore",
":(exclude).gitattributes",
]
# ---------------------------------------------------------------- git 実行
def git(repo, args, check=True):
cmd = [
"git", "-C", repo,
"-c", "core.quotepath=false",
"-c", "i18n.logOutputEncoding=utf-8",
"--no-pager",
] + args
p = subprocess.run(cmd, capture_output=True)
out = p.stdout.decode("utf-8", "replace")
if p.returncode != 0:
err = p.stderr.decode("utf-8", "replace")
if check:
sys.stderr.write(err + "\n")
raise SystemExit("git command failed: git " + " ".join(args))
return ""
return out
# ------------------------------------------------------- ファイル分類(③の下ごしらえ)
def categorize(path):
"""Spring Boot / Thymeleaf の典型構成を前提にレイヤ分類する。
プロジェクトの命名規約に合わせてここを直すのが一番効く。"""
p = path.replace("\\", "/")
low = p.lower()
name = p.rsplit("/", 1)[-1]
if "src/test/" in p:
return "テストコード"
if name.endswith("Controller.java"):
return "Controller(エンドポイント)"
if name.endswith(("Service.java", "ServiceImpl.java")) or "/service/" in low:
return "Service(業務ロジック)"
if name.endswith(("Repository.java", "Dao.java", "DaoImpl.java", "Mapper.java")) \
or "/repository/" in low or "/dao/" in low:
return "Repository/DAO"
if name.endswith("Mapper.xml"):
return "Repository/DAO(SQL)"
if name.endswith(("Form.java", "Dto.java", "Request.java", "Response.java")) \
or "/form/" in low or "/dto/" in low:
return "Form/DTO"
if name.endswith("Entity.java") or "/entity/" in low or "/domain/" in low or "/model/" in low:
return "Entity/ドメイン"
if name.endswith(("Config.java", "Configuration.java")) or "/config/" in low:
return "設定(Java Config)"
if "securit" in low and name.endswith(".java"):
return "設定(セキュリティ)"
if name.endswith(("Interceptor.java", "Filter.java", "Advice.java", "Handler.java", "Aspect.java")):
return "横断的関心事(Filter/Advice等)"
if name.endswith(("Validator.java",)) or "/validat" in low:
return "バリデーション"
if "/templates/" in low and low.endswith(".html"):
return "画面(Thymeleaf)"
if low.endswith(".html"):
return "画面(HTML)"
if low.endswith(".js"):
return "JavaScript"
if low.endswith((".css", ".scss")):
return "CSS"
if low.endswith(".sql") or "/db/migration/" in low or "flyway" in low or "liquibase" in low:
return "DB/マイグレーション"
if name.startswith("messages") and low.endswith(".properties"):
return "メッセージ定義"
if name.startswith("application") and low.endswith((".yml", ".yaml", ".properties")):
return "設定(application)"
if name in ("pom.xml", "build.gradle", "build.gradle.kts", "settings.gradle",
"package.json", "package-lock.json"):
return "ビルド/依存"
if low.endswith(".java"):
return "Java(その他)"
return "その他"
# ---------------------------------------------------------------- パース
def parse_name_status_line(line):
"""'M\tpath' / 'R100\told\tnew' を (status, path, old_path) に。"""
parts = line.split("\t")
status = parts[0]
if status.startswith(("R", "C")) and len(parts) >= 3:
return status, parts[2], parts[1]
if len(parts) >= 2:
return status, parts[1], ""
return None
def collect_commits(repo, rev_args, pathspec, no_merges, first_parent):
args = ["log"] + rev_args
if no_merges:
args.append("--no-merges")
if first_parent:
args.append("--first-parent")
args += [
"-M", "--name-status", "--date=iso-strict",
"--pretty=format:%s\t%%H\t%%h\t%%ad\t%%an\t%%ae\t%%s" % SENTINEL,
]
if pathspec:
args += ["--"] + pathspec
out = git(repo, args)
commits = []
cur = None
for raw in out.splitlines():
line = raw.rstrip("\r")
if not line.strip():
continue
if line.startswith(SENTINEL + "\t"):
f = line.split("\t")
cur = {
"hash": f[1], "short": f[2], "date": f[3],
"author": f[4], "email": f[5],
"subject": "\t".join(f[6:]) if len(f) > 6 else "",
"files": [],
}
commits.append(cur)
elif cur is not None:
rec = parse_name_status_line(line)
if rec:
cur["files"].append(rec)
return commits
def collect_net_diff(repo, base, head, pathspec):
args = ["diff", "--name-status", "-M", base, head]
if pathspec:
args += ["--"] + pathspec
out = git(repo, args)
result = []
for raw in out.splitlines():
rec = parse_name_status_line(raw.rstrip("\r"))
if rec:
result.append(rec)
return result
def resolve_range(repo, args):
"""(rev_args_for_log, base, head) を返す。"""
if args.range:
base, _, head = args.range.partition("..")
head = head or "HEAD"
base = git(repo, ["rev-parse", base]).strip()
head = git(repo, ["rev-parse", head]).strip()
return [args.range], base, head
rev = [args.branch]
if args.since:
rev.append("--since=" + args.since)
if args.until:
rev.append("--until=" + args.until)
head = ""
if args.until:
head = git(repo, ["rev-list", "-n1", "--before=" + args.until, args.branch]).strip()
if not head:
head = git(repo, ["rev-parse", args.branch]).strip()
base = ""
if args.since:
base = git(repo, ["rev-list", "-n1", "--before=" + args.since, args.branch]).strip()
if not base:
base = EMPTY_TREE # 期間開始前にコミットが無い場合
return rev, base, head
# ---------------------------------------------------------------- 出力
def write_csv(path, header, rows):
with open(path, "w", encoding="utf-8-sig", newline="") as f:
w = csv.writer(f)
w.writerow(header)
w.writerows(rows)
print(" wrote %-22s %5d rows" % (path.name, len(rows)))
def main():
ap = argparse.ArgumentParser()
ap.add_argument("-r", "--repo", required=True, help="リポジトリのパス")
ap.add_argument("-b", "--branch", default="HEAD", help="対象ブランチ (例: origin/develop)")
ap.add_argument("--since", help="開始日 YYYY-MM-DD")
ap.add_argument("--until", help="終了日 YYYY-MM-DD (この日時「未満」。終日含めるなら翌日を指定)")
ap.add_argument("--range", help="日付の代わりにリビジョン範囲で指定 (例: v1.2.0..v1.3.0)")
ap.add_argument("-o", "--out", default="./out", help="出力ディレクトリ")
ap.add_argument("--include-merges", action="store_true",
help="マージコミットを除外しない(通常は不要)")
ap.add_argument("--first-parent", action="store_true",
help="第一親のみ辿る(マージ単位で見たい場合)")
ap.add_argument("--no-default-excludes", action="store_true",
help="*.md 等のデフォルト除外を無効化")
ap.add_argument("--exclude", action="append", default=[],
help="追加除外 pathspec (例: ':(exclude)legacy/**')")
ap.add_argument("--no-fetch", action="store_true", help="git fetch を実行しない")
args = ap.parse_args()
repo = args.repo
if not (Path(repo) / ".git").exists():
sys.stderr.write("警告: %s に .git が見当たりません\n" % repo)
if not args.no_fetch:
print("fetching ...")
git(repo, ["fetch", "--all", "--prune"], check=False)
pathspec = list(args.exclude)
if not args.no_default_excludes:
pathspec = DEFAULT_EXCLUDES + pathspec
rev_args, base, head = resolve_range(repo, args)
print("range : %s .. %s" % (base[:12], head[:12]))
commits = collect_commits(repo, rev_args, pathspec,
no_merges=not args.include_merges,
first_parent=args.first_parent)
net = collect_net_diff(repo, base, head, pathspec)
outdir = Path(args.out)
outdir.mkdir(parents=True, exist_ok=True)
# --- 01 コミット一覧
write_csv(outdir / "01_commits.csv",
["hash", "short", "date", "author", "email", "subject", "changed_files"],
[[c["hash"], c["short"], c["date"], c["author"], c["email"],
c["subject"], len(c["files"])] for c in commits])
# --- 02 コミット x ファイル
rows = []
for c in commits:
for status, path, old in c["files"]:
rows.append([c["short"], c["date"], c["author"], c["subject"],
status, path, old, categorize(path)])
write_csv(outdir / "02_commit_files.csv",
["short", "date", "author", "subject", "status", "path", "old_path", "分類"],
rows)
# --- 04 累積差分
net_map = {p: (s, o) for s, p, o in net}
write_csv(outdir / "04_net_diff.csv",
["status", "path", "old_path", "分類"],
[[s, p, o, categorize(p)] for s, p, o in net])
# --- 03 ファイル別サマリ
agg = OrderedDict()
for c in commits:
for status, path, old in c["files"]:
a = agg.setdefault(path, {
"count": 0, "authors": set(), "statuses": set(),
"first": c["date"], "last": c["date"], "commits": [],
})
a["count"] += 1
a["authors"].add(c["author"])
a["statuses"].add(status[0])
a["first"] = min(a["first"], c["date"])
a["last"] = max(a["last"], c["date"])
if len(a["commits"]) < 10:
a["commits"].append(c["short"])
rows = []
for path, a in sorted(agg.items(), key=lambda kv: (-kv[1]["count"], kv[0])):
n = net_map.get(path)
rows.append([
path, categorize(path), a["count"],
"/".join(sorted(a["statuses"])),
n[0] if n else "(期間内で相殺)",
len(a["authors"]), ", ".join(sorted(a["authors"])),
a["first"][:10], a["last"][:10],
" ".join(a["commits"]),
])
write_csv(outdir / "03_files_summary.csv",
["path", "分類", "変更回数", "status集合", "net_status",
"担当者数", "担当者", "初回変更日", "最終変更日", "コミット(最大10)"],
rows)
# --- コンソールサマリ
print("\n--- summary ---")
print("commits : %d" % len(commits))
print("touched files : %d" % len(agg))
print("net changed files : %d" % len(net))
bycat = {}
for path in net_map:
c = categorize(path)
bycat[c] = bycat.get(c, 0) + 1
print("\n累積差分の分類内訳:")
for k, v in sorted(bycat.items(), key=lambda kv: -kv[1]):
print(" %-26s %3d" % (k, v))
if __name__ == "__main__":
main()
変更ファイルから影響する画面・エンドポイントへの逆引き
python impact_trace.py -r C:\work\myapp --diff .\out\04_net_diff.csv -o .\out
Impact_trace.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
impact_trace.py
Step 3: 変更ファイル -> 呼び出し元Controller -> エンドポイント -> JS -> 画面 の逆引き
前提構成 (--java-dir 等で変更可):
src/main/java/... Java
src/main/resources/templates Thymeleaf (SPAのガワ)
src/main/resources/static JS/CSS
入力:
--diff Step2 で出力した 04_net_diff.csv (列: status,path,old_path,分類)
省略時はインデックス構築のみ行う
出力 (--out):
05_endpoints.csv エンドポイント棚卸し
06_js_api_calls.csv JS -> API呼び出し -> 突合結果
07_screens.csv テンプレート -> JS / ページURL
08_impact.csv ★ 変更ファイル -> 影響エンドポイント/JS/画面
09_unresolved.csv 静的に解決できなかったもの(手動確認用)
使用例:
python impact_trace.py -r C:/work/myapp --diff ./out/04_net_diff.csv -o ./out
"""
import argparse
import csv
import re
import sys
from collections import defaultdict, deque
from pathlib import Path
# ------------------------------------------------------------------ 共通
MAPPING_ANNO = {
"GetMapping": "GET", "PostMapping": "POST", "PutMapping": "PUT",
"DeleteMapping": "DELETE", "PatchMapping": "PATCH", "RequestMapping": "",
}
JAVA_KEYWORDS = {
"String", "Integer", "Long", "Boolean", "Double", "Float", "Object", "List",
"Map", "Set", "Optional", "Override", "Exception", "RuntimeException",
"Math", "System", "Arrays", "Collections", "LocalDate", "LocalDateTime",
"BigDecimal", "Class", "Void", "Character", "Byte", "Short", "Number",
}
def read_text(p):
for enc in ("utf-8", "utf-8-sig", "cp932"):
try:
return p.read_text(encoding=enc)
except (UnicodeDecodeError, ValueError):
continue
return p.read_text(encoding="utf-8", errors="replace")
def strip_comments(src):
src = re.sub(r"/\*.*?\*/", " ", src, flags=re.S)
src = re.sub(r"//[^\n]*", " ", src)
return src
def strip_strings(src):
src = re.sub(r'"(?:\\.|[^"\\])*"', '""', src)
src = re.sub(r"'(?:\\.|[^'\\])*'", "''", src)
return src
def norm_path(p, root):
try:
return str(p.relative_to(root)).replace("\\", "/")
except ValueError:
return str(p).replace("\\", "/")
def join_url(*parts):
segs = []
for p in parts:
if not p:
continue
segs.extend([s for s in p.split("/") if s])
return "/" + "/".join(segs) if segs else "/"
# ------------------------------------------------------------------ Java 解析
def anno_arg_path(text):
"""@GetMapping("/x") / @RequestMapping(value = {"/x"}) から "/x" を取り出す"""
m = re.search(r'\(\s*(?:value\s*=\s*|path\s*=\s*)?\{?\s*"([^"]*)"', text)
return m.group(1) if m else ""
def anno_http_method(text):
m = re.search(r"RequestMethod\.(\w+)", text)
return m.group(1) if m else ""
def find_body(src, open_paren_pos):
"""メソッドの ( の位置から本体 {...} を返す"""
i = open_paren_pos
depth = 0
while i < len(src):
if src[i] == "(":
depth += 1
elif src[i] == ")":
depth -= 1
if depth == 0:
break
i += 1
j = src.find("{", i)
if j < 0:
return ""
depth = 0
k = j
while k < len(src):
if src[k] == "{":
depth += 1
elif src[k] == "}":
depth -= 1
if depth == 0:
return src[j:k + 1]
k += 1
return src[j:]
def parse_java(path, rel):
src_raw = read_text(path)
src = strip_comments(src_raw)
info = {
"file": rel,
"name": path.stem,
"package": "",
"kind": "other",
"implements": [],
"extends": "",
"refs": set(),
"endpoints": [],
"views": [],
}
m = re.search(r"^\s*package\s+([\w.]+)\s*;", src, re.M)
if m:
info["package"] = m.group(1)
header = src[: src.find("{") + 1] if "{" in src else src
if re.search(r"@RestController\b", src):
info["kind"] = "restcontroller"
elif re.search(r"@Controller\b", src):
info["kind"] = "controller"
elif re.search(r"@Service\b", src):
info["kind"] = "service"
elif re.search(r"@(Repository|Mapper)\b", src):
info["kind"] = "repository"
elif re.search(r"@(Component|Configuration|ControllerAdvice|RestControllerAdvice)\b", src):
info["kind"] = "component"
elif re.search(r"@(Entity|Table)\b", src):
info["kind"] = "entity"
m = re.search(r"\b(?:class|interface)\s+\w+[^{]*?\bimplements\s+([^{]+)\{", src)
if m:
for t in re.split(r",", re.sub(r"<[^>]*>", "", m.group(1))):
t = t.strip().split(".")[-1]
if t:
info["implements"].append(t)
m = re.search(r"\bclass\s+\w+\s+extends\s+([\w.]+)", src)
if m:
info["extends"] = m.group(1).split(".")[-1]
# 参照している型(文字列を除去してから大文字始まり識別子を収集)
body_only = strip_strings(src)
for tok in re.findall(r"\b[A-Z][A-Za-z0-9_]*\b", body_only):
if tok != info["name"] and tok not in JAVA_KEYWORDS:
info["refs"].add(tok)
# ---- クラスレベル @RequestMapping
class_pos = re.search(r"\b(class|interface)\s+" + re.escape(path.stem), src)
class_path = ""
if class_pos:
head = src[: class_pos.start()]
cm = None
for cm in re.finditer(r"@RequestMapping\s*\([^)]*\)", head):
pass
if cm:
class_path = anno_arg_path(cm.group(0))
if info["kind"] not in ("restcontroller", "controller"):
return info
# ---- メソッドレベル
anno_re = re.compile(r"@(" + "|".join(MAPPING_ANNO) + r")\b\s*(\([^()]*(?:\([^()]*\)[^()]*)*\))?")
for am in anno_re.finditer(src):
if class_pos and am.start() < class_pos.start():
continue # クラスレベルのものはスキップ
anno_name = am.group(1)
anno_text = am.group(0)
seg_start = am.end()
seg = src[seg_start: seg_start + 1200]
# 後続の他アノテーションを剥がす
extra_annos = ""
while True:
mm = re.match(r"\s*@\w+\s*(\([^()]*(?:\([^()]*\)[^()]*)*\))?\s*", seg)
if not mm:
break
extra_annos += mm.group(0)
seg = seg[mm.end():]
sig = re.match(
r"\s*(?:public|protected|private)?\s*(?:static\s+|final\s+|synchronized\s+)*"
r"([\w<>\[\],.\s?]+?)\s+(\w+)\s*\(", seg)
if not sig:
continue
ret_type, meth_name = sig.group(1).strip(), sig.group(2)
http = MAPPING_ANNO[anno_name] or anno_http_method(anno_text) or "ANY"
url = join_url(class_path, anno_arg_path(anno_text))
is_body = ("@ResponseBody" in extra_annos or "@ResponseBody" in src[:am.start()][-200:]
or info["kind"] == "restcontroller")
view = ""
if not is_body and ret_type.split("<")[0] in ("String", "ModelAndView", "Object"):
body = find_body(src, seg_start + (len(extra_annos) + sig.end() - 1))
vm = re.findall(r'return\s+(?:new\s+ModelAndView\s*\(\s*)?"([^"]+)"', body)
for v in vm:
if not v.startswith(("redirect:", "forward:")) and "/" != v:
view = v
break
ep = {
"http": http, "url": url, "cls": info["name"], "file": rel,
"method": meth_name, "kind": "api" if is_body else "page",
"view": view,
}
info["endpoints"].append(ep)
if view:
info["views"].append((view, url))
return info
# ------------------------------------------------------------------ JS 解析
STR_RE = re.compile(r"""(['"`])((?:\\.|(?!\1)[^\\])*)\1""")
METHOD_HINTS = [
(re.compile(r"\$\.getJSON\s*\($"), "GET"),
(re.compile(r"\$\.get\s*\($"), "GET"),
(re.compile(r"\$\.post\s*\($"), "POST"),
(re.compile(r"axios\.(\w+)\s*\($"), None),
(re.compile(r"fetch\s*\($"), "GET"),
]
def guess_http(js, pos):
back = js[max(0, pos - 250): pos]
m = re.search(r"axios\.(get|post|put|delete|patch)\s*\(\s*$", back)
if m:
return m.group(1).upper()
if re.search(r"\$\.getJSON\s*\(\s*$", back) or re.search(r"\$\.get\s*\(\s*$", back):
return "GET"
if re.search(r"\$\.post\s*\(\s*$", back):
return "POST"
m = re.findall(r"(?:type|method)\s*:\s*['\"](\w+)['\"]", back)
if m:
return m[-1].upper()
if re.search(r"fetch\s*\(\s*$", back):
return "GET"
fwd_ctx = js[pos: pos + 250]
m = re.findall(r"(?:type|method)\s*:\s*['\"](\w+)['\"]", fwd_ctx)
if m:
return m[0].upper()
return "?"
def normalize_url(u):
u = u.split("?")[0].split("#")[0]
u = re.sub(r"\$\{[^}]*\}", "*", u) # `${id}`
u = re.sub(r"/{2,}", "/", u)
return u
CONCAT_PLUS = re.compile(r"\s*\+\s*")
OPERAND = re.compile(r"[A-Za-z_$][\w$.]*(?:\([^()]*\))?(?:\[[^\]]*\])?")
def expand_concat(js, m):
"""'/api/users/' + id + '/orders' を /api/users/*/orders に組み立てる。
戻り値: (連結後文字列, 消費した終端位置)"""
combined = m.group(2)
end = m.end()
while True:
pm = CONCAT_PLUS.match(js, end)
if not pm:
break
p = pm.end()
sm = STR_RE.match(js, p)
if sm:
combined += sm.group(2)
end = sm.end()
continue
om = OPERAND.match(js, p)
if om:
combined += "*"
end = om.end()
continue
break
return combined, end
def parse_js(path, rel):
js = strip_comments(read_text(path))
calls = []
consumed_to = -1
for m in STR_RE.finditer(js):
if m.start() < consumed_to:
continue # 連結の一部として取り込み済み
raw, consumed_to = expand_concat(js, m)
if not raw:
continue
if not (raw.startswith("/") or raw.startswith("api/") or "/api/" in raw):
continue
if re.search(r"\.(js|css|png|jpe?g|gif|svg|ico|woff2?)$", raw, re.I):
continue
if len(raw) > 200:
continue
norm = normalize_url(raw)
if not norm.startswith("/"):
norm = "/" + norm
calls.append({
"js": rel, "raw": raw, "url": norm,
"http": guess_http(js, m.start()),
"dynamic": ("*" in norm),
})
# 同一 (url,http) は集約
seen, out = set(), []
for c in calls:
k = (c["url"], c["http"])
if k in seen:
continue
seen.add(k)
out.append(c)
return out
# ------------------------------------------------------------------ テンプレート解析
SCRIPT_RE = re.compile(r"<script[^>]*?\b(?:th:src|src|data-src)\s*=\s*[\"']([^\"']+)[\"']", re.I)
FRAG_RE = re.compile(r"(?:th:replace|th:insert|th:include|layout:decorate)\s*=\s*[\"']\s*~?\{?\s*([\w/\-.]+)", re.I)
def parse_template(path, rel):
html = read_text(path)
scripts = []
for m in SCRIPT_RE.finditer(html):
s = m.group(1)
s = re.sub(r"^@\{|\}$", "", s.strip())
s = s.split("?")[0]
if s.startswith(("http://", "https://", "//")):
continue
scripts.append(s)
frags = []
for m in FRAG_RE.finditer(html):
f = m.group(1).strip()
if f and not f.startswith("$"):
frags.append(re.sub(r"\.html$", "", f))
inline = bool(re.search(r"\$\.ajax|fetch\s*\(|axios\.", html))
return {"template": rel, "scripts": scripts, "fragments": frags, "inline_api": inline}
# ------------------------------------------------------------------ URL 突合
def to_segs(url):
return [s for s in url.split("/") if s]
def seg_match(a, b):
"""a: JS側セグメント, b: エンドポイント側セグメント"""
if a == "*" or (b.startswith("{") and b.endswith("}")) or b == "*":
return True
return a == b
def match_url(js_url, endpoints, context_path=""):
cands = [js_url]
if context_path and js_url.startswith(context_path):
cands.append(js_url[len(context_path):] or "/")
best = None
for cand in cands:
js_segs = to_segs(cand)
for ep in endpoints:
ep_segs = to_segs(ep["url"])
if len(js_segs) != len(ep_segs):
continue
if all(seg_match(a, b) for a, b in zip(js_segs, ep_segs)):
score = sum(0 if (b.startswith("{") or a == "*") else 1
for a, b in zip(js_segs, ep_segs))
if best is None or score > best[1]:
best = (ep, score, "exact")
if best:
return best[0], best[2]
# プレフィックス一致(要確認)
for cand in cands:
js_segs = to_segs(cand)
for ep in endpoints:
ep_segs = to_segs(ep["url"])
n = min(len(js_segs), len(ep_segs))
if n >= 2 and all(seg_match(a, b) for a, b in zip(js_segs[:n], ep_segs[:n])):
return ep, "prefix(要確認)"
return None, "unmatched"
# ------------------------------------------------------------------ メイン
def main():
ap = argparse.ArgumentParser()
ap.add_argument("-r", "--repo", required=True)
ap.add_argument("--diff", help="04_net_diff.csv のパス")
ap.add_argument("-o", "--out", default="./out")
ap.add_argument("--java-dir", default="src/main/java")
ap.add_argument("--templates-dir", default="src/main/resources/templates")
ap.add_argument("--static-dir", default="src/main/resources/static")
ap.add_argument("--context-path", default="", help="server.servlet.context-path (例: /myapp)")
ap.add_argument("--max-depth", type=int, default=8)
args = ap.parse_args()
repo = Path(args.repo)
outdir = Path(args.out)
outdir.mkdir(parents=True, exist_ok=True)
# ---------------- インデックス構築
print("scanning java ...")
classes = {}
dup = defaultdict(list)
jroot = repo / args.java_dir
for p in jroot.rglob("*.java"):
rel = norm_path(p, repo)
info = parse_java(p, rel)
if info["name"] in classes:
dup[info["name"]].append(rel)
classes[info["name"]] = info
print(" classes: %d" % len(classes))
# interface -> impl
impl_of = defaultdict(set)
for name, info in classes.items():
for i in info["implements"]:
if i in classes:
impl_of[i].add(name)
if info["extends"] in classes:
impl_of[info["extends"]].add(name)
# 参照グラフ(interface参照をimplにも展開)
fwd = defaultdict(set)
for name, info in classes.items():
for r in info["refs"]:
if r in classes and r != name:
fwd[name].add(r)
for im in impl_of.get(r, ()):
if im != name:
fwd[name].add(im)
rev = defaultdict(set)
for a, bs in fwd.items():
for b in bs:
rev[b].add(a)
endpoints = []
for info in classes.values():
endpoints.extend(info["endpoints"])
api_eps = [e for e in endpoints if e["kind"] == "api"]
print(" endpoints: %d (api=%d, page=%d)"
% (len(endpoints), len(api_eps), len(endpoints) - len(api_eps)))
print("scanning js ...")
js_calls = []
sroot = repo / args.static_dir
js_files = []
if sroot.exists():
for p in sroot.rglob("*.js"):
if re.search(r"[\\/](lib|vendor|node_modules|dist)[\\/]", str(p)) or p.name.endswith(".min.js"):
continue
rel = norm_path(p, repo)
js_files.append(rel)
js_calls.extend(parse_js(p, rel))
print(" js files: %d, api calls: %d" % (len(js_files), len(js_calls)))
print("scanning templates ...")
templates = {}
troot = repo / args.templates_dir
for p in troot.rglob("*.html"):
rel = norm_path(p, repo)
templates[rel] = parse_template(p, rel)
print(" templates: %d" % len(templates))
# ---- テンプレート名 -> 実ファイル
def resolve_template(view):
cand = "%s/%s.html" % (args.templates_dir, view.strip("/"))
if cand in templates:
return cand
for t in templates:
if t.endswith("/" + view.strip("/") + ".html"):
return t
return ""
# ---- fragment 展開して、テンプレートが読み込むJSを確定
def resolve_js(src):
s = src.lstrip("/")
if args.context_path and src.startswith(args.context_path):
s = src[len(args.context_path):].lstrip("/")
cand = "%s/%s" % (args.static_dir, s)
if cand in js_files:
return cand
base = s.rsplit("/", 1)[-1]
hits = [f for f in js_files if f.endswith("/" + base)]
return hits[0] if len(hits) == 1 else ""
tmpl_js = {}
for rel, t in templates.items():
seen_t, stack, acc = set(), [rel], set()
while stack:
cur = stack.pop()
if cur in seen_t or cur not in templates:
continue
seen_t.add(cur)
for s in templates[cur]["scripts"]:
r = resolve_js(s)
acc.add(r if r else "?" + s)
for f in templates[cur]["fragments"]:
rt = resolve_template(f)
if rt:
stack.append(rt)
tmpl_js[rel] = acc
js_to_tmpl = defaultdict(set)
for t, jss in tmpl_js.items():
for j in jss:
js_to_tmpl[j].add(t)
# ---- ページController: view -> URL
view_to_pageurl = defaultdict(set)
for info in classes.values():
for view, url in info["views"]:
rt = resolve_template(view)
if rt:
view_to_pageurl[rt].add(url)
# ---- JS呼び出しとエンドポイントの突合
unresolved = []
ep_key = lambda e: "%s %s" % (e["http"], e["url"])
ep_to_js = defaultdict(set)
for c in js_calls:
ep, how = match_url(c["url"], api_eps, args.context_path)
c["matched"] = ep_key(ep) if ep else ""
c["matched_cls"] = ep["cls"] if ep else ""
c["how"] = how
if ep:
ep_to_js[ep_key(ep)].add(c["js"])
else:
unresolved.append(["js->endpoint", "%s : %s" % (c["js"], c["raw"]),
"対応するエンドポイントが見つからない(動的URL/別プレフィックス等)"])
for t, jss in tmpl_js.items():
for j in jss:
if j.startswith("?"):
unresolved.append(["template->js", "%s : %s" % (t, j[1:]),
"script src から実ファイルを特定できない"])
for n, files in dup.items():
unresolved.append(["java", n, "同名クラスが複数: " + ", ".join(files)])
# ---------------- 出力: インデックス系
def w(path, header, rows):
with open(outdir / path, "w", encoding="utf-8-sig", newline="") as f:
wr = csv.writer(f)
wr.writerow(header)
wr.writerows(rows)
print(" wrote %-22s %5d rows" % (path, len(rows)))
w("05_endpoints.csv",
["http", "url", "種別", "controller", "java_method", "view", "file", "呼ぶJS"],
sorted([[e["http"], e["url"], e["kind"], e["cls"], e["method"], e["view"], e["file"],
", ".join(sorted(ep_to_js.get(ep_key(e), [])))] for e in endpoints],
key=lambda r: (r[1], r[0])))
w("06_js_api_calls.csv",
["js_file", "http(推定)", "raw_url", "normalized", "matched_endpoint", "controller", "突合"],
[[c["js"], c["http"], c["raw"], c["url"], c["matched"], c["matched_cls"], c["how"]]
for c in sorted(js_calls, key=lambda c: (c["js"], c["url"]))])
w("07_screens.csv",
["template", "画面URL", "読み込むJS", "include先", "HTML内に直書きAPI"],
[[t,
", ".join(sorted(view_to_pageurl.get(t, []))) or "(ページControllerから特定できず)",
", ".join(sorted(x for x in tmpl_js[t] if not x.startswith("?"))),
", ".join(templates[t]["fragments"]),
"yes" if templates[t]["inline_api"] else ""]
for t in sorted(templates)])
# ---------------- 影響トレース
if not args.diff:
w("09_unresolved.csv", ["種別", "対象", "理由"], unresolved)
print("\n--diff 未指定のためインデックスのみ出力しました。")
return
with open(args.diff, encoding="utf-8-sig") as f:
changed = list(csv.DictReader(f))
def controllers_for(cls_name):
"""変更クラスから逆向きBFSして到達するControllerと経路を返す"""
if cls_name not in classes:
return []
results = []
seen = {cls_name}
q = deque([(cls_name, [cls_name])])
while q:
cur, path = q.popleft()
if len(path) > args.max_depth:
continue
if classes[cur]["kind"] in ("restcontroller", "controller") and len(path) > 1:
results.append((cur, path))
continue
if classes[cur]["kind"] in ("restcontroller", "controller") and len(path) == 1:
results.append((cur, path))
continue
for nxt in rev.get(cur, ()):
if nxt not in seen:
seen.add(nxt)
q.append((nxt, path + [nxt]))
return results
rows = []
for row in changed:
path = row.get("path", "").replace("\\", "/")
status = row.get("status", "")
cat = row.get("分類", "")
eps, jss, tmpls, routes, notes = set(), set(), set(), set(), []
conf = "高"
if path.endswith(".java"):
stem = path.rsplit("/", 1)[-1][:-5]
if stem not in classes:
notes.append("クラス未検出(削除済み?)")
conf = "低"
for cls, route in controllers_for(stem):
routes.add(" -> ".join(reversed(route)))
for e in classes[cls]["endpoints"]:
eps.add(ep_key(e))
if e["kind"] == "page" and e["view"]:
rt = resolve_template(e["view"])
if rt:
tmpls.add(rt)
if not routes:
notes.append("Controllerに到達せず(バッチ/共通部品/DI未解決の可能性)")
conf = "低"
elif path.endswith(".js"):
jss.add(path)
tmpls |= js_to_tmpl.get(path, set())
for c in js_calls:
if c["js"] == path and c["matched"]:
eps.add(c["matched"])
if not tmpls:
notes.append("このJSを読むテンプレートが見つからない(動的ロード?)")
conf = "中"
elif path.endswith(".html"):
if path in templates:
tmpls.add(path)
jss |= {x for x in tmpl_js[path] if not x.startswith("?")}
for t2, tt in templates.items():
for f in tt["fragments"]:
if resolve_template(f) == path:
tmpls.add(t2)
else:
notes.append("templates配下ではない")
conf = "低"
elif path.endswith(".xml") and "Mapper" in path:
stem = path.rsplit("/", 1)[-1][:-4]
for cls, route in controllers_for(stem):
routes.add(" -> ".join(reversed(route)))
for e in classes[cls]["endpoints"]:
eps.add(ep_key(e))
conf = "中"
elif path.endswith(".sql") or cat.startswith("DB"):
notes.append("DB変更: 参照する全Repositoryから手動で辿ること")
conf = "低"
else:
notes.append("静的解析の対象外。手動判断")
conf = "低"
# エンドポイント -> JS -> 画面 へ展開
for k in list(eps):
for j in ep_to_js.get(k, ()):
jss.add(j)
for j in list(jss):
tmpls |= js_to_tmpl.get(j, set())
urls = set()
for t in tmpls:
urls |= view_to_pageurl.get(t, set())
rows.append([
path, cat, status,
"; ".join(sorted(routes)),
"\n".join(sorted(eps)),
"\n".join(sorted(jss)),
"\n".join(sorted(tmpls)),
", ".join(sorted(urls)),
len(tmpls), conf, " / ".join(notes),
])
w("08_impact.csv",
["変更ファイル", "分類", "status", "到達経路(Javaクラス)", "影響エンドポイント",
"影響JS", "影響画面(template)", "画面URL", "画面数", "確度", "備考"],
rows)
w("09_unresolved.csv", ["種別", "対象", "理由"], unresolved)
low = sum(1 for r in rows if r[9] == "低")
print("\n--- summary ---")
print("changed files : %d" % len(rows))
print("要手動確認(確度:低) : %d" % low)
print("未解決項目 : %d -> 09_unresolved.csv" % len(unresolved))
if __name__ == "__main__":
main()