refactor(agent-office): tarot 모듈 제거 (tarot-lab으로 cutover 완료)
- DELETE: app/tarot/ 디렉토리 (pipeline, prompt, schema 모듈) - DELETE: app/routers/tarot.py (FastAPI 라우터) - DELETE: 4개 tarot 테스트 파일 (test_tarot_*.py) - MODIFY: app/main.py — tarot 라우터 import + register 제거 - MODIFY: app/models.py — 5개 Tarot* 클래스 제거 - MODIFY: app/config.py — 4개 TAROT_* 환경변수 제거 - MODIFY: app/db.py — 6개 tarot_readings CRUD 함수 제거 KEEP: - tarot_readings CREATE TABLE 블록 (DB 호환성) - CREATE INDEX ... tarot_readings 인덱스 2개 - scripts/migrate_tarot_to_lab.py (cutover 마이그레이션) - tests/test_migrate_tarot.py (마이그레이션 테스트) 테스트: 88 pass (migrate_tarot tests 포함) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -38,9 +38,3 @@ LOTTO_DIGEST_HOUR = int(os.getenv("LOTTO_DIGEST_HOUR", "9"))
|
||||
LOTTO_DIGEST_MIN = int(os.getenv("LOTTO_DIGEST_MIN", "25"))
|
||||
LOTTO_THROTTLE_HOURS = int(os.getenv("LOTTO_THROTTLE_HOURS", "6"))
|
||||
LOTTO_URGENT_DAILY_MAX = int(os.getenv("LOTTO_URGENT_DAILY_MAX", "3"))
|
||||
|
||||
# Tarot Lab
|
||||
TAROT_MODEL = os.getenv("TAROT_MODEL", "claude-sonnet-4-6")
|
||||
TAROT_COST_INPUT_PER_M = float(os.getenv("TAROT_COST_INPUT_PER_M", "3.0"))
|
||||
TAROT_COST_OUTPUT_PER_M = float(os.getenv("TAROT_COST_OUTPUT_PER_M", "15.0"))
|
||||
TAROT_TIMEOUT_SEC = int(os.getenv("TAROT_TIMEOUT_SEC", "60"))
|
||||
|
||||
@@ -795,115 +795,3 @@ def get_tasks_by_agent_date_kind(agent_id: str, date_iso: str, task_type: str) -
|
||||
return [_task_to_dict(r) for r in rows]
|
||||
|
||||
|
||||
# --- tarot_readings CRUD ---
|
||||
|
||||
def save_tarot_reading(data: Dict[str, Any]) -> int:
|
||||
interp = data.get("interpretation_json") or {}
|
||||
summary = interp.get("summary", "") if isinstance(interp, dict) else ""
|
||||
with _conn() as conn:
|
||||
cur = conn.execute(
|
||||
"""INSERT INTO tarot_readings
|
||||
(spread_type, category, question, cards, interpretation_json,
|
||||
summary, model, tokens_in, tokens_out, cost_usd, confidence)
|
||||
VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
|
||||
(
|
||||
data["spread_type"],
|
||||
data.get("category"),
|
||||
data.get("question"),
|
||||
json.dumps(data.get("cards") or [], ensure_ascii=False),
|
||||
json.dumps(interp, ensure_ascii=False) if interp else None,
|
||||
summary,
|
||||
data.get("model"),
|
||||
data.get("tokens_in"),
|
||||
data.get("tokens_out"),
|
||||
data.get("cost_usd"),
|
||||
data.get("confidence"),
|
||||
),
|
||||
)
|
||||
return int(cur.lastrowid)
|
||||
|
||||
|
||||
def get_tarot_reading(reading_id: int) -> Optional[Dict[str, Any]]:
|
||||
with _conn() as conn:
|
||||
r = conn.execute("SELECT * FROM tarot_readings WHERE id=?", (reading_id,)).fetchone()
|
||||
return _tarot_row_to_dict(r) if r else None
|
||||
|
||||
|
||||
def list_tarot_readings(
|
||||
page: int = 1, size: int = 20,
|
||||
favorite: Optional[bool] = None,
|
||||
spread_type: Optional[str] = None,
|
||||
category: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
wheres, params = [], []
|
||||
if favorite is not None:
|
||||
wheres.append("favorite=?")
|
||||
params.append(1 if favorite else 0)
|
||||
if spread_type:
|
||||
wheres.append("spread_type=?")
|
||||
params.append(spread_type)
|
||||
if category:
|
||||
wheres.append("category=?")
|
||||
params.append(category)
|
||||
where_sql = ("WHERE " + " AND ".join(wheres)) if wheres else ""
|
||||
offset = (page - 1) * size
|
||||
with _conn() as conn:
|
||||
total = conn.execute(
|
||||
f"SELECT COUNT(*) c FROM tarot_readings {where_sql}", params
|
||||
).fetchone()["c"]
|
||||
rows = conn.execute(
|
||||
f"SELECT * FROM tarot_readings {where_sql} ORDER BY created_at DESC LIMIT ? OFFSET ?",
|
||||
params + [size, offset],
|
||||
).fetchall()
|
||||
return {
|
||||
"items": [_tarot_row_to_dict(r) for r in rows],
|
||||
"page": page, "size": size, "total": int(total),
|
||||
}
|
||||
|
||||
|
||||
def update_tarot_reading(reading_id: int, **kwargs) -> None:
|
||||
sets, vals = [], []
|
||||
if "favorite" in kwargs and kwargs["favorite"] is not None:
|
||||
sets.append("favorite=?")
|
||||
vals.append(1 if kwargs["favorite"] else 0)
|
||||
if "note" in kwargs and kwargs["note"] is not None:
|
||||
sets.append("note=?")
|
||||
vals.append(kwargs["note"])
|
||||
if not sets:
|
||||
return
|
||||
vals.append(reading_id)
|
||||
with _conn() as conn:
|
||||
conn.execute(f"UPDATE tarot_readings SET {','.join(sets)} WHERE id=?", vals)
|
||||
|
||||
|
||||
def delete_tarot_reading(reading_id: int) -> None:
|
||||
with _conn() as conn:
|
||||
conn.execute("DELETE FROM tarot_readings WHERE id=?", (reading_id,))
|
||||
|
||||
|
||||
def _tarot_row_to_dict(r) -> Dict[str, Any]:
|
||||
try:
|
||||
interp = json.loads(r["interpretation_json"]) if r["interpretation_json"] else None
|
||||
except (ValueError, TypeError):
|
||||
interp = None
|
||||
try:
|
||||
cards = json.loads(r["cards"]) if r["cards"] else []
|
||||
except (ValueError, TypeError):
|
||||
cards = []
|
||||
return {
|
||||
"id": r["id"],
|
||||
"created_at": r["created_at"],
|
||||
"spread_type": r["spread_type"],
|
||||
"category": r["category"],
|
||||
"question": r["question"],
|
||||
"cards": cards,
|
||||
"interpretation_json": interp,
|
||||
"summary": r["summary"],
|
||||
"model": r["model"],
|
||||
"tokens_in": r["tokens_in"],
|
||||
"tokens_out": r["tokens_out"],
|
||||
"cost_usd": r["cost_usd"],
|
||||
"confidence": r["confidence"],
|
||||
"favorite": int(r["favorite"]),
|
||||
"note": r["note"],
|
||||
}
|
||||
|
||||
@@ -12,11 +12,9 @@ from .agents import init_agents, get_agent, get_all_agent_states, AGENT_REGISTRY
|
||||
from .scheduler import init_scheduler
|
||||
from . import telegram_bot
|
||||
from .routers import notify as notify_router
|
||||
from .routers import tarot as tarot_router
|
||||
|
||||
app = FastAPI()
|
||||
app.include_router(notify_router.router)
|
||||
app.include_router(tarot_router.router)
|
||||
|
||||
_cors_origins = CORS_ALLOW_ORIGINS.split(",")
|
||||
app.add_middleware(
|
||||
|
||||
@@ -33,46 +33,3 @@ class ComposeCommand(BaseModel):
|
||||
style: Optional[str] = None
|
||||
model: Optional[str] = "V4"
|
||||
instrumental: Optional[bool] = False
|
||||
|
||||
|
||||
class TarotCardDraw(BaseModel):
|
||||
position: str
|
||||
card_id: str
|
||||
reversed: bool = False
|
||||
|
||||
|
||||
class TarotInterpretRequest(BaseModel):
|
||||
spread_type: Literal["one_card", "three_card"]
|
||||
category: Optional[str] = None
|
||||
question: Optional[str] = None
|
||||
cards: List[TarotCardDraw]
|
||||
cards_reference: str = Field(..., min_length=1)
|
||||
context_meta: dict = Field(default_factory=dict)
|
||||
|
||||
|
||||
class TarotInterpretResponse(BaseModel):
|
||||
interpretation_json: dict
|
||||
model: str
|
||||
tokens_in: int
|
||||
tokens_out: int
|
||||
cost_usd: float
|
||||
latency_ms: int
|
||||
reroll_count: int = 0
|
||||
|
||||
|
||||
class TarotSaveRequest(BaseModel):
|
||||
spread_type: Literal["one_card", "three_card"]
|
||||
category: Optional[str] = None
|
||||
question: Optional[str] = None
|
||||
cards: List[TarotCardDraw]
|
||||
interpretation_json: dict
|
||||
model: str
|
||||
tokens_in: int
|
||||
tokens_out: int
|
||||
cost_usd: float
|
||||
confidence: Optional[str] = None
|
||||
|
||||
|
||||
class TarotPatchRequest(BaseModel):
|
||||
favorite: Optional[bool] = None
|
||||
note: Optional[str] = None
|
||||
|
||||
@@ -1,70 +0,0 @@
|
||||
"""Tarot Lab 엔드포인트 — interpret + readings CRUD."""
|
||||
from fastapi import APIRouter, HTTPException
|
||||
|
||||
from ..models import (
|
||||
TarotInterpretRequest,
|
||||
TarotInterpretResponse,
|
||||
TarotSaveRequest,
|
||||
TarotPatchRequest,
|
||||
)
|
||||
from ..tarot import pipeline
|
||||
from .. import db as db_module
|
||||
|
||||
|
||||
router = APIRouter(prefix="/api/agent-office/tarot")
|
||||
|
||||
|
||||
@router.post("/interpret", response_model=TarotInterpretResponse)
|
||||
async def interpret_endpoint(req: TarotInterpretRequest):
|
||||
try:
|
||||
result = await pipeline.interpret(req)
|
||||
except pipeline.TarotError as e:
|
||||
raise HTTPException(status_code=500, detail=str(e)) from e
|
||||
return result
|
||||
|
||||
|
||||
@router.post("/readings")
|
||||
async def save_reading(req: TarotSaveRequest):
|
||||
rid = db_module.save_tarot_reading(req.model_dump())
|
||||
row = db_module.get_tarot_reading(rid)
|
||||
return {"id": rid, "created_at": row["created_at"]}
|
||||
|
||||
|
||||
@router.get("/readings")
|
||||
async def list_readings(
|
||||
page: int = 1,
|
||||
size: int = 20,
|
||||
favorite: bool | None = None,
|
||||
spread_type: str | None = None,
|
||||
category: str | None = None,
|
||||
):
|
||||
return db_module.list_tarot_readings(
|
||||
page=page, size=size,
|
||||
favorite=favorite, spread_type=spread_type, category=category,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/readings/{reading_id}")
|
||||
async def get_reading(reading_id: int):
|
||||
row = db_module.get_tarot_reading(reading_id)
|
||||
if not row:
|
||||
raise HTTPException(status_code=404, detail="reading not found")
|
||||
return row
|
||||
|
||||
|
||||
@router.patch("/readings/{reading_id}")
|
||||
async def patch_reading(reading_id: int, req: TarotPatchRequest):
|
||||
row = db_module.get_tarot_reading(reading_id)
|
||||
if not row:
|
||||
raise HTTPException(status_code=404, detail="reading not found")
|
||||
db_module.update_tarot_reading(reading_id, **req.model_dump(exclude_none=True))
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@router.delete("/readings/{reading_id}")
|
||||
async def delete_reading(reading_id: int):
|
||||
row = db_module.get_tarot_reading(reading_id)
|
||||
if not row:
|
||||
raise HTTPException(status_code=404, detail="reading not found")
|
||||
db_module.delete_tarot_reading(reading_id)
|
||||
return {"ok": True}
|
||||
@@ -1 +0,0 @@
|
||||
"""Tarot Lab — Claude Sonnet 기반 evidence·interactions 해석 파이프라인."""
|
||||
@@ -1,139 +0,0 @@
|
||||
"""Tarot 파이프라인 — Claude Sonnet 호출 + 파싱 폴백 + reroll 1회."""
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from typing import Any, Dict
|
||||
|
||||
import httpx
|
||||
|
||||
from ..config import (
|
||||
ANTHROPIC_API_KEY,
|
||||
TAROT_MODEL,
|
||||
TAROT_COST_INPUT_PER_M,
|
||||
TAROT_COST_OUTPUT_PER_M,
|
||||
TAROT_TIMEOUT_SEC,
|
||||
)
|
||||
|
||||
|
||||
logger = logging.getLogger("agent-office.tarot")
|
||||
from ..models import TarotInterpretRequest
|
||||
from .prompt import SYSTEM_PROMPT, build_user_message
|
||||
from .schema import validate_interpretation
|
||||
|
||||
|
||||
API_URL = "https://api.anthropic.com/v1/messages"
|
||||
|
||||
|
||||
class TarotError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def calc_cost(tokens_in: int, tokens_out: int) -> float:
|
||||
return (
|
||||
tokens_in / 1_000_000 * TAROT_COST_INPUT_PER_M
|
||||
+ tokens_out / 1_000_000 * TAROT_COST_OUTPUT_PER_M
|
||||
)
|
||||
|
||||
|
||||
def _strip_codeblock(text: str) -> str:
|
||||
t = text.strip()
|
||||
if t.startswith("```"):
|
||||
t = t.strip("`")
|
||||
if t.startswith("json"):
|
||||
t = t[4:]
|
||||
t = t.strip()
|
||||
return t
|
||||
|
||||
|
||||
def _extract_json(raw: str) -> dict:
|
||||
cleaned = _strip_codeblock(raw)
|
||||
try:
|
||||
return json.loads(cleaned)
|
||||
except json.JSONDecodeError:
|
||||
start, end = cleaned.find("{"), cleaned.rfind("}")
|
||||
if start >= 0 and end > start:
|
||||
try:
|
||||
return json.loads(cleaned[start : end + 1])
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
raise
|
||||
|
||||
|
||||
async def _call_claude(user_text: str, feedback: str = "") -> tuple[dict, dict, str]:
|
||||
if not ANTHROPIC_API_KEY:
|
||||
raise TarotError("ANTHROPIC_API_KEY missing")
|
||||
if feedback:
|
||||
user_text = f"이전 응답이 다음 이유로 거절됨: {feedback}\n올바른 스키마(시스템 지침)로 다시 응답.\n\n{user_text}"
|
||||
payload = {
|
||||
"model": TAROT_MODEL,
|
||||
"max_tokens": 1400, # 응답 시간 단축 — 3-card spread evidence·interactions 포함 충분
|
||||
"system": [{"type": "text", "text": SYSTEM_PROMPT,
|
||||
"cache_control": {"type": "ephemeral"}}],
|
||||
"messages": [{"role": "user", "content": [{"type": "text", "text": user_text}]}],
|
||||
}
|
||||
headers = {
|
||||
"x-api-key": ANTHROPIC_API_KEY,
|
||||
"anthropic-version": "2023-06-01",
|
||||
"anthropic-beta": "prompt-caching-2024-07-31",
|
||||
"content-type": "application/json",
|
||||
}
|
||||
started = time.monotonic()
|
||||
async with httpx.AsyncClient(timeout=TAROT_TIMEOUT_SEC) as client:
|
||||
r = await client.post(API_URL, headers=headers, json=payload)
|
||||
r.raise_for_status()
|
||||
resp = r.json()
|
||||
latency_ms = int((time.monotonic() - started) * 1000)
|
||||
raw_text = "".join(
|
||||
b.get("text", "") for b in resp.get("content", []) if b.get("type") == "text"
|
||||
)
|
||||
usage = resp.get("usage", {}) or {}
|
||||
tokens_in = int(usage.get("input_tokens", 0) or 0)
|
||||
tokens_out = int(usage.get("output_tokens", 0) or 0)
|
||||
logger.info("tarot claude call: latency=%dms, in=%d, out=%d", latency_ms, tokens_in, tokens_out)
|
||||
parsed = _extract_json(raw_text)
|
||||
meta = {
|
||||
"tokens_in": tokens_in,
|
||||
"tokens_out": tokens_out,
|
||||
"latency_ms": latency_ms,
|
||||
}
|
||||
return parsed, meta, raw_text
|
||||
|
||||
|
||||
async def interpret(req: TarotInterpretRequest) -> Dict[str, Any]:
|
||||
user_text = build_user_message(
|
||||
question=req.question or "",
|
||||
category=req.category or "",
|
||||
spread_type=req.spread_type,
|
||||
cards_reference=req.cards_reference,
|
||||
context_meta=req.context_meta or {},
|
||||
spread_count=len(req.cards),
|
||||
)
|
||||
|
||||
total_in, total_out, total_latency = 0, 0, 0
|
||||
last_error = ""
|
||||
for attempt in range(2):
|
||||
try:
|
||||
parsed, meta, _raw = await _call_claude(user_text, feedback=last_error)
|
||||
except httpx.HTTPError as e:
|
||||
raise TarotError(f"Claude HTTP error: {e}") from e
|
||||
except json.JSONDecodeError as e:
|
||||
last_error = f"JSON 파싱 실패: {e}"
|
||||
continue
|
||||
total_in += meta["tokens_in"]
|
||||
total_out += meta["tokens_out"]
|
||||
total_latency += meta["latency_ms"]
|
||||
|
||||
ok, err = validate_interpretation(parsed, req.spread_type)
|
||||
if ok:
|
||||
return {
|
||||
"interpretation_json": parsed,
|
||||
"model": TAROT_MODEL,
|
||||
"tokens_in": total_in,
|
||||
"tokens_out": total_out,
|
||||
"cost_usd": calc_cost(total_in, total_out),
|
||||
"latency_ms": total_latency,
|
||||
"reroll_count": attempt,
|
||||
}
|
||||
last_error = err
|
||||
|
||||
raise TarotError(f"검증 실패 (reroll 2회): {last_error}")
|
||||
@@ -1,108 +0,0 @@
|
||||
"""Tarot 프롬프트 — SYSTEM + build_user_message."""
|
||||
|
||||
SYSTEM_PROMPT = """당신은 라이더-웨이트(RWS) 타로 덱의 전통 상징체계에 정통한 타로 리더입니다.
|
||||
사용자의 질문, 카테고리, 뽑힌 카드 각각의 정·역방향과 위치를 받아 근거 기반으로 해석합니다.
|
||||
|
||||
# 해석 원칙
|
||||
1. 데이터 우선: "참고 카드 정보" 블록의 키워드·기본의미·상징만을 1차 근거로 사용.
|
||||
외부 변형 의미·다른 덱 해석은 사용하지 않음.
|
||||
2. 위치 의미 결합: 카드의 의미와 위치(과거/현재/미래 또는 오늘)를 명시적으로 결합해서 해석. evidence에 근거 기록.
|
||||
3. 카드 간 상호작용 분석 (3장 스프레드):
|
||||
- 시너지: 같은 슈트, 같은 원소, 메이저 비율, 정·역 흐름
|
||||
- 충돌·전환: 슈트 충돌(컵-소드, 완드-펜타클), 정→역 전환, 메이저↔마이너 전환
|
||||
4. 자기 성찰 톤: 운명론 단정 금지. "…할 가능성이 있어 보입니다" 같은 표현.
|
||||
5. 카테고리 컨텍스트: 동일 카드라도 카테고리에 따라 강조점이 달라야 함.
|
||||
6. 질문 직접 응답: 사용자 질문을 evidence·advice에서 인용·반영.
|
||||
|
||||
# 응답 형식 (strict JSON only — 코드블록 없이 raw JSON)
|
||||
{
|
||||
"summary": "전체 흐름 한 단락 (3~4문장)",
|
||||
"cards": [
|
||||
{
|
||||
"position": "<위치 라벨>",
|
||||
"card": "<card_id>",
|
||||
"reversed": <bool>,
|
||||
"interpretation": "3~4문장",
|
||||
"evidence": {
|
||||
"card_meaning_used": "참고 카드 정보에서 인용한 키워드·상징",
|
||||
"position_logic": "왜 이 위치에 이렇게 적용되는지 (1~2문장)",
|
||||
"category_lens": "카테고리 관점에서 부각되는 면 (1문장)"
|
||||
},
|
||||
"advice": "1문장"
|
||||
}
|
||||
],
|
||||
"interactions": [
|
||||
{ "type": "synergy"|"conflict"|"transition",
|
||||
"between": ["<card_id>", "<card_id>"],
|
||||
"explanation": "1~2문장" }
|
||||
],
|
||||
"advice": "2문장. interactions를 1개 이상 참조할 것.",
|
||||
"warning": "역방향·충돌 경계 (없으면 null)",
|
||||
"confidence": "high"|"medium"|"low"
|
||||
}
|
||||
|
||||
# confidence 판정 기준
|
||||
- high: 3장 모두 한 방향 서사 또는 명확한 전환
|
||||
- medium: 2장 일관, 1장 별도 신호
|
||||
- low: 카드 간 의미 충돌이 커서 명확한 흐름 잡기 어려움
|
||||
|
||||
# 금지사항
|
||||
- 참고 카드 정보에 없는 상징 도입 금지
|
||||
- 역방향 카드를 정방향처럼 다루지 말 것
|
||||
- "신비롭게 들리는" 문구로 채우지 말 것 — evidence에 인용·근거 명시
|
||||
- JSON 외 텍스트 금지
|
||||
"""
|
||||
|
||||
|
||||
SPREAD_NAMES = {
|
||||
"one_card": "오늘의 카드",
|
||||
"three_card": "3장 스프레드 (과거·현재·미래)",
|
||||
}
|
||||
|
||||
|
||||
def build_user_message(
|
||||
question: str,
|
||||
category: str,
|
||||
spread_type: str,
|
||||
cards_reference: str,
|
||||
context_meta: dict,
|
||||
spread_count: int,
|
||||
) -> str:
|
||||
q = question or "(질문 없음)"
|
||||
cat = category or "일반"
|
||||
spread_name = SPREAD_NAMES.get(spread_type, spread_type)
|
||||
|
||||
meta_lines = []
|
||||
if context_meta:
|
||||
if "major_minor_ratio" in context_meta:
|
||||
meta_lines.append(f"- 메이저:마이너 비율: {context_meta['major_minor_ratio']}")
|
||||
if "element_distribution" in context_meta:
|
||||
ed = context_meta["element_distribution"]
|
||||
meta_lines.append(
|
||||
f"- 원소 분포: 공기 {ed.get('air',0)}, 물 {ed.get('water',0)}, 불 {ed.get('fire',0)}, 흙 {ed.get('earth',0)}"
|
||||
)
|
||||
if "orientation_flow" in context_meta:
|
||||
meta_lines.append(f"- 정역 흐름: {context_meta['orientation_flow']}")
|
||||
meta_block = "\n".join(meta_lines) if meta_lines else "(추가 컨텍스트 없음)"
|
||||
|
||||
return f"""# 질문
|
||||
{q}
|
||||
|
||||
# 카테고리
|
||||
{cat}
|
||||
|
||||
# 스프레드
|
||||
{spread_name} ({spread_count}장)
|
||||
|
||||
# 뽑힌 카드와 참고 카드 정보
|
||||
{cards_reference}
|
||||
|
||||
## 추가 컨텍스트
|
||||
{meta_block}
|
||||
|
||||
# 작업
|
||||
위 정보만을 근거로 사용해, 시스템 지침의 JSON 형식으로 응답하세요.
|
||||
- 각 카드의 evidence.card_meaning_used에는 위 "참고 카드 정보"에서 발췌한 키워드·의미를 그대로 인용.
|
||||
- interactions는 3장 간 슈트·원소·정역방향 패턴을 분석해 최소 1개 이상 도출 (1장 스프레드면 빈 배열 허용).
|
||||
- confidence는 카드 흐름의 일관성에 따라 정직하게 판정.
|
||||
"""
|
||||
@@ -1,36 +0,0 @@
|
||||
"""Tarot 응답 스키마 검증 — 누락·빈 필드 reroll 트리거."""
|
||||
|
||||
VALID_CONFIDENCE = {"high", "medium", "low"}
|
||||
|
||||
|
||||
def validate_interpretation(parsed: dict, spread_type: str) -> tuple[bool, str]:
|
||||
if not isinstance(parsed, dict):
|
||||
return False, "응답이 dict가 아님"
|
||||
for k in ("summary", "cards", "interactions", "advice", "confidence"):
|
||||
if k not in parsed:
|
||||
return False, f"필수 필드 누락: {k}"
|
||||
if parsed.get("confidence") not in VALID_CONFIDENCE:
|
||||
return False, f"confidence 값 비정상: {parsed.get('confidence')}"
|
||||
cards = parsed.get("cards")
|
||||
if not isinstance(cards, list) or not cards:
|
||||
return False, "cards가 빈 배열"
|
||||
for i, c in enumerate(cards):
|
||||
if not isinstance(c, dict):
|
||||
return False, f"cards[{i}] dict 아님"
|
||||
for k in ("position", "card", "reversed", "interpretation", "advice", "evidence"):
|
||||
if k not in c:
|
||||
return False, f"cards[{i}].{k} 누락"
|
||||
ev = c["evidence"]
|
||||
if not isinstance(ev, dict):
|
||||
return False, f"cards[{i}].evidence dict 아님"
|
||||
for k in ("card_meaning_used", "position_logic", "category_lens"):
|
||||
if k not in ev:
|
||||
return False, f"cards[{i}].evidence.{k} 누락"
|
||||
if not isinstance(ev[k], str) or not ev[k].strip():
|
||||
return False, f"cards[{i}].evidence.{k} 빈 문자열"
|
||||
interactions = parsed.get("interactions")
|
||||
if not isinstance(interactions, list):
|
||||
return False, "interactions가 list 아님"
|
||||
if spread_type == "three_card" and len(interactions) == 0:
|
||||
return False, "three_card는 interactions 1개 이상 필요"
|
||||
return True, ""
|
||||
@@ -1,70 +0,0 @@
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
from app import db as db_module
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def fresh_db(monkeypatch, tmp_path):
|
||||
db_file = tmp_path / "test_tarot.db"
|
||||
monkeypatch.setattr(db_module, "DB_PATH", str(db_file))
|
||||
db_module.init_db()
|
||||
yield
|
||||
if db_file.exists():
|
||||
db_file.unlink()
|
||||
|
||||
|
||||
def test_save_and_get_tarot_reading():
|
||||
rid = db_module.save_tarot_reading({
|
||||
"spread_type": "three_card",
|
||||
"category": "연애",
|
||||
"question": "Q",
|
||||
"cards": [{"position": "과거", "card_id": "the-fool", "reversed": False}],
|
||||
"interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "medium"},
|
||||
"model": "claude-sonnet-4-6",
|
||||
"tokens_in": 100, "tokens_out": 200, "cost_usd": 0.005,
|
||||
"confidence": "medium",
|
||||
})
|
||||
assert rid > 0
|
||||
row = db_module.get_tarot_reading(rid)
|
||||
assert row["id"] == rid
|
||||
assert row["category"] == "연애"
|
||||
assert row["interpretation_json"]["summary"] == "S"
|
||||
assert row["favorite"] == 0
|
||||
|
||||
|
||||
def test_list_tarot_readings_filters_and_pagination():
|
||||
for cat in ["연애", "연애", "재물"]:
|
||||
db_module.save_tarot_reading({
|
||||
"spread_type": "three_card", "category": cat, "question": "Q",
|
||||
"cards": [], "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "", "warning": None, "confidence": "low"},
|
||||
"model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "low",
|
||||
})
|
||||
res = db_module.list_tarot_readings(page=1, size=10, category="연애")
|
||||
assert res["total"] == 2
|
||||
assert all(r["category"] == "연애" for r in res["items"])
|
||||
|
||||
|
||||
def test_update_tarot_reading_favorite_and_note():
|
||||
rid = db_module.save_tarot_reading({
|
||||
"spread_type": "one_card", "category": None, "question": None,
|
||||
"cards": [], "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "", "warning": None, "confidence": "high"},
|
||||
"model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "high",
|
||||
})
|
||||
db_module.update_tarot_reading(rid, favorite=True, note="기억하고 싶음")
|
||||
row = db_module.get_tarot_reading(rid)
|
||||
assert row["favorite"] == 1
|
||||
assert row["note"] == "기억하고 싶음"
|
||||
|
||||
|
||||
def test_delete_tarot_reading():
|
||||
rid = db_module.save_tarot_reading({
|
||||
"spread_type": "one_card", "category": None, "question": None,
|
||||
"cards": [], "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "", "warning": None, "confidence": "high"},
|
||||
"model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "high",
|
||||
})
|
||||
db_module.delete_tarot_reading(rid)
|
||||
assert db_module.get_tarot_reading(rid) is None
|
||||
@@ -1,113 +0,0 @@
|
||||
import json
|
||||
import pytest
|
||||
import respx
|
||||
from httpx import Response
|
||||
|
||||
from app.tarot import pipeline as p
|
||||
from app.models import TarotInterpretRequest
|
||||
|
||||
|
||||
def _valid_response_text():
|
||||
return json.dumps({
|
||||
"summary": "S",
|
||||
"cards": [
|
||||
{"position": "과거", "card": "the-fool", "reversed": False,
|
||||
"interpretation": "i", "advice": "a",
|
||||
"evidence": {"card_meaning_used": "k", "position_logic": "p", "category_lens": "c"}},
|
||||
{"position": "현재", "card": "the-lovers", "reversed": True,
|
||||
"interpretation": "i", "advice": "a",
|
||||
"evidence": {"card_meaning_used": "k", "position_logic": "p", "category_lens": "c"}},
|
||||
{"position": "미래", "card": "ten-of-cups", "reversed": False,
|
||||
"interpretation": "i", "advice": "a",
|
||||
"evidence": {"card_meaning_used": "k", "position_logic": "p", "category_lens": "c"}},
|
||||
],
|
||||
"interactions": [{"type": "synergy", "between": ["the-fool", "ten-of-cups"], "explanation": "."}],
|
||||
"advice": "A", "warning": None, "confidence": "medium",
|
||||
})
|
||||
|
||||
|
||||
def _claude_resp(text, in_tok=100, out_tok=200):
|
||||
return {
|
||||
"content": [{"type": "text", "text": text}],
|
||||
"usage": {"input_tokens": in_tok, "output_tokens": out_tok},
|
||||
}
|
||||
|
||||
|
||||
def _req():
|
||||
return TarotInterpretRequest(
|
||||
spread_type="three_card",
|
||||
category="연애",
|
||||
question="Q",
|
||||
cards=[
|
||||
{"position": "과거", "card_id": "the-fool", "reversed": False},
|
||||
{"position": "현재", "card_id": "the-lovers", "reversed": True},
|
||||
{"position": "미래", "card_id": "ten-of-cups", "reversed": False},
|
||||
],
|
||||
cards_reference="REFERENCE",
|
||||
context_meta={"major_minor_ratio": "2:1"},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_interpret_happy_path(monkeypatch):
|
||||
monkeypatch.setattr(p, "ANTHROPIC_API_KEY", "sk-test")
|
||||
with respx.mock(base_url="https://api.anthropic.com") as mock:
|
||||
mock.post("/v1/messages").mock(return_value=Response(200, json=_claude_resp(_valid_response_text())))
|
||||
out = await p.interpret(_req())
|
||||
assert out["interpretation_json"]["confidence"] == "medium"
|
||||
assert out["tokens_in"] == 100
|
||||
assert out["tokens_out"] == 200
|
||||
assert out["reroll_count"] == 0
|
||||
assert out["cost_usd"] > 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_interpret_codeblock_strip(monkeypatch):
|
||||
monkeypatch.setattr(p, "ANTHROPIC_API_KEY", "sk-test")
|
||||
wrapped = "```json\n" + _valid_response_text() + "\n```"
|
||||
with respx.mock(base_url="https://api.anthropic.com") as mock:
|
||||
mock.post("/v1/messages").mock(return_value=Response(200, json=_claude_resp(wrapped)))
|
||||
out = await p.interpret(_req())
|
||||
assert out["interpretation_json"]["summary"] == "S"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_interpret_reroll_on_validation_fail(monkeypatch):
|
||||
monkeypatch.setattr(p, "ANTHROPIC_API_KEY", "sk-test")
|
||||
bad = json.loads(_valid_response_text())
|
||||
bad["cards"][0]["evidence"]["card_meaning_used"] = ""
|
||||
bad_text = json.dumps(bad)
|
||||
with respx.mock(base_url="https://api.anthropic.com") as mock:
|
||||
route = mock.post("/v1/messages")
|
||||
route.side_effect = [
|
||||
Response(200, json=_claude_resp(bad_text)),
|
||||
Response(200, json=_claude_resp(_valid_response_text())),
|
||||
]
|
||||
out = await p.interpret(_req())
|
||||
assert out["reroll_count"] == 1
|
||||
assert out["interpretation_json"]["cards"][0]["evidence"]["card_meaning_used"] == "k"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_interpret_raises_when_both_attempts_fail(monkeypatch):
|
||||
monkeypatch.setattr(p, "ANTHROPIC_API_KEY", "sk-test")
|
||||
bad = json.loads(_valid_response_text())
|
||||
bad["cards"][0]["evidence"]["card_meaning_used"] = ""
|
||||
bad_text = json.dumps(bad)
|
||||
with respx.mock(base_url="https://api.anthropic.com") as mock:
|
||||
mock.post("/v1/messages").mock(return_value=Response(200, json=_claude_resp(bad_text)))
|
||||
with pytest.raises(p.TarotError):
|
||||
await p.interpret(_req())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_interpret_raises_when_api_key_missing(monkeypatch):
|
||||
monkeypatch.setattr(p, "ANTHROPIC_API_KEY", "")
|
||||
with pytest.raises(p.TarotError):
|
||||
await p.interpret(_req())
|
||||
|
||||
|
||||
def test_calc_cost():
|
||||
assert p.calc_cost(1_000_000, 0) == pytest.approx(3.0)
|
||||
assert p.calc_cost(0, 1_000_000) == pytest.approx(15.0)
|
||||
assert p.calc_cost(500_000, 500_000) == pytest.approx(9.0)
|
||||
@@ -1,86 +0,0 @@
|
||||
import json
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app import db as db_module
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def fresh_db(monkeypatch, tmp_path):
|
||||
db_file = tmp_path / "test_routes.db"
|
||||
monkeypatch.setattr(db_module, "DB_PATH", str(db_file))
|
||||
db_module.init_db()
|
||||
from app.main import app
|
||||
yield app
|
||||
|
||||
|
||||
def test_interpret_calls_pipeline(monkeypatch, fresh_db):
|
||||
async def fake_interpret(req):
|
||||
return {
|
||||
"interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "high"},
|
||||
"model": "claude-sonnet-4-6", "tokens_in": 100, "tokens_out": 200,
|
||||
"cost_usd": 0.005, "latency_ms": 1234, "reroll_count": 0,
|
||||
}
|
||||
from app.tarot import pipeline
|
||||
monkeypatch.setattr(pipeline, "interpret", fake_interpret)
|
||||
client = TestClient(fresh_db)
|
||||
r = client.post("/api/agent-office/tarot/interpret", json={
|
||||
"spread_type": "one_card",
|
||||
"category": "일반",
|
||||
"question": "Q",
|
||||
"cards": [{"position": "오늘", "card_id": "the-fool", "reversed": False}],
|
||||
"cards_reference": "REF",
|
||||
"context_meta": {},
|
||||
})
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["interpretation_json"]["confidence"] == "high"
|
||||
|
||||
|
||||
def test_save_and_list(fresh_db):
|
||||
client = TestClient(fresh_db)
|
||||
save = client.post("/api/agent-office/tarot/readings", json={
|
||||
"spread_type": "three_card", "category": "연애", "question": "Q",
|
||||
"cards": [{"position": "과거", "card_id": "the-fool", "reversed": False}],
|
||||
"interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "medium"},
|
||||
"model": "claude-sonnet-4-6", "tokens_in": 1, "tokens_out": 2, "cost_usd": 0.01,
|
||||
"confidence": "medium",
|
||||
})
|
||||
assert save.status_code == 200, save.text
|
||||
rid = save.json()["id"]
|
||||
lst = client.get("/api/agent-office/tarot/readings?page=1&size=10")
|
||||
assert lst.json()["total"] == 1
|
||||
assert lst.json()["items"][0]["id"] == rid
|
||||
|
||||
|
||||
def test_patch_favorite(fresh_db):
|
||||
client = TestClient(fresh_db)
|
||||
save = client.post("/api/agent-office/tarot/readings", json={
|
||||
"spread_type": "one_card", "cards": [],
|
||||
"interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "low"},
|
||||
"model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "low",
|
||||
})
|
||||
rid = save.json()["id"]
|
||||
p = client.patch(f"/api/agent-office/tarot/readings/{rid}", json={"favorite": True})
|
||||
assert p.status_code == 200
|
||||
g = client.get(f"/api/agent-office/tarot/readings/{rid}")
|
||||
assert g.json()["favorite"] == 1
|
||||
|
||||
|
||||
def test_delete(fresh_db):
|
||||
client = TestClient(fresh_db)
|
||||
save = client.post("/api/agent-office/tarot/readings", json={
|
||||
"spread_type": "one_card", "cards": [],
|
||||
"interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "low"},
|
||||
"model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "low",
|
||||
})
|
||||
rid = save.json()["id"]
|
||||
d = client.delete(f"/api/agent-office/tarot/readings/{rid}")
|
||||
assert d.status_code == 200
|
||||
g = client.get(f"/api/agent-office/tarot/readings/{rid}")
|
||||
assert g.status_code == 404
|
||||
|
||||
|
||||
def test_get_missing_reading_404(fresh_db):
|
||||
client = TestClient(fresh_db)
|
||||
r = client.get("/api/agent-office/tarot/readings/99999")
|
||||
assert r.status_code == 404
|
||||
@@ -1,75 +0,0 @@
|
||||
import pytest
|
||||
|
||||
from app.tarot.schema import validate_interpretation
|
||||
|
||||
|
||||
def _valid_three():
|
||||
return {
|
||||
"summary": "S",
|
||||
"cards": [
|
||||
{"position": "과거", "card": "the-fool", "reversed": False,
|
||||
"interpretation": "...", "advice": "a",
|
||||
"evidence": {"card_meaning_used": "키워드", "position_logic": "p", "category_lens": "c"}},
|
||||
{"position": "현재", "card": "the-lovers", "reversed": True,
|
||||
"interpretation": "...", "advice": "a",
|
||||
"evidence": {"card_meaning_used": "키워드", "position_logic": "p", "category_lens": "c"}},
|
||||
{"position": "미래", "card": "ten-of-cups", "reversed": False,
|
||||
"interpretation": "...", "advice": "a",
|
||||
"evidence": {"card_meaning_used": "키워드", "position_logic": "p", "category_lens": "c"}},
|
||||
],
|
||||
"interactions": [{"type": "synergy", "between": ["the-fool", "ten-of-cups"], "explanation": "..."}],
|
||||
"advice": "A",
|
||||
"warning": None,
|
||||
"confidence": "medium",
|
||||
}
|
||||
|
||||
|
||||
def test_valid_three_card_passes():
|
||||
ok, msg = validate_interpretation(_valid_three(), "three_card")
|
||||
assert ok, msg
|
||||
|
||||
|
||||
def test_missing_evidence_fails():
|
||||
bad = _valid_three()
|
||||
del bad["cards"][0]["evidence"]
|
||||
ok, msg = validate_interpretation(bad, "three_card")
|
||||
assert not ok
|
||||
assert "evidence" in msg
|
||||
|
||||
|
||||
def test_empty_card_meaning_used_fails():
|
||||
bad = _valid_three()
|
||||
bad["cards"][0]["evidence"]["card_meaning_used"] = ""
|
||||
ok, msg = validate_interpretation(bad, "three_card")
|
||||
assert not ok
|
||||
assert "card_meaning_used" in msg
|
||||
|
||||
|
||||
def test_three_card_requires_interactions():
|
||||
bad = _valid_three()
|
||||
bad["interactions"] = []
|
||||
ok, msg = validate_interpretation(bad, "three_card")
|
||||
assert not ok
|
||||
assert "interactions" in msg
|
||||
|
||||
|
||||
def test_one_card_accepts_empty_interactions():
|
||||
one = {
|
||||
"summary": "S",
|
||||
"cards": [{"position": "오늘", "card": "the-fool", "reversed": False,
|
||||
"interpretation": "...", "advice": "a",
|
||||
"evidence": {"card_meaning_used": "k", "position_logic": "p", "category_lens": "c"}}],
|
||||
"interactions": [],
|
||||
"advice": "A",
|
||||
"warning": None,
|
||||
"confidence": "high",
|
||||
}
|
||||
ok, msg = validate_interpretation(one, "one_card")
|
||||
assert ok, msg
|
||||
|
||||
|
||||
def test_invalid_confidence_fails():
|
||||
bad = _valid_three()
|
||||
bad["confidence"] = "very high"
|
||||
ok, msg = validate_interpretation(bad, "three_card")
|
||||
assert not ok
|
||||
Reference in New Issue
Block a user