o2o-triple-pick/backend/app/services/ai.py

"""AI 3모델 실연동 — GPT(OpenAI) · Claude(Anthropic) · Gemini(Google).

각 모델에 동일한 경기 컨텍스트를 주고 구조화된 예측 JSON 을 받는다.
키가 없으면 ProviderUnavailable 을 던지고(조용한 실패 0), 워커는 모델별로
독립 처리하여 가능한 것만 갱신한다.

반환 표준 dict:
  {outcome, scoreA, scoreB, confidencePct, reasonKo, reasonEn}
"""
from __future__ import annotations

import json
import logging
from dataclasses import dataclass

from ..config import settings

log = logging.getLogger("triplepick.ai")

OUTCOMES = {"TEAM_A_WIN", "DRAW", "TEAM_B_WIN"}


class ProviderUnavailable(RuntimeError):
    """API 키 미설정 등으로 해당 모델을 호출할 수 없음."""


@dataclass
class MatchContext:
    team_a: str  # 표시명 (예: Korea Republic)
    team_b: str
    venue: str
    kickoff: str  # ISO


def _prompt(ctx: MatchContext) -> str:
    return (
        f"You are a football match analyst. Predict the result of this "
        f"2026 FIFA World Cup match.\n"
        f"Team A: {ctx.team_a}\nTeam B: {ctx.team_b}\n"
        f"Venue: {ctx.venue}\nKickoff: {ctx.kickoff}\n"
        f"Important: World Cup group-stage matches are played at NEUTRAL venues. "
        f"Neither team has home advantage unless it is a host nation "
        f"(Mexico, USA, or Canada). Do NOT claim home advantage otherwise.\n\n"
        f"Predict the final regulation-time score. "
        f"Respond with a single JSON object and nothing else, with keys:\n"
        f'  "scoreA": integer 0-9 (Team A goals),\n'
        f'  "scoreB": integer 0-9 (Team B goals),\n'
        f'  "outcome": one of "TEAM_A_WIN" | "DRAW" | "TEAM_B_WIN" (must match the score),\n'
        f'  "confidencePct": integer 0-100 (confidence in the predicted winner; '
        f"for a draw, confidence in the draw),\n"
        f'  "reasonKo": a short one-line rationale in Korean (max ~30 chars),\n'
        f'  "reasonEn": a short one-line rationale in English (max ~60 chars).\n'
    )


# JSON Schema (구조화 출력용 — Anthropic/OpenAI 공통)
_SCHEMA = {
    "type": "object",
    "additionalProperties": False,
    "properties": {
        "scoreA": {"type": "integer"},
        "scoreB": {"type": "integer"},
        "outcome": {"type": "string", "enum": ["TEAM_A_WIN", "DRAW", "TEAM_B_WIN"]},
        "confidencePct": {"type": "integer"},
        "reasonKo": {"type": "string"},
        "reasonEn": {"type": "string"},
    },
    "required": ["scoreA", "scoreB", "outcome", "confidencePct", "reasonKo", "reasonEn"],
}


def _normalize(data: dict) -> dict:
    a = max(0, min(9, int(data["scoreA"])))
    b = max(0, min(9, int(data["scoreB"])))
    # outcome 은 스코어와 일관되도록 서버에서 재도출 (모델 불일치 방지)
    outcome = "TEAM_A_WIN" if a > b else "TEAM_B_WIN" if a < b else "DRAW"
    conf = max(0, min(100, int(data.get("confidencePct", 50))))
    return {
        "outcome": outcome,
        "scoreA": a,
        "scoreB": b,
        "confidencePct": conf,
        "reasonKo": str(data.get("reasonKo", "")).strip()[:120],
        "reasonEn": str(data.get("reasonEn", "")).strip()[:160],
    }


# ── GPT (OpenAI) ────────────────────────────────────────────
async def predict_gpt(ctx: MatchContext) -> dict:
    if not settings.openai_api_key:
        raise ProviderUnavailable("OPENAI_API_KEY 미설정")
    from openai import AsyncOpenAI

    client = AsyncOpenAI(api_key=settings.openai_api_key)
    resp = await client.chat.completions.create(
        model=settings.openai_model,
        messages=[
            {"role": "system", "content": "You output only valid JSON."},
            {"role": "user", "content": _prompt(ctx)},
        ],
        response_format={"type": "json_object"},
    )
    content = resp.choices[0].message.content or "{}"
    return _normalize(json.loads(content))


# ── Claude (Anthropic) ──────────────────────────────────────
async def predict_claude(ctx: MatchContext) -> dict:
    if not settings.anthropic_api_key:
        raise ProviderUnavailable("ANTHROPIC_API_KEY 미설정")
    from anthropic import AsyncAnthropic

    client = AsyncAnthropic(api_key=settings.anthropic_api_key)

    async def _create(**extra):  # noqa: ANN003
        return await client.messages.create(
            model=settings.anthropic_model,
            max_tokens=1024,
            messages=[{"role": "user", "content": _prompt(ctx)}],
            **extra,
        )

    # 1순위: 구조화 출력(output_config.format) + Opus 4.8 어댑티브 thinking.
    # SDK/모델 버전에 따라 미지원이면 평문 JSON 파싱으로 폴백.
    try:
        msg = await _create(
            thinking={"type": "adaptive"},
            output_config={"format": {"type": "json_schema", "schema": _SCHEMA}},
        )
    except TypeError:
        msg = await _create()
    text = "".join(b.text for b in msg.content if getattr(b, "type", "") == "text")
    return _normalize(json.loads(text))


# ── Gemini (Google) ─────────────────────────────────────────
async def predict_gemini(ctx: MatchContext) -> dict:
    if not settings.google_api_key:
        raise ProviderUnavailable("GOOGLE_API_KEY 미설정")
    from google import genai
    from google.genai import types

    client = genai.Client(api_key=settings.google_api_key)
    resp = await client.aio.models.generate_content(
        model=settings.google_model,
        contents=_prompt(ctx),
        config=types.GenerateContentConfig(response_mime_type="application/json"),
    )
    return _normalize(json.loads(resp.text or "{}"))


PROVIDERS = {
    "GPT": predict_gpt,
    "Claude": predict_claude,
    "Gemini": predict_gemini,
}