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

"""ScriptRepository — 테넌트 대화 스크립트 로드 + 치환 (P7 대화엔진의 데이터 계층).

- rq_type("재협상"|"재견적")별 스크립트 + 와일드카드 스크립트 + step 매핑 + 변수 매핑 로드.
- 탐색 순서: tenants/<id>/resources/ → 없으면 tenants/_base/resources/ 폴백.
- 브랜드({company_name}/{service_name})는 TenantConfig.resources 에서 주입(클린룸: 특정사 브랜드 비포함).
- 협상 변수({input_price}, {target}, {offer_1pct} 등)는 format_script 에서 치환.

Chat_server 구조 참고, 동적 import 해킹/특정사 표현은 제거(CLEANROOM.md).
"""

import json
import os
from typing import Any, Dict, Optional

from common.logger import LOG
from tenancy.config import TenantConfig

_RQ_FILES = {"재협상": "scripts_renegotiation.json", "재견적": "scripts_requote.json"}


class ScriptRepository:
    def __init__(self, config: TenantConfig, tenants_dir: str):
        self._config = config
        self._tenants_dir = tenants_dir
        self._cache: Dict[str, Any] = {}

    # ---- 경로 해석 (_base 폴백) ---------------------------------------
    def _resource_path(self, filename: str) -> Optional[str]:
        scripts_dir = self._config.resources.scripts_dir
        candidates = [
            os.path.join(self._tenants_dir, self._config.tenant_id, scripts_dir, filename),
            os.path.join(self._tenants_dir, "_base", scripts_dir, filename),
        ]
        for p in candidates:
            if os.path.exists(p):
                return p
        return None

    def _load_json(self, filename: str) -> dict:
        if filename in self._cache:
            return self._cache[filename]
        path = self._resource_path(filename)
        if not path:
            LOG.w(f"[ScriptRepository] resource not found: {filename} (tenant={self._config.tenant_id})")
            self._cache[filename] = {}
            return {}
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        data.pop("_comment", None)
        self._cache[filename] = data
        return data

    # ---- 로드 ----------------------------------------------------------
    def load_scripts(self, rq_type: str = "재협상") -> dict:
        """rq_type 스크립트 + 와일드카드 병합(가격협상_와일드 진입용)."""
        filename = _RQ_FILES.get(rq_type, _RQ_FILES["재협상"])
        scripts = dict(self._load_json(filename))
        # 와일드카드는 재협상 흐름에 병합 (Chat_server 와 동일 동작).
        if rq_type == "재협상":
            for card_id, cfg in self._load_json("scripts_wildcard.json").items():
                scripts[card_id] = cfg
        return scripts

    def wildcard_scripts(self) -> dict:
        return self._load_json("scripts_wildcard.json")

    def card_scripts(self) -> dict:
        """가격협상 카드 스크립트(action_id → 멘트). scripts_cards.json (테넌트→_base 폴백)."""
        return self._load_json("scripts_cards.json")

    def card_script(self, action_id: int, variables: Optional[Dict[str, Any]] = None) -> Optional[str]:
        """선택된 카드(action_id)의 스크립트를 변수 치환해서 반환. 없으면 None(호출부가 기본 문구 유지)."""
        text = self.card_scripts().get(str(action_id))
        if not text:
            return None
        return self.format_script(text, variables)

    def client_step_mapping(self) -> dict:
        return self._load_json("client_step_mapping.json")

    def variable_mapping(self) -> dict:
        return self._load_json("variable_mapping.json")

    # ---- 치환 ----------------------------------------------------------
    def _brand_vars(self) -> Dict[str, str]:
        return {
            "company_name": self._config.resources.company_name,
            "service_name": self._config.resources.service_name,
        }

    def format_script(self, text: str, variables: Optional[Dict[str, Any]] = None) -> str:
        """{company_name}/{service_name} + 협상 변수 치환. 누락 변수는 원형 유지(KeyError 방지)."""
        if not text:
            return text
        ctx = self._brand_vars()
        if variables:
            ctx.update({k: v for k, v in variables.items() if v is not None})

        class _Safe(dict):
            def __missing__(self, key):
                return "{" + key + "}"

        try:
            return text.format_map(_Safe(ctx))
        except (ValueError, IndexError):
            return text  # 형식 토큰 충돌 시 원형

    def get_step(self, step: str, rq_type: str = "재협상", variables: Optional[Dict[str, Any]] = None) -> Optional[dict]:
        """step 정의를 반환하되 script 를 치환해서 돌려준다."""
        scripts = self.load_scripts(rq_type)
        node = scripts.get(step)
        if node is None:
            return None
        node = dict(node)
        node["script"] = self.format_script(node.get("script", ""), variables)
        return node