src/webui/services/tavern.py

Rama main · réplica de lectura del repositorio original.

"""酒馆角色卡导入服务。"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

from src.engine.character_utils import parse_tavern_card
from src.webui.services.character_cards import commit_character_book

logger = logging.getLogger("trpg")
GameKey = tuple[str, ...]


@dataclass(frozen=True)
class TavernImportDependencies:
    lorebook: Any | None
    get_instance: Callable[[GameKey], Any | None]
    parse_game_key: Callable[[str], GameKey]
    rebuild_lorebook_index: Callable[[str], None]


async def import_tavern_card(dependencies: TavernImportDependencies, file_path: str = "", file_data: str = "",
                             file_name: str = "card.png", game_key: str = "") -> dict[str, Any]:
    """导入酒馆角色卡为 NPC 或玩家角色。

    支持两种模式：
    - file_path: 服务器上的文件路径
    - file_data + file_name: 客户端上传的 base64 数据
    """
    import base64
    import tempfile

    if file_data:
        if len(file_data) > 40_000_000:
            return {"ok": False, "error": "文件过大（上限 30MB）"}
        raw_bytes = base64.b64decode(file_data)
        safe_name = Path(file_name).name or "card.png"
        tmp_path = Path(tempfile.gettempdir()) / f"trpg_import_{safe_name}"
        tmp_path.write_bytes(raw_bytes)
        card = parse_tavern_card(str(tmp_path))
        try:
            tmp_path.unlink()
        except OSError:
            pass
    elif file_path:
        return {"ok": False, "error": "已禁用 file_path 模式（安全风险），请改用 file_data 上传"}
    else:
        return {"ok": False, "error": "未提供文件"}

    if "error" in card:
        return {"ok": False, "error": card["error"]}

    npc_info = {
        "name": card["name"],
        "type": "npc",
        "keywords": [card["name"]] + card.get("tags", []),
        "content": (
            f"描述: {card['description']}\n"
            f"性格: {card['personality']}\n"
            f"背景: {card['scenario']}\n"
            f"初次见面: {card['first_mes']}"
        ).strip(),
        "tier": "core",
    }

    lorebook = dependencies.lorebook
    world_id = ""
    if game_key and lorebook:
        inst = dependencies.get_instance(dependencies.parse_game_key(game_key))
        if inst and inst.world_id:
            world_id = str(inst.world_id)
            entry_id = f"npc_tavern_{card['name'].replace(' ', '_')}"
            existing = lorebook.get_entry(entry_id)
            npc_info["id"] = entry_id
            npc_info["world_id"] = world_id
            if existing:
                lorebook.update_entry(entry_id, npc_info)
            else:
                lorebook.add_entry(npc_info)
            dependencies.rebuild_lorebook_index(world_id)
            logger.info("酒馆角色卡已导入: %s -> world=%s", card["name"], world_id)

    # This route is a façade: the embedded character_book goes through the very
    # same canonical commit as the Character Card product flow, instead of only
    # being counted. `lorebook_entries` stays for existing callers.
    embedded = card.get("character_book")
    if embedded and lorebook:
        safe_name = str(card["name"]).replace(" ", "_") or "card"
        lore = commit_character_book(
            lorebook, name=str(card["name"]), book={"entries": embedded},
            book_id=f"character_card:{world_id or 'unbound'}:{safe_name}",
            entry_id_prefix=f"tavern_{safe_name}_book", world_id=world_id,
        )
        npc_info["lorebook_entries"] = int(lore["entries"])
        npc_info["lorebook_book_id"] = lore["book_id"]
        npc_info["lorebook"] = lore
    elif embedded:
        npc_info["lorebook_entries"] = len(embedded)

    return {"ok": True, "card": card, "npc": npc_info}


class TavernImportService:
    """Tavern-card import with explicit game and lorebook boundaries."""

    def __init__(self, dependencies: TavernImportDependencies) -> None:
        self._dependencies = dependencies

    async def import_card(
        self,
        file_path: str = "",
        file_data: str = "",
        file_name: str = "card.png",
        game_key: str = "",
    ) -> dict[str, Any]:
        return await import_tavern_card(
            self._dependencies, file_path, file_data, file_name, game_key,
        )