o2o-site-AEO/solution/backend/services/llm/perplexity.py

"""Perplexity 호출 — 이 프로젝트에서 Perplexity 로 나가는 **유일한 통로**.

여기가 책임지는 것: 주소·인증 헤더·타임아웃·응답 파싱.
여기가 책임지지 않는 것: 무엇을 물을지(services/prompts/channel_discovery.py),
답으로 받은 URL 을 믿을지(services/grounding/channels.py).

★ Perplexity 답변을 **사실로 쓰지 않는다.** 이 어댑터의 산출물은 "여기를 보라"는 URL 포인터일
  뿐이고, 실제 사실은 그 URL 을 크롤링해서 얻는다. 그래서 응답 원문(raw)을 통째로 박제해
  나중에 환각을 추적할 수 있게 한다.
"""
import json
from dataclasses import dataclass

import httpx

from config.server_configs import external_api_config

API_URL = "https://api.perplexity.ai/chat/completions"

DEFAULT_MODEL = "sonar"

# 첫 구조화 출력 요청은 스키마 준비에 10~30초가 걸린다 — 그 시간을 포함한 값이다.
DEFAULT_TIMEOUT = httpx.Timeout(90.0, connect=10.0)
DEFAULT_MAX_TOKENS = 2048


class PerplexityError(RuntimeError):
    """Perplexity 호출 실패. 호출측(collector)은 ErrorType.COLLECT_FETCH_FAILED 로 매핑한다."""


class PerplexityNotConfigured(PerplexityError):
    """PERPLEXITY_API_KEY 미설정 — 이 어댑터만 비활성. **서버 부팅을 막지 않는다.**"""


def is_configured() -> bool:
    """키가 설정돼 있는지. 어댑터 등록 여부를 판단할 때 쓴다."""
    return bool((external_api_config.perplexity_api_key or "").strip())


@dataclass
class Usage:
    """호출 1회의 실제 사용량·비용.

    ★ cost 는 우리가 계산한 값이 아니라 Perplexity 가 응답에 직접 실어주는 실제 청구액(USD)이다
    (`usage.cost.total_cost`) — 검색 컨텍스트 요금(request_cost)까지 포함된 진짜 값이라,
    Gemini·OpenAI 처럼 토큰 단가표로 역산하는 것보다 정확하다.
    ★ 실측(2026-09-16): `usage.cost` 는 평평한 숫자가 아니라
      `{"input_tokens_cost", "output_tokens_cost", "request_cost", "total_cost"}` 객체다 —
      문서 예시(평평한 숫자)와 다르다. num_search_queries 는 비용은 아니지만 검색 남용 감시용으로 같이 둔다."""

    input_tokens: int = 0
    output_tokens: int = 0
    num_search_queries: int = 0
    cost: float = 0.0


def read_usage(payload: dict) -> Usage:
    """응답의 usage 를 읽는다. 필드가 없거나 모양이 다르면 0 — 계측 실패가 본 기능을 막으면 안 된다."""
    u = payload.get("usage") or {}
    cost_field = u.get("cost")
    if isinstance(cost_field, dict):
        cost = cost_field.get("total_cost")
    else:
        cost = cost_field
    return Usage(
        input_tokens=int(u.get("prompt_tokens") or 0),
        output_tokens=int(u.get("completion_tokens") or 0),
        num_search_queries=int(u.get("num_search_queries") or 0),
        cost=float(cost) if isinstance(cost, (int, float)) else 0.0,
    )


async def call(body: dict, *, client: httpx.AsyncClient | None = None) -> dict:
    """★ LLM 이 실제로 불리는 지점. chat/completions 1회.

    Gemini 쪽(services/llm/gemini.py)과 달리 재시도하지 않는다 — 이 호출은 검색을 동반해
    한 번이 비싸고 느리다. 실패하면 수집 파이프라인이 등록된 링크로 그냥 진행한다
    (services/collect_service.py 의 discover_links 참조).

    응답 본문(dict)을 그대로 돌려준다. 해석은 부르는 쪽 몫이다.
    """
    api_key = (external_api_config.perplexity_api_key or "").strip()
    if not api_key:
        raise PerplexityNotConfigured("PERPLEXITY_API_KEY 미설정 — 채널 발견을 건너뛴다")

    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    owns_client = client is None
    http = client or httpx.AsyncClient(timeout=DEFAULT_TIMEOUT)
    try:
        resp = await http.post(API_URL, json=body, headers=headers)
    except httpx.TimeoutException as ex:
        raise PerplexityError(f"Perplexity 타임아웃: {ex}") from ex
    except httpx.HTTPError as ex:
        raise PerplexityError(f"Perplexity 호출 실패: {type(ex).__name__}: {ex}") from ex
    finally:
        if owns_client:
            await http.aclose()

    if resp.status_code != 200:
        raise PerplexityError(f"Perplexity 응답 오류: status={resp.status_code} body={resp.text[:200]}")

    try:
        payload = resp.json()
    except (json.JSONDecodeError, ValueError) as ex:
        raise PerplexityError(f"Perplexity 응답 JSON 파싱 실패: {ex}") from ex
    if not isinstance(payload, dict):
        raise PerplexityError("Perplexity 응답이 객체가 아니다")
    return payload