"""OKF v0.2 마크다운 위키 출력 (계획서 v4 §5.7, docs/logic-chunk-design.md). - 문서 단위는 프로그램. LLM 이 골라낸 로직 조각(logic_chunk)은 프로그램 문서 안의 `## 로직 조각` 섹션(조각마다 H3, resource 줄 범위)으로 들어간다. unit 별 개별 문서는 만들지 않는다 (1만 프로그램 × 수십 unit 파일 문제 회피). - 요약이 아직 없는 program 도 파서 구조 사실만으로 문서를 생성한다 (원칙: 사실은 파서가, 해석은 LLM이 — 요약이 생기면 다음 실행에서 채워짐). - 커밋·사람 교정 대상: programs/ packages/ concepts/ (merge 규칙 적용). - OKF v0.2 필드 매핑(스펙 확인 2026-08-26, docs/okf-version.md): generated: {by, at} = 생성 주체 / verified(human:) = 사람 교정 / status: draft = 재검토 필요. x-* 는 스펙이 허용하는 확장 필드(소비자는 미지 키를 보존해야 함). """ from __future__ import annotations import re import sqlite3 from datetime import datetime, timezone from pathlib import Path from config.settings import settings from index import decls as decls_mod from index.db import connect, loads from parser.refs import EXTERNAL_CALL_KINDS from .merge import merge_write PIPELINE = "abap-indexing/wiki_out" def _now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def _q(s: str) -> str: """YAML 이중따옴표 스칼라.""" return '"' + str(s).replace("\\", "\\\\").replace('"', '\\"') + '"' def _lst(items) -> str: """한 줄 YAML 리스트 — merge.py 가 줄 단위로 갱신할 수 있게 반드시 한 줄.""" return "[" + ", ".join(_q(i) for i in items) + "]" def _safe_name(name: str) -> str: return re.sub(r"[^0-9A-Za-z_.\-#]", "_", name or "_") def _fm(lines: list[str]) -> str: return "---\n" + "\n".join(lines) + "\n---\n" def _anchor(text: str) -> str: """H3 제목 → 마크다운 링크용 앵커 (영문·숫자·한글·하이픈만).""" return re.sub(r"[^0-9a-z가-힣\-]", "", text.lower().replace(" ", "-")) # ------------------------------------------------------------- 로직 조각 섹션 MAX_CHUNK_CODE_LINES = 120 # 이보다 긴 조각은 앞부분만 싣고 나머지는 API 로 안내 def _chunk_code(lines: list[str], line_start: int, line_end: int) -> tuple[list[str], int]: """조각의 소스 원문을 '줄번호| 코드' 형태로. (표시 줄, 생략된 줄 수)""" out: list[str] = [] end = min(line_end, line_start + MAX_CHUNK_CODE_LINES - 1) for no in range(line_start, end + 1): if 1 <= no <= len(lines): out.append(f"{no:6d}| {lines[no - 1]}") return out, max(0, line_end - end) def _render_chunks(con: sqlite3.Connection, program: str, unit_by_id: dict[str, sqlite3.Row]) -> tuple[list[str], dict[tuple, dict]]: """프로그램 문서의 `## 로직 조각` 섹션 — **소스 원문 + 자연어 설명**을 한 쌍으로 싣는다. 조각의 핵심은 "이 코드가 업무적으로 무엇을 하는가" 이므로 설명만 있으면 검증할 수 없다. 코드 원문을 함께 실어 읽는 사람이 설명이 맞는지 바로 대조하게 한다. 순서는 **실행 순서(topo)** 를 따른다. include 이름 순으로 놓으면 공용 인클루드(ZFIALV·ZFICOM)의 유틸 조각이 프로그램 고유 업무 로직보다 앞에 나와 문서가 엉뚱해 보인다. """ needed: dict[tuple, dict] = {} # 조각들이 실제로 쓰는 선언 (정의부 섹션이 링크 대상으로 쓴다) chunks = con.execute("SELECT * FROM logic_chunk WHERE program=?", (program,)).fetchall() if not chunks: return (["## 로직 조각", "", "(아직 추출된 로직 조각이 없다 — LLM 조각 추출 전. `python -m summarize.runner` 참고)", ""], needed) topo = {r["unit_id"]: r["ord"] for r in con.execute("SELECT unit_id, ord FROM topo WHERE program=?", (program,))} code_by_inc = {r["include"]: (r["code"] or "").split("\n") for r in con.execute("SELECT include, code FROM include WHERE program=?", (program,))} ordered = sorted(chunks, key=lambda c: (topo.get(c["unit_id"], 9_999), c["include"], c["line_start"])) n_units = len({c["unit_id"] for c in chunks}) body = [ f"## 로직 조각 ({len(chunks)}건 / unit {n_units}개)", "", "LLM 이 코드에서 골라낸 업무 로직 단위다. **조각마다 소스 원문과 자연어 설명을 함께 싣는다** — " "설명이 코드와 맞는지 바로 대조할 수 있어야 하기 때문이다.", "테이블·호출은 LLM 이 쓴 값이 아니라 파서가 조각 범위에서 다시 뽑은 사실이다. 순서는 실행 순서다.", "", ] current_unit = None for c in ordered: u = unit_by_id.get(c["unit_id"]) if c["unit_id"] != current_unit: current_unit = c["unit_id"] label = f"{u['unit_type']} {u['name']}" if u else c["unit_id"] head = f"### {label}" if u: head += f" — `{c['include']}` L{u['line_start']}-L{u['line_end']} ({u['loc']}줄)" body += [head, ""] if u: us = loads(u["summary_json"]) or {} if us.get("purpose_ko"): body += [f"*{us['purpose_ko']}*", ""] cov = us.get("coverage") if cov is not None: body += [f"> 이 unit 의 조각 {us.get('chunk_count', 0)}개가 " f"{us.get('covered_lines', 0)}/{u['loc']}줄({cov:.0%})을 덮는다. " f"나머지는 선언·화면설정 등 업무 의미가 없다고 판단해 버린 구간이다.", ""] body += [f"#### [{c['kind']}] {c['purpose_ko']}", ""] # --- 소스 원문 --- lines = code_by_inc.get(c["include"], []) chunk_code = "\n".join(lines[c["line_start"] - 1 : c["line_end"]]) need = decls_mod.for_chunk(con, c, chunk_code) shown, omitted = _chunk_code(lines, c["line_start"], c["line_end"]) if shown: body += [f"```abap", *shown] if omitted: body.append(f" … ({omitted}줄 생략 — 전체는 " f"GET /chunks/{c['chunk_id'].replace('#', '%23')})") body += ["```", ""] # --- 설명·사실 --- if c["purpose_en"]: body.append(f"- **EN**: {c['purpose_en']}") tr, tw = loads(c["tables_read"]) or [], loads(c["tables_write"]) or [] calls, objs = loads(c["calls"]) or [], loads(c["sap_objects"]) or [] kw_ko = loads(c["keywords_ko"]) or [] kw_en = loads(c["keywords_en"]) or [] if tr or tw: read_links = ", ".join(f"[{t}](/tables/{t}.md)" for t in tr) or "-" write_links = ", ".join(f"[{t}](/tables/{t}.md)" for t in tw) or "-" body.append(f"- 테이블(파서): read {read_links} / write {write_links}") if calls: body.append(f"- 호출(파서): {', '.join(calls)}") if objs: body.append(f"- SAP 객체: {', '.join(objs)}") if kw_ko: body.append(f"- 업무 키워드: {', '.join(kw_ko)}") if kw_en: body.append(f"- English: {', '.join(kw_en)}") # 정의부 — 이 조각을 복사해 갈 때 함께 필요한 선언. 원문은 아래 `## 정의부` 에 한 번만 싣고 # 여기서는 **이름 + 그 선언으로 가는 링크**만 둔다 (같은 구조 선언이 조각마다 반복되면 # 문서를 읽을 수 없다). 링크는 문서 안 앵커라 뷰어에서는 눌러서 이동하고, # LLM 이 읽을 때는 그냥 이름으로 읽힌다. if need["declarations"]: needed.update({_decl_key(d): d for d in need["declarations"]}) names = ", ".join(f"[`{d['name']}`](#{_decl_anchor(d)})" for d in need["declarations"][:12]) more = f" 외 {len(need['declarations']) - 12}건" if len(need["declarations"]) > 12 else "" body.append(f"- 정의부(복사 시 함께 필요): {names}{more}") if need["unresolved"]: body.append(f"- 선언을 못 찾은 이름: {', '.join(need['unresolved'][:8])} " f"(수집 안 된 인클루드·함수 인터페이스일 수 있다)") body.append(f"- 위치: `abap://{program}/{c['include']}#L{c['line_start']}-L{c['line_end']}` " f"· confidence {c['confidence']} · `{c['chunk_id']}`") body.append("") return body, needed # ------------------------------------------------------------- 정의부 섹션 MAX_DECL_CODE_LINES = 60 # 한 선언이 이보다 길면 앞부분만 (구조체 수백 줄짜리가 있다) def _decl_key(d: dict) -> tuple[str, str, int]: """선언 한 건을 가리키는 키. 같은 이름이 인클루드·unit 마다 따로 있을 수 있어 위치까지 쓴다.""" return (d["include"], d["name"], d["line_start"]) def _decl_anchor(d: dict) -> str: """조각 → 정의부 항목으로 가는 문서 내 앵커 id.""" inc, name, line = _decl_key(d) return re.sub(r"[^0-9a-z가-힣_-]", "-", f"decl-{inc}-{line}-{name}".lower()) def _render_declarations(con: sqlite3.Connection, program: str, needed: dict[tuple, dict]) -> list[str]: """`## 정의부` — 이 프로그램이 선언한 내부테이블·스트럭처·상수·필드심볼의 **원문**. 조각 코드는 로직만이라 그대로 붙여넣으면 컴파일되지 않는다. 조각마다 필요한 선언은 위에서 이름으로 가리키고(앵커 링크), 원문은 여기 한 번만 싣는다. 붙여넣기용 조립본은 `GET /chunks/` 의 `declaration_code` 다 (index/decls.py). 싣는 것: 전역 선언 전체 + **조각이 실제로 쓰는 unit 로컬 선언**. 후자를 빼면 조각의 정의부 링크가 갈 곳이 없어진다(FORM 머리에서 선언된 작업영역이 그렇다). """ rows = decls_mod.program_declarations(con, program, scope="global") seen = {_decl_key(r) for r in rows} rows += [d for k, d in needed.items() if k not in seen] if not rows: return [] body = [ f"## 정의부 ({len(rows)}건)", "", "로직 조각을 다른 프로그램으로 옮겨 붙일 때 **함께 가야 하는 선언**이다. " "조각별로 무엇이 필요한지는 위 조각 항목의 `정의부` 줄에 있고(이름을 누르면 여기로 온다), " "붙여넣기용으로 의존까지 묶은 코드는 `GET /chunks/` 의 `declaration_code` 로 받는다.", "", ] by_include: dict[str, list[dict]] = {} for r in rows: by_include.setdefault(r["include"], []).append(r) for include, items in by_include.items(): items.sort(key=lambda d: d["line_start"]) body += [f"### `{include}` ({len(items)}건)", ""] for d in items: code_lines = d["code"].split("\n") shown = code_lines[:MAX_DECL_CODE_LINES] scope_note = " · unit 로컬" if d["scope"] == "unit" else "" # `{#id}` 는 제목에 앵커를 다는 표기다 — 뷰어가 이걸 id 로 바꿔 링크가 걸린다. body += [f"#### {d['name']} — {d['kind']}{scope_note} · " f"L{d['line_start']}-L{d['line_end']} {{#{_decl_anchor(d)}}}", "", "```abap", *shown] if len(code_lines) > MAX_DECL_CODE_LINES: body.append(f"* … ({len(code_lines) - MAX_DECL_CODE_LINES}줄 생략)") body += ["```", ""] return body # ------------------------------------------------------------- program 문서 def _render_program(con: sqlite3.Connection, p: sqlite3.Row) -> tuple[str, str | None]: program = p["name"] summary = loads(p["summary_json"]) or {} units = con.execute( "SELECT * FROM unit WHERE program=? ORDER BY include, line_start", (program,) ).fetchall() # call_type 을 걸러야 한다 — 걸르지 않으면 CL_GUI_ALV_GRID=>MC_FC_* 상수 읽기와 # CALL SCREEN 의 화면번호('100')가 외부 호출 목록을 다 차지하고 정작 FM 호출이 상한에 밀린다. ph = ",".join("?" * len(EXTERNAL_CALL_KINDS)) ext_calls = [r["external_name"] for r in con.execute( f"SELECT DISTINCT external_name FROM call_edge WHERE program=? AND external_name IS NOT NULL " f"AND call_type IN ({ph}) ORDER BY external_name LIMIT 40", (program, *EXTERNAL_CALL_KINDS))] t_read = [r["table_name"] for r in con.execute( "SELECT DISTINCT table_name FROM table_ref WHERE program=? AND mode='read' ORDER BY table_name", (program,))] t_write = [r["table_name"] for r in con.execute( "SELECT DISTINCT table_name FROM table_ref WHERE program=? AND mode='write' ORDER BY table_name", (program,))] chunk_count = con.execute( "SELECT COUNT(*) AS c FROM logic_chunk WHERE program=?", (program,)).fetchone()["c"] decl_count = con.execute( "SELECT COUNT(*) AS c FROM declaration WHERE program=? AND scope='global'", (program,)).fetchone()["c"] chunk_keywords = [] for r in con.execute("SELECT keywords_ko FROM logic_chunk WHERE program=?", (program,)): chunk_keywords += loads(r["keywords_ko"]) or [] tags = list(dict.fromkeys( (summary.get("business_tags") or []) + ([summary.get("sap_module")] if summary.get("sap_module") else []) + chunk_keywords ))[:20] purpose = summary.get("business_purpose_ko") or "" generated_by = f"abap-indexing/{settings.llm_model}" if (summary or chunk_count) else "abap-indexing/parser" title = "{} — {}".format(program, p["title_ko"] or "").rstrip(" —") fm = [ "type: abap-program", f"title: {_q(title)}", f"description: {_q(purpose or p['title_ko'] or program)}", f"resource: {_q('abap://' + program)}", f"tags: {_lst(tags)}", f"generated: {{ by: {_q(generated_by)}, at: {_q(_now())} }}", "status: stable" if summary else "status: draft", f"x-package: {p['devclass'] or ''}", f"x-changed-on: {p['changed_on'] or ''}", f"x-tables-read: {_lst(t_read[:40])}", f"x-tables-write: {_lst(t_write[:40])}", f"x-calls: {_lst(ext_calls)}", f"x-code-hash: {_q(p['source_hash'] or '')}", f"x-unit-count: {len(units)}", f"x-chunk-count: {chunk_count}", f"x-declaration-count: {decl_count}", ] body: list[str] = ["## 업무 목적", "", purpose or p["title_ko"] or "(LLM 요약 전)", ""] flow = summary.get("main_flow") or [ f"{u['name']} ({u['include']} L{u['line_start']})" for u in units if u["unit_type"] == "EVENT" ] if flow: body += ["## 주 흐름", ""] + [f"{i}. {s}" for i, s in enumerate(flow, 1)] + [""] if summary.get("key_internal_tables"): body += ["## 핵심 내부테이블", ""] for t in summary["key_internal_tables"]: body.append(f"- `{t.get('name')}` — {t.get('desc_ko','')} " f"(채움: {', '.join(t.get('filled_by', [])) or '?'} / " f"소비: {', '.join(t.get('consumed_by', [])) or '?'})") body.append("") body += ["## 테이블 / 외부 호출", "", f"- read: {', '.join(t_read[:30]) or '-'}", f"- write: {', '.join(t_write[:30]) or '-'}", f"- 외부 호출: {', '.join(ext_calls[:30]) or '-'}", ""] unit_by_id = {u["unit_id"]: u for u in units} chunk_body, needed_decls = _render_chunks(con, program, unit_by_id) body += chunk_body body += _render_declarations(con, program, needed_decls) body += ["## Unit 목록 (컨테이너)", "", "| unit | 유형 | 위치 | 조각 | 한 줄 요약 |", "|---|---|---|---|---|"] for u in units: if u["unit_type"] not in ("FORM", "METHOD", "FUNCTION", "MODULE", "EVENT"): continue us = loads(u["summary_json"]) or {} body.append(f"| {u['name']} | {u['unit_type']} | {u['include']} " f"L{u['line_start']}-{u['line_end']} | {u['chunk_count'] or 0} | {us.get('purpose_ko','')} |") body.append("") return _fm(fm) + "\n" + "\n".join(body), p["source_hash"] # ------------------------------------------------------------- package 문서 def _render_package(con: sqlite3.Connection, devclass: str, text_ko: str) -> str: progs = con.execute( "SELECT name, title_ko, has_source FROM program WHERE devclass=? ORDER BY name", (devclass,) ).fetchall() title = "{} — {}".format(devclass, text_ko or "").rstrip(" —") fm = [ "type: abap-package", f"title: {_q(title)}", f"description: {_q(text_ko or devclass)}", f"resource: {_q('abap://package/' + devclass)}", "tags: []", f"generated: {{ by: {_q(PIPELINE)}, at: {_q(_now())} }}", f"x-program-count: {len(progs)}", ] body = [f"## 프로그램 목록 ({len(progs)}건)", "", "| 프로그램 | 타이틀 | 소스 |", "|---|---|---|"] body += [f"| [{r['name']}](/programs/{r['name']}.md) | {r['title_ko'] or ''} " f"| {'O' if r['has_source'] else '-'} |" for r in progs] body.append("") return _fm(fm) + "\n" + "\n".join(body) # ---------------------------------------------------------------- 매니페스트 def write_manifest(wiki_dir: Path, con: sqlite3.Connection | None = None) -> None: """OKF v0.2 예약 파일 index.md — 번들 루트 매니페스트 (구 계획의 okf.yaml 대체). 단순 디렉토리 나열이 아니라 **계층 진입점**이다 (수정사항 8번): SAP 모듈 → 패키지 → 프로그램으로 내려갈 수 있어야 에이전트가 위키를 따라 탐색할 수 있다. """ counts = {} for sub in ("programs", "packages", "concepts", "tables", "functions", "tcodes"): d = wiki_dir / sub counts[sub] = sum(1 for _ in d.rglob("*.md")) if d.exists() else 0 chunk_re = re.compile(r"^x-chunk-count: (\d+)$", re.M) counts["chunks"] = 0 if (wiki_dir / "programs").exists(): for f in (wiki_dir / "programs").glob("*.md"): m = chunk_re.search(f.read_text(encoding="utf-8")[:4000]) counts["chunks"] += int(m.group(1)) if m else 0 fm = [ "type: index", 'okf_version: "0.2"', f"title: {_q('ABAP 소스 위키')}", f"description: {_q('ABAP 프로그램·로직 조각·엔티티(테이블/FM)의 위키 — 모듈에서 프로그램으로 내려가는 진입점')}", f"generated: {{ by: {_q(PIPELINE)}, at: {_q(_now())} }}", ] body = [ "## 구성", "", f"- [programs/](/programs/) — 프로그램 문서 {counts['programs']}건, 로직 조각 {counts['chunks']}건 포함 (커밋·교정 대상)", f"- [packages/](/packages/) — 패키지 문서 {counts['packages']}건 (커밋·교정 대상)", f"- [concepts/](/concepts/) — 도메인 개념 {counts['concepts']}건 (질의 확장 사전과 같은 원본)", f"- [tables/](/tables/) — 테이블 엔티티 {counts['tables']}건 (이 테이블을 읽고 쓰는 프로그램 전부)", f"- [functions/](/functions/) — 펑션모듈·BAPI·RFC {counts['functions']}건 (호출처 전부)", f"- [tcodes/](/tcodes/) — T-Code {counts['tcodes']}건", "", "프로그램 문서의 `## 로직 조각` 섹션이 검색의 1차 단위다 (LLM 이 코드에서 골라낸 업무 로직).", "위키 본문은 LLM 이 쓴 **해석**이다. \"어디서 채워지나/누가 호출하나\" 같은 사실은", "index-api 의 trace/graph/usage 도구로 확인한다.", "", ] if con is not None: body += _hierarchy_section(con) wiki_dir.mkdir(parents=True, exist_ok=True) (wiki_dir / "index.md").write_text(_fm(fm) + "\n" + "\n".join(body), encoding="utf-8") def _hierarchy_section(con: sqlite3.Connection) -> list[str]: """SAP 모듈 → 패키지 → 프로그램 계층. 모듈은 요약의 sap_module, 없으면 이름 접두로 추정.""" rows = con.execute( "SELECT p.name, p.devclass, p.title_ko, p.summary_json, p.has_source, " "COALESCE(k.text_ko,'') AS pkg_text FROM program p " "LEFT JOIN package k ON k.devclass=p.devclass WHERE p.has_source=1 ORDER BY p.name" ).fetchall() tree: dict[str, dict[str, list[sqlite3.Row]]] = {} for r in rows: module = ((loads(r["summary_json"]) or {}).get("sap_module") or "").strip().upper() if not module: # 요약 전에도 계층이 서야 한다 — Z<모듈><번호> 관행에서 2글자 모듈 코드를 추정 m = re.match(r"^[YZ]([A-Z]{2})", r["name"] or "") module = m.group(1) if m else "기타" tree.setdefault(module, {}).setdefault(r["devclass"] or "(패키지 미지정)", []).append(r) out = ["## 계층 — 모듈 → 패키지 → 프로그램", ""] for module in sorted(tree): n = sum(len(v) for v in tree[module].values()) out.append(f"### {module} ({n}본)") out.append("") for devclass, progs in sorted(tree[module].items()): link = f"[{devclass}](/packages/{devclass}.md)" if devclass != "(패키지 미지정)" else devclass out.append(f"- {link}") for p in progs: title = p["title_ko"] or "" out.append(f" - [{p['name']}](/programs/{p['name']}.md)" + (f" — {title}" if title else "")) out.append("") return out # ------------------------------------------------------------------ 진입점 def write_program_wiki(program: str, con: sqlite3.Connection | None = None, wiki_dir: Path | None = None) -> dict: """프로그램 하나의 program(로직 조각 포함)/package 문서를 생성·병합한다. 조각 추출 후 호출.""" own = con is None con = con or connect() wiki_dir = wiki_dir or settings.wiki_dir stats: dict[str, int] = {"chunks": 0, "programs": 0, "packages": 0} try: p = con.execute( "SELECT p.*, COALESCE(k.text_ko,'') AS pkg_text FROM program p " "LEFT JOIN package k ON k.devclass=p.devclass WHERE p.name=?", (program.upper(),), ).fetchone() if not p: raise ValueError(f"프로그램 '{program}' 이(가) 인덱스에 없습니다") program = p["name"] stats["chunks"] = con.execute( "SELECT COUNT(*) AS c FROM logic_chunk WHERE program=?", (program,)).fetchone()["c"] # program 문서 — 사람 교정 보존 병합 text, source_hash = _render_program(con, p) rel = f"programs/{program}.md" merge_write(wiki_dir / rel, text, code_hash=source_hash, wiki_dir=wiki_dir, doc_rel=rel) stats["programs"] += 1 # package 문서 — 사람 교정 보존 병합 (해시 개념 없음 → 본문 보존만) if p["devclass"]: rel = f"packages/{p['devclass']}.md" merge_write(wiki_dir / rel, _render_package(con, p["devclass"], p["pkg_text"]), code_hash=None, wiki_dir=wiki_dir, doc_rel=rel) stats["packages"] += 1 write_manifest(wiki_dir, con) return stats finally: if own: con.close()