byeongwook.choiandClaude Fable 5.1
11ae3629b2
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
277 lines
12 KiB
Python
277 lines
12 KiB
Python
"""Stage 2 실행 — data/normalized/<PROGRAM>/ → data/parsed/<PROGRAM>.parse.json
|
|
|
|
python -m parser.run [--program ZFIR10070] [--limit N] [--dry-run]
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from collections import Counter
|
|
from dataclasses import asdict
|
|
from pathlib import Path
|
|
|
|
from .dataflow import SymbolDecl, extract_declarations, extract_writes
|
|
from .declarations import Declaration, extract_declaration_blocks, unit_declaration
|
|
from .refs import extract_refs
|
|
from .statements import Statement, assignment_eq_index, split_statements
|
|
from .units import Unit, extract_units, finalize_units
|
|
|
|
# 파서가 의미를 아는 문장 머리 키워드 (미인식 리포트용 — 계획서 §4.6)
|
|
RECOGNIZED_HEADS = {
|
|
"REPORT", "PROGRAM", "INCLUDE", "TYPE-POOLS", "TYPES", "DATA", "CONSTANTS", "STATICS",
|
|
"TABLES", "NODES", "FIELD-SYMBOLS", "RANGES", "CLASS-DATA", "PARAMETERS", "PARAMETER",
|
|
"CONTROLS", "SELECT-OPTIONS",
|
|
"SELECTION-SCREEN", "INITIALIZATION", "START-OF-SELECTION", "END-OF-SELECTION",
|
|
"TOP-OF-PAGE", "END-OF-PAGE", "LOAD-OF-PROGRAM", "AT", "GET", "FORM", "ENDFORM",
|
|
"FUNCTION", "ENDFUNCTION", "MODULE", "ENDMODULE", "METHOD", "ENDMETHOD", "CLASS",
|
|
"ENDCLASS", "INTERFACE", "ENDINTERFACE", "DEFINE", "END-OF-DEFINITION",
|
|
"SELECT", "ENDSELECT", "FETCH", "OPEN", "CLOSE", "INSERT", "UPDATE", "MODIFY", "DELETE",
|
|
"COMMIT", "ROLLBACK", "APPEND", "COLLECT", "READ", "LOOP", "ENDLOOP", "SORT", "CLEAR",
|
|
"REFRESH", "FREE", "MOVE", "MOVE-CORRESPONDING", "SPLIT", "CONCATENATE", "CONDENSE",
|
|
"TRANSLATE", "REPLACE", "SHIFT", "OVERLAY", "SEARCH", "FIND", "ASSIGN", "UNASSIGN",
|
|
"IF", "ELSEIF", "ELSE", "ENDIF", "CASE", "WHEN", "ENDCASE", "DO", "ENDDO", "WHILE",
|
|
"ENDWHILE", "CHECK", "EXIT", "CONTINUE", "RETURN", "LEAVE", "STOP", "REJECT",
|
|
"PERFORM", "CALL", "SUBMIT", "EXPORT", "IMPORT", "SET", "AUTHORITY-CHECK", "MESSAGE",
|
|
"WRITE", "ULINE", "SKIP", "NEW-LINE", "NEW-PAGE", "FORMAT", "HIDE", "WINDOW",
|
|
"DESCRIBE", "ADD", "SUBTRACT", "MULTIPLY", "DIVIDE", "COMPUTE", "CONVERT", "CREATE",
|
|
"RAISE", "TRY", "CATCH", "CLEANUP", "ENDTRY", "WAIT", "SUPPRESS", "SCROLL",
|
|
"PRINT-CONTROL", "BREAK-POINT", "ASSERT", "FIELD", "CHAIN", "ENDCHAIN", "ENDAT",
|
|
"SUMMARY", "DETAIL", "BACK", "RESERVE", "POSITION", "EDITOR-CALL", "RETURN",
|
|
# 클래스/인터페이스 정의부
|
|
"METHODS", "CLASS-METHODS", "CLASS-EVENTS", "EVENTS", "INTERFACES", "ALIASES",
|
|
"PUBLIC", "PROTECTED", "PRIVATE", "SECTION", "IMPLEMENTATION", "DEFINITION",
|
|
"TYPE-POOL", "FUNCTION-POOL",
|
|
}
|
|
|
|
EVENT_ORDER = ["LOAD-OF-PROGRAM", "INITIALIZATION", "AT SELECTION-SCREEN", "START-OF-SELECTION",
|
|
"GET", "END-OF-SELECTION", "TOP-OF-PAGE", "END-OF-PAGE", "AT LINE-SELECTION",
|
|
"AT USER-COMMAND"]
|
|
|
|
|
|
def _event_rank(u: Unit) -> int:
|
|
for i, ev in enumerate(EVENT_ORDER):
|
|
if u.name.startswith(ev):
|
|
return i
|
|
return len(EVENT_ORDER)
|
|
|
|
|
|
def parse_program(prog_dir: Path) -> dict:
|
|
program = prog_dir.name
|
|
meta = json.loads((prog_dir / f"{program}.meta.json").read_text(encoding="utf-8"))
|
|
|
|
code_by_include: dict[str, str] = {}
|
|
statements_by_include: dict[str, list[Statement]] = {}
|
|
for inc in meta["includes"]:
|
|
name = inc["include"]
|
|
code = (prog_dir / f"{name}.abap").read_text(encoding="utf-8")
|
|
code_by_include[name] = code
|
|
statements_by_include[name], _ = split_statements(code, name)
|
|
|
|
# 1) 매크로 이름 선수집 (매크로 호출 인식용)
|
|
macro_names: set[str] = set()
|
|
for sts in statements_by_include.values():
|
|
for st in sts:
|
|
if st.upper and st.upper[0] == "DEFINE" and len(st.upper) >= 2:
|
|
macro_names.add(st.upper[1])
|
|
|
|
# 2) unit 추출
|
|
units: list[Unit] = []
|
|
for name, sts in statements_by_include.items():
|
|
units.extend(extract_units(program, name, sts, code_by_include[name].count("\n") + 1))
|
|
finalize_units(units, statements_by_include, code_by_include)
|
|
|
|
# unit_id 유일화 — 같은 이벤트(AT SELECTION-SCREEN ON VALUE-REQUEST 등)가 반복되면 라인 번호를 붙인다
|
|
seen_ids: set[str] = set()
|
|
for u in units:
|
|
if u.unit_id in seen_ids:
|
|
u.unit_id = f"{u.unit_id}@L{u.line_start}"
|
|
seen_ids.add(u.unit_id)
|
|
|
|
# 3) 심볼 선언 — 전역(DECLARATION unit) + unit 로컬
|
|
symbols: list[SymbolDecl] = []
|
|
for u in units:
|
|
sts = statements_by_include[u.include]
|
|
if u.unit_type == "DECLARATION":
|
|
symbols.extend(extract_declarations(sts, u.stmt_indexes, None, "global"))
|
|
elif u.unit_type in {"FORM", "METHOD", "FUNCTION", "MODULE", "EVENT"}:
|
|
symbols.extend(extract_declarations(sts, u.stmt_indexes, u.unit_id, "unit",
|
|
unit_type=u.unit_type, signature=u.signature))
|
|
# `TABLES: zfit0030.` / `NODES: ...` 는 **DDIC 테이블 작업영역** 선언이다. 이 이름을
|
|
# known_symbols 에 넣으면 extract_refs 의 "심볼이면 DB 테이블이 아니다" 가드가
|
|
# `SELECT ... FROM zfit0030` 을 걸러버린다 — 정작 업무의 핵심인 커스텀 Z 테이블이 통째로
|
|
# 누락된다(실측: ZFIR0010 의 tables_read 에 ZFIT0030 이 없었다).
|
|
# 가드의 목적은 내부테이블(DATA gt_x TYPE TABLE OF ...)을 DB 테이블로 오인하지 않는 것이므로
|
|
# DDIC 작업영역은 제외한다.
|
|
DDIC_WORK_AREA = {"tables", "nodes"}
|
|
global_syms = {s.name for s in symbols if s.scope == "global" and s.kind not in DDIC_WORK_AREA}
|
|
local_syms_by_unit: dict[str, set[str]] = {}
|
|
for s in symbols:
|
|
if s.scope == "unit" and s.unit_id:
|
|
local_syms_by_unit.setdefault(s.unit_id, set()).add(s.name)
|
|
|
|
# 3-b) 정의부(선언 블록) — 붙여넣기용 원문. symbol 과 목적이 다르다(parser/declarations.py)
|
|
lines_by_include = {name: code.split("\n") for name, code in code_by_include.items()}
|
|
declarations: list[Declaration] = []
|
|
for u in units:
|
|
sts = statements_by_include[u.include]
|
|
lns = lines_by_include[u.include]
|
|
if u.unit_type == "DECLARATION":
|
|
declarations.extend(extract_declaration_blocks(sts, u.stmt_indexes, lns, u.include, "global"))
|
|
elif u.unit_type == "CLASS_DEF":
|
|
declarations.append(unit_declaration(u, lns, "class"))
|
|
elif u.unit_type == "MACRO":
|
|
declarations.append(unit_declaration(u, lns, "macro"))
|
|
elif u.unit_type in {"FORM", "METHOD", "FUNCTION", "MODULE", "EVENT"}:
|
|
declarations.extend(
|
|
extract_declaration_blocks(sts, u.stmt_indexes, lns, u.include, "unit", u.unit_id))
|
|
|
|
# 4) unit 별 참조/데이터플로우
|
|
all_writes, all_reads, unit_refs = [], [], {}
|
|
for u in units:
|
|
if u.unit_type in {"DECLARATION", "CLASS_DEF"}:
|
|
continue
|
|
sts = statements_by_include[u.include]
|
|
known = global_syms | local_syms_by_unit.get(u.unit_id, set())
|
|
r = extract_refs(sts, u.stmt_indexes, macro_names, known)
|
|
unit_refs[u.unit_id] = r
|
|
writes, reads = extract_writes(sts, u.stmt_indexes, u.unit_id, known)
|
|
all_writes.extend(writes)
|
|
all_reads.extend(reads)
|
|
|
|
# 5) 호출 그래프
|
|
form_by_name = {u.name: u.unit_id for u in units if u.unit_type == "FORM"}
|
|
module_by_name = {u.name.split(" ")[0]: u.unit_id for u in units if u.unit_type == "MODULE"}
|
|
edges: list[dict] = []
|
|
for u in units:
|
|
r = unit_refs.get(u.unit_id)
|
|
if not r:
|
|
continue
|
|
for c in r.calls:
|
|
kind, target = c["kind"], c["target"]
|
|
edge = {"from_unit": u.unit_id, "to_unit": None, "external_name": None,
|
|
"call_type": kind, "line": c.get("line", 0)}
|
|
if kind == "perform":
|
|
if target in form_by_name:
|
|
edge["to_unit"] = form_by_name[target]
|
|
else:
|
|
edge["external_name"] = target
|
|
edge["call_type"] = "perform_unresolved"
|
|
elif kind == "perform_external":
|
|
edge["external_name"] = f"{c.get('program', '?')}:{target}"
|
|
else:
|
|
edge["external_name"] = target
|
|
edges.append(edge)
|
|
|
|
# 6) 실행 순서(topo_order) — 이벤트 순 → PERFORM 체인 DFS (순환은 끊음)
|
|
adj: dict[str, list[str]] = {}
|
|
for e in edges:
|
|
if e["to_unit"]:
|
|
adj.setdefault(e["from_unit"], []).append(e["to_unit"])
|
|
topo: list[str] = []
|
|
visited: set[str] = set()
|
|
|
|
def dfs(uid: str) -> None:
|
|
if uid in visited:
|
|
return
|
|
visited.add(uid)
|
|
topo.append(uid)
|
|
for nxt in adj.get(uid, []):
|
|
dfs(nxt)
|
|
|
|
for u in sorted([x for x in units if x.unit_type == "EVENT"], key=_event_rank):
|
|
dfs(u.unit_id)
|
|
for u in units: # 이벤트에서 도달 못한 unit (PBO/PAI 모듈 등)
|
|
if u.unit_type in {"MODULE", "FORM", "METHOD", "FUNCTION"} and u.unit_id not in visited:
|
|
dfs(u.unit_id)
|
|
|
|
# 7) 미인식 문장 리포트
|
|
unknown = Counter()
|
|
total_stmts = 0
|
|
for sts in statements_by_include.values():
|
|
for st in sts:
|
|
if not st.upper:
|
|
continue
|
|
total_stmts += 1
|
|
head = st.upper[0]
|
|
base = head.split("(")[0]
|
|
if (
|
|
base in RECOGNIZED_HEADS
|
|
or base in macro_names
|
|
or base.startswith("&") # 매크로 치환 파라미터(&1 = …) — 템플릿이라 분석 대상 아님
|
|
or "->" in base or "=>" in base # 메서드 호출식 문장
|
|
or assignment_eq_index(st.upper) is not None
|
|
):
|
|
continue
|
|
unknown[base] += 1
|
|
|
|
return {
|
|
"program": program,
|
|
"description": meta.get("description", ""),
|
|
"includes": meta["includes"],
|
|
"text_symbols": meta.get("text_symbols", []),
|
|
"units": [
|
|
{**{k: v for k, v in asdict(u).items() if k != "stmt_indexes"},
|
|
"refs": asdict(unit_refs[u.unit_id]) if u.unit_id in unit_refs else None}
|
|
for u in units
|
|
],
|
|
"symbols": [asdict(s) for s in symbols],
|
|
"declarations": [asdict(d) for d in declarations],
|
|
"writes": [asdict(w) for w in all_writes],
|
|
"reads": all_reads,
|
|
"call_edges": edges,
|
|
"topo_order": topo,
|
|
"stats": {
|
|
"units": len(units),
|
|
"statements": total_stmts,
|
|
"symbols": len(symbols),
|
|
"declarations": len(declarations),
|
|
"writes": len(all_writes),
|
|
"unknown_statements": sum(unknown.values()),
|
|
"unknown_ratio": round(sum(unknown.values()) / total_stmts, 4) if total_stmts else 0,
|
|
"unknown_top": unknown.most_common(15),
|
|
},
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(description="Stage 2 — 구조 파싱")
|
|
ap.add_argument("--program", default=None, help="특정 프로그램만")
|
|
ap.add_argument("--limit", type=int, default=None)
|
|
ap.add_argument("--dry-run", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
from config.settings import settings
|
|
|
|
norm, out = settings.data_normalized, settings.data_parsed
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
prog_dirs = sorted(d for d in norm.iterdir() if d.is_dir()) if norm.exists() else []
|
|
if args.program:
|
|
prog_dirs = [d for d in prog_dirs if d.name == args.program.upper()]
|
|
if args.limit:
|
|
prog_dirs = prog_dirs[: args.limit]
|
|
|
|
agg = {"programs": 0, "units": 0, "statements": 0, "unknown": 0, "errors": 0}
|
|
for d in prog_dirs:
|
|
try:
|
|
result = parse_program(d)
|
|
except Exception as e: # noqa: BLE001
|
|
agg["errors"] += 1
|
|
print(f"[FAIL] {d.name}: {type(e).__name__}: {e}")
|
|
continue
|
|
agg["programs"] += 1
|
|
agg["units"] += result["stats"]["units"]
|
|
agg["statements"] += result["stats"]["statements"]
|
|
agg["unknown"] += result["stats"]["unknown_statements"]
|
|
if not args.dry_run:
|
|
(out / f"{d.name}.parse.json").write_text(
|
|
json.dumps(result, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
print(f"[OK] {d.name}: units={result['stats']['units']} stmts={result['stats']['statements']} "
|
|
f"unknown={result['stats']['unknown_statements']} ({result['stats']['unknown_ratio']:.1%}) "
|
|
f"top={result['stats']['unknown_top'][:5]}")
|
|
print(json.dumps(agg, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|