All checks were successful
Deploy Noval to K3s / deploy (push) Successful in 54s
- **子 Agent 支持**: - 新增 `dispatch_subagent` 工具,支持分配特定任务给子 Agent,由其独立操作。 - 前端新增 `ChatSubAgent` 组件,支持子 Agent 任务状态展示(运行中/已完成)、工具调用进度及输出预览等。 - **技能管理扩展**: - 新增 Skill 编辑器页面(`SkillEditorView.vue`),支持创建、编辑、删除技能。 - 完善技能工具权限分组选择功能,按类型灵活分配工具。 - **工具增强**: - 新增工具接口:`get_outline_detail`、`get_character_detail`、`get_chapter_detail`,用于获取详情内容。 - 优化部分工具返回数据,仅输出基础信息以减少上下文体积。 - 后端添加 `SPECIAL_TOOLS` 支持,增强 `dispatch_subagent` 的特殊处理能力。 同时优化部分前后端交互,提升代码维护性与可读性。
761 lines
28 KiB
Python
761 lines
28 KiB
Python
"""工具注册表和执行器 —— 小说信息、大纲 & 世界观 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、排序和摘要标题,不含详情)。如需查看某节点的详情内容,请使用 get_outline_detail 工具。",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {},
|
||
},
|
||
},
|
||
{
|
||
"name": "get_outline_detail",
|
||
"description": "查看指定大纲节点的完整详情内容",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"outline_id": {"type": "integer", "description": "大纲节点 ID"},
|
||
},
|
||
"required": ["outline_id"],
|
||
},
|
||
},
|
||
{
|
||
"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 和名称,不含描述详情)。如需查看某角色的完整描述,请使用 get_character_detail 工具。",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {},
|
||
},
|
||
},
|
||
{
|
||
"name": "get_character_detail",
|
||
"description": "查看指定角色的完整描述信息",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"character_id": {"type": "integer", "description": "角色 ID"},
|
||
},
|
||
"required": ["character_id"],
|
||
},
|
||
},
|
||
{
|
||
"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、排序和标题,不含正文)。如需查看某章节的完整正文,请使用 get_chapter_detail 工具。",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {},
|
||
},
|
||
},
|
||
{
|
||
"name": "get_chapter_detail",
|
||
"description": "查看指定章节的完整正文内容",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"chapter_id": {"type": "integer", "description": "章节 ID"},
|
||
},
|
||
"required": ["chapter_id"],
|
||
},
|
||
},
|
||
{
|
||
"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"],
|
||
},
|
||
},
|
||
{
|
||
"name": "dispatch_subagent",
|
||
"description": "派遣一个子 Agent 执行特定任务。子 Agent 拥有独立的系统提示词和工具集(由 Skill 定义),适合将复杂任务拆分为子任务分别处理。",
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"skill_id": {"type": "integer", "description": "要使用的 Skill ID(从可用 Skill 列表中选择)"},
|
||
"task": {"type": "string", "description": "交给子 Agent 的具体任务描述"},
|
||
},
|
||
"required": ["skill_id", "task"],
|
||
},
|
||
},
|
||
]
|
||
|
||
# dispatch_subagent 不由 execute_tool 处理,它在 chat.py 中特殊处理
|
||
SPECIAL_TOOLS = {"dispatch_subagent"}
|
||
|
||
|
||
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,
|
||
"get_outline_detail": _get_outline_detail,
|
||
"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,
|
||
"get_character_detail": _get_character_detail,
|
||
"create_character": _create_character,
|
||
"update_character": _update_character,
|
||
"delete_character": _delete_character,
|
||
"list_chapters": _list_chapters,
|
||
"get_chapter_detail": _get_chapter_detail,
|
||
"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:
|
||
"""列表仅返回 id + 排序 + 摘要标题(不含 detail),节省上下文。"""
|
||
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()
|
||
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: dict = {"id": o.id, "sort_order": o.sort_order, "summary": o.summary}
|
||
kids = children_map.get(o.id, [])
|
||
if kids:
|
||
node["children"] = [
|
||
{"id": c.id, "sort_order": c.sort_order, "summary": c.summary, "parent_id": c.parent_id}
|
||
for c in kids
|
||
]
|
||
result.append(node)
|
||
return json.dumps({"outlines": result, "total": len(result)}, ensure_ascii=False)
|
||
|
||
|
||
def _get_outline_detail(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)
|
||
return json.dumps({
|
||
"id": outline.id, "sort_order": outline.sort_order,
|
||
"summary": outline.summary, "detail": outline.detail,
|
||
"parent_id": outline.parent_id,
|
||
}, 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:
|
||
"""列表仅返回 id + 名称(不含 description),节省上下文。"""
|
||
items = session.exec(
|
||
select(Character)
|
||
.where(Character.novel_id == novel_id)
|
||
.order_by(Character.created_at)
|
||
).all()
|
||
result = [{"id": c.id, "name": c.name} for c in items]
|
||
return json.dumps({"characters": result, "total": len(result)}, ensure_ascii=False)
|
||
|
||
|
||
def _get_character_detail(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)
|
||
return json.dumps({
|
||
"id": char.id, "name": char.name, "description": char.description,
|
||
}, 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:
|
||
"""列表仅返回 id + 排序 + 标题(不含正文),节省上下文。"""
|
||
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} for c in items]
|
||
return json.dumps({"chapters": result, "total": len(result)}, ensure_ascii=False)
|
||
|
||
|
||
def _get_chapter_detail(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)
|
||
return json.dumps({
|
||
"id": chapter.id, "sort_order": chapter.sort_order,
|
||
"title": chapter.title, "content": chapter.content,
|
||
}, 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)
|