byeongwook.choiandClaude Fable 5.1
fdae34161e
- GET_PROGRAM_LIST 는 TADIR 의 PROG 만 주므로 GET_OBJECT_LIST_BY_PACKAGE(FUGR) 로 그룹을 받아 SAPL<그룹> 을 프로그램처럼 저장 (설명에 펑션 이름·텍스트 목록) - 펑션 그룹은 서버 INCLUDE_LIST 가 비어 온다(실측) → 소스의 INCLUDE 문을 깊이 4 까지 따라가며 고객 인클루드(Z/Y/네임스페이스, L<그룹>…)만 받는다. 표준 생성기 인클루드(LSVIM…)는 제외 - 실측 ZFI01: 펑션 그룹 14, SAPLZFG_FI01 인클루드 47 → 파서 FUNCTION 38 unit, 미인식 0 - --no-fugr 로 끌 수 있다 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
332 lines
15 KiB
Python
332 lines
15 KiB
Python
"""Stage 0 — SAP 에서 패키지 단위로 프로그램 소스를 받아 data/raw/ 에 저장한다.
|
|
|
|
abap-mcp 의 SAP 클라이언트(sap/sap_client.py, ZAA_ICF = ZCL_ZAA_AGENT_API)를 그대로 쓴다.
|
|
저장 형식은 `ingest/normalize.py` 가 읽는 수집 JSON 과 같다 — 그래서 받은 뒤엔 평소대로
|
|
`python -m ingest.normalize` 부터 돌리면 된다.
|
|
|
|
python -m ingest.from_sap ZFI01 # 패키지 하나
|
|
python -m ingest.from_sap ZFI01 ZCO01 ZSD01 # 여러 개
|
|
python -m ingest.from_sap ZFI01 --limit 3 # 시험: 프로그램 3본만
|
|
python -m ingest.from_sap ZFI01 --force # 이미 받은 파일도 다시 받기
|
|
python -m ingest.from_sap --pattern "ZFIR1*" # 패키지 대신 이름 패턴
|
|
|
|
접속 정보는 `.env` 의 SAP_URL / SAP_USER / SAP_PASS (sap/README.md). 고객사 SAP 에
|
|
ZAA_ICF 서비스(ZCL_ZAA_AGENT_API)가 설치돼 있어야 한다.
|
|
|
|
만드는 파일 (기본 data/raw/):
|
|
<PACKAGE>.txt 프로그램 목록 [{DEVCLASS, OBJ_NAME, TEXT, CREATED_ON, CHANGED_ON}]
|
|
<PROGRAM>.txt 프로그램 소스 {MAIN_PROGRAM, DESCRIPTION, TEXT_SYMBOL: [],
|
|
INCLUDE_PROGRAM: [{INCLUDE, SOURCE_CODE}], TCODE_LIST: [...]}
|
|
_errors.log 실패한 프로그램 (계속 진행하고 여기 기록)
|
|
|
|
- 펑션 모듈: 프로그램 목록 API 에는 안 나온다(TADIR 의 FUGR). 패키지의 펑션 그룹을 따로 받아
|
|
그룹 메인 프로그램 SAPL<그룹> 을 프로그램처럼 저장한다 (펑션 소스는 그 인클루드 L<그룹>U01 …).
|
|
펑션 그룹은 서버 INCLUDE_LIST 가 비어 오므로 소스의 INCLUDE 문을 따라 받는다. --no-fugr 로 끈다.
|
|
- SAP 목록 API 는 SUBC 를 생략하면 실행 프로그램(1)과 모듈 풀(M)만 준다. Include 는 각 프로그램의
|
|
INCLUDE_LIST 로 따라오므로 따로 받지 않는다.
|
|
- 텍스트 심볼(TEXT-nnn)은 이 API 에 없다 → TEXT_SYMBOL 은 빈 배열 (from_dir 와 같은 한계).
|
|
- 이미 받은 프로그램(.txt 존재)은 건너뛴다. 다시 받으려면 --force.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
|
|
def _log(msg: str) -> None:
|
|
print(msg, file=sys.stderr, flush=True)
|
|
|
|
|
|
def _call(method: str, params: dict, retries: int = 3, timeout: int = 180) -> dict:
|
|
"""sap_call + 네트워크 오류 재시도. 결과는 sap_client.sap_call 과 같은 dict."""
|
|
from sap.sap_client import sap_call
|
|
|
|
last: dict = {}
|
|
for attempt in range(1, retries + 1):
|
|
last = sap_call(method, params, timeout=timeout)
|
|
# error 는 네트워크/인증/파싱 실패. RETURN.TYPE=E 는 SAP 이 정상 응답한 것이라 재시도 안 함
|
|
if not last.get("error"):
|
|
return last
|
|
_log(f" ! {method} {params} 실패({attempt}/{retries}): {last['error']}")
|
|
if attempt < retries:
|
|
time.sleep(2 * attempt)
|
|
return last
|
|
|
|
|
|
def list_programs(package: str | None = None, pattern: str | None = None,
|
|
subc: str | None = None) -> list[dict]:
|
|
"""GET_PROGRAM_LIST → [{OBJ_NAME, TEXT, SUBC, UDAT, UNAM, ...}]"""
|
|
params = {"IV_PACKAGE": package or "", "IV_PATTERN": pattern or "", "IV_SUBC": subc or ""}
|
|
res = _call("GET_PROGRAM_LIST", params)
|
|
if res.get("error"):
|
|
raise RuntimeError(res["error"])
|
|
ret = res["return"]
|
|
if ret["type"] == "E":
|
|
raise RuntimeError(f"SAP 오류: {ret['message']}")
|
|
rows = (res.get("parsed") or {}).get("RESULT") or []
|
|
if ret["type"] == "W":
|
|
_log(f" ! 목록이 잘렸습니다: {ret['message']}")
|
|
return [r for r in rows if isinstance(r, dict) and r.get("OBJ_NAME")]
|
|
|
|
|
|
# 소스 안의 INCLUDE 문. 주석줄(* 로 시작)은 제외한다
|
|
_INCLUDE_STMT = re.compile(r"^(?![ \t]*\*)[ \t]*INCLUDE[ \t]+([A-Za-z0-9_/]+)[ \t]*\.", re.I | re.M)
|
|
MAX_INCLUDE_DEPTH = 4
|
|
|
|
|
|
def is_customer_include(name: str, program: str) -> bool:
|
|
"""받아올 인클루드인가 — 고객 개발(Z/Y/네임스페이스)과 그 펑션 그룹의 L<그룹>… 인클루드만.
|
|
|
|
SAP 표준(LSVIMFXX, lsvimdat 같은 생성기 인클루드)은 소스가 크고 업무 로직이 아니라 뺀다.
|
|
"""
|
|
n = name.upper()
|
|
if n.startswith(("Z", "Y", "/")):
|
|
return True
|
|
p = program.upper()
|
|
if p.startswith("SAPL") and n.startswith("L" + p[4:]): # SAPLZFG_FI01 → LZFG_FI01TOP / U01 / F01 …
|
|
return True
|
|
return False
|
|
|
|
|
|
def _resolve_includes_from_source(program: str, includes: list[dict], seen: set[str]) -> int:
|
|
"""INCLUDE 문을 따라 인클루드를 받아 includes 에 덧붙인다 (D010INC 가 비어 있을 때의 대안).
|
|
|
|
펑션 그룹(SAPL*)은 서버의 INCLUDE_LIST 가 비어 온다(실측 2026-09-22). 메인 → LxxxUXX → LxxxU01 …
|
|
처럼 단계가 있어 깊이 MAX_INCLUDE_DEPTH 까지 따라간다. 받은 개수를 돌려준다.
|
|
"""
|
|
added = 0
|
|
frontier = list(includes)
|
|
for _ in range(MAX_INCLUDE_DEPTH):
|
|
next_frontier: list[dict] = []
|
|
for inc in frontier:
|
|
for ref in _INCLUDE_STMT.findall(inc["SOURCE_CODE"]):
|
|
ref = ref.upper()
|
|
if ref in seen or not is_customer_include(ref, program):
|
|
continue
|
|
seen.add(ref)
|
|
res = _call("GET_PROGRAM_SOURCE", {"IV_PROGRAM": ref})
|
|
code = ((res.get("parsed") or {}).get("SOURCE_CODE") or "") if not res.get("error") else ""
|
|
if res.get("error") or res["return"]["type"] == "E" or not code.strip():
|
|
_log(f" include {ref} 못 받음: {res.get('error') or res['return']['message']}")
|
|
continue
|
|
item = {"INCLUDE": ref, "SOURCE_CODE": code}
|
|
includes.append(item)
|
|
next_frontier.append(item)
|
|
added += 1
|
|
if not next_frontier:
|
|
break
|
|
frontier = next_frontier
|
|
return added
|
|
|
|
|
|
def fetch_program(name: str, description: str = "") -> dict:
|
|
"""GET_PROGRAM_SOURCE(+Include) → 수집 JSON 모양의 payload. 실패하면 RuntimeError."""
|
|
res = _call("GET_PROGRAM_SOURCE", {"IV_PROGRAM": name, "IV_WITH_INCLUDE": "X"})
|
|
if res.get("error"):
|
|
raise RuntimeError(res["error"])
|
|
ret = res["return"]
|
|
if ret["type"] == "E":
|
|
raise RuntimeError(f"SAP 오류: {ret['message']}")
|
|
data = res.get("parsed") or {}
|
|
main_code = data.get("SOURCE_CODE") or ""
|
|
if not main_code.strip():
|
|
raise RuntimeError("소스가 비어 있음")
|
|
|
|
program = (data.get("PROGRAM") or name).strip().upper()
|
|
includes = [{"INCLUDE": program, "SOURCE_CODE": main_code}]
|
|
seen = {program}
|
|
for inc in data.get("INCLUDE_LIST") or []:
|
|
if not isinstance(inc, dict):
|
|
continue
|
|
inc_name = (inc.get("INCL_NAME") or "").strip().upper()
|
|
code = inc.get("SOURCE") or ""
|
|
# 이름 없음 / 빈 소스(권한 없음·삭제됨) / 중복은 뺀다
|
|
if not inc_name or inc_name in seen or not code.strip():
|
|
continue
|
|
seen.add(inc_name)
|
|
includes.append({"INCLUDE": inc_name, "SOURCE_CODE": code})
|
|
# 서버 INCLUDE_LIST 에 없는 인클루드를 소스의 INCLUDE 문으로 보충 (펑션 그룹은 이 경로로만 받아진다)
|
|
_resolve_includes_from_source(program, includes, seen)
|
|
|
|
tcodes = [
|
|
{"TCODE": (t.get("TCODE") or "").strip().upper(), "TTEXT": t.get("TTEXT") or ""}
|
|
for t in data.get("TCODE_LIST") or [] if isinstance(t, dict) and t.get("TCODE")
|
|
]
|
|
return {
|
|
"MAIN_PROGRAM": program,
|
|
"DESCRIPTION": description or "",
|
|
"TEXT_SYMBOL": [],
|
|
"INCLUDE_PROGRAM": includes,
|
|
"TCODE_LIST": tcodes,
|
|
}
|
|
|
|
|
|
def list_function_groups(package: str) -> list[dict]:
|
|
"""패키지의 펑션 그룹 → [{OBJ_NAME: 'SAPLZFG_FI01', TEXT: '펑션 그룹 ZFG_FI01: FN1, FN2 …'}]
|
|
|
|
펑션 모듈 소스는 그룹 메인 프로그램 SAPL<그룹> 의 인클루드(L<그룹>U01 …)에 있다.
|
|
GET_PROGRAM_LIST 는 TADIR 의 PROG 만 주므로 펑션 그룹(FUGR)은 여기서 따로 받는다.
|
|
"""
|
|
res = _call("GET_OBJECT_LIST_BY_PACKAGE", {"IV_PACKAGE": package, "IV_OBJ_TYPE": "FUGR"})
|
|
if res.get("error"):
|
|
raise RuntimeError(res["error"])
|
|
if res["return"]["type"] == "E":
|
|
raise RuntimeError(f"SAP 오류: {res['return']['message']}")
|
|
out = []
|
|
for r in (res.get("parsed") or {}).get("RESULT") or []:
|
|
if not isinstance(r, dict) or not r.get("OBJ_NAME"):
|
|
continue
|
|
group = r["OBJ_NAME"].strip().upper()
|
|
# 그룹 안의 펑션 이름·설명을 프로그램 설명으로 — 펑션 이름 검색이 프로그램 단위에서도 되게
|
|
names: list[str] = []
|
|
fl = _call("GET_FUNCTION_LIST", {"IV_FUNC_GROUP": group})
|
|
if not fl.get("error"):
|
|
for f in (fl.get("parsed") or {}).get("RESULT") or []:
|
|
if isinstance(f, dict) and f.get("FUNCNAME"):
|
|
text = (f.get("STEXT") or f.get("FUNC_TEXT") or "").strip()
|
|
names.append(f["FUNCNAME"].strip().upper() + (f" ({text})" if text else ""))
|
|
desc = f"펑션 그룹 {group}" + (": " + ", ".join(names[:30]) if names else "")
|
|
out.append({"OBJ_NAME": "SAPL" + group, "TEXT": desc, "SUBC": "F", "UDAT": r.get("UDAT") or ""})
|
|
return out
|
|
|
|
|
|
def package_rows(package: str, rows: list[dict]) -> list[dict]:
|
|
"""프로그램 목록 → normalize 가 packages.jsonl 로 만드는 package_list 모양."""
|
|
return [
|
|
{
|
|
"DEVCLASS": package.upper(),
|
|
"OBJ_NAME": r["OBJ_NAME"].upper(),
|
|
"TEXT": r.get("TEXT") or "",
|
|
"CREATED_ON": "",
|
|
"CHANGED_ON": r.get("UDAT") or "",
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
def _write_json(path: Path, obj) -> None:
|
|
# 한 줄 JSON. normalize 는 "줄바꿈+패딩 → 공백" 치환을 먼저 하므로 들여쓰기(indent)를 쓰면 안 된다.
|
|
path.write_text(json.dumps(obj, ensure_ascii=False), encoding="utf-8", newline="\n")
|
|
|
|
|
|
def run(packages: list[str], out_dir: Path, pattern: str | None = None, subc: str | None = None,
|
|
limit: int | None = None, force: bool = False, dry_run: bool = False,
|
|
fugr: bool = True) -> dict:
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
stats = {"packages": 0, "listed": 0, "function_groups": 0, "fetched": 0, "skipped": 0, "errors": 0,
|
|
"includes": 0}
|
|
errors: list[str] = []
|
|
|
|
targets: list[tuple[str, str]] = [] # (program, description)
|
|
if packages:
|
|
for pkg in packages:
|
|
pkg = pkg.strip().upper()
|
|
if not pkg:
|
|
continue
|
|
_log(f"[{pkg}] 프로그램 목록 조회")
|
|
try:
|
|
rows = list_programs(package=pkg, pattern=pattern, subc=subc)
|
|
except Exception as e: # noqa: BLE001
|
|
stats["errors"] += 1
|
|
errors.append(f"{pkg}\tLIST\t{e}")
|
|
_log(f" ! 목록 실패: {e}")
|
|
continue
|
|
stats["packages"] += 1
|
|
_log(f" 프로그램 {len(rows)}건")
|
|
if fugr and not pattern:
|
|
# 펑션 그룹 → SAPL<그룹> 을 프로그램처럼 받는다 (펑션 모듈 소스가 그 인클루드에 있다)
|
|
try:
|
|
groups = list_function_groups(pkg)
|
|
_log(f" 펑션 그룹 {len(groups)}건")
|
|
stats["function_groups"] += len(groups)
|
|
rows = rows + groups
|
|
except Exception as e: # noqa: BLE001
|
|
stats["errors"] += 1
|
|
errors.append(f"{pkg}\tFUGR_LIST\t{e}")
|
|
_log(f" ! 펑션 그룹 목록 실패: {e}")
|
|
if not dry_run:
|
|
_write_json(out_dir / f"{pkg}.txt", package_rows(pkg, rows))
|
|
targets += [(r["OBJ_NAME"].upper(), r.get("TEXT") or "") for r in rows]
|
|
else:
|
|
_log(f"[패턴 {pattern}] 프로그램 목록 조회")
|
|
rows = list_programs(pattern=pattern, subc=subc)
|
|
_log(f" 프로그램 {len(rows)}건")
|
|
targets += [(r["OBJ_NAME"].upper(), r.get("TEXT") or "") for r in rows]
|
|
|
|
# 같은 프로그램이 두 패키지 목록에 있을 일은 없지만 안전하게 중복 제거
|
|
uniq: dict[str, str] = {}
|
|
for name, desc in targets:
|
|
uniq.setdefault(name, desc)
|
|
targets = list(uniq.items())
|
|
if limit:
|
|
targets = targets[:limit]
|
|
stats["listed"] = len(targets)
|
|
|
|
for i, (name, desc) in enumerate(targets, 1):
|
|
path = out_dir / f"{name}.txt"
|
|
if path.exists() and not force:
|
|
stats["skipped"] += 1
|
|
continue
|
|
if dry_run:
|
|
_log(f" ({i}/{len(targets)}) {name} {desc}")
|
|
continue
|
|
try:
|
|
payload = fetch_program(name, desc)
|
|
except Exception as e: # noqa: BLE001 — 실패 기록 후 계속
|
|
stats["errors"] += 1
|
|
errors.append(f"{name}\tSOURCE\t{e}")
|
|
_log(f" ({i}/{len(targets)}) {name} 실패: {e}")
|
|
continue
|
|
_write_json(path, payload)
|
|
stats["fetched"] += 1
|
|
stats["includes"] += len(payload["INCLUDE_PROGRAM"])
|
|
_log(f" ({i}/{len(targets)}) {name} include {len(payload['INCLUDE_PROGRAM'])} {desc}")
|
|
|
|
if errors and not dry_run:
|
|
with (out_dir / "_errors.log").open("a", encoding="utf-8") as f:
|
|
f.write("\n".join(errors) + "\n")
|
|
return stats
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(
|
|
description="Stage 0 — SAP(ZAA_ICF) 에서 패키지의 프로그램 소스를 data/raw 로 받는다",
|
|
epilog="받은 뒤: python -m ingest.normalize && python -m parser.run && python -m index.loader",
|
|
)
|
|
ap.add_argument("packages", nargs="*", help="패키지명 (예: ZFI01). 여러 개 가능")
|
|
ap.add_argument("--pattern", default=None, help="프로그램명 패턴 (예: ZFIR1*). 패키지와 같이 쓰면 그 안에서 필터")
|
|
ap.add_argument("--subc", default=None, help="프로그램 유형. 생략=1(실행)+M(모듈풀). 1/M/I/F/K")
|
|
ap.add_argument("--out", default=None, help="저장 디렉토리 (기본 data/raw)")
|
|
ap.add_argument("--limit", type=int, default=None, help="받을 프로그램 수 상한 (시험용)")
|
|
ap.add_argument("--force", action="store_true", help="이미 받은 파일도 다시 받는다")
|
|
ap.add_argument("--dry-run", action="store_true", help="목록만 보고 저장하지 않는다")
|
|
ap.add_argument("--no-fugr", action="store_true",
|
|
help="펑션 그룹(SAPL<그룹>, 펑션 모듈 소스)은 받지 않는다 (기본은 같이 받음)")
|
|
args = ap.parse_args()
|
|
|
|
if not args.packages and not args.pattern:
|
|
ap.error("패키지명 또는 --pattern 중 하나는 있어야 합니다")
|
|
|
|
from config.settings import settings
|
|
from sap.sap_client import CONFIG
|
|
|
|
if not CONFIG["user"] or not CONFIG["password"]:
|
|
print("SAP_USER / SAP_PASS 가 비어 있습니다 — .env 에 채우세요 (sap/README.md)", file=sys.stderr)
|
|
sys.exit(1)
|
|
_log(f"SAP: {CONFIG['base_url']} 사용자 {CONFIG['user']}")
|
|
|
|
out_dir = Path(args.out) if args.out else settings.data_raw
|
|
stats = run(args.packages, out_dir, pattern=args.pattern, subc=args.subc,
|
|
limit=args.limit, force=args.force, dry_run=args.dry_run, fugr=not args.no_fugr)
|
|
print(json.dumps(stats, ensure_ascii=False))
|
|
if stats["errors"]:
|
|
print(f"실패 {stats['errors']}건 → {out_dir / '_errors.log'}", file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|