"""工具注册表和执行器 —— 小说信息、大纲 & 世界观 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, NovelFile from ..database import DATA_DIR # ── 工具定义（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"], }, }, # ── 文件工具 ── { "name": "list_files", "description": "列出当前小说关联的所有上传文件，返回文件名、大小和纯文本字符数", "parameters": { "type": "object", "properties": {}, }, }, { "name": "read_file", "description": "读取指定上传文件的纯文本内容。支持分段读取（offset/limit 按字符数），适合大文件。返回文本片段和总字符数。", "parameters": { "type": "object", "properties": { "file_id": {"type": "integer", "description": "文件 ID（从 list_files 获取）"}, "offset": {"type": "integer", "description": "起始字符位置（默认 0）"}, "limit": {"type": "integer", "description": "读取字符数（默认 5000，最大 10000）"}, }, "required": ["file_id"], }, }, ] def _filter_defs(allowed: list[str] | None) -> list[dict]: """根据白名单过滤工具定义""" if not allowed: return TOOL_DEFINITIONS return [t for t in TOOL_DEFINITIONS if t["name"] in allowed] def get_openai_tools(allowed: list[str] | None = None) -> list[dict]: """转换为 OpenAI function calling 格式""" return [ {"type": "function", "function": t} for t in _filter_defs(allowed) ] def get_anthropic_tools(allowed: list[str] | None = None) -> list[dict]: """转换为 Anthropic tool_use 格式""" return [ {"name": t["name"], "description": t["description"], "input_schema": t["parameters"]} for t in _filter_defs(allowed) ] def get_tools_description(allowed: list[str] | None = None) -> str: """动态生成工具能力描述文本（用于系统提示词）""" defs = _filter_defs(allowed) if not defs: return "" lines = ["你拥有以下工具能力，可以直接操作小说数据："] for t in defs: lines.append(f"- {t['name']}: {t['description']}") lines.append("\n请根据用户需求主动使用工具，不要仅给出建议而不执行。") return "\n".join(lines) # ── 工具执行 ── 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, "list_files": _list_files, "read_file": _read_file, } 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) # ── 文件处理函数 ── def _list_files(session: Session, novel_id: int, _args: dict) -> str: items = session.exec( select(NovelFile) .where(NovelFile.novel_id == novel_id) .order_by(NovelFile.created_at.desc()) ).all() result = [ { "id": f.id, "filename": f.filename, "file_size": f.file_size, "text_length": f.text_length, "created_at": f.created_at.isoformat() if f.created_at else None, } for f in items ] return json.dumps({"files": result, "total": len(result)}, ensure_ascii=False) def _read_file(session: Session, novel_id: int, args: dict) -> str: file_id = args["file_id"] offset = max(0, args.get("offset", 0)) limit = min(10000, max(1, args.get("limit", 5000))) novel_file = session.get(NovelFile, file_id) if not novel_file or novel_file.novel_id != novel_id: return json.dumps({"error": "文件不存在"}, ensure_ascii=False) # 读取预解析的纯文本缓存 text_path = DATA_DIR / "uploads" / str(novel_id) / f"{novel_file.stored_name}.txt" if not text_path.exists(): return json.dumps({"error": "文件纯文本缓存不存在，可能需要重新上传"}, ensure_ascii=False) full_text = text_path.read_text(encoding="utf-8") total_length = len(full_text) segment = full_text[offset:offset + limit] return json.dumps({ "filename": novel_file.filename, "text": segment, "offset": offset, "limit": limit, "total_length": total_length, "has_more": offset + limit < total_length, }, ensure_ascii=False)