"""Step 6: anchor_sets 검증 + 근거 첨부. 원칙: - AI 는 의미 묶음 판단 안 함. anchor_sets_input.yaml 이 사람이 편집한 source. - pipeline_06 은 term 존재 검증 + 근거(local/df/source) 첨부 + 보고서 생성만. - 없는 term 은 not_in_frame 플래그 (에러 아님, 사용자 재검토용). 입력: - anchor_sets_input.yaml (사용자 편집) - anchor_candidates.yaml (Step 5.1) - normalized_text_tokens.yaml (corpus 통계) 산출: - anchor_sets_draft.yaml - anchor_sets_report.md - anchor_sets_report.html """ from pathlib import Path import yaml import markdown HERE = Path(__file__).parent INPUT = HERE / "anchor_sets_input.yaml" CANDIDATES = HERE / "anchor_candidates.yaml" NORMALIZED = HERE / "normalized_text_tokens.yaml" OUT_YAML = HERE / "anchor_sets_draft.yaml" OUT_MD = HERE / "anchor_sets_report.md" OUT_HTML = HERE / "anchor_sets_report.html" EXAMPLE_TRUNCATE = 80 def truncate(s, n): return s if len(s) <= n else s[:n - 1] + '…' def build_candidate_index(frame_info): """frame 의 full_candidates 를 token → candidate dict 로 인덱싱.""" idx = {c['token']: c for c in frame_info['full_candidates']} frequent_set = {c['token'] for c in frame_info['frequent_candidates']} special_set = {c['token'] for c in frame_info['special_candidates']} unique_set = {c['token'] for c in frame_info['unique_to_frame_candidates']} return idx, frequent_set, special_set, unique_set def verify_term(token, cand_idx, freq_set, spec_set, uniq_set, corpus_fd, corpus_md_df): """term 검증 + 근거 수집.""" if token in cand_idx: c = cand_idx[token] sources = [] if token in freq_set: sources.append('Frequent') if token in spec_set: sources.append('Special') if token in uniq_set: sources.append('Unique') if not sources: sources.append('Full') example = c['examples'][0] if c.get('examples') else None return { 'token': token, 'in_frame': True, 'local_count': c['frame_local_count'], 'frame_df': c['frame_df'], 'mdx_df': c['mdx_df'], 'mdx_hit': c['mdx_hit'], 'is_special': c['is_special'], 'source_groups': sources, 'example': example, } else: # corpus 에 있는지 확인 (다른 frame 에만 있는 경우) in_corpus = token in corpus_fd return { 'token': token, 'in_frame': False, 'in_corpus': in_corpus, 'corpus_frame_df': corpus_fd.get(token, 0) if in_corpus else 0, 'corpus_mdx_df': corpus_md_df.get(token, 0) if in_corpus else 0, 'note': ( 'token 이 해당 frame 의 실제 후보에 없음. ' '코퍼스에 있지만 다른 frame/source 에만 등장.' if in_corpus else 'token 이 전체 코퍼스에 없음. 철자/정규화 확인 필요.' ), } def verify_flagged_terms(flagged_list, cand_idx, corpus_fd, corpus_md_df): """flagged_terms 검증 + corpus 존재 여부 첨부.""" out = [] for ft in flagged_list or []: tok = ft.get('token') out.append({ 'token': tok, 'reason': ft.get('reason', ''), 'in_frame': tok in cand_idx, 'in_corpus': tok in corpus_fd, 'corpus_frame_df': corpus_fd.get(tok, 0), 'corpus_mdx_df': corpus_md_df.get(tok, 0), }) return out def verify_review_needed(review_list, cand_idx, corpus_fd, corpus_md_df): """review_needed 검증 (frame 에는 있으나 의미상 확인 필요).""" out = [] for rv in review_list or []: tok = rv.get('token') cand = cand_idx.get(tok) out.append({ 'token': tok, 'reason': rv.get('reason', ''), 'in_frame': tok in cand_idx, 'local_count': cand['frame_local_count'] if cand else 0, 'frame_df': cand['frame_df'] if cand else corpus_fd.get(tok, 0), 'mdx_df': cand['mdx_df'] if cand else corpus_md_df.get(tok, 0), }) return out def process_frame(fid, frame_input, candidates_data, corpus_fd, corpus_md_df): frame_info = candidates_data['frames'][fid] cand_idx, freq_set, spec_set, uniq_set = build_candidate_index(frame_info) anchor_sets_out = [] for aset in frame_input.get('anchor_sets', []): verified_terms = [] for tok in aset.get('terms', []): verified_terms.append( verify_term(tok, cand_idx, freq_set, spec_set, uniq_set, corpus_fd, corpus_md_df) ) in_frame_count = sum(1 for t in verified_terms if t['in_frame']) not_in_frame_count = len(verified_terms) - in_frame_count anchor_sets_out.append({ 'id': aset['id'], 'term_count': len(verified_terms), 'in_frame_count': in_frame_count, 'not_in_frame_count': not_in_frame_count, 'terms': verified_terms, }) result = { 'frame_number': frame_info['frame_number'], 'total_unique_tokens': frame_info['total_unique_tokens'], 'anchor_sets': anchor_sets_out, } if 'notes' in frame_input: result['notes'] = frame_input['notes'] if 'review_needed' in frame_input: result['review_needed'] = verify_review_needed( frame_input['review_needed'], cand_idx, corpus_fd, corpus_md_df ) if 'flagged_terms' in frame_input: result['flagged_terms'] = verify_flagged_terms( frame_input['flagged_terms'], cand_idx, corpus_fd, corpus_md_df ) return result # ---------- 렌더 ---------- def term_row_md(t): if t['in_frame']: flags = [] if t['is_special']: flags.append('S') if t['mdx_hit']: flags.append('M') flag_str = ''.join(flags) if flags else '—' ex = truncate(t['example'], EXAMPLE_TRUNCATE) if t.get('example') else '—' return (f"| ✓ {t['token']} | {t['local_count']} | {t['frame_df']} | " f"{t['mdx_df']} | {flag_str} | {'+'.join(t['source_groups'])} | {ex} |") else: note = t.get('note', '') return (f"| ⚠ {t['token']} | — | — | — | — | **not_in_frame** | {note} |") def anchor_set_md(aset): lines = [ f"##### anchor_set: `{aset['id']}` " f"({aset['in_frame_count']}/{aset['term_count']} in_frame" + (f", ⚠ {aset['not_in_frame_count']} not_in_frame" if aset['not_in_frame_count'] else "") + ")", "", "| token | local | df/32 | mdx/3 | flags | source | example |", "|---|---|---|---|---|---|---|", ] for t in aset['terms']: lines.append(term_row_md(t)) return '\n'.join(lines) def frame_section_md(fid, frame_out): head = (f"### Frame {frame_out['frame_number']} / {fid} " f"(total_unique {frame_out['total_unique_tokens']})") sets = [anchor_set_md(a) for a in frame_out['anchor_sets']] return head + "\n\n" + "\n\n".join(sets) def main(): input_data = yaml.safe_load(INPUT.read_text(encoding='utf-8')) candidates_data = yaml.safe_load(CANDIDATES.read_text(encoding='utf-8')) normalized = yaml.safe_load(NORMALIZED.read_text(encoding='utf-8')) corpus_fd = normalized['corpus']['token_frame_df'] corpus_md_df = normalized['corpus']['token_mdx_df'] output = { 'meta': { 'pipeline_step': 6, 'scope': 'TARGET_FRAMES (13, 14, 18, 29)', 'note': ( 'AI 자동 묶음 없음. anchor_sets_input.yaml 이 사람 편집 source. ' '이 파이프라인은 존재 검증 + 근거 첨부만 수행. ' 'not_in_frame 은 에러가 아니라 사용자 재검토 플래그.' ), }, 'frames': {}, } total_sets = 0 total_terms = 0 total_in_frame = 0 total_not_in_frame = 0 not_in_frame_details = [] # (frame_id, set_id, token, note) for fid, frame_input in input_data.get('frames', {}).items(): fid_str = str(fid) if fid_str not in candidates_data['frames']: print(f"WARNING: frame {fid_str} 이 anchor_candidates.yaml 에 없음") continue frame_out = process_frame(fid_str, frame_input, candidates_data, corpus_fd, corpus_md_df) output['frames'][fid_str] = frame_out for aset in frame_out['anchor_sets']: total_sets += 1 total_terms += aset['term_count'] total_in_frame += aset['in_frame_count'] total_not_in_frame += aset['not_in_frame_count'] for t in aset['terms']: if not t['in_frame']: not_in_frame_details.append({ 'frame_id': fid_str, 'frame_number': frame_out['frame_number'], 'set_id': aset['id'], 'token': t['token'], 'in_corpus': t.get('in_corpus', False), 'note': t.get('note', ''), }) output['meta']['totals'] = { 'frames': len(output['frames']), 'anchor_sets': total_sets, 'terms': total_terms, 'in_frame': total_in_frame, 'not_in_frame': total_not_in_frame, } output['meta']['not_in_frame_details'] = not_in_frame_details OUT_YAML.write_text( yaml.safe_dump(output, allow_unicode=True, sort_keys=False, width=300), encoding='utf-8', ) # md 생성 md_sections = [ "# Anchor Sets Report — Step 6", "", f"- scope: TARGET_FRAMES 4개 (13, 14, 18, 29)", f"- frames: {len(output['frames'])}", f"- anchor_sets: {total_sets}", f"- terms: {total_terms} ({total_in_frame} in_frame, {total_not_in_frame} not_in_frame)", f"- source: `anchor_sets_input.yaml` (사람이 직접 편집)", f"- 이 보고서: `pipeline_06_anchor_sets.py` 가 검증 + 근거 첨부 후 생성", "", ] if not_in_frame_details: md_sections.append("## ⚠ not_in_frame terms (사용자 재검토)") md_sections.append("") md_sections.append("| Frame | set_id | token | in_corpus | note |") md_sections.append("|---|---|---|---|---|") for d in not_in_frame_details: md_sections.append( f"| Frame {d['frame_number']} / {d['frame_id']} | {d['set_id']} | " f"{d['token']} | {'Y' if d['in_corpus'] else 'N'} | {d['note']} |" ) md_sections.append("") md_sections.append("## Frame 별 anchor_sets 상세") md_sections.append("") md_sections.append("**flags**: S = special token, M = mdx_hit") md_sections.append("") for fid, frame_out in output['frames'].items(): md_sections.append(frame_section_md(fid, frame_out)) md_sections.append("") md_text = '\n'.join(md_sections) OUT_MD.write_text(md_text, encoding='utf-8') # html html_body = markdown.markdown(md_text, extensions=['tables']) html = f""" Anchor Sets Report — Step 6 {html_body} """ OUT_HTML.write_text(html, encoding='utf-8') print(f"[Step 6] anchor_sets 검증 + 근거 첨부 완료") print(f" frames: {len(output['frames'])}") print(f" anchor_sets: {total_sets}") print(f" terms: {total_terms}") print(f" in_frame: {total_in_frame}") print(f" not_in_frame: {total_not_in_frame}") if not_in_frame_details: print() print(f" [⚠ not_in_frame 상세]") for d in not_in_frame_details: print(f" Frame {d['frame_number']} [{d['set_id']}] {d['token']!r} " f"corpus={'Y' if d['in_corpus'] else 'N'} {d['note']}") print() print(f"산출:") print(f" yaml: {OUT_YAML}") print(f" md: {OUT_MD}") print(f" html: {OUT_HTML}") if __name__ == "__main__": main()