"""工具注册表和执行器 —— 小说信息、大纲 & 世界观 CRUD""" import json from datetime import datetime, timezone from sqlmodel import Session, select, func from ..database import engine from ..models import Novel, Outline, WorldSetting, Character, Chapter # ── 工具定义（provider 无关格式）── TOOL_DEFINITIONS: list[dict] = [ { "name": "get_novel_info", "description": "获取当前小说的基础信息，包括标题、简介、创建时间等", "parameters": { "type": "object", "properties": {}, }, }, { "name": "update_novel_info", "description": "更新当前小说的基础信息（标题或简介）", "parameters": { "type": "object", "properties": { "title": {"type": "string", "description": "新的小说标题（可选，1-100字）"}, "description": {"type": "string", "description": "新的小说简介（可选）"}, }, }, }, { "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字以内）"}, "parent_id": {"type": "integer", "description": "父节点 ID，创建子节点时需提供"}, }, "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": "reorder_outlines", "description": "调整大纲节点的排列顺序。传入按期望顺序排列的节点 ID 数组即可。可以排序顶级节点，也可以排序某个父节点下的子节点。", "parameters": { "type": "object", "properties": { "ordered_ids": { "type": "array", "items": {"type": "integer"}, "description": "按期望顺序排列的大纲节点 ID 数组", }, "parent_id": { "type": "integer", "description": "父节点 ID。如果排序的是某个父节点下的子节点，需提供此参数；排序顶级节点时不传", }, }, "required": ["ordered_ids"], }, }, { "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"], }, }, # ── 角色工具 ── { "name": "list_characters", "description": "列出当前小说的所有角色，返回每个角色的 id、名称和描述", "parameters": { "type": "object", "properties": {}, }, }, { "name": "create_character", "description": "为当前小说创建一个新角色", "parameters": { "type": "object", "properties": { "name": {"type": "string", "description": "角色名称（100字以内）"}, "description": {"type": "string", "description": "角色描述：外貌、性格、背景等（可选，1000字以内）"}, }, "required": ["name"], }, }, { "name": "update_character", "description": "更新指定角色的名称或描述", "parameters": { "type": "object", "properties": { "character_id": {"type": "integer", "description": "要更新的角色 ID"}, "name": {"type": "string", "description": "新的角色名称（可选）"}, "description": {"type": "string", "description": "新的角色描述（可选）"}, }, "required": ["character_id"], }, }, { "name": "delete_character", "description": "删除指定的角色", "parameters": { "type": "object", "properties": { "character_id": {"type": "integer", "description": "要删除的角色 ID"}, }, "required": ["character_id"], }, }, # ── 章节工具 ── { "name": "list_chapters", "description": "列出当前小说的所有章节，按排序顺序返回每个章节的 id、标题和内容摘要", "parameters": { "type": "object", "properties": {}, }, }, { "name": "create_chapter", "description": "为当前小说创建一个新章节", "parameters": { "type": "object", "properties": { "title": {"type": "string", "description": "章节标题（200字以内）"}, "content": {"type": "string", "description": "章节正文（可选，5000字以内）"}, }, "required": ["title"], }, }, { "name": "update_chapter", "description": "更新指定章节的标题或正文内容", "parameters": { "type": "object", "properties": { "chapter_id": {"type": "integer", "description": "要更新的章节 ID"}, "title": {"type": "string", "description": "新的章节标题（可选）"}, "content": {"type": "string", "description": "新的章节正文（可选）"}, }, "required": ["chapter_id"], }, }, { "name": "delete_chapter", "description": "删除指定的章节", "parameters": { "type": "object", "properties": { "chapter_id": {"type": "integer", "description": "要删除的章节 ID"}, }, "required": ["chapter_id"], }, }, ] 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 = { "get_novel_info": _get_novel_info, "update_novel_info": _update_novel_info, "list_outlines": _list_outlines, "create_outline": _create_outline, "update_outline": _update_outline, "delete_outline": _delete_outline, "reorder_outlines": _reorder_outlines, "get_world_setting": _get_world_setting, "save_world_setting": _save_world_setting, "list_characters": _list_characters, "create_character": _create_character, "update_character": _update_character, "delete_character": _delete_character, "list_chapters": _list_chapters, "create_chapter": _create_chapter, "update_chapter": _update_chapter, "delete_chapter": _delete_chapter, } 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 _get_novel_info(session: Session, novel_id: int, _args: dict) -> str: novel = session.get(Novel, novel_id) if not novel: return json.dumps({"error": "小说不存在"}, ensure_ascii=False) return json.dumps({ "id": novel.id, "title": novel.title, "description": novel.description or "", "created_at": novel.created_at.isoformat() if novel.created_at else None, "updated_at": novel.updated_at.isoformat() if novel.updated_at else None, }, ensure_ascii=False) def _update_novel_info(session: Session, novel_id: int, args: dict) -> str: novel = session.get(Novel, novel_id) if not novel: return json.dumps({"error": "小说不存在"}, ensure_ascii=False) if "title" in args: title = args["title"].strip() if not title or len(title) > 100: return json.dumps({"error": "标题长度需在1-100字之间"}, ensure_ascii=False) novel.title = title if "description" in args: novel.description = args["description"] novel.updated_at = datetime.now(timezone.utc) session.add(novel) session.commit() session.refresh(novel) return json.dumps({ "id": novel.id, "title": novel.title, "description": novel.description or "", "updated": True, }, ensure_ascii=False) def _list_outlines(session: Session, novel_id: int, _args: dict) -> str: # 查询顶级节点 top_items = session.exec( select(Outline) .where(Outline.novel_id == novel_id, Outline.parent_id.is_(None)) # type: ignore .order_by(Outline.sort_order) ).all() # 查询所有子节点，按 parent_id 分组 all_children = session.exec( select(Outline) .where(Outline.novel_id == novel_id, Outline.parent_id.isnot(None)) # type: ignore .order_by(Outline.sort_order) ).all() children_map: dict[int, list[Outline]] = {} for child in all_children: children_map.setdefault(child.parent_id, []).append(child) result = [] for o in top_items: node = {"id": o.id, "sort_order": o.sort_order, "summary": o.summary, "detail": o.detail} kids = children_map.get(o.id, []) if kids: node["children"] = [ {"id": c.id, "sort_order": c.sort_order, "summary": c.summary, "detail": c.detail, "parent_id": c.parent_id} for c in kids ] result.append(node) return json.dumps({"outlines": result, "total": len(result)}, ensure_ascii=False) def _create_outline(session: Session, novel_id: int, args: dict) -> str: parent_id = args.get("parent_id") # 如果指定了 parent_id，校验父节点存在且是顶级节点 if parent_id is not None: parent = session.get(Outline, parent_id) if not parent or parent.novel_id != novel_id: return json.dumps({"error": "父节点不存在"}, ensure_ascii=False) if parent.parent_id is not None: return json.dumps({"error": "仅支持两级结构，不能在子节点下创建子节点"}, ensure_ascii=False) # 同级内自增 sort_order if parent_id is not None: max_order = session.exec( select(func.max(Outline.sort_order)) .where(Outline.novel_id == novel_id, Outline.parent_id == parent_id) ).one() else: max_order = session.exec( select(func.max(Outline.sort_order)) .where(Outline.novel_id == novel_id, Outline.parent_id.is_(None)) # type: ignore ).one() next_order = (max_order or 0) + 1 outline = Outline( novel_id=novel_id, parent_id=parent_id, sort_order=next_order, summary=args["summary"], detail=args.get("detail", ""), ) session.add(outline) session.commit() session.refresh(outline) result = {"id": outline.id, "sort_order": outline.sort_order, "summary": outline.summary, "detail": outline.detail} if parent_id is not None: result["parent_id"] = parent_id return json.dumps(result, 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) # 删除父节点时先删除所有子节点 children = session.exec( select(Outline) .where(Outline.novel_id == novel_id, Outline.parent_id == outline.id) ).all() for child in children: session.delete(child) session.delete(outline) session.commit() return json.dumps({"deleted": True, "id": args["outline_id"]}, ensure_ascii=False) def _reorder_outlines(session: Session, novel_id: int, args: dict) -> str: ordered_ids = args["ordered_ids"] parent_id = args.get("parent_id") now = datetime.now(timezone.utc) for idx, oid in enumerate(ordered_ids): outline = session.get(Outline, oid) if not outline or outline.novel_id != novel_id: return json.dumps({"error": f"大纲节点 {oid} 不存在"}, ensure_ascii=False) # 校验层级一致性 if parent_id is None and outline.parent_id is not None: return json.dumps({"error": f"节点 {oid} 不是顶级节点"}, ensure_ascii=False) if parent_id is not None and outline.parent_id != parent_id: return json.dumps({"error": f"节点 {oid} 不属于父节点 {parent_id}"}, ensure_ascii=False) outline.sort_order = idx + 1 outline.updated_at = now session.add(outline) session.commit() return json.dumps({ "reordered": True, "count": len(ordered_ids), "parent_id": parent_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) # ── 角色处理函数 ── def _list_characters(session: Session, novel_id: int, _args: dict) -> str: items = session.exec( select(Character) .where(Character.novel_id == novel_id) .order_by(Character.created_at) ).all() result = [ {"id": c.id, "name": c.name, "description": c.description} for c in items ] return json.dumps({"characters": result, "total": len(result)}, ensure_ascii=False) def _create_character(session: Session, novel_id: int, args: dict) -> str: name = args["name"].strip() if not name or len(name) > 100: return json.dumps({"error": "角色名称长度需在1-100字之间"}, ensure_ascii=False) char = Character( novel_id=novel_id, name=name, description=args.get("description", ""), ) session.add(char) session.commit() session.refresh(char) return json.dumps( {"id": char.id, "name": char.name, "description": char.description}, ensure_ascii=False, ) def _update_character(session: Session, novel_id: int, args: dict) -> str: char = session.get(Character, args["character_id"]) if not char or char.novel_id != novel_id: return json.dumps({"error": "角色不存在"}, ensure_ascii=False) if "name" in args: name = args["name"].strip() if not name or len(name) > 100: return json.dumps({"error": "角色名称长度需在1-100字之间"}, ensure_ascii=False) char.name = name if "description" in args: char.description = args["description"] char.updated_at = datetime.now(timezone.utc) session.add(char) session.commit() session.refresh(char) return json.dumps( {"id": char.id, "name": char.name, "description": char.description}, ensure_ascii=False, ) def _delete_character(session: Session, novel_id: int, args: dict) -> str: char = session.get(Character, args["character_id"]) if not char or char.novel_id != novel_id: return json.dumps({"error": "角色不存在"}, ensure_ascii=False) session.delete(char) session.commit() return json.dumps({"deleted": True, "id": args["character_id"]}, ensure_ascii=False) # ── 章节处理函数 ── def _list_chapters(session: Session, novel_id: int, _args: dict) -> str: items = session.exec( select(Chapter) .where(Chapter.novel_id == novel_id) .order_by(Chapter.sort_order) ).all() result = [ { "id": c.id, "sort_order": c.sort_order, "title": c.title, "content_preview": c.content[:100] + "…" if len(c.content) > 100 else c.content, } for c in items ] return json.dumps({"chapters": result, "total": len(result)}, ensure_ascii=False) def _create_chapter(session: Session, novel_id: int, args: dict) -> str: title = args["title"].strip() if not title or len(title) > 200: return json.dumps({"error": "章节标题长度需在1-200字之间"}, ensure_ascii=False) # 自增 sort_order max_order = session.exec( select(func.max(Chapter.sort_order)) .where(Chapter.novel_id == novel_id) ).one() next_order = (max_order or 0) + 1 chapter = Chapter( novel_id=novel_id, sort_order=next_order, title=title, content=args.get("content", ""), ) session.add(chapter) session.commit() session.refresh(chapter) return json.dumps( {"id": chapter.id, "sort_order": chapter.sort_order, "title": chapter.title}, ensure_ascii=False, ) def _update_chapter(session: Session, novel_id: int, args: dict) -> str: chapter = session.get(Chapter, args["chapter_id"]) if not chapter or chapter.novel_id != novel_id: return json.dumps({"error": "章节不存在"}, ensure_ascii=False) if "title" in args: title = args["title"].strip() if not title or len(title) > 200: return json.dumps({"error": "章节标题长度需在1-200字之间"}, ensure_ascii=False) chapter.title = title if "content" in args: chapter.content = args["content"] chapter.updated_at = datetime.now(timezone.utc) session.add(chapter) session.commit() session.refresh(chapter) return json.dumps( {"id": chapter.id, "title": chapter.title, "content_length": len(chapter.content)}, ensure_ascii=False, ) def _delete_chapter(session: Session, novel_id: int, args: dict) -> str: chapter = session.get(Chapter, args["chapter_id"]) if not chapter or chapter.novel_id != novel_id: return json.dumps({"error": "章节不存在"}, ensure_ascii=False) session.delete(chapter) session.commit() return json.dumps({"deleted": True, "id": args["chapter_id"]}, ensure_ascii=False)