C.E.L_Slide_test2/tests/matching/text_pipeline.py

"""텍스트 추출 → 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()