solar_eval/pipelines/rules.py

"""multi_step 파이프라인의 rule 스텝 구현체.

`multi_step.py` 가 518줄이라 (`.claude/rules/code-style.md` 의 400줄 권장선을
넘어) 룰 구현만 따로 뽑아냈다. vocabulary/post_process/correction_filter 는
동작을 바꾸지 않고 `MultiStepPipeline._apply_*` 메서드에서 그대로 옮겨온
것이다 — 순수 텍스트 변환 로직이라 `self` 상태가 필요 없다(`apply_vocabulary`
만 예외 — 데이터셋에서 로드한 어휘 캐시를 인자로 받고 갱신된 캐시를 되돌려준다).

`apply_judge_decisions`/`apply_change_rate_guard` 는 v24 계열 포팅 단계(§5-F)에서
추가됐다. 전자는 `EvalSample.artifacts` 를 읽어야 해서(judge 스텝의 구조화 결정,
self-consistency 의 corrections 목록) `sample` 을 인자로 받는다 — 다른 rule
함수들과 시그니처가 다른 유일한 이유다.
"""

from __future__ import annotations

import logging
import re
from pathlib import Path
from typing import TYPE_CHECKING, Any

from solar_eval.models.sample import EvalSample
from solar_eval.pipelines.errors import KnowledgeAssetError

if TYPE_CHECKING:
    from solar_eval.core.dataset_loader import DatasetLoader

logger = logging.getLogger(__name__)


def apply_vocabulary(
    text: str,
    config: dict[str, Any],
    vocabulary: list[dict[str, str]] | None,
    dataset_loader: DatasetLoader | None,
    config_dir: Path | None = None,
) -> tuple[str, list[dict[str, str]] | None]:
    """`vocabulary_sub`/`vocabulary_reapply` rule_type 구현.

    호출자(`VocabularyStep`)가 run 수명 캐시를 들고 있으므로, 최초 호출 시 로드한
    어휘 목록을 반환값으로 되돌려줘 캐시로 저장하게 한다.

    Args:
        config_dir: 레포의 config 정본 디렉토리(`03-evaluation`). 치환 사전은
            데이터가 아니라 **설정**이라(바뀌면 같은 문제를 다르게 푼다) 여기를
            먼저 찾는다. 데이터 루트 기준 레거시 선언도 계속 지원한다.

    Returns:
        (치환된 text, 갱신된 vocabulary 캐시) 튜플.

    Raises:
        KnowledgeAssetError: `vocab_dataset` 을 선언했는데 못 읽을 때.
    """
    if vocabulary is None:
        vocabulary = _load_vocabulary(config, dataset_loader, config_dir)

    for row in vocabulary:
        wrong = row.get("wrong", row.get("before", ""))
        correct = row.get("correct", row.get("after", ""))
        if wrong and correct:
            text = text.replace(wrong, correct)

    return text, vocabulary


def _load_vocabulary(
    config: dict[str, Any],
    dataset_loader: DatasetLoader | None,
    config_dir: Path | None,
) -> list[dict[str, str]]:
    """치환 사전을 로드한다 — config 정본 우선, 데이터 루트 폴백.

    선언이 없으면 빈 목록(정상적인 no-op)이고, 선언했는데 못 읽으면 예외다.
    """
    vocab_path = config.get("vocab_dataset", "")
    if not vocab_path:
        return []

    searched: list[str] = []

    # 1. config 정본 (레포의 03-evaluation/) — 치환 사전은 설정이다.
    if config_dir is not None:
        candidate = Path(config_dir) / vocab_path
        searched.append(str(candidate))
        if candidate.is_file() and dataset_loader is not None:
            return dataset_loader.load_csv(str(config_dir), vocab_path, source="local")

    # 2. 레거시 선언 (repo/source 로 데이터 루트나 HF 를 가리키는 경우)
    repo = config.get("repo", "")
    source = config.get("source", "huggingface")
    if dataset_loader is not None:
        searched.append(f"{source}:{repo}/{vocab_path}" if repo else f"{source}:{vocab_path}")
        try:
            return dataset_loader.load_csv(repo, vocab_path, source=source)
        except Exception as e:  # noqa: BLE001 — 원인을 KnowledgeAssetError 에 실어 다시 던진다
            logger.debug("vocabulary load failed (%s): %s", vocab_path, e)
    else:
        searched.append("(dataset_loader 없음)")

    raise KnowledgeAssetError(vocab_path, searched)


def apply_pronoun_postprocess(text: str, replacements: dict[str, str]) -> str:
    """대명사/직함 치환 + 한국어 조사 교정.

    단어를 치환할 때(예: "이재명 대표" → "이재명 대통령") 뒤따르는 조사도
    새 단어의 받침에 맞춰 바꿔야 한다.
    """
    # 모음으로 끝나는 조사 → 자음으로 끝나는 단어 뒤에 붙는 대응 조사
    correction_map = {
        "는": "은",
        "가": "이",
        "를": "을",
        "와": "과",
        "로": "으로",
        "여": "이여",
        "라": "이라",
        "랑": "이랑",
        "다": "이다",
        "였다": "이었다",
        "라면": "이라면",
        "라서": "이라서",
        "로부터": "으로부터",
    }
    all_particles = (
        list(correction_map.keys()) + list(correction_map.values()) + ["만", "도", "께서"]
    )
    particle_pattern = "|".join(re.escape(p) for p in sorted(all_particles, key=len, reverse=True))

    for old, new in replacements.items():
        regex = re.compile(f"({re.escape(old)})({particle_pattern})?")

        def replacer(m: re.Match, new_word: str = new) -> str:
            particle = m.group(2)
            if particle:
                return new_word + correction_map.get(particle, particle)
            return new_word

        text = regex.sub(replacer, text)
    return text


def apply_currency_compact(text: str) -> str:
    """통화 표현에서 숫자와 단위 사이 공백을 제거한다."""
    _unit = r"(원|명|개|대|위안|달러|유로|엔|톤|권|건|회|장|마리|그루|송이|필)"
    _tail = r"(?=[\s.,!?()\"\'가-힣]|$)"
    # "3억 5000만 원" → "3억5000만원"
    text = re.sub(
        r"(\d+(?:\.\d+)?)(만|억|조)\s*(\d+)\s*" + _unit + _tail,
        r"\1\2\3\4",
        text,
    )
    # "3억 원" → "3억원"
    text = re.sub(
        r"(\d+(?:\.\d+)?)(만|억|조)\s*" + _unit + _tail,
        r"\1\2\3",
        text,
    )
    return text


def apply_comma_removal(text: str) -> str:
    """숫자 사이의 천단위 콤마를 제거한다 (1,000 → 1000)."""
    return re.sub(r"(?<=\d),(?=\d)", "", text)


def apply_correction_filter(text: str, config: dict[str, Any], original: str = "") -> str:
    """Filter corrections by comparing with original text.

    Keeps only 'safe' corrections (spacing changes, small character diffs).
    Reverts larger changes that are likely unnecessary expression substitutions.

    Config options:
        max_char_diff: Maximum character differences to allow (default: 2)
        allow_spacing: Keep spacing-only changes (default: true)
    """
    from solar_eval.evaluators.scoring import compute_lcs

    if not original:
        return text

    max_cd = config.get("max_char_diff", 2)
    allow_spacing = config.get("allow_spacing", True)

    orig_tokens = original.split()
    pred_tokens = text.split()

    if orig_tokens == pred_tokens:
        return text

    lcs_pairs = compute_lcs(orig_tokens, pred_tokens)

    def is_safe(o_text: str, c_text: str) -> bool:
        if allow_spacing and o_text.replace(" ", "") == c_text.replace(" ", ""):
            return True
        if abs(len(o_text) - len(c_text)) <= 1:
            d = sum(1 for a, b in zip(o_text, c_text) if a != b)
            d += abs(len(o_text) - len(c_text))
            if d <= max_cd:
                return True
        return False

    result_tokens: list[str] = []
    prev_orig = -1
    prev_pred = -1

    for oi, pi in lcs_pairs:
        orig_gap = orig_tokens[prev_orig + 1 : oi]
        pred_gap = pred_tokens[prev_pred + 1 : pi]
        if orig_gap or pred_gap:
            o_text = " ".join(orig_gap)
            p_text = " ".join(pred_gap)
            if o_text == p_text or is_safe(o_text, p_text):
                result_tokens.extend(pred_gap)
            else:
                result_tokens.extend(orig_gap)
        result_tokens.append(orig_tokens[oi])
        prev_orig = oi
        prev_pred = pi

    orig_tail = orig_tokens[prev_orig + 1 :]
    pred_tail = pred_tokens[prev_pred + 1 :]
    if orig_tail or pred_tail:
        o_text = " ".join(orig_tail)
        p_text = " ".join(pred_tail)
        if o_text == p_text or is_safe(o_text, p_text):
            result_tokens.extend(pred_tail)
        else:
            result_tokens.extend(orig_tail)

    return " ".join(result_tokens)


def apply_judge_decisions(text: str, config: dict[str, Any], sample: EvalSample) -> str:
    """`tool_calling_judge` 의 verdict="revert" 판정을 텍스트에 되돌린다.

    `judge_step`(기본 "tool_calling_judge")의 판정 목록과 `corrections_step`
    (기본 "basic_correction")의 corrections 목록을 `sample.artifacts` 에서 읽는다
    — 이 두 artifact 는 각각 `pipelines/judge.py::run_tool_calling_judge` 와
    `MultiStepPipeline` 의 self-consistency 경로가 채운다. 어느 한쪽이라도 비어
    있으면(judge 스텝이 없었거나 corrections 가 하나도 없었던 경우) 조용히
    원문(text)을 그대로 돌려준다 — revert 할 대상 자체가 없는 정상 상태다.

    HF 구현(`huggingface/pipelines_v24.py::apply_judge_decisions_for_bulk`)을
    포팅. bulk 단위였던 원본과 달리 여기서는 문서 전체 text 에 적용한다
    (§5-F 설계 결정 — 이 엔진은 tool_calling_judge/basic_correction 을 이미
    문서 단위로 처리하므로 bulk 정렬 문제 자체가 없다).
    """
    judge_step = config.get("judge_step", "tool_calling_judge")
    corrections_step = config.get("corrections_step", "basic_correction")
    decisions = sample.artifacts.get("judge_decisions", {}).get(judge_step, [])
    corrections = sample.artifacts.get("corrections", {}).get(corrections_step, [])
    if not decisions or not corrections:
        return text

    revert_pairs: list[tuple[str, str]] = []
    for d in decisions:
        if not isinstance(d, dict) or d.get("verdict") != "revert":
            continue
        idx = d.get("correction_index")
        if not isinstance(idx, int) or idx < 0 or idx >= len(corrections):
            continue
        entry = corrections[idx]
        if not isinstance(entry, dict):
            continue
        corrected = str(entry.get("corrected", ""))
        original = str(entry.get("original", ""))
        if corrected and original and corrected != original:
            revert_pairs.append((corrected, original))

    # 긴 것부터 되돌려야 짧은 교정이 긴 교정의 부분 문자열일 때 잘못 되돌리지 않는다.
    revert_pairs.sort(key=lambda p: -len(p[0]))
    for corrected, original in revert_pairs:
        if corrected in text:
            text = text.replace(corrected, original, 1)
    return text


def apply_change_rate_guard(text: str, original: str, config: dict[str, Any]) -> str:
    """원문 대비 토큰 변경 비율이 `max_change_rate` 를 넘으면 원문으로 되돌리는 안전망.

    v17 인사이트 문서(`insights/260504_solar-pro2_essentials.md`)가 "change_rate_guard
    (rule, bulk-level safety net, max_change_rate: 0.45)" 라고만 설명하고, 실제
    구현은 HF 아카이브(`huggingface/pipelines.py`/`pipelines_v24.py`) 어디에도 없다
    (레포 전체 검색으로 확인). 그래서 이 함수는 그 설명만 보고 새로 작성한 것이다
    — v17 수치 재현을 약속하지 않는다는 전제로 포팅 대상에 포함됐다
    (`.agents/01-plans/harness/2026-08-09-chosun-proofread-evalsample-migration.md` §9-3).

    변경 비율은 `evaluators/scoring.compute_lcs` 로 원문·현재 텍스트의 토큰 단위
    LCS 를 구해 `1 - |LCS| / |원문 토큰|` 으로 정의한다 — 이 프로젝트의 다른 모든
    교열 diff 로직(`compute_proofread_metrics`, `apply_correction_filter`)과 같은
    척도를 쓴다.
    """
    from solar_eval.evaluators.scoring import compute_lcs

    max_rate = config.get("max_change_rate", 0.45)
    orig_tokens = original.split()
    if not orig_tokens:
        return text

    text_tokens = text.split()
    lcs_pairs = compute_lcs(orig_tokens, text_tokens)
    change_rate = 1 - (len(lcs_pairs) / len(orig_tokens))

    if change_rate > max_rate:
        logger.warning(
            "[change_rate_guard] change_rate=%.3f > max=%.3f, reverting to original",
            change_rate,
            max_rate,
        )
        return original
    return text


def apply_unit_unicode(text: str) -> str:
    """단위 텍스트를 유니코드 결합 문자로 변환한다."""
    replacements = {
        "mm²": "㎟",
        "cm²": "㎠",
        "m²": "㎡",
        "km²": "㎢",
        "mm": "㎜",
        "cm": "㎝",
        "km": "㎞",
        "mg": "㎎",
        "kg": "㎏",
        "ml": "㎖",
        "dl": "㎗",
        "cc": "㏄",
    }
    for k, v in replacements.items():
        text = text.replace(k, v)
    return text


def load_compound_whitelist(config: dict[str, Any], config_dir: Path | None) -> set[str]:
    """복합어 화이트리스트(탭 구분 ``표제어\\t빈도``)를 로드한다.

    선언이 없으면 빈 set(정상적인 no-op)이고, 선언했는데 못 읽으면 예외다 —
    `KnowledgeAssetError` 계약(`errors.py`)을 치환 사전과 동일하게 따른다.
    화이트리스트는 사람이 편집하는 지식 자산이라 config 정본(`03-evaluation/`)
    기준 상대경로만 지원한다.
    """
    wl_path = config.get("whitelist_path", "")
    if not wl_path:
        return set()

    searched: list[str] = []
    if config_dir is not None:
        candidate = Path(config_dir) / wl_path
        searched.append(str(candidate))
        if candidate.is_file():
            entries: set[str] = set()
            for line in candidate.read_text(encoding="utf-8").splitlines():
                form = line.split("\t")[0].strip()
                if form:
                    entries.add(form)
            return entries
    else:
        searched.append("(config_dir 없음)")

    raise KnowledgeAssetError(wl_path, searched)


def _has_whitelist_prefix(joined: str, split_at: int, whitelist: set[str]) -> bool:
    """결합형의 접두(분리점 너머 ~ 전체)가 화이트리스트 표제어인지.

    조사·어미가 붙은 어절("삼성"+"전자는")도 표제어("삼성전자")로 잡기 위해
    정확 일치가 아니라 접두 일치를 본다. 분리점(`split_at`) 안쪽에서 끝나는
    접두는 첫 어절만으로 성립하므로 후보에서 뺀다.
    """
    return any(joined[:k] in whitelist for k in range(split_at + 1, len(joined) + 1))


def apply_compound_whitelist_restore(text: str, original: str, whitelist: set[str]) -> str:
    """`compound_whitelist_restore` rule_type 구현 (v19/v32 계보의 현 엔진 포팅).

    파이프라인(FT·step1/2/3)이 갈라놓은 화이트리스트 복합어를 다시 붙인다.
    인접 어절 쌍의 결합이 (1) 화이트리스트 표제어를 접두로 갖고 — 분리점이
    표제어 안쪽 — (2) 원문에 그대로 존재할 때만 붙인다. 조건 (2) 때문에 이
    룰은 파이프라인이 만든 분리를 되돌릴 뿐, 새 교정을 만들지 않는다.

    2026-09-02 재생 실측(combo 체인 출력 위): paragraph +2.91 f1
    (redundant 408→342), ci-v0 +0.43, ci-v2/ci-v1/yul-ryul 무변화.
    """
    if not whitelist:
        return text

    restored_lines: list[str] = []
    for line in text.split("\n"):
        tokens = line.split()
        out: list[str] = []
        i = 0
        while i < len(tokens):
            if i + 1 < len(tokens):
                joined = tokens[i] + tokens[i + 1]
                if joined in original and _has_whitelist_prefix(joined, len(tokens[i]), whitelist):
                    out.append(joined)
                    i += 2
                    continue
            out.append(tokens[i])
            i += 1
        restored_lines.append(" ".join(out))
    return "\n".join(restored_lines)


#: 율/률 정규화가 "명사 + 조사" 문맥으로 인정하는 어절 꼬리(한글 잔여) 목록.
#: 이 목록 밖의 한글이 이어지면(예: 인명 '김율희가'의 '희가') 건드리지 않는다.
_YUL_RYUL_JOSA = frozenset(
    {
        "은",
        "는",
        "이",
        "가",
        "을",
        "를",
        "의",
        "에",
        "도",
        "만",
        "과",
        "와",
        "로",
        "으로",
        "보다",
        "부터",
        "까지",
        "조차",
        "마저",
        "밖에",
        "처럼",
        "대로",
        "만큼",
        "뿐",
        "나",
        "이나",
        "란",
        "이란",
        "라",
        "이라",
        "라고",
        "이라고",
        "라는",
        "이라는",
        "라며",
        "이라며",
        "며",
        "이며",
        "고",
        "이고",
        "다",
        "이다",
        "입니다",
        "에서",
        "에게",
        "에는",
        "에도",
        "에서는",
        "에서도",
        "로는",
        "으로는",
        "로도",
        "으로도",
        "로서",
        "으로서",
        "로써",
        "으로써",
        "와의",
        "과의",
        "과는",
        "와는",
        "은요",
        "요",
        "대비",
    }
)


def apply_yul_ryul_normalize(text: str, exceptions: tuple[str, ...] = ()) -> str:
    """`yul_ryul_normalize` rule_type 구현 — 율/률 받침 규칙의 결정론적 적용.

    한글 맞춤법 제11항: 앞 음절에 받침이 없거나 ㄴ 받침이면 '율'(비율·환율·
    전환율), 그 외 받침이면 '률'(법률·확률·출석률). 받침은 유니코드 연산으로
    판정되므로 LLM 없이 최종 텍스트에서 강제할 수 있다.

    인명 오탐 방지 2중 가드:
    1. 어절 꼬리 검사 — 율/률 뒤의 한글 잔여가 조사 목록(`_YUL_RYUL_JOSA`)에
       없으면 접미사 용법이 아닌 것으로 보고 건너뛴다 (재생 실측의 유일한 오탐
       '김율희가'→'김률희가' 가 여기서 걸러진다).
    2. `exceptions` — 어절에 포함되면 무조건 건너뛰는 문자열 목록 (율/률로
       끝나는 인명 '김석율' 류를 위한 훅. 기본 비어 있음).

    2026-09-02 재생 실측(v33/combo 출력 위, 조사 가드 포함): yul-ryul 놓침 2건
    전량 회수(+3.07), ci-v2 +0.59, ci-v1 +0.53, paragraph·ci-v0 변경 0건.
    """
    out = list(text)
    n = len(text)
    for i, ch in enumerate(text):
        if ch not in ("율", "률"):
            continue
        prev = text[i - 1] if i > 0 else ""
        if not ("가" <= prev <= "힣"):
            continue
        jong = (ord(prev) - 0xAC00) % 28
        want = "율" if jong in (0, 4) else "률"  # 받침 없음(0)/ㄴ(4) → 율
        if want == ch:
            continue

        j = i + 1
        while j < n and not text[j].isspace():
            j += 1
        residue = "".join(c for c in text[i + 1 : j] if "가" <= c <= "힣")
        if residue and residue not in _YUL_RYUL_JOSA:
            continue

        if exceptions:
            k = i
            while k > 0 and not text[k - 1].isspace():
                k -= 1
            token = text[k:j]
            if any(e in token for e in exceptions):
                continue

        out[i] = want
    return "".join(out)