"""Generate MDX_MATCHING_REPORT. The report compares MDX section keywords against frame keywords and shows the top matching frames with Korean labels that are readable outside the codebase. """ from __future__ import annotations import hashlib from collections import defaultdict from datetime import datetime from pathlib import Path import markdown import yaml HERE = Path(__file__).parent INPUT_AUTO = HERE / "auto_anchor_candidates.yaml" INPUT_NORMALIZED = HERE / "normalized_text_tokens.yaml" OUT_YAML = HERE / "mdx_matching_result.yaml" OUT_MD = HERE / "MDX_MATCHING_REPORT.md" OUT_HTML = HERE / "MDX_MATCHING_REPORT.html" # Current review weights. These are intentionally shown in the report. WEIGHT_STANDALONE = 0.414 # Logistic Regression 학습 (TARGET 4, LOOCV 4/4) WEIGHT_GROUP = 0.320 WEIGHT_RELATED = 0.265 # Review targets with known expected frames (ANSWER_MAP 잠금: 수정 금지). ANSWER_MAP = { "01-2": 18, "02-2.2": 14, "03-1": 13, "03-2": 29, } # 홀드아웃 섹션 (블라인드 검증, 기대 프레임 미지정) # 04-* 는 MDX 04 (DX 지연 요인) 신규 일반화 테스트 — 기대 프레임 없음 HOLDOUT_SECTIONS = ['01-1', '02-1', '02-2.1', '04-1', '04-2.1', '04-2.2'] # 기준 잠금 대상 파일 (홀드아웃 실행 시 sha256 기록) LOCK_SNAPSHOT_FILES = [ 'keyword_normalizer.py', 'synonyms.yaml', 'pipeline_01_extract_nodes.py', 'pipeline_04_normalize.py', 'pipeline_06_2_mdx_matching.py', 'pipeline_07_auto_anchor_candidates.py', ] def sha256_file(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def build_lock_snapshot() -> dict: """홀드아웃 실행 시점의 기준 잠금 스냅샷.""" return { 'timestamp': datetime.now().isoformat(timespec='seconds'), 'files': { name: sha256_file(HERE / name) if (HERE / name).exists() else None for name in LOCK_SNAPSHOT_FILES }, 'config': { 'weights': { 'W1_standalone': WEIGHT_STANDALONE, 'W2_group': WEIGHT_GROUP, 'W3_related': WEIGHT_RELATED, }, 'answer_map': dict(ANSWER_MAP), 'holdout_sections': list(HOLDOUT_SECTIONS), }, 'principle': [ '검증 대상 고정 — ANSWER_MAP 4개 유지', '기준 잠금 — 전처리/anchor/가중치 수정 금지', '블라인드 실행 — 홀드아웃 기대 프레임 미지정', '사후 평가 — 실행 후 사람이 결과 해석', '수정 규칙 제한 — 섹션 맞추기 보정 금지, 공통 규칙만', ], } def is_standalone_keyword(frame_appearances: int, is_important: bool) -> bool: return frame_appearances == 1 or (is_important and frame_appearances <= 2) def write_html(md_text: str) -> None: html_body = markdown.markdown(md_text, extensions=["tables"]) html = f""" 키워드 기반 MDX 매칭 {html_body} """ OUT_HTML.write_text(html, encoding="utf-8") def build_frame_layers(auto: dict) -> tuple[dict, dict]: frame_groups = defaultdict(list) frame_keyword_info = defaultdict(dict) for set_id, info in auto["source_text_sets"].items(): frame_id = info["frame_id"] frame_groups[frame_id].append( { "source_text_raw": info["source_text_raw"], "keywords": info["term_values"], } ) for term in info["terms"]: keyword = term["token"] if keyword not in frame_keyword_info[frame_id]: frame_keyword_info[frame_id][keyword] = { "frame_appearances": term["frame_df"], "mdx_appearances": term["mdx_df"], "is_important": term["is_special"], } frame_layers = {} for frame_id, keyword_map in frame_keyword_info.items(): standalone = sorted( keyword for keyword, info in keyword_map.items() if is_standalone_keyword(info["frame_appearances"], info["is_important"]) ) related = sorted( keyword for keyword, info in keyword_map.items() if not is_standalone_keyword(info["frame_appearances"], info["is_important"]) ) frame_layers[frame_id] = { "standalone_keywords": standalone, "related_keywords": related, "keyword_groups": frame_groups[frame_id], } frame_numbers = { frame_id: auto["frame_stats"][frame_id]["frame_number"] for frame_id in auto["frame_stats"] } return frame_layers, frame_numbers def score_frame(mdx_keywords: set[str], layers: dict) -> dict: standalone_total = layers["standalone_keywords"] standalone_hit = [kw for kw in standalone_total if kw in mdx_keywords] standalone_score = len(standalone_hit) / len(standalone_total) if standalone_total else 0.0 group_details = [] for group in layers["keyword_groups"]: hit = [kw for kw in group["keywords"] if kw in mdx_keywords] coverage = len(hit) / len(group["keywords"]) if group["keywords"] else 0.0 group_details.append( { "source_text_raw": group["source_text_raw"], "keywords": group["keywords"], "hit_keywords": hit, "coverage": round(coverage, 3), } ) group_score_avg = ( sum(group["coverage"] for group in group_details) / len(group_details) if group_details else 0.0 ) group_score_max = max((group["coverage"] for group in group_details), default=0.0) related_total = layers["related_keywords"] related_hit = [kw for kw in related_total if kw in mdx_keywords] related_score = len(related_hit) / len(related_total) if related_total else 0.0 total_score = ( WEIGHT_STANDALONE * standalone_score + WEIGHT_GROUP * group_score_avg + WEIGHT_RELATED * related_score ) return { "standalone": { "total": standalone_total, "total_count": len(standalone_total), "hit": standalone_hit, "hit_count": len(standalone_hit), "score": round(standalone_score, 3), }, "keyword_group": { "total_count": len(group_details), "score_avg": round(group_score_avg, 3), "score_max": round(group_score_max, 3), "groups": group_details, }, "related": { "total_count": len(related_total), "hit": related_hit, "hit_count": len(related_hit), "score": round(related_score, 3), }, "matching_score": round(total_score, 3), } def build_frame_descriptions(auto: dict) -> dict: """Frame 번호 → 첫 source_text (설명용).""" out = {} for fid, info in auto["frame_stats"].items(): fnum = info["frame_number"] # 해당 frame 의 source_text_sets 중 index=1 인 것 for set_id, sinfo in auto["source_text_sets"].items(): if sinfo["frame_id"] == fid and sinfo["source_text_index"] == 1: raw = sinfo["source_text_raw"] if len(raw) > 35: raw = raw[:34] + "…" out[fnum] = raw break if fnum not in out: out[fnum] = f"Frame {fnum}" return out def cell_html(rank_row: dict, descriptions: dict, is_answer: bool = False, is_holdout_top: bool = False) -> str: """Top-N 매트릭스 한 칸의 HTML.""" fn = rank_row["frame_number"] score = rank_row["matching_score"] desc = descriptions.get(fn, "") img_src = f"../../data/figma_previews/{fn:02d}.png" inner = ( f'{fn:02d}' f'Frame {fn} ' f'{score:.3f}' f'{desc}' ) if is_answer: return f'
🎯 정답
{inner}
' if is_holdout_top: return f'
{inner}
' return f'{inner}' def main() -> None: auto = yaml.safe_load(INPUT_AUTO.read_text(encoding="utf-8")) normalized = yaml.safe_load(INPUT_NORMALIZED.read_text(encoding="utf-8")) frame_layers, frame_numbers = build_frame_layers(auto) frame_descriptions = build_frame_descriptions(auto) results = {} # TARGET + 홀드아웃 모두 처리. ANSWER_MAP 없는 섹션은 answer 없이 Top 만. all_mdx_sections = list(ANSWER_MAP.keys()) + [ s for s in HOLDOUT_SECTIONS if s not in ANSWER_MAP ] for mdx_id in all_mdx_sections: if mdx_id not in normalized["mdx"]: continue answer_frame_number = ANSWER_MAP.get(mdx_id) # 홀드아웃은 None mdx_keywords = set(normalized["mdx"][mdx_id]["unique_tokens"]) per_frame = {} for frame_id, layers in frame_layers.items(): scored = score_frame(mdx_keywords, layers) scored["frame_number"] = frame_numbers[frame_id] per_frame[frame_id] = scored def rank_by(key_fn): return sorted(per_frame.items(), key=lambda x: (-key_fn(x[1]), x[0])) rank_total = rank_by(lambda d: d["matching_score"]) rank_group = rank_by(lambda d: d["keyword_group"]["score_avg"]) rank_standalone = rank_by(lambda d: d["standalone"]["score"]) rank_related = rank_by(lambda d: d["related"]["score"]) def find_rank(ranked_list, frame_number: int) -> int | None: for idx, (_frame_id, info) in enumerate(ranked_list, start=1): if info["frame_number"] == frame_number: return idx return None is_holdout = mdx_id in HOLDOUT_SECTIONS # 홀드아웃 섹션은 answer_frame_number 가 None 이라 answer_ranks 계산 불가 → None 기록 answer_ranks_val = None if answer_frame_number is not None: answer_ranks_val = { "by_matching_score": find_rank(rank_total, answer_frame_number), "by_keyword_group_score": find_rank(rank_group, answer_frame_number), "by_standalone_keyword_score": find_rank(rank_standalone, answer_frame_number), "by_related_keyword_score": find_rank(rank_related, answer_frame_number), } results[mdx_id] = { "mdx_keywords_count": len(mdx_keywords), "section_type": "holdout" if is_holdout else "target", "answer_frame_number": answer_frame_number, "answer_ranks": answer_ranks_val, "rank_by_matching_score": [ { "frame_id": frame_id, "frame_number": info["frame_number"], "matching_score": info["matching_score"], "standalone": info["standalone"]["score"], "keyword_group": info["keyword_group"]["score_avg"], "related": info["related"]["score"], } for frame_id, info in rank_total ], "rank_by_keyword_group_score": [ { "frame_id": frame_id, "frame_number": info["frame_number"], "score_avg": info["keyword_group"]["score_avg"], "score_max": info["keyword_group"]["score_max"], } for frame_id, info in rank_group ], "rank_by_standalone_keyword_score": [ { "frame_id": frame_id, "frame_number": info["frame_number"], "score": info["standalone"]["score"], "hit": info["standalone"]["hit"], } for frame_id, info in rank_standalone ], "rank_by_related_keyword_score": [ { "frame_id": frame_id, "frame_number": info["frame_number"], "score": info["related"]["score"], "hit_count": info["related"]["hit_count"], } for frame_id, info in rank_related ], "per_frame_detail": per_frame, } output = { "meta": { "pipeline_step": "6.2", "description": "MDX 키워드와 프레임별 키워드 매칭 결과 (TARGET + 홀드아웃)", "score_parts": ["단독 대표 키워드", "대표 키워드 묶음", "연관 키워드"], "weights": { "standalone_keyword": WEIGHT_STANDALONE, "keyword_group": WEIGHT_GROUP, "related_keyword": WEIGHT_RELATED, }, "formula": "matching_score = 0.30*standalone + 0.50*keyword_group + 0.20*related", "answer_map": ANSWER_MAP, "holdout_sections": HOLDOUT_SECTIONS, "source_mdx_keywords": "normalized_text_tokens.yaml", "source_frame_keywords": "auto_anchor_candidates.yaml", "lock_snapshot": build_lock_snapshot(), "note": "규칙 기반 점수입니다. 홀드아웃 섹션(01-1/02-1/02-2.1)은 기대 프레임 미지정 — 블라인드 검증용.", }, "mdx_sections": results, } OUT_YAML.write_text( yaml.safe_dump(output, allow_unicode=True, sort_keys=False, width=300), encoding="utf-8", ) md: list[str] = [ "# 키워드 기반 MDX 매칭", "", "이 문서는 **키워드 수준** 의 매칭 결과입니다. MDX 구간에서 추출한 키워드와 각 프레임에 적용된 키워드를 비교해, 어떤 프레임이 가장 잘 맞는지 1~3순위로 보여줍니다.", "", "_향후 단계에서 의미(semantic) / 구조(structure) / 적용가능성(template fit) 매칭이 추가될 예정입니다._", "", "점수는 세 가지 단서를 합산합니다.", "", "| 단서 | 의미 | 가중치 |", "|---|---|---:|", f"| 단독 대표 키워드 | 특정 프레임을 혼자서도 강하게 가리키는 키워드가 MDX에 얼마나 맞는지 | {WEIGHT_STANDALONE} |", f"| 대표 키워드 묶음 | 프레임 원문에서 함께 등장한 키워드 조합이 MDX에 얼마나 같이 맞는지 | {WEIGHT_GROUP} |", f"| 연관 키워드 | 그 외 참고 키워드가 MDX와 얼마나 겹치는지 | {WEIGHT_RELATED} |", "", "```text", f"종합 매칭 점수 = {WEIGHT_STANDALONE} x 단독 대표 키워드 점수", f" + {WEIGHT_GROUP} x 대표 키워드 묶음 점수", f" + {WEIGHT_RELATED} x 연관 키워드 점수", "```", "", "### 검증용 정답 프레임", "", "| MDX 구간 | 기대 프레임 |", "|---|---:|", ] for mdx_id, frame_number in ANSWER_MAP.items(): md.append(f"| {mdx_id} | Frame {frame_number} |") md += ["", "---", ""] # ============================================================ # 🖼 Top-3 매트릭스 (이미지 포함, TARGET + 홀드아웃) # ============================================================ md.append("## 🖼 Top-3 매칭 매트릭스") md.append("") md.append("**TARGET 섹션** (ANSWER_MAP 4개): 정답 프레임은 🎯 노란 박스. ") md.append("**홀드아웃 섹션** (기대 프레임 미지정): 1위는 파란 박스, 해석은 사용자 판단.") md.append("") def _matrix_table(section_order, label_fn): rows = [ '' '' ] for mdx_id in section_order: if mdx_id not in results: continue r = results[mdx_id] top3 = r["rank_by_matching_score"][:3] label = label_fn(mdx_id, r) tds = [f''] for idx, row in enumerate(top3): is_answer = (r["answer_frame_number"] is not None and row["frame_number"] == r["answer_frame_number"]) is_hold_top = (r["answer_frame_number"] is None and idx == 0) tds.append(cell_html(row, frame_descriptions, is_answer=is_answer, is_holdout_top=is_hold_top)) rows.append('' + ''.join(tds) + '') rows.append('
콘텐츠1순위2순위3순위

{label}
') return '\n'.join(rows) target_order = ["01-2", "02-2.2", "03-1", "03-2"] holdout_order = HOLDOUT_SECTIONS md.append("### TARGET 4 섹션") md.append("") md.append(_matrix_table( target_order, lambda mid, r: f'MDX {mid}
정답 Frame {r["answer_frame_number"]}', )) md.append("") md.append("### 🔒 홀드아웃 3 섹션 (블라인드)") md.append("") md.append(_matrix_table( holdout_order, lambda mid, r: f'MDX {mid}
기대 프레임 미지정', )) md.append("") md.append("---") md.append("") for mdx_id in ["01-2", "02-2.2", "03-1", "03-2"]: result = results[mdx_id] ranks = result["answer_ranks"] answer_frame = result["answer_frame_number"] md.append(f"## MDX {mdx_id} (기대 프레임: Frame {answer_frame})") md.append("") md.append("**기대 프레임의 순위**") md.append("") md.append(f"- 종합 매칭 점수 기준: **{ranks['by_matching_score']}위**") md.append(f"- 대표 키워드 묶음 기준: {ranks['by_keyword_group_score']}위") md.append(f"- 단독 대표 키워드 기준: {ranks['by_standalone_keyword_score']}위") md.append(f"- 연관 키워드 기준: {ranks['by_related_keyword_score']}위") md.append("") md.append("**종합 매칭 점수 Top 3**") md.append("") md.append("| 순위 | 프레임 | 종합 점수 | 단독 대표 키워드 | 대표 키워드 묶음 | 연관 키워드 |") md.append("|---:|---|---:|---:|---:|---:|") for idx, row in enumerate(result["rank_by_matching_score"][:3], start=1): mark = " ✓" if row["frame_number"] == answer_frame else "" md.append( f"| **{idx}**{mark} | Frame {row['frame_number']} / {row['frame_id']} | " f"**{row['matching_score']}** | {row['standalone']} | {row['keyword_group']} | {row['related']} |" ) md.append("") md.append("
Top 10 전체 보기") md.append("") md.append("| 순위 | 프레임 | 종합 점수 | 단독 대표 키워드 | 대표 키워드 묶음 | 연관 키워드 |") md.append("|---:|---|---:|---:|---:|---:|") for idx, row in enumerate(result["rank_by_matching_score"][:10], start=1): mark = " ✓" if row["frame_number"] == answer_frame else "" md.append( f"| {idx}{mark} | Frame {row['frame_number']} / {row['frame_id']} | " f"{row['matching_score']} | {row['standalone']} | {row['keyword_group']} | {row['related']} |" ) md.append("") md.append("
") md.append("") answer_frame_id = [ frame_id for frame_id, info in result["per_frame_detail"].items() if info["frame_number"] == answer_frame ][0] detail = result["per_frame_detail"][answer_frame_id] md.append(f"
기대 프레임 Frame {answer_frame} / {answer_frame_id} 상세 보기") md.append("") md.append( f"- 단독 대표 키워드: {detail['standalone']['hit_count']} / {detail['standalone']['total_count']} = {detail['standalone']['score']}" ) md.append(f" - 맞은 키워드: {', '.join(detail['standalone']['hit']) or '없음'}") md.append( f"- 대표 키워드 묶음: 평균 {detail['keyword_group']['score_avg']} / 최고 {detail['keyword_group']['score_max']}" ) for group in detail["keyword_group"]["groups"]: terms = ", ".join(group["keywords"]) hits = ", ".join(group["hit_keywords"]) or "없음" md.append(f" - `{group['source_text_raw']}` → [{terms}] / 맞은 키워드 [{hits}] / 일치율 **{group['coverage']}**") md.append( f"- 연관 키워드: {detail['related']['hit_count']} / {detail['related']['total_count']} = {detail['related']['score']}" ) md.append(f" - 맞은 키워드: {', '.join(detail['related']['hit']) or '없음'}") md.append(f"- 종합 매칭 점수: **{detail['matching_score']}**") md.append("") md.append("
") md.append("") md.append("---") md.append("") # ============================================================ # 홀드아웃 섹션 (블라인드, 기대 프레임 없음) # ============================================================ md.append("") md.append("# 🔒 홀드아웃 검증 섹션 (블라인드)") md.append("") md.append("아래 섹션들은 **기대 프레임을 사전에 지정하지 않은** 블라인드 검증 대상입니다.") md.append("현재 파이프라인으로 그대로 돌려 Top 1~3 결과와 점수를 raw 로 기록합니다.") md.append("결과 해석은 사람이 사후에 수행합니다.") md.append("") md.append("### 기준 잠금 스냅샷") md.append("") snap = output['meta']['lock_snapshot'] md.append(f"- **실행 시각**: {snap['timestamp']}") md.append(f"- **가중치**: W1={snap['config']['weights']['W1_standalone']} / " f"W2={snap['config']['weights']['W2_group']} / " f"W3={snap['config']['weights']['W3_related']}") md.append("") md.append("**파일 지문 (sha256, 앞 16자)**:") md.append("") md.append("| 파일 | sha256 |") md.append("|---|---|") for name, h in snap['files'].items(): h_short = h[:16] + '...' if h else '(없음)' md.append(f"| `{name}` | `{h_short}` |") md.append("") md.append("**원칙**:") md.append("") for p in snap['principle']: md.append(f"- {p}") md.append("") md.append("---") md.append("") for mdx_id in HOLDOUT_SECTIONS: if mdx_id not in results: continue result = results[mdx_id] md.append(f"## 🔒 MDX {mdx_id} (홀드아웃, 기대 프레임 미지정)") md.append("") md.append(f"- MDX 키워드 수: {result['mdx_keywords_count']}") md.append("") md.append("**Top 3 (기대 프레임 없음, 해석은 사후)**") md.append("") md.append("| 순위 | 프레임 | 종합 점수 | 단독 대표 키워드 | 대표 키워드 묶음 | 연관 키워드 |") md.append("|---:|---|---:|---:|---:|---:|") for idx, row in enumerate(result["rank_by_matching_score"][:3], start=1): md.append( f"| **{idx}** | Frame {row['frame_number']} / {row['frame_id']} | " f"**{row['matching_score']}** | {row['standalone']} | {row['keyword_group']} | {row['related']} |" ) md.append("") md.append("
Top 10 전체 보기") md.append("") md.append("| 순위 | 프레임 | 종합 점수 | 단독 대표 키워드 | 대표 키워드 묶음 | 연관 키워드 |") md.append("|---:|---|---:|---:|---:|---:|") for idx, row in enumerate(result["rank_by_matching_score"][:10], start=1): md.append( f"| {idx} | Frame {row['frame_number']} / {row['frame_id']} | " f"{row['matching_score']} | {row['standalone']} | {row['keyword_group']} | {row['related']} |" ) md.append("") md.append("
") md.append("") md.append("---") md.append("") md_text = "\n".join(md) OUT_MD.write_text(md_text, encoding="utf-8") write_html(md_text) print("산출 완료:") print(f" yaml: {OUT_YAML}") print(f" md: {OUT_MD}") print(f" html: {OUT_HTML}") print() print("기준 잠금 스냅샷:") print(f" 실행시각: {output['meta']['lock_snapshot']['timestamp']}") for name, h in output['meta']['lock_snapshot']['files'].items(): if h: print(f" {name:40s} {h[:16]}...") if __name__ == "__main__": main()