app.py

"""Gradio Server backend for the NPCverse hackathon app."""

from __future__ import annotations

import base64
import io
import json
import os
from typing import Any

from dotenv import load_dotenv
from fastapi.responses import HTMLResponse, JSONResponse
from gradio import Server

load_dotenv()

from model_engine import (  # noqa: E402
    DEFAULT_NPC,
    analyze_image,
    chat_respond,
    check_new_secrets,
    generate_npc,
    get_friendship_label,
)
from share_card import generate_share_card  # noqa: E402

app = Server(title="NPCverse")


@app.get("/", response_class=HTMLResponse)
async def homepage() -> HTMLResponse:
    """Serve the custom NPCverse frontend."""
    index_path = os.path.join(os.path.dirname(__file__), "index.html")
    try:
        with open(index_path, "r", encoding="utf-8") as index_file:
            return HTMLResponse(index_file.read())
    except FileNotFoundError:
        return HTMLResponse("<main><h1>NPCverse</h1><p>UI loading...</p></main>")


@app.get("/health")
async def health() -> JSONResponse:
    """Return a lightweight health check for deployment probes."""
    return JSONResponse({"status": "ok", "model": "MiniCPM-V-4.6"})


@app.api(name="summon_npc")
def summon_npc(image_path: Any) -> dict:
    """Analyze an uploaded image and summon a complete NPC profile."""
    import traceback
    try:
        path = _extract_file_path(image_path)
        print(f"[NPCverse] summon_npc: extracted path = {path!r}")
        description = analyze_image(path)
        print(f"[NPCverse] summon_npc: image description = {description!r}")
        npc = generate_npc(description)
        print(f"[NPCverse] summon_npc: generated NPC name = {npc.get('name')!r}")
        return _json_safe_dict(npc)
    except Exception as e:
        print(f"[NPCverse] summon_npc FAILED: {e}")
        traceback.print_exc()
        return _json_safe_dict(DEFAULT_NPC)


@app.api(name="chat_with_npc")
def chat_with_npc_api(
    npc_json: str,
    history_json: str,
    user_message: str,
    msg_count: int,
    unlocked_secrets_json: str,
) -> dict:
    """Continue an in-character NPC conversation and update progression state."""
    npc = _loads_or_default(npc_json, DEFAULT_NPC)
    history = _loads_or_default(history_json, [])
    unlocked_secrets = _loads_or_default(unlocked_secrets_json, [])

    if not isinstance(npc, dict):
        npc = DEFAULT_NPC
    if not isinstance(history, list):
        history = []
    if not isinstance(unlocked_secrets, list):
        unlocked_secrets = []

    current_msg_count = _safe_int(msg_count)
    new_msg_count = current_msg_count + 1
    newly_unlocked = check_new_secrets(new_msg_count, unlocked_secrets)
    secret_texts = _get_secret_texts(npc, newly_unlocked)

    response = chat_respond(
        npc=npc,
        history=history,
        user_message=user_message,
        msg_count=new_msg_count,
        unlocked_secrets=[*unlocked_secrets, *newly_unlocked],
    )

    return {
        "response": str(response),
        "new_msg_count": new_msg_count,
        "friendship_label": get_friendship_label(new_msg_count),
        "friendship_pct": min(100.0, (current_msg_count / 55) * 100),
        "newly_unlocked": newly_unlocked,
        "secret_texts": secret_texts,
    }


@app.api(name="generate_share_card")
def get_share_card(npc_json: str) -> dict:
    """Generate a base64 PNG social share card for an NPC profile."""
    npc = _loads_or_default(npc_json, DEFAULT_NPC)
    if not isinstance(npc, dict):
        npc = DEFAULT_NPC

    image = generate_share_card(npc)
    buffer = io.BytesIO()
    image.save(buffer, format="PNG")
    image_b64 = base64.b64encode(buffer.getvalue()).decode("utf-8")
    return {"image_b64": f"data:image/png;base64,{image_b64}"}


def _extract_file_path(file_value: Any) -> str:
    """Extract an uploaded file path from Gradio's possible file payloads."""
    if isinstance(file_value, str):
        return file_value
    if isinstance(file_value, dict):
        return str(file_value.get("path") or file_value.get("name") or "")
    path = getattr(file_value, "path", None) or getattr(file_value, "name", None)
    return str(path or "")


def _loads_or_default(raw_json: str, default: Any) -> Any:
    """Parse JSON input from the frontend, returning a fallback on invalid data."""
    try:
        return json.loads(raw_json)
    except (TypeError, json.JSONDecodeError):
        return default


def _safe_int(value: Any) -> int:
    """Convert a value to a non-negative integer message count."""
    try:
        return max(0, int(value))
    except (TypeError, ValueError):
        return 0


def _get_secret_texts(npc: dict, secret_indices: list[int]) -> list[str]:
    """Resolve unlocked secret indices to their corresponding secret text."""
    secrets = npc.get("secrets", [])
    if not isinstance(secrets, list):
        return []

    secret_texts: list[str] = []
    for index in secret_indices:
        try:
            secret_texts.append(str(secrets[int(index)]))
        except (TypeError, ValueError, IndexError):
            continue
    return secret_texts


def _json_safe_dict(payload: dict) -> dict:
    """Round-trip a dictionary through JSON to ensure API-safe primitives."""
    return json.loads(json.dumps(payload, ensure_ascii=False))


if __name__ == "__main__":
    app.launch(server_name="0.0.0.0", server_port=7860, show_error=True)