shadow_brain_core/utils/emotion.py

"""
🔱 Imperial Emotion Utility — 단일 진실 저장소
감정 키워드 목록, 태그 추출, 태그 제거 함수를 여기서만 정의합니다.
chat_routes.py / web_server.py / discord_service.py 모두 이 모듈을 임포트하십시오.
"""
import re
import time
import logging

_emotion_logger = logging.getLogger(__name__)

EMOTION_KEYWORDS = [
    'happy', 'sad', 'angry', 'surprised', 'disgust', 'smirk', 'neutral',
    'peaceful', 'excited', 'serious', 'thinking', 'love', 'shy', 'fear',
    'fearful', 'joy', 'joyful', 'pain', 'tired', 'bored', 'playful',
    'confident', 'anxious', 'relief', 'devoted', 'loyal', 'proud', 'sympathetic',
]

# [EMOTION:xxx] / [EMOTION Board:xxx]
EMOTION_BOARD_RE = re.compile(r'\[EMOTION(?:\s+Board)?:([^\]]*)\]', re.IGNORECASE)

# 레거시 인라인 태그 [happy] 등
INLINE_EMOTION_RE = re.compile(
    r'\[(' + '|'.join(EMOTION_KEYWORDS) + r')\]',
    re.IGNORECASE,
)

# 프롬프트용 허용 감정 목록 문자열
ALLOWED_EMOTION_LIST = ', '.join(EMOTION_KEYWORDS)

# 🎭 가독성·감정 지침의 '하드코딩 폴백'은 제거됨(2026-06-29).
#   과거 브레인이 Aegis를 못 읽으면(토큰 누락 401 등) 조용히 이 하드코딩 값을 주입해,
#   '공통 지침 관리자' 편집이 안 먹히는 footgun이었다. 이제 지침 텍스트의 '유일한 진실'은
#   Aegis DB(공통 지침 관리자)이며, 조회 실패 시 빈 값(주입 안 함)으로 실패를 드러낸다.
#   (DB 시드 기본값은 Aegis system_views.DEFAULT_*가 담당 — UI 초기 콘텐츠.)


def _append_emotion_example(instruction: str, example: str | None) -> str:
    """감정 지침 본문 끝에 말투 예시 한 줄을 덧붙인다(예시가 비면 그대로).
    DB의 emotion_example(토글 ON)을 본문에 붙일 때 사용."""
    ex = (example or "").strip()
    if not ex:
        return instruction
    return instruction.rstrip() + f"\n  (예: \"{ex}\")"

# ─────────────────────────────────────────────────────────────────────────────
# 🔱 동적 지침 로더 (Aegis REST API → 메모리 캐시 → 하드코딩 폴백)
# 편집 즉시성은 '무효화 훅'이 담당한다: Aegis가 공통 지침 저장 직후
#   POST /api/shadow/guidelines/invalidate 로 캐시를 비워, 편집이 '다음 메시지 1번'에 바로
#   반영된다(TTL 만료를 기다릴 필요 없음 — '두 번 말 걸기' 불편 해소).
#   따라서 TTL은 '대화 중 중복 호출 절감 + 훅이 닿지 않는 경우(브레인 재기동/클라우드 NAT)의
#   안전망' 역할만 한다. DB(cache_ttl_seconds, 관리자 편집)가 우선이며 기본 60초.
#   완전 무캐시는 매 메시지마다 3종(가독성/감정/검색정책)을 Aegis로 GET해 비효율적이다.
#   Aegis 미기동 시에도 하드코딩 폴백으로 정상 작동한다.
# ─────────────────────────────────────────────────────────────────────────────
_GUIDELINES_CACHE: dict = {}
_GUIDELINES_TTL: float = 60.0  # 기본 60초(안전망). 즉시성은 무효화 훅이 담당. DB값이 우선.
# ⏱️ DB(공통 지침 관리자)에서 받은 TTL을 런타임에 보관. 첫 조회 전엔 기본값 사용.
_RUNTIME_TTL: float | None = None


def _eff_ttl() -> float:
    """캐시 유효 시간(초). DB에서 받은 값 우선, 없으면 기본값."""
    return _RUNTIME_TTL if _RUNTIME_TTL is not None else _GUIDELINES_TTL


def invalidate_guidelines_cache() -> None:
    """공통 지침 캐시를 즉시 비운다. Aegis가 지침 저장 직후 무효화 훅으로 호출 →
    다음 메시지 1번에 새 지침이 바로 반영된다(편집 후 '두 번 말 걸기' 불편 해소).
    TTL은 안전망으로만 남고, 평소엔 이 훅이 갱신을 담당한다."""
    _GUIDELINES_CACHE.clear()
    _emotion_logger.info("[Guidelines] 캐시 무효화 — 다음 조회에서 최신 지침을 다시 가져옵니다.")


_FETCH_FAIL_CACHE: dict = {}  # aegis_url → 실패 시각 (네거티브 캐시)


def _fetch_guidelines_from_aegis(aegis_url: str) -> dict | None:
    """Aegis REST API에서 공통 지침을 가져옵니다. 실패 시 None 반환.

    ⚠️ GET /api/system/guidelines 는 인증을 요구한다(_require_auth). 토큰 없이 호출하면
    401 → None → 하드코딩 폴백이 쓰여, '공통 지침 관리자' 편집이 브레인에 전혀 반영되지
    않는다. 반드시 서비스 토큰을 실어 보낸다(저장 경로 get_job_perf_flags와 동일 패턴)."""
    global _RUNTIME_TTL
    # 🛡️ 실패 네거티브 캐시 — 401/네트워크 실패가 반복되는 동안 모든 소비자
    # (retrieval_policy/emotion/readability)가 매 턴 클라우드 왕복을 반복하지 않게.
    _fail_at = _FETCH_FAIL_CACHE.get(aegis_url)
    if _fail_at and (time.time() - _fail_at) < _eff_ttl():
        return None
    try:
        import os
        import requests
        headers = {}
        tok = os.environ.get("IMPERIAL_API_TOKEN") or os.environ.get("SHADOWBRAIN_TOKEN")
        if tok:
            headers["X-Imperial-Token"] = tok
            headers["Authorization"] = f"Bearer {tok}"
        url = aegis_url.rstrip('/') + '/api/system/guidelines'
        resp = requests.get(url, headers=headers, timeout=3)
        if resp.status_code == 200:
            data = resp.json()
            # ⏱️ 관리자에서 설정한 캐시 TTL을 런타임에 반영(다음 캐시 판정부터 적용).
            if isinstance(data, dict) and data.get('cache_ttl_seconds') is not None:
                try:
                    _RUNTIME_TTL = float(data['cache_ttl_seconds'])
                except (TypeError, ValueError):
                    pass
            _FETCH_FAIL_CACHE.pop(aegis_url, None)
            return data  # {"readability": "...", "emotion": "...", "cache_ttl_seconds": N}
        _emotion_logger.debug(f"[Guidelines] Aegis HTTP {resp.status_code} — 네거티브 캐시")
    except Exception as e:
        _emotion_logger.debug(f"[Guidelines] Aegis 조회 실패 (폴백 사용): {e}")
    _FETCH_FAIL_CACHE[aegis_url] = time.time()
    return None


def get_dynamic_retrieval_policy(aegis_url: str | None = None) -> dict:
    """DB 공통 검색 정책. 조회 실패/누락 시 자동 검색을 중단한다."""
    if not aegis_url:
        return {}
    cache_key = f"{aegis_url}:retrieval_policy"
    cached = _GUIDELINES_CACHE.get(cache_key)
    if cached and (time.time() - cached['ts']) < _eff_ttl():
        return cached['value']
    data = _fetch_guidelines_from_aegis(aegis_url)
    value = data.get('retrieval_policy') if data else None
    if isinstance(value, dict):
        _GUIDELINES_CACHE[cache_key] = {'value': value, 'ts': time.time()}
        return value
    # 🛡️ 네거티브 캐시 — Aegis 실패/누락도 TTL 동안 기억. 안 하면 의도 감지 헬퍼
    # (_tool_policy/_intent_cfg 등)가 매 턴 클라우드 왕복(~840ms)을 반복해 TTFT를
    # 수 초씩 태운다(2026-07-10 실측).
    _GUIDELINES_CACHE[cache_key] = {'value': {}, 'ts': time.time()}
    return {}


def get_dynamic_emotion_instruction(aegis_url: str | None = None) -> str:
    """
    감정 연기 지침을 반환합니다. DB(Aegis '공통 지침 관리자')가 '유일한 진실'.
    - aegis_url 제공 + 조회 성공 시: DB 값(+ 토글 켜졌으면 말투 예시) 반환.
    - aegis_url 없거나 조회 실패 시: 빈 문자열(주입 안 함). 하드코딩 폴백 없음 —
      예전엔 폴백이 '조용히 다른 값'을 주입해 관리자 편집이 안 먹히는 footgun이었다.
    """
    if not aegis_url:
        return ""

    cache_key = f"{aegis_url}:emotion"
    cached = _GUIDELINES_CACHE.get(cache_key)
    if cached and (time.time() - cached['ts']) < _eff_ttl():
        return cached['value']

    data = _fetch_guidelines_from_aegis(aegis_url)
    if data and 'emotion' in data and data['emotion'] is not None:
        value = data['emotion']
        # 🎭 말투 예시는 별도 필드 + 토글(emotion_example_enabled). 켜졌을 때만 덧붙인다.
        if data.get('emotion_example_enabled', True):
            value = _append_emotion_example(value, data.get('emotion_example'))
        _GUIDELINES_CACHE[cache_key] = {'value': value, 'ts': time.time()}
        return value

    return ""


def get_dynamic_readability_instruction(aegis_url: str | None = None) -> str:
    """
    가독성 지침을 반환합니다. DB(Aegis '공통 지침 관리자')가 '유일한 진실'.
    - aegis_url 제공 + 조회 성공 시: DB 값 반환.
    - aegis_url 없거나 조회 실패 시: 빈 문자열(주입 안 함). 하드코딩 폴백 없음.
    """
    if not aegis_url:
        return ""

    cache_key = f"{aegis_url}:readability"
    cached = _GUIDELINES_CACHE.get(cache_key)
    if cached and (time.time() - cached['ts']) < _eff_ttl():
        return cached['value']

    data = _fetch_guidelines_from_aegis(aegis_url)
    if data and 'readability' in data and data['readability'] is not None:
        value = data['readability']
        _GUIDELINES_CACHE[cache_key] = {'value': value, 'ts': time.time()}
        return value

    return ""


def extract_emotion(text: str, default: str = 'peaceful') -> str:
    """AI 응답에서 감정값을 추출합니다. [EMOTION:xxx] 우선, 레거시 인라인 폴백."""
    if not text:
        return default
    matches = EMOTION_BOARD_RE.findall(text)
    if matches:
        return matches[-1].strip().lower()
    matches = INLINE_EMOTION_RE.findall(text)
    if matches:
        return matches[-1].lower()
    return default


def strip_emotion_tags(text: str) -> str:
    """표시용 텍스트에서 모든 감정 태그를 제거합니다. extract_emotion 후 호출하십시오."""
    t = EMOTION_BOARD_RE.sub('', text)
    t = INLINE_EMOTION_RE.sub('', t)
    return t.strip()