ABAP-Indexing/ingest/from_dir.py

"""Stage 1 대안 입력 — 프로그램별 디렉토리에 인클루드가 .txt 로 있는 덤프를 정규화한다.

`ingest/normalize.py` 는 수집 API 의 JSON(.txt) 을 받는다. 이 모듈은 다른 형태를 받는다:

    <root>/<program>/<include>.txt        인클루드별 ABAP 원문 (CP949)
    <root>/<program>/screens/*.txt        화면·GUI 타이틀 텍스트 (참고용, 파싱 대상 아님)
    <root>/<program>/dictionary_objects/  DDIC 오브젝트 HTML (참고용)

  python -m ingest.from_dir <root> [--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()