C.E.L_Slide_test2/src/kei_client.py

"""DA-12: 1단계 — Kei 실장 (꼭지 추출 + 분석).

1차: Kei API를 통해 Kei persona가 사고하여 꼭지를 추출한다.
fallback: Kei API 실패 시 Anthropic API 직접 호출.
"""
from __future__ import annotations

import json
import logging
import re
from typing import Any

import anthropic
import httpx

from src.config import settings

logger = logging.getLogger(__name__)

KEI_PROMPT = (
    "다음 콘텐츠로 슬라이드 1장을 만들려고 해. 어떤 스토리로 구성할지 설계해줘.\n\n"
    "## 1단계: 핵심 메시지 파악\n"
    "- 이 콘텐츠가 전달하려는 **핵심 메시지**를 한 줄로 파악해줘.\n"
    "- 슬라이드를 본 사람이 기억해야 할 단 하나의 문장.\n"
    "- core_message 필드에 기록.\n\n"
    "## 2단계: 정보 구조 파악\n"
    "- 본문 흐름(flow)과 참조 정보(reference)로 분리되는 구조인가?\n"
    "- 독립적으로 참조되는 정보(용어 정의, 부록)가 있는가?\n"
    "- info_structure 필드에 기술.\n\n"
    "## 3단계: 슬라이드 스토리라인 설계\n"
    "핵심 메시지를 전달하기 위한 **흐름**을 설계해줘. 각 위치에 **목적(purpose)**을 부여:\n"
    "- 문제제기: 왜 이 주제가 중요한가? 현재 무엇이 잘못되고 있는가?\n"
    "- 근거사례: 문제의 근거, 사례, 증거 (출처 포함)\n"
    "- 핵심전달: 그래서 사실은 이거다. 핵심 내용 전달.\n"
    "- 용어정의: 사용된 용어를 구체적으로 설명 (보조 참조, sidebar 배치)\n"
    "- 결론강조: 핵심 메시지 강조. 슬라이드 하단.\n"
    "- 구조시각화: 관계도, 프로세스 등 시각화가 필요한 경우\n\n"
    "## 원본 텍스트 보존 원칙\n"
    "- 원본의 논리 흐름과 정보를 빠뜨리지 마라\n"
    "- 원본 텍스트는 최대한 보존. 약간의 편집만.\n"
    "- 원본에 있는 내용을 임의로 제거하거나 다른 의미로 바꾸지 마라\n"
    "- 각 꼭지의 source_hint에 원본의 어떤 부분이 가는지 명시\n\n"
    "## 배치 규칙\n"
    "- 참조 정보(용어 정의 등)는 role: 'reference'로 표시 → 사이드바 배치\n"
    "- 본문 흐름은 role: 'flow' → 메인 영역 배치\n"
    "- 결론은 layer: 'conclusion' → 하단 배치\n"
    "- detail_target: true는 정말로 별도로 봐야 하는 상세 데이터에만 사용\n"
    "- 이미지/표가 있으면 images[], tables[]에 기록\n"
    "- 1페이지 적정 꼭지: 5개. 분량 적으면 1페이지로.\n\n"
    "## 출력 형식 (JSON만)\n"
    "```json\n"
    '{"title": "제목", '
    '"core_message": "이 슬라이드의 핵심 메시지 한 줄", '
    '"total_pages": 1, '
    '"info_structure": "정보 구조 설명", '
    '"topics": ['
    '{"id": 1, "title": "꼭지 제목", "summary": "요약", '
    '"purpose": "문제제기|근거사례|핵심전달|용어정의|결론강조|구조시각화", '
    '"source_hint": "원본에서 이 위치에 가져올 텍스트 범위 설명", '
    '"layer": "intro|core|supporting|conclusion", '
    '"role": "flow|reference", '
    '"emphasis": true, "direction": "vertical|horizontal|flexible", '
    '"content_type": "text|image|table|mixed", '
    '"detail_target": false, "page": 1}], '
    '"images": [{"topic_id": 1, "role": "key|supporting", "has_text": false, "description": "이미지 설명"}], '
    '"tables": [{"topic_id": 2, "rows": 5, "cols": 3, "fits_single_page": true, "description": "표 설명"}]}\n'
    "```\n\n"
    "## 콘텐츠:\n"
)


async def classify_content(content: str) -> dict[str, Any] | None:
    """1단계: Kei API를 통해 꼭지를 추출하고 분석한다.

    Kei API만 사용. Sonnet fallback 없음.
    Kei API 실패 시 None 반환 → pipeline.py에서 manual_classify() 안전망.
    """
    result = await _call_kei_api(content)
    if result:
        logger.info(
            f"[Kei API] 꼭지 추출 완료: {result.get('title', '')}, "
            f"{len(result.get('topics', []))}개 꼭지"
        )
        return result

    logger.warning("[Kei API] 꼭지 추출 실패. manual_classify로 안전망 적용.")
    return None


async def _call_kei_api(content: str) -> dict[str, Any] | None:
    """Kei API를 통해 꼭지 추출. SSE 스트리밍으로 실시간 수신."""
    kei_url = getattr(settings, "kei_api_url", "http://localhost:8000")

    try:
        async with httpx.AsyncClient(timeout=None) as client:
            async with client.stream(
                "POST",
                f"{kei_url}/api/message",
                json={
                    "message": KEI_PROMPT + content,
                    "session_id": "design-agent",
                    "mode_hint": "chat",
                },
                timeout=None,
            ) as response:
                if response.status_code != 200:
                    logger.warning(f"Kei API HTTP {response.status_code}")
                    return None

                full_text = await _stream_sse_tokens(response)

            if not full_text:
                logger.warning("Kei API 응답에서 텍스트 추출 실패")
                return None

            result = _parse_json(full_text)
            if result and "topics" in result:
                return result

            logger.warning(f"Kei API JSON 파싱 실패. 텍스트: {full_text[:200]}")
            return None

    except Exception as e:
        logger.warning(f"Kei API 호출 실패: {e}")
        return None


async def _stream_sse_tokens(response: httpx.Response) -> str:
    """SSE 스트리밍 응답에서 토큰을 실시간 수집한다.

    persona_agent의 SSE 이벤트:
    - token: 텍스트 토큰 수집
    - done: 완료, 중단
    - error: 에러, 즉시 중단
    - planning/planning_done/research_progress/warning: 스킵
    """
    tokens: list[str] = []
    event_type = ""

    async for line in response.aiter_lines():
        line = line.strip()
        if not line:
            event_type = ""
            continue
        if line.startswith("event:"):
            event_type = line[6:].strip()
        elif line.startswith("data:"):
            data = line[5:].strip()
            if event_type == "token" and data:
                try:
                    token = json.loads(data)
                    if isinstance(token, str):
                        tokens.append(token)
                except json.JSONDecodeError:
                    tokens.append(data)
            elif event_type == "done":
                break
            elif event_type == "error":
                logger.warning(f"Kei API SSE 에러: {data}")
                break

    return "".join(tokens)


def _extract_sse_text(raw: str) -> str:
    """SSE 응답에서 토큰 텍스트를 수집한다. CRLF/LF 모두 처리."""
    tokens = []
    # CRLF 또는 LF로 이벤트 분리
    events = re.split(r'\r?\n\r?\n', raw)

    for event in events:
        if not event.strip():
            continue

        event_type = ""
        event_data = ""

        for line in event.split('\n'):
            line = line.strip('\r')
            if line.startswith('event:'):
                event_type = line[6:].strip()
            elif line.startswith('data:'):
                event_data = line[5:].strip()

        if not event_data:
            continue

        if event_type == 'token':
            try:
                token = json.loads(event_data)
                if isinstance(token, str):
                    tokens.append(token)
            except json.JSONDecodeError:
                tokens.append(event_data)
        elif event_type == 'done':
            break

    return "".join(tokens)


async def _call_anthropic_direct(content: str) -> dict[str, Any] | None:
    """Anthropic API 직접 호출 (Kei API fallback)."""
    if not settings.anthropic_api_key:
        return None

    system_prompt = (
        "당신은 콘텐츠를 분석하여 슬라이드 구조를 설계하는 실장이다.\n\n"
        "## 핵심 원칙\n"
        "- 원본의 논리 흐름과 정보를 빠뜨리지 마라\n"
        "- 원본에 있는 내용을 임의로 제거하거나 다른 의미로 바꾸지 마라\n"
        "- 슬라이드에 맞게 정리하되, 원본이 말하려는 흐름은 유지\n\n"
        "## 꼭지 추출 규칙\n"
        "- 본문에서 2~5개의 핵심 꼭지를 추출한다\n"
        "- 참조 정보는 role: 'reference', 본문 흐름은 role: 'flow'로 표시\n"
        "- 1페이지 적정 꼭지 수: 5개\n"
        "- 초과 시 2페이지 분리\n"
        "- 이미지가 있으면 images[]에, 표가 있으면 tables[]에 판단 기록\n\n"
        "## 출력 형식 (JSON만. 설명 없이.)\n"
        '{"title": "제목", "total_pages": 1, '
        '"info_structure": "정보 구조 설명", '
        '"topics": ['
        '{"id": 1, "title": "꼭지 제목", "summary": "요약", '
        '"layer": "intro|core|supporting|conclusion", '
        '"role": "flow|reference", '
        '"emphasis": true, "direction": "vertical|horizontal|flexible", '
        '"content_type": "text|image|table|mixed", '
        '"detail_target": false, "page": 1}], '
        '"images": [], "tables": []}'
    )

    try:
        client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
        response = await client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            system=system_prompt,
            messages=[{"role": "user", "content": f"다음 콘텐츠의 꼭지를 추출해줘:\n\n{content}"}],
        )

        result_text = response.content[0].text
        result = _parse_json(result_text)

        if result and "topics" in result:
            return result

        return None

    except Exception as e:
        logger.warning(f"Anthropic 직접 호출 실패: {e}")
        return None


def _parse_json(text: str) -> dict[str, Any] | None:
    """텍스트에서 JSON을 추출한다.

    Kei API가 마크다운 리스트 접두사(- )를 붙여 응답하는 경우에도 처리.
    """
    # 전처리: 각 줄 앞의 마크다운 리스트 접두사(- ) 제거
    # Kei API가 JSON을 마크다운 리스트로 감싸서 응답하는 경우 대응
    lines = text.split("\n")
    cleaned_lines = []
    for line in lines:
        stripped = line.lstrip()
        if stripped.startswith("- "):
            cleaned_lines.append(stripped[2:])
        elif stripped.startswith("* "):
            cleaned_lines.append(stripped[2:])
        else:
            cleaned_lines.append(stripped)
    cleaned = "\n".join(cleaned_lines)

    # 원본 + 클린 버전 둘 다 시도
    for target in [text, cleaned]:
        patterns = [
            r"```json\s*(.*?)```",
            r"```\s*(.*?)```",
            r"(\{.*\})",
        ]
        for pattern in patterns:
            match = re.search(pattern, target, re.DOTALL)
            if match:
                try:
                    return json.loads(match.group(1).strip())
                except json.JSONDecodeError:
                    continue
    return None


def manual_classify(content: str) -> dict[str, Any]:
    """분류 실패 시 기본 구조 fallback."""
    return {
        "title": "슬라이드",
        "total_pages": 1,
        "info_structure": "",
        "topics": [
            {
                "id": 1,
                "title": "핵심 내용",
                "summary": content[:100],
                "layer": "core",
                "role": "flow",
                "emphasis": False,
                "direction": "flexible",
                "content_type": "text",
                "detail_target": False,
                "page": 1,
            },
        ],
        "images": [],
        "tables": [],
    }