"""로직 조각 후처리 — LLM 이 준 조각을 코드와 대조해 확정한다 (docs/logic-chunk-design.md). 원칙: 사실은 파서가, 해석은 LLM 이. - 줄 번호: first_line 앵커로 검증·보정, 실패하면 버림 - 테이블 · 호출: 조각 코드 범위를 파서로 다시 돌려 채움 (LLM 값은 쓰지 않음) - 해시: 조각 코드 sha256 """ from __future__ import annotations import hashlib import re from dataclasses import dataclass, field from parser.refs import extract_refs from parser.statements import split_statements from .schemas import CHUNK_KINDS, LogicChunk MIN_LINES = 1 _WS = re.compile(r"\s+") _SQL_HEAD = re.compile( r"^\s*(SELECT|OPEN\s+CURSOR|INSERT|UPDATE|MODIFY|DELETE|COMMIT\s+WORK|ROLLBACK\s+WORK|" r"CALL\s+FUNCTION|CALL\s+TRANSACTION|CALL\s+METHOD|SUBMIT|PERFORM|AUTHORITY-CHECK|" r"LOOP\s+AT|COLLECT|MESSAGE)\b", re.I, ) def _norm(s: str) -> str: return _WS.sub(" ", (s or "").strip()).lower() @dataclass class ResolvedChunk: seq: int line_start: int line_end: int code: str code_hash: str kind: str purpose_ko: str purpose_en: str keywords_ko: list[str] keywords_en: list[str] sap_objects: list[str] tables_read: list[str] = field(default_factory=list) tables_write: list[str] = field(default_factory=list) calls: list[str] = field(default_factory=list) confidence: float = 0.5 def numbered_code(lines: list[str], line_start: int, line_end: int) -> str: """include 전체 줄 목록에서 [line_start, line_end] 를 ' 12| code' 형식으로.""" out = [] for no in range(line_start, line_end + 1): if 1 <= no <= len(lines): out.append(f"{no:5d}| {lines[no - 1]}") return "\n".join(out) def parser_hints(lines: list[str], line_start: int, line_end: int, limit: int = 40) -> list[str]: """DB 접근 · 호출 · 검증 문장이 시작되는 줄 — LLM 이 자를 후보 지점 힌트.""" hints = [] for no in range(line_start, line_end + 1): if 1 <= no <= len(lines): m = _SQL_HEAD.match(lines[no - 1]) if m: hints.append(f"L{no} {lines[no - 1].strip()[:80]}") if len(hints) >= limit: break return hints def _find_anchor(lines: list[str], anchor: str, lo: int, hi: int) -> int | None: """[lo, hi] 안에서 anchor(정규화) 와 같은 줄 번호. 없으면 접두 일치.""" a = _norm(anchor) if not a: return None for no in range(lo, hi + 1): if _norm(lines[no - 1]) == a: return no head = a[:20] if len(head) >= 8: for no in range(lo, hi + 1): if _norm(lines[no - 1]).startswith(head): return no return None def resolve_chunks(raw: list[LogicChunk], lines: list[str], unit_start: int, unit_end: int, include: str, known_symbols: set[str]) -> tuple[list[ResolvedChunk], list[str]]: """LLM 조각 → 검증·보정·사실 채움. 반환: (확정 조각(줄 순), 버린 사유 목록).""" dropped: list[str] = [] resolved: list[ResolvedChunk] = [] seen_ranges: set[tuple[int, int]] = set() for c in raw: s, e = int(c.line_start), int(c.line_end) if e < s: s, e = e, s length = e - s # 1) 앵커 검증 — line_start 줄이 first_line 과 다르면 unit 안에서 찾아 보정 if c.first_line: ok = unit_start <= s <= unit_end and _norm(lines[s - 1]) == _norm(c.first_line) \ if 1 <= s <= len(lines) else False if not ok: found = _find_anchor(lines, c.first_line, unit_start, unit_end) if found is None: dropped.append(f"L{s}-L{e}: first_line 을 unit 안에서 찾지 못함 ({c.first_line[:40]!r})") continue s, e = found, found + length # 2) 범위 클램프 if s < unit_start or s > unit_end: dropped.append(f"L{s}-L{e}: unit 범위(L{unit_start}-L{unit_end}) 밖") continue e = min(e, unit_end) if e - s + 1 < MIN_LINES: dropped.append(f"L{s}-L{e}: 너무 짧음") continue if (s, e) in seen_ranges: dropped.append(f"L{s}-L{e}: 중복 범위") continue seen_ranges.add((s, e)) # 3) 코드 · 해시 · 파서 사실 code = "\n".join(lines[s - 1 : e]) stmts, _ = split_statements(code, include) refs = extract_refs(stmts, list(range(len(stmts))), set(), known_symbols) kind = c.kind if c.kind in CHUNK_KINDS else "other" resolved.append(ResolvedChunk( seq=0, line_start=s, line_end=e, code=code, code_hash=hashlib.sha256(code.encode("utf-8")).hexdigest(), kind=kind, purpose_ko=c.purpose_ko.strip(), purpose_en=c.purpose_en.strip(), keywords_ko=list(dict.fromkeys(k.strip() for k in c.keywords_ko if k.strip()))[:15], keywords_en=list(dict.fromkeys(k.strip() for k in c.keywords_en if k.strip()))[:15], sap_objects=list(dict.fromkeys(k.strip().upper() for k in c.sap_objects if k.strip()))[:20], tables_read=refs.tables_read, tables_write=refs.tables_write, calls=[x["target"] for x in refs.calls][:20], confidence=max(0.0, min(1.0, float(c.confidence))), )) resolved.sort(key=lambda r: (r.line_start, r.line_end)) for i, r in enumerate(resolved, 1): r.seq = i return resolved, dropped def covered_lines(chunks: list[ResolvedChunk]) -> int: covered: set[int] = set() for c in chunks: covered.update(range(c.line_start, c.line_end + 1)) return len(covered)