fateforge/server.py

"""FateForge v2 — FastAPI 后端。

四关卡流转 (Vue 3 SPA):
  画面一 缘起:  GET /
  画面二 窥机:  POST /api/cast/seed   (前端摇 6 次后,生成卦象)
  画面三 破迷:  POST /api/read        (SSE 流式返回 LLM 解读)
  画面四 自渡:  前端展示自渡锦囊 + 拂袖

免登录, 卦象/解读客户端只放 localStorage。
"""
from __future__ import annotations

import json
import logging
from typing import AsyncIterator

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

from .config import STATIC_DIR
from .divination import (
    all_lines_glyph,
    cast_six_lines,
    get_changed_hex,
    hex_auspicious,
    hex_auspicious_level,
    lines_to_hex,
    moving_indices,
)
from .llm import stream_fate_reading
from .interpret import build_full_reading, build_full_reading_en
from .constants import CATEGORY_UI, HEX_LEVELS, LEVEL_CATEGORY

logging.basicConfig(level=logging.INFO, format="[FF] %(message)s")
log = logging.getLogger("fateforge")

app = FastAPI(title="FateForge", version="2.0.0")


# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class CastRequest(BaseModel):
    seed: int | None = Field(default=None, description="可选随机种子,用于复现")


class ReadRequest(BaseModel):
    question: str = Field(..., min_length=1, max_length=200)
    lines: list[int] = Field(..., min_length=6, max_length=6)
    lang: str = Field(default="zh", description="界面语言: zh | en")


# ---------------------------------------------------------------------------
# 静态资源
# ---------------------------------------------------------------------------
@app.get("/", include_in_schema=False)
async def index() -> FileResponse:
    return FileResponse(STATIC_DIR / "index.html")


# favicon
@app.get("/favicon.ico", include_in_schema=False)
async def favicon() -> FileResponse:
    f = STATIC_DIR / "favicon.svg"
    return FileResponse(f) if f.exists() else FileResponse(STATIC_DIR / "index.html")


# ---------------------------------------------------------------------------
# API: 起卦 (画面二 → 画面三 的过渡接口)
# ---------------------------------------------------------------------------
@app.post("/api/cast/seed")
async def cast_seed(req: CastRequest) -> dict:
    """后端帮前端生成 6 爻(避免前端随机数可预测)。

    返回:
      lines:        [6,7,8,9,7,8]    # 自下而上
      glyphs:       [{value, yang, moving, mark, label}, ...]
      hex:          主卦完整信息
      changed:      变卦 (无动爻时为 null)
      moving:       [3,5]            # 动爻位置 (1-indexed)
      auspicious:   '吉' | '平' | '凶'
    """
    lines = cast_six_lines(req.seed)
    glyphs = all_lines_glyph(lines)
    hex_info = lines_to_hex(lines)
    changed = get_changed_hex(lines)
    moving = moving_indices(lines)
    level = hex_auspicious_level(hex_info)
    level_info = HEX_LEVELS.get(hex_info["name"], {})
    return {
        "lines": lines,
        "glyphs": glyphs,
        "hex": hex_info,
        "changed": changed,
        "moving": moving,
        "auspicious": hex_auspicious(hex_info),  # 3 档 (吉/平/凶) 兼容旧字段
        "level": level,                            # 7 档 (大吉/吉/.../大凶)
        "levelInfo": {
            "tag": level_info.get("tag", "潜龙勿用"),
            "reason": level_info.get("reason", ""),
        },
        "ui": CATEGORY_UI.get(LEVEL_CATEGORY.get(level, "平"), CATEGORY_UI["平"]),
    }


# ---------------------------------------------------------------------------
# API: 解读 (SSE)
# ---------------------------------------------------------------------------
SSE_HEADERS = {
    "Cache-Control": "no-cache",
    "X-Accel-Buffering": "no",   # 禁止 nginx 等代理缓冲
    "Connection": "keep-alive",
}


def _sse_pack(data: dict) -> str:
    """SSE 协议: data: <json>\n\n"""
    return f"data: {json.dumps(data, ensure_ascii=False)}\n\n"


@app.post("/api/read")
async def read(req: ReadRequest) -> StreamingResponse:
    """流式推演缘主困惑 → 卦象 → LLM 解读 (SSE)。

    LLM 在线时:直接流式输出 LLM 文本 (含 <听天命> 与 <运在人为> 两段标签)。
    LLM 离线 / 超时:降级到 fateforge.interpret 的本地卦象解读,按字流式输出,
                     保持「水墨渗字」的视觉效果。
    """
    if not req.lines or len(req.lines) != 6:
        raise HTTPException(status_code=400, detail="lines 必须是长度为 6 的数组")
    for v in req.lines:
        if v not in (6, 7, 8, 9):
            raise HTTPException(status_code=400, detail="爻值必须为 6/7/8/9")

    hex_info = lines_to_hex(req.lines)
    changed = get_changed_hex(req.lines)
    moving = moving_indices(req.lines)
    level = hex_auspicious_level(hex_info)
    level_info = HEX_LEVELS.get(hex_info["name"], {})
    ui = CATEGORY_UI.get(LEVEL_CATEGORY.get(level, "平"), CATEGORY_UI["平"])

    async def gen() -> AsyncIterator[str]:
        # 1) 卦象元数据
        yield _sse_pack({
            "type": "meta",
            "hex": hex_info,
            "changed": changed,
            "moving": moving,
            "auspicious": hex_auspicious(hex_info),  # 3 档
            "level": level,                            # 7 档
            "levelInfo": {
                "tag": level_info.get("tag", ui["tag"]),
                "reason": level_info.get("reason", ""),
            },
            "ui": ui,
        })
        # 2) 先尝试 LLM (短超时,失败立即降级)
        import asyncio
        llm_chunks: list[str] = []
        llm_failed = True
        try:
            async with asyncio.timeout(20):
                async for chunk in stream_fate_reading(
                    req.question, hex_info, changed, moving,
                    level=level,
                    reason=level_info.get("reason", ""),
                    ui=ui,
                    lang=req.lang,
                ):
                    llm_chunks.append(chunk)
                    llm_failed = False
                    yield _sse_pack({"type": "delta", "text": chunk, "source": "llm"})
        except (asyncio.TimeoutError, Exception) as e:  # noqa: BLE001
            if llm_failed:
                log.warning(f"LLM 不可用,降级本地解读: {type(e).__name__}: {e}")
            else:
                # 半路断开,已流出的部分保留,降级补全剩余
                log.warning(f"LLM 流中断,补全降级: {e}")

        # 3) 降级: 本地解读 (整篇一次生成,按字流式)
        if llm_failed:
            yield _sse_pack({
                "type": "warn",
                "message": "天机暂不可闻,以下为本座亲推" if req.lang == "zh" else "The way is silent. These notes are mine.",
            })
        # 3a) 先把已收的 LLM 块补上
        full = "".join(llm_chunks)
        # 3b) 若 LLM 失败,补上完整本地解读;若 LLM 半路断,补上缺的两段
        if llm_failed or "<听天命" not in full or "<运在人为" not in full:
            if req.lang == "en":
                # 英文 fallback: 用 <Reading> / <Way Through> 标签
                local_full = build_full_reading_en(
                    req.question, hex_info, moving, changed,
                    level=level, ui=ui,
                )
                if llm_failed:
                    yield _sse_pack({"type": "delta", "text": local_full, "source": "local"})
            else:
                local_full = build_full_reading(
                    req.question, hex_info, moving, changed,
                    level=level, ui=ui,
                )
                if llm_failed:
                    yield _sse_pack({"type": "delta", "text": local_full, "source": "local"})
                else:
                    # 半路断:补缺哪段发哪段(仍带标签以便前端 split)
                    import re
                    if "<听天命" not in full:
                        m1 = re.search(r"<听天命[^>]*>[\s\S]*?</听天命[^>]*>", local_full)
                        if m1:
                            yield _sse_pack({"type": "delta", "text": "\n\n" + m1.group(0), "source": "local"})
                    if "<运在人为" not in full:
                        m2 = re.search(r"<运在人为[^>]*>[\s\S]*?</运在人为[^>]*>", local_full)
                        if m2:
                            yield _sse_pack({"type": "delta", "text": "\n\n" + m2.group(0), "source": "local"})

        # 4) 收尾
        yield _sse_pack({"type": "done", "source": "local" if llm_failed else "llm"})

    return StreamingResponse(gen(), media_type="text/event-stream", headers=SSE_HEADERS)


# ---------------------------------------------------------------------------
# 静态资源挂载 (放最后, 不挡 API)
# ---------------------------------------------------------------------------
if STATIC_DIR.exists():
    app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")