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 에서 치환.
- 표현 규칙: 스크립트는 의미(텍스트)만 소유한다. 문장 내 강조는 `**굵게**` 경량 마커만 허용
  — 렌더(굵기·색)는 프론트 소유(frontend emphasis.tsx). 선행 Chat_server 의 리치텍스트 JSON
  이중 관리(response_scripts.json)는 채택하지 않는다.

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

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

from common.database.db_session_manager import DB_SESSION_MNG
from common.enums import DBType, DBWRType, ErrorType
from common.logger import LOG
from negotiation.cards.adapters.card_script_db import CardScriptDbRepository
from negotiation.cards.ports.card_script_port import ICardScriptRepository
from tenancy.config import TenantConfig

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


class ScriptRepository:
    def __init__(self, config: TenantConfig, tenants_dir: str,
                 card_repo: Optional[ICardScriptRepository] = None):
        self._config = config
        self._tenants_dir = tenants_dir
        self._cache: Dict[str, Any] = {}
        # 카드 멘트 DB 소스(backoffice_db). file 모드면 미사용.
        self._card_repo: ICardScriptRepository = card_repo or CardScriptDbRepository()

    # ---- 경로 해석 (_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)

    async def resolve_card_script(self, action_id: int, card_id: Optional[str],
                                  variables: Optional[Dict[str, Any]] = None) -> Optional[str]:
        """카드 멘트 해석. cards.source_type == 'backoffice_db' 면 card.nego_cards.script(DB)를
        우선 조회하고, 없거나 file 모드면 scripts_cards.json(파일) 폴백. 변수 치환 후 반환.

        DB 멘트는 백오피스(negodata)가 편집한 정본이라 파일보다 우선한다. 마커(**굵게** 등)가
        섞여 있어도 agent 는 불투명 텍스트로 취급 — 표현 렌더는 프론트 소유.
        """
        if self._config.cards.source_type == _CARD_SOURCE_DB and card_id:
            db_text = await self._fetch_card_script_db(card_id)
            if db_text:
                return self.format_script(db_text, variables)
        return self.card_script(action_id, variables)  # 파일 폴백

    async def _fetch_card_script_db(self, card_id: str) -> Optional[str]:
        async def _q(s):
            _, text = await self._card_repo.get_script_by_number(s, card_id)
            return text

        try:
            return await DB_SESSION_MNG.execute_lambda(DBType.MAIN.value, DBWRType.DB_READ.value, _q)
        except Exception as ex:  # DB 불가 등 — 파일 폴백으로 협상은 계속되어야 함
            LOG.e_no_callstack(f"[ScriptRepository] 카드 멘트 DB 조회 실패 card_id={card_id}: {ex}")
            return None

    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