Aislo/resources/knowledge/original/_pipeline/hwp5_text.py

# -*- coding: utf-8 -*-
import os as _os
from pathlib import Path as _P

# 이 스크립트는 original/_pipeline/ 에 위치. 코퍼스 루트 = 상위 폴더.
ROOT_DIR = _P(__file__).resolve().parent.parent  # ...\original
DATA_DIR = _P(__file__).resolve().parent / "data"  # 생성 데이터(JSON)


# API 키: 커밋 금지 파일(law/.secrets.local.md) 또는 환경변수에서 읽는다.
def _load_key(name):
    v = _os.environ.get(name)
    if v:
        return v.strip()
    sec = ROOT_DIR.parent / ".secrets.local.md"
    if sec.exists():
        import re as _re

        for pat in (r"KCSC[\s\S]*?`([A-Za-z0-9]{20,})`", r"인증키[:\s]*`?([A-Za-z0-9]{30,})`?"):
            m = _re.search(pat, sec.read_text(encoding="utf-8"))
            if m:
                return m.group(1)
    return ""


"""HWP5(OLE) 본문 텍스트 추출 — 순수 파이썬.

BodyText/Section* 스트림을 (필요시 raw-deflate 해제) 레코드 파싱해
문단 텍스트(HWPTAG_PARA_TEXT)를 UTF-16LE로 뽑는다.
"""
import sys, re, zlib, struct
import olefile

HWPTAG_BEGIN = 0x10
HWPTAG_PARA_HEADER = HWPTAG_BEGIN + 50  # 0x42
HWPTAG_PARA_TEXT = HWPTAG_BEGIN + 51  # 0x43
HWPTAG_CTRL_HEADER = HWPTAG_BEGIN + 55  # 0x47
HWPTAG_LIST_HEADER = HWPTAG_BEGIN + 56  # 0x48
HWPTAG_TABLE = HWPTAG_BEGIN + 61  # 0x4d


def is_compressed(ole):
    with ole.openstream("FileHeader") as f:
        data = f.read()
    # 36바이트 오프셋의 속성 플래그 bit0 = 압축여부
    flags = struct.unpack("<I", data[36:40])[0]
    return bool(flags & 1)


def records(buf):
    i, n = 0, len(buf)
    while i + 4 <= n:
        header = struct.unpack("<I", buf[i : i + 4])[0]
        i += 4
        tag = header & 0x3FF
        level = (header >> 10) & 0x3FF
        size = (header >> 20) & 0xFFF
        if size == 0xFFF:
            size = struct.unpack("<I", buf[i : i + 4])[0]
            i += 4
        yield tag, level, buf[i : i + size]
        i += size


# 인라인 확장 제어문자(뒤에 14 WCHAR = 28바이트 추가로 따라옴)
EXT_CTRL = {1, 2, 3, 4, 5, 6, 7, 8, 9, 11, 12, 14, 15, 16, 17, 18, 21, 22, 23}
# 인라인 문자 제어(그 자체로 1 WCHAR)
INLINE = {0, 10, 13, 24, 25, 26, 27, 28, 29, 30, 31}


def para_text(data):
    out = []
    i, n = 0, len(data)
    while i + 1 < n:
        code = struct.unpack("<H", data[i : i + 2])[0]
        if code in EXT_CTRL:
            i += 16 * 2  # 제어문자 1 + 확장 14 + 종료 1 = 16 WCHAR
            continue
        if code in INLINE:
            if code == 13:  # 문단 구분? (para text엔 없음)
                out.append("\n")
            i += 2
            continue
        out.append(chr(code))
        i += 2
    return "".join(out)


def extract(path):
    """PARA_TEXT 문단만 평문 리스트로(하위호환)."""
    ole = olefile.OleFileIO(path)
    comp = is_compressed(ole)
    secs = sorted(e[-1] for e in ole.listdir() if e[0] == "BodyText")
    paras = []
    for sec in secs:
        with ole.openstream(f"BodyText/{sec}") as f:
            raw = f.read()
        buf = zlib.decompress(raw, -15) if comp else raw
        for tag, level, data in records(buf):
            if tag == HWPTAG_PARA_TEXT:
                paras.append(para_text(data).strip())
    ole.close()
    return paras


def _table_md(cells, ncols):
    """셀 텍스트 리스트를 nCols 기준으로 md 표로."""
    cells = [re.sub(r"\s+", " ", (c or "").replace("|", "／")).strip() for c in cells]
    if ncols < 1 or len(cells) < ncols * 2:
        # 표라기엔 빈약 → 줄 텍스트
        return None
    rows = [cells[i : i + ncols] for i in range(0, len(cells), ncols)]
    rows = [r + [""] * (ncols - len(r)) for r in rows]
    out = ["| " + " | ".join(rows[0]) + " |", "|" + "|".join(["---"] * ncols) + "|"]
    for r in rows[1:]:
        out.append("| " + " | ".join(r) + " |")
    return "\n".join(out)


def extract_items(path):
    """(kind, val) 아이템 리스트. kind='text'|'table'. 표를 복원한다."""
    ole = olefile.OleFileIO(path)
    comp = is_compressed(ole)
    secs = sorted(e[-1] for e in ole.listdir() if e[0] == "BodyText")
    items = []
    for sec in secs:
        with ole.openstream(f"BodyText/{sec}") as f:
            raw = f.read()
        buf = zlib.decompress(raw, -15) if comp else raw
        recs = list(records(buf))
        i, n = 0, len(recs)
        while i < n:
            tag, level, data = recs[i]
            # 표 컨트롤: CTRL_HEADER ctrl_id 'tbl ' (역순 b' lbt')
            if tag == HWPTAG_CTRL_HEADER and data[:4] == b" lbt":
                tbl_level = level
                # 다음 TABLE 레코드에서 열수
                ncols = 0
                j = i + 1
                while j < n and recs[j][0] != HWPTAG_TABLE:
                    j += 1
                if j < n:
                    d = recs[j][2]
                    if len(d) >= 8:
                        ncols = struct.unpack("<H", d[6:8])[0]
                    j += 1
                # 셀 수집: level > tbl_level 인 동안, LIST_HEADER마다 새 셀
                cells, cur = [], None
                while j < n and recs[j][1] > tbl_level:
                    t2, l2, d2 = recs[j]
                    if t2 == HWPTAG_LIST_HEADER:
                        if cur is not None:
                            cells.append(cur)
                        cur = ""
                    elif t2 == HWPTAG_PARA_TEXT and cur is not None:
                        seg = para_text(d2).strip()
                        cur = (cur + " " + seg).strip() if cur else seg
                    j += 1
                if cur is not None:
                    cells.append(cur)
                md = _table_md(cells, ncols) if ncols else None
                if md:
                    items.append(("table", md))
                else:
                    for c in cells:
                        if c.strip():
                            items.append(("text", c))
                i = j
                continue
            if tag == HWPTAG_PARA_TEXT:
                s = para_text(data).strip()
                if s:
                    items.append(("text", s))
            i += 1
    ole.close()
    return items


if __name__ == "__main__":
    paras = extract(sys.argv[1])
    text = "\n".join(p for p in paras if p)
    if len(sys.argv) > 2:
        open(sys.argv[2], "w", encoding="utf-8").write(text)
    print(f"문단 {len(paras)} / 문자 {len(text)}")
    print("--- 처음 40줄 ---")
    for p in [x for x in paras if x][:40]:
        print(p[:100])