"""텍스트 추출 → Kiwi 전처리 → 중복 제거 → synonym 매칭 정리 (투명 파이프라인). 원칙: - 프레임별 texts.md 에서 `- ` list item 만 추출 (메타/섹션 제목 배제) - BEPS(1171281171) + 32 frames + 3 MDX corpus 전체 - 각 단계 counts 명시 출력: - TEXT_PIPELINE_REPORT.md — 4단계 진행 상황 + 결과 - text_canonical.yaml — Step 4 최종 canonical 목록 (검수용) """ import re import sys from pathlib import Path import yaml from collections import Counter, defaultdict sys.path.insert(0, str(Path(__file__).parent)) from methods import _get_kiwi ROOT = Path(r"d:\ad-hoc\kei\design_agent") BLOCKS_DIR = ROOT / "figma_to_html_agent" / "blocks" MDX_DIR = ROOT / "samples" / "mdx_batch" HERE = Path(__file__).parent BEPS_ID = "1171281171" FRAME_IDS = sorted([ d.name for d in BLOCKS_DIR.iterdir() if d.is_dir() and d.name.startswith("1171") and d.name != BEPS_ID ]) # Kiwi 명사/외국어/숫자 tag NOUN_TAGS = {'NNG', 'NNP', 'NR', 'SL', 'SN', 'SH'} MIN_TOKEN_LEN = 2 # Step 2 어휘 노이즈 (너무 빈번하거나 의미 약한 조사/관형어) KIWI_STOPWORDS = { '것', '수', '등', '때', '중', '후', '전', '및', '위', '내', '때문', '통해', '대한', '관련', '따라', '위해', '이런', '저런', '그런', '이것', '그것', '저것', '무엇', '어디', '언제', '어떻게', # text 파싱 artifacts 'br', 'td', 'tr', 'lt', 'gt', 'px', 'rem', 'em', } # ═══ Step 1: raw list items 수집 ═══ def extract_list_items(path): """texts.md 또는 MDX 에서 `- xxx` 형태 list item 만 추출.""" items = [] for line in path.read_text(encoding='utf-8').split('\n'): stripped = line.strip() if not stripped: continue # heading / blockquote 배제 if stripped.startswith('#') or stripped.startswith('>'): continue m = re.match(r'^[-*]\s+(.+)$', stripped) if m: content = m.group(1).strip() # HTML tag 제거 content = re.sub(r'<[^>]+>', ' ', content) # markdown bold 제거 content = re.sub(r'\*\*([^*]+)\*\*', r'\1', content) content = content.strip() if content and len(content) >= 2 and content not in {'-', '–', '—'}: items.append(content) return items def step1_gather(): """frames + BEPS + MDX 에서 list items 수집.""" items_per_source = {} # BEPS p = BLOCKS_DIR / BEPS_ID / "texts.md" if p.exists(): items_per_source[f'BEPS'] = extract_list_items(p) # 32 frames for fid in FRAME_IDS: p = BLOCKS_DIR / fid / "texts.md" if p.exists(): items_per_source[f'Frame/{fid}'] = extract_list_items(p) # 3 MDX for n in ['01', '02', '03']: p = MDX_DIR / f'{n}.mdx' if p.exists(): items_per_source[f'MDX/{n}'] = extract_list_items(p) return items_per_source # ═══ Step 2: Kiwi 전처리 + 정리 ═══ def step2_preprocess(items_per_source): """각 item 에 대해: - Kiwi 로 의미있는 토큰(명사/외국어/숫자) 추출 - 짧은 compound/label 은 원형 그대로 보존 (Kiwi 가 쪼개는 문제 보호) """ kiwi = _get_kiwi() tokens_with_source = [] # [(token, source)] atomics_with_source = [] # [(short_atomic, source)] for source, items in items_per_source.items(): for item in items: # Kiwi tokenize for tok in kiwi.tokenize(item): if tok.tag not in NOUN_TAGS: continue if len(tok.form) < MIN_TOKEN_LEN: continue if tok.form in KIWI_STOPWORDS: continue # 순수 숫자/단위 배제 (너무 generic) if tok.form.isdigit(): continue tokens_with_source.append((tok.form, source)) # 짧은 label/compound 그대로 보존 (Kiwi 분해 대비) # 공백 1~2개 이하 + 20자 이하 + 문장부호 적음 if len(item) <= 20 and item.count(' ') <= 1 and ',' not in item: if item and item not in KIWI_STOPWORDS: atomics_with_source.append((item, source)) return tokens_with_source, atomics_with_source # ═══ Step 3: 중복 제거 ═══ def step3_dedup(tokens_with_source, atomics_with_source): """token 단위 + atomic 단위 각각 dedup, 합쳐서 최종 unique set.""" token_set = set(t for t, _ in tokens_with_source) atomic_set = set(a for a, _ in atomics_with_source) # source 별 추적 (어느 파일에서 나왔는지) source_map = defaultdict(set) for t, src in tokens_with_source: source_map[t].add(src) for a, src in atomics_with_source: source_map[a].add(src) combined = token_set | atomic_set return combined, source_map # ═══ Step 4: synonym 매칭 정리 ═══ def step4_synonym_collapse(unique_texts, source_map): """synonyms.yaml 기준으로 variants → canonical 치환. 추가 규칙 (compound 보호): - 띄어쓰기 제거 시 canonical 이면 매핑 (결과 혁신 → 결과혁신) - case-insensitive variant 매핑 (As-is → AS-IS 등) """ # synonyms.yaml 로드 syn_path = HERE / "synonyms.yaml" with open(syn_path, encoding='utf-8') as f: syn_data = yaml.safe_load(f) synonyms = syn_data.get('synonyms', {}) # reverse map: variant → canonical reverse = {} for canonical, variants in synonyms.items(): reverse[canonical] = canonical for v in variants: reverse[v] = canonical canonical_set = set() collapse_log = [] # [(original, canonical)] for text in unique_texts: # 1) exact variant match if text in reverse: canonical = reverse[text] # 2) 공백 제거 변형 (결과 혁신 → 결과혁신) elif text.replace(' ', '') in reverse: compact = text.replace(' ', '') canonical = reverse[compact] # 3) case-insensitive (As-is → AS-IS) elif text.upper() in reverse: canonical = reverse[text.upper()] # 4) as-is else: canonical = text canonical_set.add(canonical) if canonical != text: collapse_log.append((text, canonical)) return canonical_set, collapse_log # ═══ 리포트 ═══ def write_report(items_per_source, s2_tokens, s2_atomics, s3_combined, s3_source_map, s4_canonical, s4_log): lines = [] lines.append("# 텍스트 파이프라인 리포트 (Step 1 → 4)") lines.append("") lines.append("**목적**: BEPS + 32 frames + 3 MDX 의 list item 에서 canonical 키워드를 " "투명한 4단계로 추출. 각 단계 count 제공.") lines.append("") # ─── Step 1 ─── total_raw = sum(len(v) for v in items_per_source.values()) lines.append(f"## Step 1. 프레임별 texts.md → `- ` list items 수집") lines.append("") lines.append(f"- **총 raw items**: **{total_raw}개**") lines.append(f"- Sources: BEPS 1 + 32 frames + 3 MDX = {len(items_per_source)}") lines.append("") # per-source count lines.append("**Source 별 items 수 (상위 10)**:") lines.append("") lines.append("| Source | items |") lines.append("|--------|------|") for src, items in sorted(items_per_source.items(), key=lambda x: -len(x[1]))[:10]: lines.append(f"| {src} | {len(items)} |") lines.append(f"| ... | ... |") lines.append("") # ─── Step 2 ─── unique_tokens = set(t for t, _ in s2_tokens) unique_atomics = set(a for a, _ in s2_atomics) lines.append(f"## Step 2. Kiwi 전처리 + 정리") lines.append("") lines.append(f"- **Kiwi 토큰** (명사/외국어/숫자, len≥2, stopword 제외): **{len(s2_tokens)}개** (중복 포함)") lines.append(f" - 고유 토큰: **{len(unique_tokens)}개**") lines.append(f"- **Atomic 보존** (≤20자, 공백≤1): **{len(s2_atomics)}개** (중복 포함)") lines.append(f" - 고유 atomic: **{len(unique_atomics)}개**") lines.append(f"- **중복 포함 총합**: {len(s2_tokens) + len(s2_atomics)}개") lines.append("") lines.append("**샘플**:") lines.append("") lines.append("- Kiwi 토큰 예: " + ", ".join(sorted(unique_tokens)[:15]) + " ...") lines.append("- Atomic 예: " + ", ".join(sorted(unique_atomics)[:15]) + " ...") lines.append("") # ─── Step 3 ─── lines.append(f"## Step 3. 중복 제거") lines.append("") lines.append(f"- Kiwi 토큰 고유 ({len(unique_tokens)}) ∪ Atomic 고유 ({len(unique_atomics)})") lines.append(f"- **합친 후 고유**: **{len(s3_combined)}개**") lines.append("") # ─── Step 4 ─── lines.append(f"## Step 4. Synonym 매칭 정리") lines.append("") lines.append(f"- synonyms.yaml 의 canonical 로 collapse") lines.append(f"- **collapse 적용 전**: {len(s3_combined)}개") lines.append(f"- **collapse 수**: **{len(s4_log)}개** (variant → canonical 치환)") lines.append(f"- **최종 canonical**: **{len(s4_canonical)}개**") lines.append("") lines.append("**collapse 사례 (상위 20)**:") lines.append("") lines.append("| variant | canonical |") lines.append("|---------|-----------|") # sort: canonical 별 묶기 by_canon = defaultdict(list) for v, c in s4_log: by_canon[c].append(v) shown = 0 for c in sorted(by_canon.keys()): for v in by_canon[c]: lines.append(f"| `{v}` | `{c}` |") shown += 1 if shown >= 20: break if shown >= 20: break lines.append("") # Compound 관심 항목 검증 lines.append("### 사용자님 우려 검증 (결과혁신 / DX / 필수조건)") lines.append("") test_canonicals = ['결과혁신', '과정혁신', 'DX', '필수조건', '3D모델', '2D도면'] lines.append("| Canonical | collapse 된 원본 | canonical set 포함? |") lines.append("|-----------|----------------|-------------------|") for tc in test_canonicals: mapped = by_canon.get(tc, []) in_set = '✓' if tc in s4_canonical else '✗' mapped_str = ', '.join(mapped) if mapped else '_(매핑 없음)_' lines.append(f"| `{tc}` | {mapped_str} | {in_set} |") lines.append("") # ─── 최종 canonical list ─── lines.append("## 최종 canonical 전체 (alphabetical)") lines.append("") canonical_sorted = sorted(s4_canonical) lines.append(", ".join(f"`{c}`" for c in canonical_sorted[:80])) if len(canonical_sorted) > 80: lines.append(f" ... (총 {len(canonical_sorted)}개)") lines.append("") out = HERE / "TEXT_PIPELINE_REPORT.md" out.write_text("\n".join(lines), encoding='utf-8') print(f"리포트: {out}") # yaml 도 저장 yaml_out = HERE / "text_canonical.yaml" with open(yaml_out, 'w', encoding='utf-8') as f: yaml.safe_dump({ 'meta': { 'step1_raw_items': total_raw, 'step2_unique_tokens': len(unique_tokens), 'step2_unique_atomics': len(unique_atomics), 'step3_combined_unique': len(s3_combined), 'step4_final_canonical': len(s4_canonical), }, 'canonical': sorted(s4_canonical), }, f, allow_unicode=True, sort_keys=False) print(f"yaml: {yaml_out}") # ═══ 엔트리 포인트 ═══ def main(): print("[Step 1] list items 수집...") items_per_source = step1_gather() total_raw = sum(len(v) for v in items_per_source.values()) print(f" → 총 raw items: {total_raw} (sources: {len(items_per_source)})") print("[Step 2] Kiwi 전처리 + atomic 보존...") s2_tokens, s2_atomics = step2_preprocess(items_per_source) unique_tokens = set(t for t, _ in s2_tokens) unique_atomics = set(a for a, _ in s2_atomics) print(f" → Kiwi 토큰 {len(s2_tokens)} (고유 {len(unique_tokens)})") print(f" → Atomic {len(s2_atomics)} (고유 {len(unique_atomics)})") print("[Step 3] 중복 제거 + 합치기...") s3_combined, s3_source_map = step3_dedup(s2_tokens, s2_atomics) print(f" → 고유 합치기: {len(s3_combined)}개") print("[Step 4] synonym 매칭 collapse...") s4_canonical, s4_log = step4_synonym_collapse(s3_combined, s3_source_map) print(f" → collapse {len(s4_log)}개 → 최종 canonical: {len(s4_canonical)}개") print() print("[리포트 작성]") write_report(items_per_source, s2_tokens, s2_atomics, s3_combined, s3_source_map, s4_canonical, s4_log) if __name__ == "__main__": main()