feat(agent-office): tarot_readings 테이블 + CRUD (T1)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

agent-office/app/db.py

@@ -131,6 +131,33 @@ def init_db() -> None:
                 updated_at          TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
             )
         """)
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS tarot_readings (
+                id                  INTEGER PRIMARY KEY AUTOINCREMENT,
+                created_at          TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
+                spread_type         TEXT NOT NULL,
+                category            TEXT,
+                question            TEXT,
+                cards               TEXT NOT NULL,
+                interpretation_json TEXT,
+                summary             TEXT,
+                model               TEXT,
+                tokens_in           INTEGER,
+                tokens_out          INTEGER,
+                cost_usd            REAL,
+                confidence          TEXT,
+                favorite            INTEGER NOT NULL DEFAULT 0,
+                note                TEXT
+            )
+        """)
+        conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_tarot_created
+            ON tarot_readings(created_at DESC)
+        """)
+        conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_tarot_favorite
+            ON tarot_readings(favorite, created_at DESC)
+        """)
         # Seed default agent configs
         for agent_id, name in [
             ("stock", "주식 트레이더"),
@@ -766,3 +793,117 @@ def get_tasks_by_agent_date_kind(agent_id: str, date_iso: str, task_type: str) -
             (agent_id, task_type, date_iso),
         ).fetchall()
     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"],
+    }

agent-office/tests/test_tarot_db.py

@@ -0,0 +1,70 @@
+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