"""Stage 1 대안 입력 — 프로그램별 디렉토리에 인클루드가 .txt 로 있는 덤프를 정규화한다. `ingest/normalize.py` 는 수집 API 의 JSON(.txt) 을 받는다. 이 모듈은 다른 형태를 받는다: //.txt 인클루드별 ABAP 원문 (CP949) //screens/*.txt 화면·GUI 타이틀 텍스트 (참고용, 파싱 대상 아님) //dictionary_objects/ DDIC 오브젝트 HTML (참고용) python -m ingest.from_dir [--out DIR] [--limit N] [--dry-run] 핵심 차이 두 가지: 1. **인코딩** — 파일이 CP949(EUC-KR) 다. UTF-8 로 읽으면 깨진다. 2. **메타가 코드 주석에 있다** — 수집 JSON 의 DESCRIPTION/T_CODE 에 해당하는 값이 주 프로그램 헤더 주석 박스에 들어 있다: * Report : ZFIR0010 * Module/Sub-Module : FI / GL * T_CODE : ZFIR0010 * Description : 거래처마스터 I/F 이력조회 이걸 긁어 수집 JSON 과 같은 모양의 payload 로 만들고 `normalize.write_program` 을 그대로 재사용한다 — 정규화 경로를 두 개로 갈라두지 않기 위해서다. 텍스트 심볼(TEXT-nnn → 한국어)은 이 덤프에 없다. 텍스트 풀이 별도 오브젝트라 소스에 포함되지 않는다 → `text_symbols` 는 빈 배열이고, 그만큼 조각의 자연어 앵커가 줄어든다. """ from __future__ import annotations import argparse import json import re import sys from pathlib import Path from .normalize import clean_text_field, write_program ENCODINGS = ("cp949", "utf-8-sig", "utf-8", "latin-1") SKIP_DIRS = {"screens", "dictionary_objects"} # 헤더 주석 박스의 "* 키 : 값 *" 한 줄. 세 가지 서식이 섞여 있다: # * Description : 거래처마스터 I/F 이력조회 * # * P/G desc : 매입 채무 Interface 내역 # *& 프로그램 명 : [FI] 전자세금계산서 매입전표 생성 _HEADER_FIELD = re.compile( r"^\*&?\s*([A-Za-z가-힣_/][A-Za-z0-9가-힣 _/\-]*?)\s*:\s*(.*?)\s*\*?\s*$" ) _REPORT_STMT = re.compile(r"^\s*(?:REPORT|PROGRAM)\s+([A-Za-z0-9_/]+)", re.I | re.M) # 헤더에서 의미 있게 쓰는 키만 남긴다 — 주석 처리된 ABAP 선언문(`* TABLES: BKPF,`)이 # 키:값 모양이라 그대로 두면 잡동사니가 섞인다. _KNOWN_KEYS = { "DESCRIPTION", "P/G_DESC", "PROGRAM_DESC", "TITLE", "프로그램_명", "프로그램명", "개요", "T_CODE", "TCODE", "MODULE/SUB-MODULE", "MODULE", "모듈", "TYPE", "REPORT", "PROGRAM_ID", "프로그램_ID", "시스템", "AUTHOR", "생성자", "DATE", "생성일", "PROJECT", } # 설명으로 쓸 키의 우선순위 _DESC_KEYS = ("DESCRIPTION", "P/G_DESC", "PROGRAM_DESC", "프로그램_명", "프로그램명", "개요", "TITLE") _TCODE_KEYS = ("T_CODE", "TCODE") _MODULE_KEYS = ("MODULE/SUB-MODULE", "MODULE", "모듈") # 덤프 도구가 모든 인클루드 끝에 붙이는 꼬리말. ABAP 이 아니라 추출기 서명이다: # ---------------------------------------------------------------------- # Extracted by Mass Download version 1.4.4 - E.G.Mellodew. 1998-2013. Sap Release 731 # 그냥 두면 인클루드마다 미인식 문장 7건이 생기고(176개 × 7 ≈ 1,200), 마지막 unit 의 줄 범위에 # 섞여 들어가 LLM 프롬프트에도 그대로 보인다. _TRAILER = re.compile( r"\n-{20,}\s*\n\s*Extracted by Mass Download.*?$", re.S | re.I ) def strip_trailer(text: str) -> str: """덤프 도구 꼬리말 제거. 없으면 그대로 돌려준다.""" stripped = _TRAILER.sub("\n", text) if stripped != text: return stripped # 구분선이 없는 변형도 처리 — 마지막 'Extracted by ...' 줄만 떼낸다 lines = text.split("\n") while lines and (not lines[-1].strip() or lines[-1].lstrip().lower().startswith("extracted by ")): if lines[-1].lstrip().lower().startswith("extracted by "): lines.pop() while lines and set(lines[-1].strip()) <= {"-"} and lines[-1].strip(): lines.pop() return "\n".join(lines) + "\n" lines.pop() return text def read_text(path: Path) -> str: """CP949 우선으로 디코드. 줄바꿈 통일 + 덤프 도구 꼬리말 제거.""" raw = path.read_bytes() for enc in ENCODINGS: try: text = raw.decode(enc) break except UnicodeDecodeError: continue else: # pragma: no cover — latin-1 은 실패하지 않는다 text = raw.decode("cp949", errors="replace") return strip_trailer(text.replace("\r\n", "\n").replace("\r", "\n")) def header_fields(text: str, max_lines: int = 60) -> dict[str, str]: """주 프로그램 앞부분 주석 박스에서 키:값 을 긁는다 (알려진 키만).""" out: dict[str, str] = {} for line in text.split("\n")[:max_lines]: if not line.startswith("*"): continue m = _HEADER_FIELD.match(line) if not m: continue key = m.group(1).strip().upper().replace(" ", "_") if key not in _KNOWN_KEYS: continue val = clean_text_field(m.group(2).rstrip("*").strip()) # 구분선(*----*)이나 빈 값, N/A 는 버린다 if val and not set(val) <= {"-", "*", "="} and val.upper() != "N/A" and key not in out: out[key] = val return out def _pick(fields: dict[str, str], keys: tuple[str, ...]) -> str: for k in keys: if fields.get(k): return fields[k] return "" def include_files(prog_dir: Path) -> list[Path]: """인클루드 후보 .txt — 주 프로그램 파일을 맨 앞에 둔다 (REPORT 문이 먼저 보이도록).""" main = prog_dir / f"{prog_dir.name}.txt" others = sorted( p for p in prog_dir.glob("*.txt") if p.is_file() and p != main and p.parent.name not in SKIP_DIRS ) return ([main] if main.exists() else []) + others def program_payload(prog_dir: Path) -> dict: """프로그램 디렉토리 → 수집 JSON 과 같은 모양의 payload.""" files = include_files(prog_dir) if not files: raise ValueError(f"인클루드 .txt 가 없습니다: {prog_dir}") main_text = read_text(files[0]) fields = header_fields(main_text) program = prog_dir.name.upper() m = _REPORT_STMT.search(main_text) if m: program = m.group(1).upper() # REPORT 문이 진짜 프로그램명 (디렉토리명과 다를 수 있다) description = _pick(fields, _DESC_KEYS) return { "MAIN_PROGRAM": program, "DESCRIPTION": description, "TEXT_SYMBOL": [], # 이 덤프에는 텍스트 풀이 없다 "INCLUDE_PROGRAM": [ {"INCLUDE": p.stem.upper(), "SOURCE_CODE": read_text(p)} for p in files ], # 아래는 이 포맷에서만 나오는 부가 정보 — meta 에 실어 로더가 tcode 로 적재한다 "_HEADER": fields, "_TCODE": _pick(fields, _TCODE_KEYS), "_MODULE": _pick(fields, _MODULE_KEYS), } def run(root: Path, out_dir: Path, limit: int | None = None, dry_run: bool = False) -> dict: prog_dirs = sorted(d for d in root.iterdir() if d.is_dir() and d.name not in SKIP_DIRS) if limit: prog_dirs = prog_dirs[:limit] stats = {"programs": 0, "includes": 0, "total_lines": 0, "tcodes": 0, "errors": 0} errors: list[str] = [] extras: list[dict] = [] for d in prog_dirs: try: payload = program_payload(d) except Exception as e: # noqa: BLE001 — 실패 기록 후 계속 stats["errors"] += 1 errors.append(f"{d}\t{type(e).__name__}: {e}") continue if dry_run: stats["programs"] += 1 stats["includes"] += len(payload["INCLUDE_PROGRAM"]) continue meta = write_program(payload, out_dir) stats["programs"] += 1 stats["includes"] += len(meta["includes"]) stats["total_lines"] += meta["total_lines"] # 헤더에서 뽑은 부가 정보를 meta.json 에 덧붙인다 (write_program 이 쓴 뒤에 갱신) meta_path = out_dir / meta["program"] / f"{meta['program']}.meta.json" meta["tcode"] = payload["_TCODE"] meta["sap_module"] = payload["_MODULE"] meta["header"] = payload["_HEADER"] meta["source_dir"] = str(d) meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8") if payload["_TCODE"]: stats["tcodes"] += 1 extras.append({"tcode": payload["_TCODE"], "program": meta["program"], "text_ko": meta["description"]}) if not dry_run: out_dir.mkdir(parents=True, exist_ok=True) if extras: # 로더가 읽어 tcode 테이블에 넣는다 (ASSUMPTIONS.md §5 의 TSTC 미확보를 일부 대체) (out_dir / "tcodes.jsonl").write_text( "\n".join(json.dumps(e, ensure_ascii=False) for e in extras) + "\n", encoding="utf-8") if errors: (out_dir / "_errors.log").write_text("\n".join(errors) + "\n", encoding="utf-8") return stats def main() -> None: ap = argparse.ArgumentParser(description="Stage 1 대안 — 프로그램 디렉토리 덤프 정규화") ap.add_argument("root", help="프로그램 디렉토리들이 들어 있는 루트") ap.add_argument("--out", default=None, help="출력 디렉토리 (기본 data/normalized)") 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 root = Path(args.root) if not root.is_dir(): print(f"입력 디렉토리 없음: {root}", file=sys.stderr) sys.exit(1) out_dir = Path(args.out) if args.out else settings.data_normalized print(json.dumps(run(root, out_dir, args.limit, args.dry_run), ensure_ascii=False)) if __name__ == "__main__": main()