引入二级大纲功能,新增父子节点关系及子节点排序,同时调整API与UI以支持嵌套大纲管理

This commit is contained in:
2026-03-22 16:53:33 +08:00
parent 20c759150a
commit 098ff4e16d
33 changed files with 2516 additions and 381 deletions

View File

@@ -1,4 +1,4 @@
"""工具注册表和执行器 —— 大纲 & 世界观 CRUD"""
"""工具注册表和执行器 —— 小说信息、大纲 & 世界观 CRUD"""
import json
from datetime import datetime, timezone
@@ -6,12 +6,31 @@ from datetime import datetime, timezone
from sqlmodel import Session, select, func
from ..database import engine
from ..models import Novel, Outline, WorldSetting
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、排序、摘要和详情",
@@ -28,6 +47,7 @@ TOOL_DEFINITIONS: list[dict] = [
"properties": {
"summary": {"type": "string", "description": "大纲摘要简短标题200字以内"},
"detail": {"type": "string", "description": "大纲详情可选1000字以内"},
"parent_id": {"type": "integer", "description": "父节点 ID创建子节点时需提供"},
},
"required": ["summary"],
},
@@ -56,6 +76,25 @@ TOOL_DEFINITIONS: list[dict] = [
"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": "获取当前小说的世界观设定内容",
@@ -75,6 +114,96 @@ TOOL_DEFINITIONS: list[dict] = [
"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"],
},
},
]
@@ -101,12 +230,23 @@ 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:
@@ -117,27 +257,100 @@ def execute_tool(name: str, arguments: dict, novel_id: int) -> str:
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:
items = session.exec(
# 查询顶级节点
top_items = session.exec(
select(Outline)
.where(Outline.novel_id == novel_id)
.where(Outline.novel_id == novel_id, Outline.parent_id.is_(None)) # type: ignore
.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
]
# 查询所有子节点,按 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:
max_order = session.exec(
select(func.max(Outline.sort_order)).where(Outline.novel_id == novel_id)
).one()
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", ""),
@@ -145,10 +358,10 @@ def _create_outline(session: Session, novel_id: int, args: dict) -> str:
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,
)
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:
@@ -175,11 +388,43 @@ def _delete_outline(session: Session, novel_id: int, args: dict) -> str:
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)
@@ -203,3 +448,148 @@ def _save_world_setting(session: Session, novel_id: int, args: dict) -> str:
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)