hyungi_document_server/app/services/hier_decomp/builder.py

"""Hierarchical decomposition rule builder (PR-DocSrv-Hierarchical-Decomposition-1 c3).

텍스트(주로 md_content 마크다운) → heading 경계 segment 트리.
- 규칙 우선 경계 탐지: ATX 마크다운(#{1,6}) > 한국 구조(제N장/절/조) > 영문(Chapter/Section/Article).
- 각 segment = heading 라인 + 다음 heading 전까지 본문 (서로 disjoint, 100% 커버).
- parent/level = heading 깊이 기반 네비 트리. preamble(첫 heading 이전) = level 0 root 직속.
- 과대 segment(>LEAF_HARD_MAX, 더 깊은 heading 없음) = window fallback: 본문을 무overlap
  window 로 분해해 child leaf 생성, 부모는 is_leaf=false(heading 만 보유, 코퍼스 제외).
- is_leaf = 코퍼스 편입 대상 (replace predicate). window-split 부모만 false.

순수 함수 — DB 미접근. c4 에서 이 트리를 document_chunks 에 insert(parent_id 해소).
"""
from __future__ import annotations
import re
import hashlib
from dataclasses import dataclass, field

STRUCTURE_SPLIT_THRESHOLD = 4000
LEAF_TARGET_MAX = 3000
LEAF_HARD_MAX = 5000
MAX_DEPTH = 6

# 경계 패턴 (우선순위 순). group 'title' = 표시용, level 은 매처가 결정.
_ATX = re.compile(r'^(#{1,6})\s+(?P<title>\S.*?)\s*#*\s*$')
_KO_JANG = re.compile(r'^\s*(?P<title>제\s*\d+\s*장\b.*)$')
_KO_JEOL = re.compile(r'^\s*(?P<title>제\s*\d+\s*절\b.*)$')
_KO_JO = re.compile(r'^\s*(?P<title>제\s*\d+\s*조\b.*)$')
_ENG = re.compile(r'^\s*(?P<title>(?:Chapter|Section|Article|Part|PART)\s+[\dIVXLA-Z]+\b.*)$')


@dataclass
class HierNode:
    idx: int
    parent_idx: int | None
    level: int
    node_type: str | None
    section_title: str | None
    heading_path: str | None
    text: str
    is_leaf: bool = True
    chunk_content_hash: str = field(default="")

    def finalize_hash(self):
        self.chunk_content_hash = hashlib.sha256(self.text.encode("utf-8")).hexdigest()


def _detect_heading(line: str) -> tuple[int, str, str] | None:
    """(level, title, node_type) 또는 None. level 은 상대 깊이."""
    m = _ATX.match(line)
    if m:
        return (len(m.group(1)), m.group("title").strip(), None)  # node_type 은 후처리에서
    for pat, lvl, nt in ((_KO_JANG, 1, "chapter"), (_KO_JEOL, 2, "section"),
                          (_KO_JO, 3, "clause"), (_ENG, 1, "chapter")):
        m = pat.match(line)
        if m:
            return (lvl, m.group("title").strip()[:200], nt)
    return None


def _segment(text: str) -> list[tuple[int, str | None, str | None, str]]:
    """heading 경계로 분할 → [(level, title, node_type, segment_text), ...].

    preamble(첫 heading 이전 본문) = (0, None, None, text).
    """
    lines = text.splitlines(keepends=True)
    segs: list[tuple[int, str | None, str | None, list[str]]] = []
    cur: tuple[int, str | None, str | None, list[str]] | None = None
    preamble: list[str] = []
    for ln in lines:
        h = _detect_heading(ln.rstrip("\n"))
        if h:
            if cur is not None:
                segs.append(cur)
            elif preamble and "".join(preamble).strip():
                segs.append((0, None, None, preamble))
            cur = (h[0], h[1], h[2], [ln])
        else:
            if cur is None:
                preamble.append(ln)
            else:
                cur[3].append(ln)
    if cur is not None:
        segs.append(cur)
    elif preamble and "".join(preamble).strip():
        segs.append((0, None, None, preamble))
    return [(lvl, title, nt, "".join(body)) for (lvl, title, nt, body) in segs]


def _window_split(body: str, target: int) -> list[str]:
    """무overlap, 문단 우선 window 분해 (과대 segment fallback)."""
    paras = re.split(r'(\n\s*\n)', body)  # 구분자 보존
    chunks: list[str] = []
    buf = ""
    for p in paras:
        if len(buf) + len(p) <= target:
            buf += p
        else:
            if buf.strip():
                chunks.append(buf)
            if len(p) <= target:
                buf = p
            else:  # 단일 문단이 target 초과 → 문자 단위 hard split
                for i in range(0, len(p), target):
                    chunks.append(p[i:i + target])
                buf = ""
    if buf.strip():
        chunks.append(buf)
    return [c for c in chunks if c.strip()]


def build_hier_tree(
    text: str, *,
    split_threshold: int = STRUCTURE_SPLIT_THRESHOLD,
    leaf_target_max: int = LEAF_TARGET_MAX,
    leaf_hard_max: int = LEAF_HARD_MAX,
    max_depth: int = MAX_DEPTH,
) -> list[HierNode]:
    """텍스트 → HierNode 리스트 (idx 순, parent_idx 로 트리)."""
    if not text or not text.strip():
        return []
    segs = _segment(text)
    nodes: list[HierNode] = []
    # heading 깊이 정규화: 관측된 distinct level(>0) 을 1..k 로 매핑(절대 # 수 gap 제거).
    distinct = sorted({lvl for lvl, *_ in segs if lvl > 0})
    level_map = {raw: i + 1 for i, raw in enumerate(distinct)}

    # 부모 찾기용 스택: (norm_level, idx)
    stack: list[tuple[int, int]] = []

    def _heading_path(parent_idx: int | None, title: str | None) -> str | None:
        chain = []
        pi = parent_idx
        while pi is not None:
            if nodes[pi].section_title:
                chain.append(nodes[pi].section_title)
            pi = nodes[pi].parent_idx
        chain.reverse()
        if title:
            chain.append(title)
        return " > ".join(chain) if chain else None

    for lvl, title, nt, body in segs:
        norm = 0 if lvl == 0 else min(level_map[lvl], max_depth)
        # 부모 = 스택에서 norm 보다 작은 가장 가까운 노드
        while stack and stack[-1][0] >= norm:
            stack.pop()
        parent_idx = stack[-1][1] if stack else None
        idx = len(nodes)
        hp = _heading_path(parent_idx, title)
        node = HierNode(idx=idx, parent_idx=parent_idx, level=norm, node_type=nt,
                        section_title=title, heading_path=hp, text=body, is_leaf=True)
        nodes.append(node)
        if norm > 0:
            stack.append((norm, idx))

        # 과대 segment fallback (window-split) — 이 segment 가 leaf 일 때만(자식 heading 이
        # 뒤에 오면 자연히 분할되므로, 여기선 일단 생성 후 후처리에서 자식 유무로 판정).
    has_child = {n.parent_idx for n in nodes if n.parent_idx is not None}
    MIN_LEAF_BODY = 30  # heading 제외 own body 가 이보다 짧고 자식 있으면 구조 전용(코퍼스 제외)

    def _body_only(n: HierNode) -> str:
        lines = n.text.splitlines(keepends=True)
        if n.section_title and lines:  # 첫 줄 = heading
            return "".join(lines[1:])
        return n.text

    final: list[HierNode] = list(nodes)
    for n in list(final):
        is_nav_internal = n.idx in has_child
        # (B) 구조 전용 heading (자식 보유 + own body 빈약) → 코퍼스 제외. heading 은 자식 heading_path 에 보존.
        if is_nav_internal and len(_body_only(n).strip()) < MIN_LEAF_BODY:
            n.is_leaf = False
            continue
        # (A) own text 과대 → 자식 heading 유무 무관 window 분해. 부모는 heading 마커로 강등(코퍼스 제외).
        if len(n.text) > leaf_hard_max:
            wins = _window_split(n.text, leaf_target_max)
            if len(wins) > 1:
                n.is_leaf = False
                heading_line = (n.text.splitlines() or [""])[0]
                n.text = heading_line  # 중복 저장 회피 (full body 는 window child 가 보유)
                n.node_type = (n.node_type or "section") + "_split"
                base_level = min(n.level + 1, max_depth)
                for wtext in wins:
                    ci = len(final)
                    final.append(HierNode(
                        idx=ci, parent_idx=n.idx, level=base_level, node_type="window",
                        section_title=n.section_title, heading_path=n.heading_path,
                        text=wtext, is_leaf=True))
    for n in final:
        n.finalize_hash()
    return final


def coverage_stats(text: str, nodes: list[HierNode]) -> dict:
    """G2 검증 지표."""
    leaves = [n for n in nodes if n.is_leaf]
    leaf_chars = sum(len(n.text) for n in leaves)
    base = len(text)
    hashes = [n.chunk_content_hash for n in leaves]
    dup = len(hashes) - len(set(hashes))
    empty = sum(1 for n in leaves if not n.text.strip())
    # parent/level 무결성
    dangling = sum(1 for n in nodes if n.parent_idx is not None and (n.parent_idx < 0 or n.parent_idx >= len(nodes)))
    bad_level = 0
    for n in nodes:
        if n.parent_idx is not None:
            if n.level != nodes[n.parent_idx].level + 1 and nodes[n.parent_idx].node_type and "split" in (nodes[n.parent_idx].node_type or ""):
                pass  # window child 는 base_level 규칙
            # 일반 네비: 자식 level > 부모 level 만 보장
            if n.level <= nodes[n.parent_idx].level and nodes[n.parent_idx].level > 0:
                bad_level += 1
    return {
        "nodes": len(nodes), "leaves": len(leaves),
        "coverage_ratio": round(leaf_chars / base, 4) if base else 0,
        "dup_leaf_hash": dup, "empty_leaf": empty,
        "dangling_parent": dangling, "bad_level": bad_level,
        "level_dist": {l: sum(1 for n in nodes if n.level == l) for l in sorted({n.level for n in nodes})},
        "leaf_len_min": min((len(n.text) for n in leaves), default=0),
        "leaf_len_max": max((len(n.text) for n in leaves), default=0),
    }