NovalRedot/backend/app/services/tools.py

"""工具注册表和执行器 —— 大纲 & 世界观 CRUD"""

import json
from datetime import datetime, timezone

from sqlmodel import Session, select, func

from ..database import engine
from ..models import Novel, Outline, WorldSetting


# ── 工具定义（provider 无关格式）──

TOOL_DEFINITIONS: list[dict] = [
    {
        "name": "list_outlines",
        "description": "列出当前小说的所有大纲节点，返回每个节点的 id、排序、摘要和详情",
        "parameters": {
            "type": "object",
            "properties": {},
        },
    },
    {
        "name": "create_outline",
        "description": "为当前小说创建一个新的大纲节点",
        "parameters": {
            "type": "object",
            "properties": {
                "summary": {"type": "string", "description": "大纲摘要（简短标题，200字以内）"},
                "detail": {"type": "string", "description": "大纲详情（可选，1000字以内）"},
            },
            "required": ["summary"],
        },
    },
    {
        "name": "update_outline",
        "description": "更新指定大纲节点的摘要或详情",
        "parameters": {
            "type": "object",
            "properties": {
                "outline_id": {"type": "integer", "description": "要更新的大纲节点 ID"},
                "summary": {"type": "string", "description": "新的摘要（可选）"},
                "detail": {"type": "string", "description": "新的详情（可选）"},
            },
            "required": ["outline_id"],
        },
    },
    {
        "name": "delete_outline",
        "description": "删除指定的大纲节点",
        "parameters": {
            "type": "object",
            "properties": {
                "outline_id": {"type": "integer", "description": "要删除的大纲节点 ID"},
            },
            "required": ["outline_id"],
        },
    },
    {
        "name": "get_world_setting",
        "description": "获取当前小说的世界观设定内容",
        "parameters": {
            "type": "object",
            "properties": {},
        },
    },
    {
        "name": "save_world_setting",
        "description": "保存/更新当前小说的世界观设定",
        "parameters": {
            "type": "object",
            "properties": {
                "content": {"type": "string", "description": "世界观设定内容（5000字以内）"},
            },
            "required": ["content"],
        },
    },
]


def get_openai_tools() -> list[dict]:
    """转换为 OpenAI function calling 格式"""
    return [
        {"type": "function", "function": t}
        for t in TOOL_DEFINITIONS
    ]


def get_anthropic_tools() -> list[dict]:
    """转换为 Anthropic tool_use 格式"""
    return [
        {"name": t["name"], "description": t["description"], "input_schema": t["parameters"]}
        for t in TOOL_DEFINITIONS
    ]


# ── 工具执行 ──


def execute_tool(name: str, arguments: dict, novel_id: int) -> str:
    """执行工具，返回结果字符串（JSON）"""
    with Session(engine) as session:
        handlers = {
            "list_outlines": _list_outlines,
            "create_outline": _create_outline,
            "update_outline": _update_outline,
            "delete_outline": _delete_outline,
            "get_world_setting": _get_world_setting,
            "save_world_setting": _save_world_setting,
        }
        handler = handlers.get(name)
        if not handler:
            return json.dumps({"error": f"未知工具: {name}"}, ensure_ascii=False)
        try:
            return handler(session, novel_id, arguments)
        except Exception as e:
            return json.dumps({"error": str(e)}, ensure_ascii=False)


def _list_outlines(session: Session, novel_id: int, _args: dict) -> str:
    items = session.exec(
        select(Outline)
        .where(Outline.novel_id == novel_id)
        .order_by(Outline.sort_order)
    ).all()
    result = [
        {"id": o.id, "sort_order": o.sort_order, "summary": o.summary, "detail": o.detail}
        for o in items
    ]
    return json.dumps({"outlines": result, "total": len(result)}, ensure_ascii=False)


def _create_outline(session: Session, novel_id: int, args: dict) -> str:
    max_order = session.exec(
        select(func.max(Outline.sort_order)).where(Outline.novel_id == novel_id)
    ).one()
    next_order = (max_order or 0) + 1

    outline = Outline(
        novel_id=novel_id,
        sort_order=next_order,
        summary=args["summary"],
        detail=args.get("detail", ""),
    )
    session.add(outline)
    session.commit()
    session.refresh(outline)
    return json.dumps(
        {"id": outline.id, "sort_order": outline.sort_order, "summary": outline.summary, "detail": outline.detail},
        ensure_ascii=False,
    )


def _update_outline(session: Session, novel_id: int, args: dict) -> str:
    outline = session.get(Outline, args["outline_id"])
    if not outline or outline.novel_id != novel_id:
        return json.dumps({"error": "大纲不存在"}, ensure_ascii=False)

    if "summary" in args:
        outline.summary = args["summary"]
    if "detail" in args:
        outline.detail = args["detail"]
    outline.updated_at = datetime.now(timezone.utc)
    session.add(outline)
    session.commit()
    session.refresh(outline)
    return json.dumps(
        {"id": outline.id, "summary": outline.summary, "detail": outline.detail},
        ensure_ascii=False,
    )


def _delete_outline(session: Session, novel_id: int, args: dict) -> str:
    outline = session.get(Outline, args["outline_id"])
    if not outline or outline.novel_id != novel_id:
        return json.dumps({"error": "大纲不存在"}, ensure_ascii=False)

    session.delete(outline)
    session.commit()
    return json.dumps({"deleted": True, "id": args["outline_id"]}, ensure_ascii=False)


def _get_world_setting(session: Session, novel_id: int, _args: dict) -> str:
    ws = session.exec(
        select(WorldSetting).where(WorldSetting.novel_id == novel_id)
    ).first()
    if not ws:
        return json.dumps({"content": "", "exists": False}, ensure_ascii=False)
    return json.dumps({"content": ws.content, "exists": True}, ensure_ascii=False)


def _save_world_setting(session: Session, novel_id: int, args: dict) -> str:
    ws = session.exec(
        select(WorldSetting).where(WorldSetting.novel_id == novel_id)
    ).first()
    now = datetime.now(timezone.utc)
    if ws:
        ws.content = args["content"]
        ws.updated_at = now
    else:
        ws = WorldSetting(novel_id=novel_id, content=args["content"])
    session.add(ws)
    session.commit()
    session.refresh(ws)
    return json.dumps({"saved": True, "content_length": len(ws.content)}, ensure_ascii=False)