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

"""Phase 23 — Domain Terms Mining (검수용 지식 후보)
목적: corpus 전체에서 도메인 용어 자동 추출 + 관계 힌트 자동 분류.
비목적: synonyms.yaml 업데이트 (synonyms.yaml은 6개 동결 유지).

산출물:
- domain_terms.yaml: 용어 풀 + 분류 힌트 (safe_normalization / abbrev-fullname / semantic / context / hierarchy)
- DOMAIN_TERMS_REPORT.md(.html): 사람 검수용
"""
import sys
import re
import json
import collections
from pathlib import Path
import yaml

sys.path.insert(0, str(Path(__file__).parent))

from methods import _get_kiwi, _extract_content_tokens

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

# 노이즈 필터
_NOISE_PATTERNS = [
    re.compile(r"^\d+$"),                        # 순수 숫자
    re.compile(r"^\d+[년월일]"),                   # 날짜 단편
    re.compile(r"^\d{2}[\.\-/]\d{2}"),            # YY.MM
    re.compile(r"^[①②③④⑤⑥⑦⑧⑨⑩]+$"),
    re.compile(r"^[가-힣]$"),                    # 단일 한글
    re.compile(r"^[A-Za-z]$"),                   # 단일 영문
]
_STOP = {"것", "수", "등", "때", "중", "후", "전", "및", "이", "그", "저",
         "위", "아래", "바", "뿐", "또", "또한", "그리고", "하지만",
         "있", "없", "되", "하", "이", "같"}


def is_noise(token):
    if len(token) <= 1: return True
    if token in _STOP: return True
    for p in _NOISE_PATTERNS:
        if p.match(token): return True
    return False


def load_corpus():
    """BEPS + 32 Figma (analysis + texts + flat) + MDX 3개"""
    sources = {}
    # BEPS (1171281171은 32개에 없음 — 별도 취급)
    beps = BLOCKS_DIR / "1171281171" / "texts.md"
    if beps.exists():
        sources["BEPS/texts.md"] = beps.read_text(encoding="utf-8")
    # 32 Figma
    for d in sorted(BLOCKS_DIR.iterdir()):
        if not d.is_dir(): continue
        fid = d.name
        if not fid.startswith("1171") or fid == "1171281171": continue
        for fn in ["texts.md", "analysis.md", "flat.md"]:
            p = d / fn
            if p.exists():
                sources[f"Figma/{fid}/{fn}"] = p.read_text(encoding="utf-8")
    # MDX
    for mdx in ["01.mdx", "02.mdx", "03.mdx"]:
        p = MDX_DIR / mdx
        if p.exists():
            sources[f"MDX/{mdx}"] = p.read_text(encoding="utf-8")
    return sources


def mine_repeated_terms(corpus):
    """2회 이상 등장하는 내용어 추출"""
    kiwi = _get_kiwi()
    counter = collections.Counter()
    term_sources = collections.defaultdict(set)  # {term: {source_ids}}
    for src_id, text in corpus.items():
        tokens = _extract_content_tokens(text, kiwi)
        for t in tokens:
            if is_noise(t): continue
            counter[t] += 1
            term_sources[t].add(src_id)
    return counter, term_sources


def find_spacing_variants(terms, corpus_text):
    """'기술개발' 과 '기술 개발' 처럼 공백만 다른 쌍"""
    pairs = []
    term_set = set(terms)
    seen = set()
    for t in terms:
        if " " in t: continue
        if len(t) < 4: continue  # 너무 짧으면 노이즈
        # 글자 사이에 공백 삽입한 버전이 corpus에 등장하는지
        for split_pos in range(1, len(t)):
            spaced = t[:split_pos] + " " + t[split_pos:]
            if spaced in corpus_text and (t, spaced) not in seen:
                pairs.append((t, spaced))
                seen.add((t, spaced))
                seen.add((spaced, t))
                break
    return pairs


def find_ko_en_paren(corpus_text):
    """'한글 (English)' 또는 'English (한글)'"""
    out = []
    # 한글 (영문)
    p1 = re.compile(r"([가-힣][가-힣\s]{1,15}[가-힣])\s*\(([A-Za-z][A-Za-z\s]{1,30}[A-Za-z])\)")
    # 영문 (한글)
    p2 = re.compile(r"([A-Za-z][A-Za-z\s]{1,30}[A-Za-z])\s*\(([가-힣][가-힣\s]{1,15}[가-힣])\)")
    for m in p1.finditer(corpus_text):
        out.append((m.group(1).strip(), m.group(2).strip(), "ko→en"))
    for m in p2.finditer(corpus_text):
        out.append((m.group(2).strip(), m.group(1).strip(), "en→ko"))
    return out


def find_abbrev_expansion(corpus_text):
    """'Long Full Name (SHORT)' 패턴 — 약어 풀네임"""
    p = re.compile(r"([A-Z][A-Za-z\s]{5,50})\s*\(([A-Z]{2,6})\)")
    out = []
    for m in p.finditer(corpus_text):
        full = m.group(1).strip()
        abbr = m.group(2).strip()
        if 2 <= len(abbr) <= 6 and len(full.split()) >= 2:
            out.append((abbr, full))
    return out


def classify_relation(a, b):
    """두 용어 관계 힌트"""
    # 공백만 다름
    if a.replace(" ", "") == b.replace(" ", "") and a != b:
        return "safe_normalization"
    # 영문/한글 혼용
    is_en_a = bool(re.match(r"^[A-Za-z\s&\./]+$", a))
    is_en_b = bool(re.match(r"^[A-Za-z\s&\./]+$", b))
    if is_en_a != is_en_b:
        return "ko_en_pair"
    # 길이 차이 크면 abbrev-full
    if abs(len(a) - len(b)) > 4:
        return "abbrev_fullname_candidate"
    return "semantic_candidate"


def main():
    corpus = load_corpus()
    all_text = "\n".join(corpus.values())
    print(f"Corpus: {len(corpus)}개 소스 (BEPS 1 + Figma 31×3 + MDX 3)")
    print()

    # 1. 반복 용어 추출
    counter, term_sources = mine_repeated_terms(corpus)
    filtered = [(t, c) for t, c in counter.items() if c >= 2]
    filtered.sort(key=lambda x: -x[1])
    print(f"2회+ 등장 내용어: {len(filtered)}개")
    print()

    # 2. 공백 변형 쌍
    spacing = find_spacing_variants([t for t, _ in filtered], all_text)
    print(f"공백 변형 쌍 (safe_normalization 후보): {len(spacing)}개")
    for a, b in spacing[:15]:
        print(f"  {a}  ↔  {b}")
    print()

    # 3. 한영 병기
    ko_en = find_ko_en_paren(all_text)
    # 빈번한 것 top
    ko_en_counter = collections.Counter(
        (k, e) for k, e, _ in ko_en
    )
    print(f"한영 병기 패턴 (unique pairs): {len(ko_en_counter)}개")
    for (k, e), c in ko_en_counter.most_common(15):
        print(f"  '{k}'  ↔  '{e}'  ({c}회)")
    print()

    # 4. 약어 풀네임
    abbrev = find_abbrev_expansion(all_text)
    abbrev_counter = collections.Counter(abbrev)
    print(f"약어 풀네임 패턴: {len(abbrev_counter)}개")
    for (a, f), c in abbrev_counter.most_common(15):
        print(f"  '{a}'  ≡  '{f}'  ({c}회)")
    print()

    # 5. domain_terms.yaml 조립
    safe_norm = [{"canonical": a.replace(" ", ""), "variants": [a, b],
                  "evidence_count_a": counter.get(a, 0), "evidence_count_b": counter.get(b, 0)}
                 for a, b in spacing]
    # 중복 canonical 병합
    canonical_groups = collections.defaultdict(set)
    for item in safe_norm:
        c = item["canonical"]
        canonical_groups[c].update(item["variants"])
    safe_norm_clean = [{"canonical": c, "variants": sorted(vs),
                        "evidence": counter.get(c, 0)}
                        for c, vs in canonical_groups.items()]
    safe_norm_clean.sort(key=lambda x: -x["evidence"])

    domain = {
        "meta": {
            "phase": 23,
            "purpose": "검수용 도메인 용어 후보. synonyms.yaml은 별도 동결.",
            "corpus": f"BEPS 1 + Figma 31×3 + MDX 3 = {len(corpus)} sources",
            "total_unique_terms_2plus": len(filtered),
        },
        "top_terms_by_frequency": [
            {"term": t, "count": c, "sources": len(term_sources[t])}
            for t, c in filtered[:50]
        ],
        "classified": {
            "safe_normalization": safe_norm_clean,
            "ko_en_paren_pairs": [
                {"ko": k, "en": e, "count": c}
                for (k, e), c in ko_en_counter.most_common(40)
            ],
            "abbrev_fullname_patterns": [
                {"abbr": a, "full": f, "count": c}
                for (a, f), c in abbrev_counter.most_common(40)
            ],
            # 수동 분류 대기 — 참고용
            "needs_human_review": [
                "발주자 vs 발주처 (역할 vs 기관)",
                "시공자 vs 시공사 (역할 vs 회사)",
                "설계자 vs 설계사 (역할 vs 회사)",
                "Engineering S/W vs Engn. S/W (약어 — 동의어 가능성)",
                "Engineering S/W vs Solution S/W (상하위 or 분류 관계)",
                "Engineering S/W vs Application S/W (상하위 or 분류 관계)",
            ],
        },
    }
    out_yaml = HERE / "domain_terms.yaml"
    with open(out_yaml, "w", encoding="utf-8") as f:
        yaml.safe_dump(domain, f, allow_unicode=True, sort_keys=False, width=120)
    print(f"완료: {out_yaml}")


if __name__ == "__main__":
    main()