159 lines
7.7 KiB
Python
159 lines
7.7 KiB
Python
"""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
|
|
import re
|
|
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"
|
|
# `{name}` 단일 토큰만 매칭(색 마커 `{{강조|...}}` 는 내부 `|` 로 인해 비매칭 → 보존).
|
|
_VAR_TOKEN_RE = re.compile(r"\{(\w+)\}")
|
|
|
|
|
|
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,
|
|
prefer_db: bool = False) -> Optional[str]:
|
|
"""카드 멘트 해석. cards.source_type == 'backoffice_db' 면 card.nego_cards.script(DB)를
|
|
우선 조회하고, 없거나 file 모드면 scripts_cards.json(파일) 폴백. 변수 치환 후 반환.
|
|
|
|
DB 멘트는 백오피스(negodata)가 편집한 정본이라 파일보다 우선한다. 마커(**굵게** 등)가
|
|
섞여 있어도 agent 는 불투명 텍스트로 취급 — 표현 렌더는 프론트 소유.
|
|
"""
|
|
if (prefer_db or 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} + 협상 변수 치환. 누락 변수는 원형 유지.
|
|
|
|
정규식으로 `{name}` 단일 토큰만 치환한다 — str.format_map 은 `{{`·`}}` 를 이스케이프로
|
|
해석해 색 마커 `{{강조|...}}` 를 `{강조|...}` 로 붕괴시키므로 쓰지 않는다.
|
|
색 마커는 내부에 `|`(비-\\w)가 있어 `\\{(\\w+)\\}` 에 매칭되지 않아 그대로 보존된다.
|
|
"""
|
|
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})
|
|
|
|
def _repl(m):
|
|
key = m.group(1)
|
|
return str(ctx[key]) if key in ctx else m.group(0) # 미등록 토큰은 원형
|
|
|
|
return _VAR_TOKEN_RE.sub(_repl, 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
|