app.py

"""Confidence Buddy API on Gradio Spaces (DeepSeek proxy + optional UI)."""

from __future__ import annotations

import json
import os
from pathlib import Path
from typing import Any

import gradio as gr
import gradio.routes
import httpx
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse

try:
    import spaces
except ImportError:
    spaces = None

PERSONALITIES_PATH = Path(__file__).with_name("personalities.json")
PERSONALITIES: dict[str, dict[str, Any]] = json.loads(
    PERSONALITIES_PATH.read_text(encoding="utf-8")
)

DEEPSEEK_URL = os.environ.get(
    "DEEPSEEK_API_URL", "https://api.deepseek.com/chat/completions"
)
MODEL = os.environ.get("DEEPSEEK_MODEL", "deepseek-chat")


def _api_key() -> str | None:
    return os.environ.get("DEEPSEEK_API_KEY")


def _app_secret() -> str | None:
    return os.environ.get("CHAT_API_SECRET")


def list_personalities() -> list[dict[str, Any]]:
    return [
        {
            "id": p["id"],
            "name": p["name"],
            "category": p["category"],
            "subcategory": p.get("subcategory"),
            "greeting": p["greeting"],
            "suggestions": p.get("suggestions", []),
        }
        for p in PERSONALITIES.values()
    ]


def get_personality(personality_id: str) -> dict[str, Any] | None:
    return PERSONALITIES.get(personality_id)


async def deepseek_chat(
    personality_id: str,
    message: str,
    history: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    api_key = _api_key()
    if not api_key:
        raise HTTPException(status_code=500, detail="DEEPSEEK_API_KEY is not configured")

    if not personality_id or not isinstance(message, str) or not message.strip():
        raise HTTPException(
            status_code=400, detail="personalityId and message are required"
        )

    personality = get_personality(personality_id)
    if not personality:
        raise HTTPException(status_code=404, detail="Personality not found")

    history = history or []
    messages: list[dict[str, str]] = [
        {"role": "system", "content": personality["systemPrompt"]},
    ]
    for item in history:
        if not item:
            continue
        role = item.get("role")
        content = item.get("content")
        if role in ("user", "assistant") and isinstance(content, str):
            messages.append({"role": role, "content": content})
    messages = messages[:1] + messages[1:][-20:]
    messages.append({"role": "user", "content": message.strip()})

    try:
        async with httpx.AsyncClient(timeout=60.0) as client:
            response = await client.post(
                DEEPSEEK_URL,
                headers={
                    "Content-Type": "application/json",
                    "Authorization": f"Bearer {api_key}",
                },
                json={
                    "model": MODEL,
                    "messages": messages,
                    "temperature": 0.7,
                    "max_tokens": 1024,
                },
            )
            data = response.json()
    except Exception as exc:  # noqa: BLE001
        raise HTTPException(status_code=502, detail=str(exc)) from exc

    if response.status_code >= 400:
        detail = (
            data.get("error", {}).get("message")
            if isinstance(data, dict)
            else None
        ) or response.reason_phrase
        raise HTTPException(status_code=response.status_code, detail=detail)

    reply = (
        ((data.get("choices") or [{}])[0].get("message") or {}).get("content") or ""
    ).strip()
    if not reply:
        raise HTTPException(status_code=502, detail="Empty response from DeepSeek")

    return {
        "reply": reply,
        "personality": {
            "id": personality["id"],
            "name": personality["name"],
            "category": personality["category"],
            "subcategory": personality.get("subcategory"),
        },
    }


if spaces is not None:

    @spaces.GPU(duration=60)
    def _zero_gpu_placeholder() -> str:
        return "ok"


PERSONALITY_CHOICES = [
    (f"{p['name']} ({p['id']})", p["id"]) for p in PERSONALITIES.values()
]


async def ui_chat(
    personality_id: str,
    message: str,
    history: list[dict[str, str]],
):
    if not message or not message.strip():
        return history, ""
    try:
        result = await deepseek_chat(personality_id, message, history)
        reply = result["reply"]
    except HTTPException as exc:
        reply = f"Error: {exc.detail}"
    history = history + [
        {"role": "user", "content": message},
        {"role": "assistant", "content": reply},
    ]
    return history, ""


def _attach_api_routes(app) -> None:
    """Flutter-compatible REST routes. Must be attached when Gradio builds the app."""

    @app.middleware("http")
    async def optional_app_key(request: Request, call_next):
        if request.url.path == "/health":
            return await call_next(request)
        secret = _app_secret()
        if secret and request.headers.get("x-app-key") != secret:
            return JSONResponse({"error": "Unauthorized"}, status_code=401)
        return await call_next(request)

    @app.get("/health")
    async def health():
        return {"ok": True, "hasApiKey": bool(_api_key())}

    @app.get("/personalities")
    async def personalities_list():
        return list_personalities()

    @app.get("/personalities/{personality_id}")
    async def personality_detail(personality_id: str):
        personality = get_personality(personality_id)
        if not personality:
            return JSONResponse({"error": "Personality not found"}, status_code=404)
        return {
            "id": personality["id"],
            "name": personality["name"],
            "category": personality["category"],
            "subcategory": personality.get("subcategory"),
            "greeting": personality["greeting"],
            "suggestions": personality.get("suggestions", []),
        }

    @app.post("/chat")
    async def chat_endpoint(request: Request):
        body = await request.json()
        try:
            return await deepseek_chat(
                personality_id=body.get("personalityId", ""),
                message=body.get("message", ""),
                history=body.get("history") or [],
            )
        except HTTPException as exc:
            return JSONResponse({"error": exc.detail}, status_code=exc.status_code)


_original_create_app = gradio.routes.App.create_app


def _create_app_with_api_routes(blocks, *args, **kwargs):
    app = _original_create_app(blocks, *args, **kwargs)
    _attach_api_routes(app)
    return app


gradio.routes.App.create_app = staticmethod(_create_app_with_api_routes)

# Gradio SSR captures GET /health as an HTML page; disable it so REST routes win.
_original_launch = gr.Blocks.launch


def _launch_without_ssr(self, *args, **kwargs):
    kwargs.setdefault("ssr_mode", False)
    return _original_launch(self, *args, **kwargs)


gr.Blocks.launch = _launch_without_ssr


with gr.Blocks(title="Confidence Buddy API") as demo:
    gr.Markdown(
        "## Confidence Buddy API\n"
        "Flutter uses `POST /chat`. This UI is for quick manual checks.\n\n"
        f"Personalities loaded: **{len(PERSONALITIES)}** · "
        f"API key configured: **{bool(_api_key())}**"
    )
    personality = gr.Dropdown(
        choices=PERSONALITY_CHOICES,
        value=PERSONALITY_CHOICES[0][1] if PERSONALITY_CHOICES else None,
        label="Personality",
    )
    chatbot = gr.Chatbot(type="messages", height=420)
    msg = gr.Textbox(label="Message", placeholder="Type a message…")
    clear = gr.Button("Clear")
    msg.submit(ui_chat, [personality, msg, chatbot], [chatbot, msg])
    clear.click(lambda: ([], ""), outputs=[chatbot, msg])


if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", "7860")))