o2o-negosium-original/agent/negotiation/chat/service/script_naturalizer.py

"""ScriptNaturalizer — 협상 카드 멘트를 LLM 으로 상황에 맞게 자연화 (Phase 2 표현층).

원칙: "숫자와 결정은 결정론이, 말은 LLM 이" —
- 입력은 **치환 전 템플릿**({input_price} 등 placeholder 유지 상태). LLM 은 숫자를 절대 만들지 않는다.
- 상황(라운드·가격구간·수용률)은 **정성 라벨**로만 전달(수치 미노출 → 숫자 환각 원천 차단).
- 검증 실패/타임아웃/미설정 시 None → 호출부가 원본 템플릿 폴백(협상은 절대 안 멈춤).

검증(ScriptVerifier 철학의 플레인 텍스트판):
① {placeholder} 집합이 원본과 정확히 동일(누락·추가 금지)
② 원본에 없던 숫자 등장 금지(가격 환각 차단)
③ 비어있지 않고 길이 폭주 금지

게이트: tenant llm.enabled(기본 false) + 전역 LLM 자격증명(config.local.toml [OpenAIConfig]
또는 OPENAI_API_KEY env). 호출은 스레드로 넘겨 이벤트루프 비차단 + 타임아웃.
"""

import asyncio
import json
import re
from typing import Any, Callable, Dict, List, Optional

from common.logger import LOG
from negotiation.profiling.config import LlmCredentials

_PLACEHOLDER_RE = re.compile(r"\{(\w+)\}")
_DIGITS_RE = re.compile(r"\d+")
# 강조 마커(negodata Slate 편집 정본) — 고객사가 지정한 표시라 자연화가 보존해야 한다.
#   색 마커 {{강조|..}} {{안내|..}} 는 여는 토큰 개수로, 굵게/밑줄은 구분자 개수(짝수=쌍)로 센다.
_COLOR_OPEN_RE = re.compile(r"\{\{(강조|안내)\|")

# 카드 메타 코드 → 프롬프트 라벨 (init-data.sql CardTone/CardStrategyType 정의와 동일)
_TONE_LABEL = {1: "강경", 2: "정중", 3: "우호", 4: "중립", 5: "단호"}
_STRATEGY_LABEL = {1: "경쟁", 2: "수용", 3: "고수", 4: "협력", 5: "선점", 6: "종결"}

_SYSTEM = """너는 B2B 구매 협상 챗봇의 문장 작성기다. 주어진 협상 카드 멘트 '템플릿'을 협상 상황에 맞게 자연스럽게 다시 쓴다.

규칙 (하나라도 어기면 출력은 폐기된다):
- {변수명} 치환자는 철자 그대로 유지한다. 추가/삭제/변경 금지.
- 숫자를 직접 쓰지 마라. 가격·비율 등 모든 수치는 치환자로만 표현한다.
- 강조 마커(**굵게** __밑줄__ {{강조|...}} {{안내|...}})는 **개수와 종류를 그대로 유지**한다.
  고객사가 지정한 강조 표시이므로 삭제·추가·종류변경 금지. 감싼 문구는 자연스럽게 바꿔도 되지만
  강조된 구절 수만큼 같은 마커로 반드시 다시 감싼다(예: **굵게** 2개면 결과도 **…** 2쌍).
- 새로운 약속·할인 조건·법적 표현을 만들지 마라. 원 템플릿의 협상 의도(전술)는 유지한다.
- 한국어 존댓말, 2~5문장, 채팅 말풍선에 어울리게 간결히.
- 출력은 JSON 하나만: {"script": "다시 쓴 멘트"}"""


def _default_llm_call(messages: List[dict]) -> dict:
    """기본 LLM 호출(동기) — 전역 자격증명으로 chat_json. 테스트에서 monkeypatch 지점."""
    from negotiation.profiling.infra.llm_adapter import chat_json

    return chat_json(messages, temperature=0.5, max_tokens=600)


class ScriptNaturalizer:
    def __init__(self, llm_call: Optional[Callable[[List[dict]], dict]] = None,
                 timeout_seconds: float = 8.0):
        self._llm_call = llm_call or _default_llm_call
        self._timeout = timeout_seconds

    @staticmethod
    def available() -> bool:
        """전역 LLM 자격증명이 설정돼 있는가 (테넌트 enabled 게이트는 호출부 몫)."""
        return LlmCredentials.from_config().is_configured()

    async def naturalize(self, template: str, *, situation: Optional[Dict[str, Any]] = None,
                         tone: Optional[int] = None, strategy: Optional[int] = None) -> Optional[str]:
        """템플릿(치환 전)을 상황 맞춤 문장으로 재작성. 실패/검증불통과 시 None(호출부 폴백)."""
        if not template or not template.strip():
            return None
        ctx = dict(situation or {})
        if tone in _TONE_LABEL:
            ctx["톤"] = _TONE_LABEL[tone]
        if strategy in _STRATEGY_LABEL:
            ctx["전략"] = _STRATEGY_LABEL[strategy]
        messages = [
            {"role": "system", "content": _SYSTEM},
            {"role": "user", "content":
                "템플릿:\n" + template +
                "\n\n협상 상황:\n" + json.dumps(ctx, ensure_ascii=False) +
                '\n\n규칙대로 다시 써서 {"script": "..."} 로만 출력.'},
        ]
        try:
            result = await asyncio.wait_for(asyncio.to_thread(self._llm_call, messages), self._timeout)
        except Exception as ex:  # 타임아웃 포함 — 폴백
            LOG.w(f"[ScriptNaturalizer] LLM 호출 실패(폴백): {ex}")
            return None

        text = result.get("script") if isinstance(result, dict) else None
        if not isinstance(text, str) or not text.strip():
            return None
        return text if self._verify(template, text) else None

    # ---- 검증 ----------------------------------------------------------
    @staticmethod
    def _verify(template: str, rewritten: str) -> bool:
        orig = set(_PLACEHOLDER_RE.findall(template))
        new = set(_PLACEHOLDER_RE.findall(rewritten))
        if orig != new:
            LOG.w(f"[ScriptNaturalizer] 검증 실패: 치환자 불일치 (누락={orig - new}, 추가={new - orig})")
            return False
        # 원본에 없던 숫자 금지 — 가격/비율 환각 차단 (수치는 치환자로만).
        orig_digits = set(_DIGITS_RE.findall(template))
        new_digits = set(_DIGITS_RE.findall(rewritten)) - orig_digits
        if new_digits:
            LOG.w(f"[ScriptNaturalizer] 검증 실패: 새 숫자 등장 {new_digits}")
            return False
        if len(rewritten) > max(600, len(template) * 4):
            LOG.w("[ScriptNaturalizer] 검증 실패: 길이 폭주")
            return False
        # 강조 마커 보존 — 고객사가 지정한 볼드/밑줄/색을 LLM 이 떨어뜨리면 폐기(원본 폴백).
        #   굵게/밑줄: 구분자 총 개수가 같아야 짝(쌍)이 보존됨. 색: 여는 토큰 개수 동일.
        if (template.count("**") != rewritten.count("**")
                or template.count("__") != rewritten.count("__")
                or len(_COLOR_OPEN_RE.findall(template)) != len(_COLOR_OPEN_RE.findall(rewritten))):
            LOG.w("[ScriptNaturalizer] 검증 실패: 강조 마커 불일치(볼드/색 소실) → 원본 유지")
            return False
        return True


def build_situation(context: Dict[str, Any]) -> Dict[str, Any]:
    """세션 컨텍스트 → 정성 상황 라벨 (수치 미노출 — 숫자 환각 차단의 핵심).

    가격구간: 제시가 vs 앵커/목표 관계, 라운드: 협상 진행 단계, 인하 진행: 협상 기준가 대비.
    """
    out: Dict[str, Any] = {}
    rnd = context.get("round") or 0
    if rnd:
        out["라운드"] = "첫 제안" if rnd <= 1 else ("초반 조율" if rnd == 2 else "막바지 조율")
    price = context.get("input_price") or 0
    anchor = context.get("anchor_price") or 0
    target = context.get("target_price") or 0
    if price and anchor and target:
        if price <= anchor:
            out["가격구간"] = "목표 범위 도달(마무리 국면)"
        elif price <= target:
            out["가격구간"] = "목표 범위 근접(조율 국면)"
        else:
            out["가격구간"] = "목표 상회(추가 인하 필요)"
    base = context.get("item_price") or 0
    if base and price:
        rate = (base - price) / base
        out["인하 진행"] = "아직 미미" if rate < 0.01 else ("일부 진행" if rate < 0.05 else "상당히 진행")
    return out