W10 作品骨架与人工正文版本:作品/章节骨架、目录版本与人工正文版本链。

按 R2 串行阶段整理提交;包内文件为该阶段交付(含后续小增量),状态以工作包清单为准。
This commit is contained in:
zizi 2026-09-10 19:25:41 +08:00
parent a6f4f14a81
commit 8353d11d40
25 changed files with 3949 additions and 0 deletions

View File

@ -0,0 +1 @@
"""作品规划模块;公开接口见接口.py。"""

View File

@ -0,0 +1,141 @@
"""作品档案经S04校验与S01提交;业务实例值仅由B01写入。"""
from dataclasses import asdict, dataclass
from muse.作品规划.存储 import 作品存储
from muse.作品规划.模型 import 作品错误, 档案保存, 档案结构
from muse.元数据.接口 import (
元数据服务,
字段变更,
投影字段,
授权摘要,
校验字段变更,
选择实例维护者,
)
from muse.正式变更.接口 import (
依赖引用,
修改影响,
参与操作,
固定哈希,
审阅内容,
提交计划,
)
def 核对作者(身份) -> None:
if not 身份.允许写正式内容:
raise 作品错误("SCOPE_DENIED", "作品档案只供当前已认证作者操作")
def 档案授权(作者: str, work_id: str) -> 授权摘要:
return 授权摘要(固定哈希({"author": 作者}), 固定哈希({"work": work_id, "owner": 作者}))
def 读取档案结构(
元数据: 元数据服务,
作者: str,
work_id: str,
选择: 档案结构,
*,
字段用途名: str = "userEditable",
):
结构 = 元数据.读取有效结构(
选择.schema_id, 选择.base_version, work_id=work_id, extension_version=选择.extension_version
)
类型 = 元数据.读取类型(结构.基础.type_id)
家族 = next((f for f in ("work", "planning") if f in 类型.实例族), None)
if 类型.实例粒度 != "work" or 家族 is None or 选择实例维护者(类型, 家族) != "B01":
raise 作品错误("SCOPE_DENIED", "档案结构必须属于作品级的B01生命周期")
投影 = 投影字段(
结构,
元数据.当前策略(类型.type_id),
档案授权(作者, work_id),
字段用途名=字段用途名,
内容用途="planning",
运行用途="production",
)
return 结构, 投影
class 档案入口:
身份 = "B01.save_profile"
请求类型 = 档案保存
def 查看(self, 连, 身份, 命令):
核对作者(身份)
请求 = 命令.业务请求
if 请求.target_ref != 命令.目标:
raise 作品错误("SCOPE_DENIED", "作品请求与目标不一致")
行 = 作品存储(连).读取(请求.work_id)
if 行 is not None and 行["author_id"] != 身份.作者:
raise 作品错误("SCOPE_DENIED", "作品不属于当前作者")
结构, 投影 = 读取档案结构(元数据服务(连), 身份.作者, 请求.work_id, 请求.schema)
版本 = 行["current_revision"] if 行 else 0
return 审阅内容(
请求.target_ref,
版本,
None,
None,
None,
固定哈希(
[行["content"] if 行 else {}, 请求.content, [asdict(v) for v in 请求.changes]]
),
结构.effective_schema_hash,
投影.projection_version,
固定哈希({"owner": 身份.作者, "work": 请求.work_id}),
("profile",),
(依赖引用("B01.work", 请求.target_ref, str(版本)),),
)
def 准备(self, 连, 身份, 命令, 当前):
return 提交计划(
(参与操作("B01.write_profile", 档案写入(身份.作者, 命令.业务请求)),),
(修改影响(命令.目标, "profile_changed", str(当前.数据版本), str(当前.数据版本 + 1)),),
)
@dataclass(frozen=True)
class 档案写入:
作者: str
请求: 档案保存
class 档案参与者:
身份 = "B01.write_profile"
请求类型 = 档案写入
def 提交(self, 连, 操作):
请求 = 操作.请求
元数据 = 元数据服务(连)
原 = 作品存储(连).读取(请求.work_id)
结构, 投影 = 读取档案结构(元数据, 操作.作者, 请求.work_id, 请求.schema)
if 原 is None:
绑定 = 元数据.建立初始绑定(请求.target_ref, 请求.work_id, 结构)
else:
绑定 = 元数据.读取绑定(请求.target_ref)
结构, 投影 = 元数据.保护提交(
请求.target_ref,
请求.schema.effective_schema_hash,
请求.schema.projection_version,
档案授权(操作.作者, 请求.work_id),
字段用途名="userEditable",
内容用途="planning",
运行用途="production",
)
修改 = (*tuple(字段变更((k,), "set", v) for k, v in 请求.content.items()), *请求.changes)
完整 = 校验字段变更(结构, 投影, 原["content"] if 原 else {}, 修改)
if 完整.引用需求:
raise 作品错误("PRECONDITION_PENDING", "本档案引用需要对应业务维护者解析后才能保存")
版本 = 作品存储(连).写档案(
请求.work_id, 操作.作者, 请求.expected_revision, 完整.值, 绑定, 固定哈希(完整.值)
)
return {"work_id": 请求.work_id, "revision": 版本, "schema_binding": asdict(绑定)}
class 作品依赖:
种类 = "B01.work"
def 锁定当前(self, 连, 对象ID):
if not 对象ID.startswith("work:"):
raise 作品错误("SCOPE_DENIED", "不是作品目标身份")
return str(作品存储(连).锁作品(对象ID.removeprefix("work:")))

View File

@ -0,0 +1,362 @@
"""B01所属SQL;当前指针、版本和目录写入均沿用调用方事务。"""
import hashlib
import json
from dataclasses import asdict
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb
from muse.作品规划.模型 import 作品错误
from muse.元数据.接口 import 结构绑定
def 解析规划载荷(值) -> dict:
"""归一化规划载荷:文本投影的JSON字符串转回对象;非法载荷失败关闭,不猜测。"""
if isinstance(值, str):
try:
值 = json.loads(值)
except json.JSONDecodeError as exc:
raise 作品错误("INVALID_OUTPUT", "规划载荷不是合法JSON") from exc
if not isinstance(值, dict):
raise 作品错误("INVALID_OUTPUT", "规划载荷必须是对象")
return 值
class 作品存储:
def __init__(self, 连):
self.连 = 连
def 读取探索选定(self, work_id: str) -> dict | None:
return (
self.连.cursor(row_factory=dict_row)
.execute(
"SELECT v.work_id,v.revision,v.snapshot,v.selected_by,v.selected_at::text "
"FROM muse_exploration_selection s JOIN muse_exploration_selection_version v "
"ON v.work_id=s.work_id AND v.revision=s.current_revision WHERE s.work_id=%s",
(work_id,),
)
.fetchone()
)
def 锁探索选定(self, work_id: str) -> int:
key = int.from_bytes(
hashlib.sha256(("B01:exploration-selection:" + work_id).encode()).digest()[:8],
signed=True,
)
self.连.execute("SELECT pg_advisory_xact_lock(%s)", (key,))
row = self.连.execute(
"SELECT current_revision FROM muse_exploration_selection WHERE work_id=%s FOR UPDATE",
(work_id,),
).fetchone()
return row[0] if row else 0
def 写探索选定(self, 作者: str, 请求, 快照: dict) -> dict:
version = 请求.expected_revision + 1
if 请求.expected_revision == 0:
self.连.execute(
"INSERT INTO muse_exploration_selection(work_id,current_revision) VALUES (%s,%s)",
(请求.work_id, version),
)
else:
row = self.连.execute(
"UPDATE muse_exploration_selection SET current_revision=%s "
"WHERE work_id=%s AND current_revision=%s RETURNING work_id",
(version, 请求.work_id, 请求.expected_revision),
).fetchone()
if row is None:
raise 作品错误("REVISION_CONFLICT", "作品已选定另一份探索版本")
self.连.execute(
"INSERT INTO muse_exploration_selection_version"
"(work_id,revision,snapshot,selected_by) VALUES (%s,%s,%s,%s)",
(请求.work_id, version, Jsonb(快照), 作者),
)
return {
"work_id": 请求.work_id,
"revision": version,
"draft_id": 快照["draft_id"],
"draft_hash": 快照["draft_hash"],
}
def 列出作品(self, 作者: str) -> list[str]:
return [
行[0]
for 行 in self.连.execute(
"SELECT work_id FROM muse_work WHERE author_id=%s ORDER BY work_id", (作者,)
).fetchall()
]
def 读取(self, work_id: str) -> dict | None:
return (
self.连.cursor(row_factory=dict_row)
.execute(
"SELECT w.*,v.content,v.schema_binding,v.content_hash "
"FROM muse_work w JOIN muse_work_version v ON v.work_id=w.work_id "
"AND v.revision=w.current_revision WHERE w.work_id=%s",
(work_id,),
)
.fetchone()
)
def 锁作品(self, work_id: str) -> int:
# 新作品尚无行;所有B01当前指针更新也使用此事务锁,关闭首次创建竞争窗口。
key = int.from_bytes(
hashlib.sha256(("B01:work:" + work_id).encode()).digest()[:8], signed=True
)
self.连.execute("SELECT pg_advisory_xact_lock(%s)", (key,))
行 = self.连.execute(
"SELECT current_revision FROM muse_work WHERE work_id=%s FOR UPDATE", (work_id,)
).fetchone()
return 行[0] if 行 else 0
def 写档案(
self, work_id: str, 作者: str, 预期版本: int, 内容: dict, 绑定: 结构绑定, 哈希: str
) -> int:
新版 = 预期版本 + 1
if 预期版本 == 0:
self.连.execute(
"INSERT INTO muse_work(work_id,author_id,current_revision) VALUES (%s,%s,%s)",
(work_id, 作者, 新版),
)
self.连.execute(
"INSERT INTO muse_directory_version(work_id,revision,chapters) VALUES (%s,0,'[]')",
(work_id,),
)
else:
行 = self.连.execute(
"UPDATE muse_work SET current_revision=%s WHERE work_id=%s "
"AND author_id=%s AND current_revision=%s RETURNING work_id",
(新版, work_id, 作者, 预期版本),
).fetchone()
if 行 is None:
raise 作品错误("REVISION_CONFLICT", "作品档案已被其他保存更新")
self.连.execute(
"INSERT INTO muse_work_version(work_id,revision,content,schema_binding,content_hash) "
"VALUES (%s,%s,%s,%s,%s)",
(work_id, 新版, Jsonb(内容), Jsonb(asdict(绑定)), 哈希),
)
return 新版
def 读取目录(self, work_id: str) -> dict:
行 = self.连.execute(
"SELECT v.revision,v.chapters FROM muse_work w JOIN muse_directory_version v "
"ON v.work_id=w.work_id AND v.revision=w.directory_revision WHERE w.work_id=%s",
(work_id,),
).fetchone()
if 行 is None:
raise 作品错误("WORK_NOT_FOUND", "作品目录不存在")
return {"work_id": work_id, "revision": 行[0], "chapters": 行[1]}
def 读取节点目录(self, work_id: str, 版本: int | None = None) -> dict:
row = self.连.execute(
"SELECT v.revision,v.nodes,v.chapters FROM muse_work w JOIN muse_directory_version v "
"ON v.work_id=w.work_id AND v.revision=COALESCE(%s,w.directory_revision) "
"WHERE w.work_id=%s",
(版本, work_id),
).fetchone()
if row is None:
raise 作品错误("WORK_NOT_FOUND", "该作品目录版本不存在")
# 早期单层目录也解释为同一组稳定章节点;不把旧正文迁入规划。
nodes = row[1] or [
{
"node_id": c["chapter_id"],
"kind": "chapter",
"title": c["title"],
"parent_id": None,
"position": c["position"],
}
for c in row[2]
]
return {"work_id": work_id, "revision": row[0], "nodes": nodes}
def 写完整目录(
self, work_id: str, 预期版本: int, nodes: list[dict], chapters: list[dict]
) -> dict:
for chapter in chapters:
updated = self.连.execute(
"INSERT INTO muse_chapter(chapter_id,work_id,title,position) VALUES (%s,%s,%s,%s) "
"ON CONFLICT(chapter_id) DO UPDATE "
"SET title=EXCLUDED.title,position=EXCLUDED.position "
"WHERE muse_chapter.work_id=EXCLUDED.work_id RETURNING chapter_id",
(chapter["chapter_id"], work_id, chapter["title"], chapter["position"]),
).fetchone()
if updated is None:
raise 作品错误("SCOPE_DENIED", "章节身份已属于其他作品")
新版 = 预期版本 + 1
self.连.execute(
"INSERT INTO muse_directory_version(work_id,revision,chapters,nodes) "
"VALUES (%s,%s,%s,%s)",
(work_id, 新版, Jsonb(chapters), Jsonb(nodes)),
)
row = self.连.execute(
"UPDATE muse_work SET directory_revision=%s WHERE work_id=%s AND directory_revision=%s "
"RETURNING work_id",
(新版, work_id, 预期版本),
).fetchone()
if row is None:
raise 作品错误("REVISION_CONFLICT", "目录已改变")
return {
"work_id": work_id,
"revision": 新版,
"chapter_ids": [v["chapter_id"] for v in chapters],
}
def 读取章节归属(self, chapter_id: str) -> dict | None:
return (
self.连.cursor(row_factory=dict_row)
.execute(
"SELECT c.chapter_id,c.work_id,c.title,c.position,w.author_id,w.directory_revision "
"FROM muse_chapter c JOIN muse_work w USING(work_id) WHERE c.chapter_id=%s",
(chapter_id,),
)
.fetchone()
)
class 规划存储:
def __init__(self, 连):
self.连 = 连
def 列出候选(self, work_id: str) -> list[str]:
return [
str(r[0])
for r in self.连.execute(
"SELECT candidate_id FROM muse_plan_candidate "
"WHERE work_id=%s ORDER BY candidate_id",
(work_id,),
).fetchall()
]
def 查章规划(self, work_id: str, chapter_id: str, type_id: str):
row = self.连.execute(
"SELECT plan_id FROM muse_plan WHERE work_id=%s AND chapter_id=%s AND type_id=%s",
(work_id, chapter_id, type_id),
).fetchone()
return self.读取(row[0]) if row else None
def 读取(self, plan_id: str, 版本: int | None = None):
return (
self.连.cursor(row_factory=dict_row)
.execute(
"SELECT p.*,v.revision,v.request,v.content,v.content_hash,v.schema_binding,"
"v.source_dependencies FROM muse_plan p JOIN muse_plan_version v "
"ON v.plan_id=p.plan_id AND v.revision=COALESCE(%s,p.current_revision) "
"WHERE p.plan_id=%s",
(版本, plan_id),
)
.fetchone()
)
def 读取候选(self, candidate_id: str, 版本: int | None = None):
return (
self.连.cursor(row_factory=dict_row)
.execute(
"SELECT p.*,v.revision,v.request,v.content,v.source_dependencies,v.candidate_hash "
"FROM muse_plan_candidate p JOIN muse_plan_candidate_version v "
"ON v.candidate_id=p.candidate_id AND v.revision=COALESCE(%s,p.current_revision) "
"WHERE p.candidate_id=%s",
(版本, candidate_id),
)
.fetchone()
)
def 列出(self, work_id: str) -> list[str]:
return [
r[0]
for r in self.连.execute(
"SELECT plan_id FROM muse_plan WHERE work_id=%s ORDER BY plan_id", (work_id,)
).fetchall()
]
def 锁规划(self, plan_id: str) -> int:
key = int.from_bytes(
hashlib.sha256(("B01:plan:" + plan_id).encode()).digest()[:8], signed=True
)
self.连.execute("SELECT pg_advisory_xact_lock(%s)", (key,))
r = self.连.execute(
"SELECT current_revision FROM muse_plan WHERE plan_id=%s FOR UPDATE", (plan_id,)
).fetchone()
return r[0] if r else 0
def 锁候选(self, candidate_id: str) -> int:
key = int.from_bytes(
hashlib.sha256(("B01:plan-candidate:" + candidate_id).encode()).digest()[:8],
signed=True,
)
self.连.execute("SELECT pg_advisory_xact_lock(%s)", (key,))
r = self.连.execute(
"SELECT current_revision FROM muse_plan_candidate WHERE candidate_id=%s FOR UPDATE",
(candidate_id,),
).fetchone()
return r[0] if r else 0
def 写候选(self, candidate_id: str, plan_id: str, 请求, 内容, 来源, 哈希, 预期版本: int = 0):
revision = 预期版本 + 1
if 预期版本 == 0:
self.连.execute(
"INSERT INTO muse_plan_candidate(candidate_id,work_id,plan_id,current_revision) "
"VALUES (%s,%s,%s,%s)",
(candidate_id, 请求.work_id, plan_id, revision),
)
else:
changed = self.连.execute(
"UPDATE muse_plan_candidate SET current_revision=%s WHERE candidate_id=%s "
"AND work_id=%s AND plan_id=%s AND current_revision=%s RETURNING candidate_id",
(revision, candidate_id, 请求.work_id, plan_id, 预期版本),
).fetchone()
if changed is None:
raise 作品错误("REVISION_CONFLICT", "规划候选已被编辑")
self.连.execute(
"INSERT INTO muse_plan_candidate_version"
"(candidate_id,revision,request,content,source_dependencies,candidate_hash) "
"VALUES (%s,%s,%s,%s,%s,%s)",
(
candidate_id,
revision,
Jsonb(asdict(请求)),
Jsonb(内容),
Jsonb([asdict(s) for s in 来源]),
哈希,
),
)
return {
"candidate_id": candidate_id,
"plan_id": plan_id,
"revision": revision,
"candidate_hash": 哈希,
}
def 写规划(self, plan_id, 请求, 类型ID, 内容, 绑定, 来源, 作者, 哈希, 章节ID=None):
revision = 请求.expected_revision + 1
if 请求.expected_revision == 0:
self.连.execute(
"INSERT INTO muse_plan"
"(plan_id,work_id,type_id,chapter_id,node_id,current_revision) "
"VALUES (%s,%s,%s,%s,%s,%s)",
(plan_id, 请求.work_id, 类型ID, 章节ID, 请求.node_id or 请求.chapter_id, revision),
)
else:
r = self.连.execute(
"UPDATE muse_plan SET current_revision=%s,chapter_id=%s WHERE plan_id=%s "
"AND current_revision=%s RETURNING plan_id",
(revision, 章节ID, plan_id, 请求.expected_revision),
).fetchone()
if r is None:
raise 作品错误("REVISION_CONFLICT", "规划正式版本已改变")
self.连.execute(
"INSERT INTO muse_plan_version"
"(plan_id,revision,request,content,content_hash,schema_binding,"
"source_dependencies,author_id) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s)",
(
plan_id,
revision,
Jsonb(asdict(请求)),
Jsonb(内容),
哈希,
Jsonb(asdict(绑定)),
Jsonb([asdict(s) for s in 来源]),
作者,
),
)
return {"plan_id": plan_id, "revision": revision, "content_hash": 哈希}

View File

@ -0,0 +1,287 @@
"""作品身份、动态档案及目录的唯一公开入口。"""
from dataclasses import asdict
from muse.作品规划.作品档案 import (
作品依赖,
核对作者,
档案入口,
档案参与者,
档案授权,
读取档案结构,
)
from muse.作品规划.存储 import 作品存储
from muse.作品规划.探索交接 import 探索选定依赖, 探索选定入口, 探索选定参与者, 校验交付格式
from muse.作品规划.探索草稿 import 探索服务
from muse.作品规划.模型 import (
作品错误,
探索选择,
档案保存,
档案结构,
目录修改,
目录节点,
节点目录保存,
规划保存,
规划定位,
)
from muse.作品规划.目录调整 import 目录依赖, 目录入口, 目录参与者, 节点目录入口, 读取作者目录
from muse.作品规划.规划版本 import 登记规划参与者, 规划服务
from muse.作品规划.进度查询 import 创作进度服务
from muse.元数据.接口 import 元数据服务, 投影值, 投影字段, 生成结构合同
from muse.基础设施.数据库.连接 import 数据库工厂
from muse.正式变更.接口 import 作者动作, 参与者目录, 变更命令, 正式变更服务
def 登记作品参与者(目录: 参与者目录, 探索: 探索服务 | None = None, 规划引用解析=None) -> None:
目录.登记入口(档案入口())
目录.登记参与者(档案参与者())
目录.登记依赖(作品依赖())
目录.登记入口(目录入口())
目录.登记入口(节点目录入口())
目录.登记参与者(目录参与者())
目录.登记依赖(目录依赖())
目录.登记入口(探索选定入口(探索))
目录.登记参与者(探索选定参与者())
目录.登记依赖(探索选定依赖())
登记规划参与者(目录, 规划引用解析)
class 作品服务:
def __init__(self, 数据库: 数据库工厂, 正式: 正式变更服务):
self.数据库, self.正式 = 数据库, 正式
def 读取节点目录(self, 身份, work_id: str, *, 版本: int | None = None) -> dict:
核对作者(身份)
with self.数据库.连接(只读=True) as 连:
读取作者目录(连, 身份.作者, work_id)
return 作品存储(连).读取节点目录(work_id, 版本)
def 保存节点目录(self, 身份, 命令ID: str, 请求: 节点目录保存) -> dict:
return self.正式.提交变更(
身份,
变更命令(
命令ID,
"B01.change_plan_nodes",
请求.target_ref,
作者动作.人工保存,
请求.expected_revision,
请求,
),
)
def 选定探索稿(self, 身份, 命令ID: str, 请求: 探索选择) -> dict:
return self.正式.提交变更(
身份,
变更命令(
命令ID,
"B01.select_exploration",
请求.target_ref,
作者动作.人工保存,
请求.expected_revision,
请求,
),
)
def 读取探索选定(self, 身份, work_id: str) -> dict:
核对作者(身份)
with self.数据库.连接(只读=True) as 连:
读取作者目录(连, 身份.作者, work_id)
当前 = 作品存储(连).读取探索选定(work_id)
return {
"work_id": work_id,
"revision": 当前["revision"] if 当前 else 0,
"selection": 当前,
}
def 列出作品(self, 身份) -> list[dict]:
核对作者(身份)
with self.数据库.连接(只读=True) as 连:
ids = 作品存储(连).列出作品(身份.作者)
return [self.读取作品(身份, id_) for id_ in ids]
def 读取表单(
self, 身份, work_id: str, *, schema_id: str | None = None, base_version: int | None = None
) -> dict:
核对作者(身份)
with self.数据库.连接(只读=True) as 连:
行 = 作品存储(连).读取(work_id)
元数据 = 元数据服务(连)
if 行 is not None:
if 行["author_id"] != 身份.作者:
raise 作品错误("SCOPE_DENIED", "作品不属于当前作者")
绑定 = 元数据.读取绑定("work:" + work_id)
选择 = 档案结构(
绑定.schema_id,
绑定.base_version,
绑定.effective_schema_hash,
"",
绑定.extension_version,
)
elif schema_id is not None and base_version is not None:
选择 = 档案结构(schema_id, base_version, "", "")
else:
raise 作品错误("WORK_NOT_FOUND", "作品不存在;新建时需要选择档案结构")
结构, 编辑投影 = 读取档案结构(元数据, 身份.作者, work_id, 选择)
_, 可见投影 = 读取档案结构(元数据, 身份.作者, work_id, 选择, 字段用途名="uiVisible")
return {
"work_id": work_id,
"revision": 行["current_revision"] if 行 else 0,
"content": 投影值(可见投影, 行["content"]) if 行 else {},
"schema_binding": asdict(
档案结构(
选择.schema_id,
选择.base_version,
结构.effective_schema_hash,
编辑投影.projection_version,
选择.extension_version,
)
),
"visible_schema": 生成结构合同(可见投影),
"editable_schema": 生成结构合同(编辑投影),
"visible_projection_version": 可见投影.projection_version,
}
def 新档案结构(
self, 身份, work_id: str, schema_id: str, 版本: int, *, 扩展版本: int | None = None
) -> 档案结构:
核对作者(身份)
with self.数据库.连接(只读=True) as 连:
行 = 作品存储(连).读取(work_id)
if 行 is not None and 行["author_id"] != 身份.作者:
raise 作品错误("SCOPE_DENIED", "作品不属于当前作者")
选择 = 档案结构(schema_id, 版本, "", "", 扩展版本)
结构, 投影 = 读取档案结构(元数据服务(连), 身份.作者, work_id, 选择)
return 档案结构(
schema_id, 版本, 结构.effective_schema_hash, 投影.projection_version, 扩展版本
)
def 保存档案(self, 身份, 命令ID: str, 请求: 档案保存) -> dict:
核对作者(身份)
return self.正式.提交变更(
身份,
变更命令(
命令ID,
"B01.save_profile",
请求.target_ref,
作者动作.人工保存,
请求.expected_revision,
请求,
),
)
def 读取作品(self, 身份, work_id: str) -> dict:
核对作者(身份)
with self.数据库.连接(只读=True) as 连:
行 = 作品存储(连).读取(work_id)
if 行 is None:
raise 作品错误("WORK_NOT_FOUND", "作品不存在")
if 行["author_id"] != 身份.作者:
raise 作品错误("SCOPE_DENIED", "作品不属于当前作者")
binding = asdict(元数据服务(连).读取绑定("work:" + work_id))
结构, 投影 = 读取档案结构(
元数据服务(连),
身份.作者,
work_id,
档案结构(
binding["schema_id"],
binding["base_version"],
binding["effective_schema_hash"],
"",
binding["extension_version"],
),
)
可见投影 = 投影字段(
结构,
元数据服务(连).当前策略(结构.基础.type_id),
档案授权(身份.作者, work_id),
字段用途名="uiVisible",
内容用途="planning",
运行用途="production",
)
return {
"work_id": work_id,
"revision": 行["current_revision"],
"content": 投影值(可见投影, 行["content"]),
"schema_binding": binding,
"projection_version": 投影.projection_version,
"visible_projection_version": 可见投影.projection_version,
}
def 添加章节(
self,
身份,
命令ID: str,
work_id: str,
chapter_id: str,
标题: str,
*,
预期目录版本: int,
) -> dict:
return self._提交目录(
身份, 命令ID, 目录修改(work_id, 预期目录版本, "append", (chapter_id,), 标题)
)
def 重排章节(
self,
身份,
命令ID: str,
work_id: str,
章节顺序: tuple[str, ...],
*,
预期目录版本: int,
) -> dict:
return self._提交目录(身份, 命令ID, 目录修改(work_id, 预期目录版本, "reorder", 章节顺序))
def _提交目录(self, 身份, 命令ID: str, 请求: 目录修改) -> dict:
核对作者(身份)
return self.正式.提交变更(
身份,
变更命令(
命令ID,
"B01.change_directory",
请求.target_ref,
作者动作.人工保存,
请求.expected_revision,
请求,
),
)
def 读取目录(self, 身份, work_id: str) -> dict:
核对作者(身份)
with self.数据库.连接(只读=True) as 连:
return 读取作者目录(连, 身份.作者, work_id)
def 读取章节归属(连, 作者: str, chapter_id: str) -> dict:
"""同事务参与者读取真实章节范围;不允许B05直接读B01表。"""
行 = 作品存储(连).读取章节归属(chapter_id)
if 行 is None:
raise 作品错误("CHAPTER_NOT_FOUND", "章节不存在")
if 行["author_id"] != 作者:
raise 作品错误("SCOPE_DENIED", "章节不属于当前作者")
return {k: v for k, v in 行.items() if k != "author_id"}
def 读取作品范围(连, 作者: str, work_id: str) -> dict:
"""跨模块同事务读取目录与作者归属,不暴露作品私表。"""
return 读取作者目录(连, 作者, work_id)
__all__ = [
"作品服务",
"档案保存",
"档案结构",
"作品错误",
"登记作品参与者",
"读取章节归属",
"创作进度服务",
"读取作品范围",
"探索服务",
"校验交付格式",
"探索选择",
"规划保存",
"规划定位",
"规划服务",
"目录节点",
"节点目录保存",
]

View File

@ -0,0 +1,143 @@
"""作品身份、目录与动态档案请求;内容字段由S04描述。"""
from typing import Any, Literal
from pydantic import ConfigDict
from pydantic.dataclasses import dataclass
from muse.元数据.接口 import 字段变更
from muse.共享.错误 import Muse错误
class 作品错误(Muse错误):
def __init__(self, 代码: str, 说明: str):
super().__init__(说明)
self.错误码 = 代码
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 档案结构:
schema_id: str
base_version: int
effective_schema_hash: str
projection_version: str
extension_version: int | None = None
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 档案保存:
work_id: str
expected_revision: int
content: dict[str, Any]
schema: 档案结构
changes: tuple[字段变更, ...] = ()
def __post_init__(self):
if not self.work_id or self.expected_revision < 0:
raise 作品错误("INVALID_REQUEST", "档案保存需要稳定作品身份与预期版本")
@property
def target_ref(self) -> str:
return "work:" + self.work_id
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 目录修改:
work_id: str
expected_revision: int
operation: str
chapter_ids: tuple[str, ...]
new_title: str = ""
def __post_init__(self):
if self.operation not in {"append", "reorder"} or self.expected_revision < 0:
raise 作品错误("INVALID_REQUEST", "目录操作或预期版本无效")
if len(set(self.chapter_ids)) != len(self.chapter_ids) or any(
not id_ for id_ in self.chapter_ids
):
raise 作品错误("INVALID_REQUEST", "章节身份不能为空或重复")
if self.operation == "append" and len(self.chapter_ids) != 1:
raise 作品错误("INVALID_REQUEST", "新增章节需要一个稳定章节身份")
@property
def target_ref(self) -> str:
return "directory:" + self.work_id
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 探索选择:
work_id: str
draft_id: str
expected_hash: str
expected_revision: int
def __post_init__(self):
if (
not self.work_id
or not self.draft_id.strip()
or not self.expected_hash
or self.expected_revision < 0
):
raise 作品错误("INVALID_REQUEST", "探索交接需要作品、明确稿件、内容哈希和选定版本")
@property
def target_ref(self) -> str:
return "exploration-selection:" + self.work_id
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 目录节点:
node_id: str
kind: Literal["volume", "chapter", "scene"]
title: str
parent_id: str | None = None
def __post_init__(self):
if not self.node_id or not self.title.strip():
raise 作品错误("INVALID_REQUEST", "目录节点需要稳定身份和名称")
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 节点目录保存:
work_id: str
expected_revision: int
nodes: tuple[目录节点, ...]
def __post_init__(self):
if not self.work_id or self.expected_revision < 0:
raise 作品错误("INVALID_REQUEST", "节点目录需要作品身份和预期版本")
@property
def target_ref(self) -> str:
return "directory:" + self.work_id
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 规划保存:
work_id: str
expected_revision: int
schema: 档案结构
content: dict[str, Any]
chapter_id: str | None = None
selection_revision: int | None = None
changes: tuple[字段变更, ...] = ()
node_id: str | None = None
def __post_init__(self):
if not self.work_id or self.expected_revision < 0:
raise 作品错误("INVALID_REQUEST", "规划需要作品身份与预期版本")
if self.selection_revision is not None and self.selection_revision < 1:
raise 作品错误("INVALID_REQUEST", "采用探索来源时需要真实的已选定版本")
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 规划定位:
candidate_id: str
candidate_revision: int
candidate_hash: str
plan_id: str
expected_revision: int
@property
def target_ref(self) -> str:
return "plan:" + self.plan_id

View File

@ -0,0 +1,131 @@
"""稳定章节身份与有版本目录;重排不会碰正文版本。"""
from dataclasses import asdict, dataclass
from muse.作品规划.作品档案 import 核对作者
from muse.作品规划.存储 import 作品存储
from muse.作品规划.模型 import 作品错误, 目录修改, 目录节点, 节点目录保存
from muse.正式变更.接口 import 依赖引用, 修改影响, 参与操作, 固定哈希, 审阅内容, 提交计划
def 读取作者目录(连, 作者: str, work_id: str) -> dict:
当前 = 作品存储(连).读取(work_id)
if 当前 is None:
raise 作品错误("WORK_NOT_FOUND", "作品不存在")
if 当前["author_id"] != 作者:
raise 作品错误("SCOPE_DENIED", "作品不属于当前作者")
return 作品存储(连).读取目录(work_id)
@dataclass(frozen=True)
class 目录写入:
作者: str
请求: 目录修改 | 节点目录保存
def 组织目录(nodes: tuple[目录节点, ...], 原节点: list[dict]) -> tuple[list[dict], list[dict]]:
by_id = {n.node_id: n for n in nodes}
if len(by_id) != len(nodes):
raise 作品错误("INVALID_REQUEST", "目录节点身份不能重复")
for old in 原节点:
if old["node_id"] not in by_id or by_id[old["node_id"]].kind != old["kind"]:
raise 作品错误("SCOPE_DENIED", "目录保存必须保留已有节点身份与种类")
children: dict[str | None, list[目录节点]] = {}
for node in nodes:
parent = by_id.get(node.parent_id) if node.parent_id else None
allowed = {"volume": {None}, "chapter": {None, "volume"}, "scene": {"chapter"}}
if (node.parent_id is not None and parent is None) or (
parent.kind if parent else None
) not in allowed[node.kind]:
raise 作品错误("PLAN_SCOPE_INVALID", "卷属于作品,章属于作品或卷,场景属于章")
children.setdefault(node.parent_id, []).append(node)
ordered, chapters = [], []
def 展开(parent):
for position, node in enumerate(children.get(parent, []), 1):
ordered.append({**asdict(node), "position": position})
if node.kind == "chapter":
chapters.append(
{"chapter_id": node.node_id, "title": node.title, "position": len(chapters) + 1}
)
展开(node.node_id)
展开(None)
return ordered, chapters
class 目录入口:
身份 = "B01.change_directory"
请求类型: type[目录修改] | type[节点目录保存] = 目录修改
def 查看(self, 连, 身份, 命令):
核对作者(身份)
请求 = 命令.业务请求
if 请求.target_ref != 命令.目标:
raise 作品错误("SCOPE_DENIED", "目录目标不一致")
目录 = 读取作者目录(连, 身份.作者, 请求.work_id)
return 审阅内容(
请求.target_ref,
目录["revision"],
None,
None,
None,
固定哈希([目录, 固定哈希(请求)]),
None,
None,
固定哈希({"owner": 身份.作者, "work": 请求.work_id}),
("directory",),
(依赖引用("B01.directory", 请求.target_ref, str(目录["revision"])),),
)
def 准备(self, 连, 身份, 命令, 当前):
return 提交计划(
(参与操作("B01.write_directory", 目录写入(身份.作者, 命令.业务请求)),),
(修改影响(命令.目标, "directory_changed", str(当前.数据版本), str(当前.数据版本 + 1)),),
)
class 目录参与者:
身份 = "B01.write_directory"
请求类型 = 目录写入
def 提交(self, 连, 操作):
当前 = 读取作者目录(连, 操作.作者, 操作.请求.work_id)
if 当前["revision"] != 操作.请求.expected_revision:
raise 作品错误("REVISION_CONFLICT", "目录已改变")
store = 作品存储(连)
previous = store.读取节点目录(操作.请求.work_id)["nodes"]
req = 操作.请求
if isinstance(req, 节点目录保存):
nodes = req.nodes
else:
nodes = tuple(
目录节点(n["node_id"], n["kind"], n["title"], n["parent_id"]) for n in previous
)
if req.operation == "append":
nodes = (*nodes, 目录节点(req.chapter_ids[0], "chapter", req.new_title))
else:
if set(req.chapter_ids) != {c["chapter_id"] for c in 当前["chapters"]}:
raise 作品错误("REVISION_CONFLICT", "重排必须包含当前目录全部章节")
if any(n.kind == "volume" for n in nodes):
raise 作品错误("PLAN_SCOPE_INVALID", "分卷目录请在节点视图中重排")
rank = {id_: i for i, id_ in enumerate(req.chapter_ids)}
nodes = tuple(sorted(nodes, key=lambda n: rank.get(n.node_id, len(rank))))
ordered, chapters = 组织目录(nodes, previous)
return store.写完整目录(req.work_id, req.expected_revision, ordered, chapters)
class 节点目录入口(目录入口):
身份 = "B01.change_plan_nodes"
请求类型 = 节点目录保存
class 目录依赖:
种类 = "B01.directory"
def 锁定当前(self, 连, 对象ID):
if not 对象ID.startswith("directory:"):
raise 作品错误("SCOPE_DENIED", "不是作品目录身份")
work_id = 对象ID.removeprefix("directory:")
作品存储(连).锁作品(work_id)
return str(作品存储(连).读取目录(work_id)["revision"])

View File

@ -0,0 +1,54 @@
"""只读组装实际正文、候选与任务版本,不持久化第二份进度值。"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from muse.任务运行.接口 import 任务服务
from muse.作品规划.接口 import 作品服务
from muse.共享.调用身份 import 调用身份
from muse.正文写作.接口 import 正文服务
class 创作进度服务:
def __init__(self, 作品: 作品服务, 正文: 正文服务, 任务: 任务服务):
self.作品, self.正文, self.任务 = 作品, 正文, 任务
def 查询(self, 身份: 调用身份, work_id: str) -> dict:
# B01先验证作者与作品,B05及S02仍按自己的公开接口核对实际数据。
目录 = self.作品.读取目录(身份, work_id)
assert 身份.作者 is not None
章节 = []
for 章 in 目录["chapters"]:
文稿 = self.正文.读取正文(身份, 章["chapter_id"])
候选 = self.正文.列出候选(身份, 章["chapter_id"])
章节.append(
{
"chapter_id": 章["chapter_id"],
"title": 章["title"],
"position": 章["position"],
"document_revision": 文稿["revision"],
"document_hash": 文稿["document_hash"],
"visible_characters": len(文稿["visible_text"]),
"candidates": [
{k: v[k] for k in ("candidate_id", "revision", "decision", "stale")}
for v in 候选
],
}
)
任务 = self.任务.列出任务(身份.作者, 作品ID=work_id)
return {
"work_id": work_id,
"directory_revision": 目录["revision"],
"total_chapters": len(章节),
"saved_chapters": sum(章["document_revision"] > 0 for 章 in 章节),
"undecided_candidates": sum(
c["decision"] in {"undecided", "defer"} for 章 in 章节 for c in 章["candidates"]
),
"chapters": 章节,
"tasks": [
{"task_id": t.任务ID, "state": t.状态.value, "event_cursor": t.最后序号}
for t in 任务
],
}

View File

@ -0,0 +1,171 @@
"""作品、目录与人工正文的CLI适配;不写SQL,不替作者采纳模型候选。"""
from dataclasses import asdict
from pathlib import Path
from pydantic import TypeAdapter, ValidationError
from muse.作品规划.接口 import 规划定位
from muse.共享.错误 import 配置错误
from muse.接入.主会话 import 参数合同错误
from muse.接入.创作操作 import (
人工保存请求,
创作身份,
回退正文请求,
探索保存请求,
探索选定请求,
新增章节请求,
档案保存请求,
节点目录请求,
规划候选请求,
规划决定请求,
规划编辑请求,
重排章节请求,
)
from muse.正式变更.接口 import 作者动作
def 运行作品命令(
装配, 动作: str, 目标: str, *, 结构ID="work_core", 版本: int = 1, 分支="main"
) -> dict:
if 装配.配置.HTTP is None:
raise 配置错误("人工创作需要本地作者身份配置")
身份 = 创作身份(装配.配置.HTTP.作者ID, 装配.配置)
作品 = 装配.要求作品()
正文 = 装配.要求正文()
if 动作 == "结构":
return asdict(作品.新档案结构(身份, 目标, 结构ID, 版本))
if 动作 == "查看":
return 作品.读取作品(身份, 目标)
if 动作 == "目录":
return 作品.读取目录(身份, 目标)
if 动作 == "节点目录":
return 作品.读取节点目录(身份, 目标)
if 动作 == "进度":
return 装配.要求作品进度().查询(身份, 目标)
if 动作 == "正文":
return 正文.读取正文(身份, 目标, 分支=分支)
类型 = {
"保存档案": 档案保存请求,
"新增章节": 新增章节请求,
"重排章节": 重排章节请求,
"保存节点": 节点目录请求,
"保存正文": 人工保存请求,
"回退正文": 回退正文请求,
}[动作]
try:
请求 = 类型.model_validate_json(Path(目标).read_text(encoding="utf-8"))
except ValidationError as exc:
raise 参数合同错误(exc.errors(), 前缀=("body",)) from None
if isinstance(请求, 档案保存请求):
return 作品.保存档案(身份, 请求.command_id, 请求.业务请求())
if isinstance(请求, 节点目录请求):
return 作品.保存节点目录(身份, 请求.command_id, 请求.directory)
if isinstance(请求, 新增章节请求):
return 作品.添加章节(
身份,
请求.command_id,
请求.work_id,
请求.chapter_id,
请求.title,
预期目录版本=请求.expected_directory_revision,
)
if isinstance(请求, 重排章节请求):
return 作品.重排章节(
身份,
请求.command_id,
请求.work_id,
tuple(请求.chapter_ids),
预期目录版本=请求.expected_directory_revision,
)
if isinstance(请求, 人工保存请求):
return 正文.保存人工(
身份,
请求.command_id,
请求.chapter_id,
请求.expected_revision,
请求.document,
分支=请求.branch_id,
)
return 正文.回退正文(
身份,
请求.command_id,
请求.chapter_id,
请求.expected_revision,
请求.restore_from,
分支=请求.branch_id,
)
def 运行探索命令(装配, 动作: str, 目标: str | None, *, 预期哈希: str | None = None):
if 装配.配置.HTTP is None:
raise 配置错误("私人探索需要本地作者身份配置")
身份 = 创作身份(装配.配置.HTTP.作者ID, 装配.配置)
if 动作 == "查看选定" and 目标:
return 装配.要求作品().读取探索选定(身份, 目标)
if 动作 == "列表":
return 装配.要求探索().列出(身份)
if not 目标:
raise 配置错误("当前探索动作需要稿件身份或保存请求文件")
if 动作 == "查看":
return 装配.要求探索().读取(身份, 目标)
if 动作 == "检查":
if not 预期哈希:
raise 配置错误("整理稿检查需要固定稿件哈希")
return 装配.要求探索().检查交付(身份, 目标, 预期哈希)
try:
请求类型 = 探索选定请求 if 动作 == "选定" else 探索保存请求
请求 = 请求类型.model_validate_json(Path(目标).read_text(encoding="utf-8"))
except (OSError, UnicodeError):
raise 配置错误("无法读取私人探索保存请求文件") from None
except ValidationError as exc:
raise 参数合同错误(exc.errors(), 前缀=("body",)) from None
if isinstance(请求, 探索选定请求):
return 装配.要求作品().选定探索稿(身份, 请求.command_id, 请求.selection)
return 装配.要求探索().保存(身份, 请求.draft_id, 请求.title, 请求.content, 请求.expected_hash)
def 运行规划命令(装配, 动作: str, 目标: str, *, 结构ID: str, 版本: int, 章节ID: str | None = None):
if 装配.配置.HTTP is None:
raise 配置错误("规划操作需要本地作者身份配置")
actor = 创作身份(装配.配置.HTTP.作者ID, 装配.配置)
service = 装配.要求规划()
if 动作 == "读取细纲":
if not 章节ID:
raise 配置错误("读取可用细纲需要明确章节身份")
return service.读取可用细纲(actor, 目标, 章节ID)
if 动作 == "结构":
return asdict(service.新规划结构(actor, 目标, 结构ID, 版本))
if 动作 == "列表":
return service.列出规划(actor, 目标)
if 动作 == "查看":
return service.读取规划(actor, 目标)
if 动作 == "查看候选":
return service.读取候选(actor, 目标)
类型 = {
"创建候选": 规划候选请求,
"编辑候选": 规划编辑请求,
"打开审阅": 规划定位,
"决定": 规划决定请求,
}[动作]
try:
req = TypeAdapter(类型).validate_json(Path(目标).read_text(encoding="utf-8"))
except (OSError, UnicodeError):
raise 配置错误("无法读取规划操作请求文件") from None
except ValidationError as exc:
raise 参数合同错误(exc.errors(), 前缀=("body",)) from None
if isinstance(req, 规划候选请求):
return service.保存人工候选(actor, req.command_id, req.request)
if isinstance(req, 规划编辑请求):
return service.编辑候选(actor, req.command_id, req.candidate, tuple(req.changes))
if isinstance(req, 规划定位):
return service.打开审阅(actor, req)
return service.决定(
actor,
req.command_id,
req.candidate,
作者动作(req.action),
req.author_review_id,
req.review_hash,
tuple(req.approved_changes),
)

View File

@ -0,0 +1,315 @@
"""作品与目录HTTP适配;身份来自会话,读写由B01拥有。"""
from fastapi import APIRouter, Query, Request
from muse.作品规划.接口 import 档案结构, 规划定位
from muse.接入.http.作者会话 import 作者依赖
from muse.接入.创作操作 import (
作品概要,
作品表单结果,
作品进度结果,
创作身份,
探索保存请求,
探索检查结果,
探索检查请求,
探索稿概要,
探索稿结果,
探索选定状态,
探索选定请求,
新增章节请求,
核对路径,
档案保存请求,
正式回执结果,
目录结果,
节点目录结果,
节点目录请求,
规划候选结果,
规划候选请求,
规划决定请求,
规划审阅结果,
规划结果,
规划编辑请求,
规划表单结果,
重排章节请求,
)
from muse.正式变更.接口 import 作者动作
路由 = APIRouter(prefix="/api/v1", tags=["作品规划"])
@路由.get(
"/works/{work_id}/plan-nodes", response_model=节点目录结果, operation_id="read_plan_nodes"
)
def 读取规划节点(
work_id: str, request: Request, 作者: 作者依赖, version: int | None = Query(default=None, ge=0)
):
app = request.app.state.装配
return app.要求作品().读取节点目录(创作身份(作者, app.配置), work_id, 版本=version)
@路由.post(
"/works/{work_id}/plan-nodes", response_model=正式回执结果, operation_id="save_plan_nodes"
)
def 保存规划节点(work_id: str, 请求: 节点目录请求, request: Request, 作者: 作者依赖):
核对路径(work_id, 请求.directory.work_id)
app = request.app.state.装配
return app.要求作品().保存节点目录(创作身份(作者, app.配置), 请求.command_id, 请求.directory)
@路由.get(
"/works/{work_id}/progress", response_model=作品进度结果, operation_id="read_work_progress"
)
def 读取进度(work_id: str, request: Request, 作者: 作者依赖) -> dict:
装配 = request.app.state.装配
return 装配.要求作品进度().查询(创作身份(作者, 装配.配置), work_id)
@路由.get("/works", response_model=list[作品概要], operation_id="list_works")
def 列出作品(request: Request, 作者: 作者依赖) -> list[dict]:
装配 = request.app.state.装配
return 装配.要求作品().列出作品(创作身份(作者, 装配.配置))
@路由.get("/works/{work_id}/form", response_model=作品表单结果, operation_id="read_work_form")
def 读取表单(
work_id: str,
request: Request,
作者: 作者依赖,
schema_id: str | None = None,
base_version: int | None = Query(default=None, ge=1),
) -> dict:
装配 = request.app.state.装配
return 装配.要求作品().读取表单(
创作身份(作者, 装配.配置),
work_id,
schema_id=schema_id,
base_version=base_version,
)
@路由.post(
"/works/{work_id}/profile", response_model=正式回执结果, operation_id="save_work_profile"
)
def 保存档案(work_id: str, 请求: 档案保存请求, request: Request, 作者: 作者依赖) -> dict:
核对路径(work_id, 请求.work_id)
装配 = request.app.state.装配
return 装配.要求作品().保存档案(创作身份(作者, 装配.配置), 请求.command_id, 请求.业务请求())
@路由.get("/works/{work_id}/chapters", response_model=目录结果, operation_id="read_work_chapters")
def 读取目录(work_id: str, request: Request, 作者: 作者依赖) -> dict:
装配 = request.app.state.装配
return 装配.要求作品().读取目录(创作身份(作者, 装配.配置), work_id)
@路由.post("/works/{work_id}/chapters", response_model=正式回执结果, operation_id="append_chapter")
def 新增章节(work_id: str, 请求: 新增章节请求, request: Request, 作者: 作者依赖) -> dict:
核对路径(work_id, 请求.work_id)
装配 = request.app.state.装配
return 装配.要求作品().添加章节(
创作身份(作者, 装配.配置),
请求.command_id,
work_id,
请求.chapter_id,
请求.title,
预期目录版本=请求.expected_directory_revision,
)
@路由.post(
"/works/{work_id}/chapters/reorder",
response_model=正式回执结果,
operation_id="reorder_chapters",
)
def 重排章节(work_id: str, 请求: 重排章节请求, request: Request, 作者: 作者依赖) -> dict:
核对路径(work_id, 请求.work_id)
装配 = request.app.state.装配
return 装配.要求作品().重排章节(
创作身份(作者, 装配.配置),
请求.command_id,
work_id,
tuple(请求.chapter_ids),
预期目录版本=请求.expected_directory_revision,
)
@路由.get("/explorations", response_model=list[探索稿概要], operation_id="list_explorations")
def 列出探索(request: Request, 作者: 作者依赖):
装配 = request.app.state.装配
return 装配.要求探索().列出(创作身份(作者, 装配.配置))
@路由.get("/explorations/{draft_id}", response_model=探索稿结果, operation_id="read_exploration")
def 读取探索(draft_id: str, request: Request, 作者: 作者依赖):
装配 = request.app.state.装配
return 装配.要求探索().读取(创作身份(作者, 装配.配置), draft_id)
@路由.post("/explorations/{draft_id}", response_model=探索稿结果, operation_id="save_exploration")
def 保存探索(draft_id: str, 请求: 探索保存请求, request: Request, 作者: 作者依赖):
核对路径(draft_id, 请求.draft_id)
装配 = request.app.state.装配
return 装配.要求探索().保存(
创作身份(作者, 装配.配置), draft_id, 请求.title, 请求.content, 请求.expected_hash
)
@路由.post(
"/explorations/{draft_id}/check",
response_model=探索检查结果,
operation_id="check_exploration_handoff",
)
def 检查探索(draft_id: str, 请求: 探索检查请求, request: Request, 作者: 作者依赖):
装配 = request.app.state.装配
return 装配.要求探索().检查交付(创作身份(作者, 装配.配置), draft_id, 请求.expected_hash)
@路由.get(
"/works/{work_id}/exploration-selection",
response_model=探索选定状态,
operation_id="read_exploration_selection",
)
def 读取选定(work_id: str, request: Request, 作者: 作者依赖):
装配 = request.app.state.装配
return 装配.要求作品().读取探索选定(创作身份(作者, 装配.配置), work_id)
@路由.post(
"/works/{work_id}/exploration-selection",
response_model=正式回执结果,
operation_id="select_exploration",
)
def 选定探索(work_id: str, 请求: 探索选定请求, request: Request, 作者: 作者依赖):
核对路径(work_id, 请求.selection.work_id)
装配 = request.app.state.装配
return 装配.要求作品().选定探索稿(创作身份(作者, 装配.配置), 请求.command_id, 请求.selection)
@路由.get("/works/{work_id}/plans/schema", response_model=档案结构, operation_id="read_plan_schema")
def 读取规划结构入口(
work_id: str, request: Request, 作者: 作者依赖, schema_id: str, base_version: int = Query(ge=1)
):
app = request.app.state.装配
return app.要求规划().新规划结构(创作身份(作者, app.配置), work_id, schema_id, base_version)
@路由.get("/works/{work_id}/plans", response_model=list[规划结果], operation_id="list_plans")
def 列出规划入口(work_id: str, request: Request, 作者: 作者依赖):
app = request.app.state.装配
return app.要求规划().列出规划(创作身份(作者, app.配置), work_id)
@路由.post(
"/works/{work_id}/plans/candidates",
response_model=正式回执结果,
operation_id="create_author_plan_candidate",
)
def 创建规划候选(work_id: str, 请求: 规划候选请求, request: Request, 作者: 作者依赖):
核对路径(work_id, 请求.request.work_id)
app = request.app.state.装配
return app.要求规划().保存人工候选(创作身份(作者, app.配置), 请求.command_id, 请求.request)
@路由.get(
"/plan-candidates/{candidate_id}",
response_model=规划候选结果,
operation_id="read_plan_candidate",
)
def 读取规划候选(
candidate_id: str,
request: Request,
作者: 作者依赖,
version: int | None = Query(default=None, ge=1),
):
app = request.app.state.装配
return app.要求规划().读取候选(创作身份(作者, app.配置), candidate_id, 版本=version)
@路由.post(
"/plan-candidates/{candidate_id}/versions",
response_model=正式回执结果,
operation_id="edit_plan_candidate",
)
def 编辑规划候选(candidate_id: str, 请求: 规划编辑请求, request: Request, 作者: 作者依赖):
核对路径(candidate_id, 请求.candidate.candidate_id)
app = request.app.state.装配
return app.要求规划().编辑候选(
创作身份(作者, app.配置), 请求.command_id, 请求.candidate, tuple(请求.changes)
)
@路由.get("/plans/{plan_id}", response_model=规划结果, operation_id="read_plan")
def 读取正式规划(plan_id: str, request: Request, 作者: 作者依赖):
app = request.app.state.装配
return app.要求规划().读取规划(创作身份(作者, app.配置), plan_id)
@路由.post(
"/plan-candidates/{candidate_id}/reviews",
response_model=规划审阅结果,
operation_id="open_plan_review",
)
def 打开规划审阅(candidate_id: str, 请求: 规划定位, request: Request, 作者: 作者依赖):
核对路径(candidate_id, 请求.candidate_id)
app = request.app.state.装配
return app.要求规划().打开审阅(创作身份(作者, app.配置), 请求)
@路由.post(
"/plan-candidates/{candidate_id}/decisions",
response_model=正式回执结果,
operation_id="decide_plan_candidate",
)
def 决定规划候选(candidate_id: str, 请求: 规划决定请求, request: Request, 作者: 作者依赖):
核对路径(candidate_id, 请求.candidate.candidate_id)
app = request.app.state.装配
return app.要求规划().决定(
创作身份(作者, app.配置),
请求.command_id,
请求.candidate,
作者动作(请求.action),
请求.author_review_id,
请求.review_hash,
tuple(请求.approved_changes),
)
@路由.get(
"/works/{work_id}/chapters/{chapter_id}/fine-outline",
response_model=规划结果,
operation_id="read_usable_fine_outline",
)
def 读取可用细纲入口(work_id: str, chapter_id: str, request: Request, 作者: 作者依赖):
app = request.app.state.装配
return app.要求规划().读取可用细纲(创作身份(作者, app.配置), work_id, chapter_id)
@路由.get("/works/{work_id}/plans/form", response_model=规划表单结果, operation_id="read_plan_form")
def 读取规划表单(
work_id: str,
request: Request,
作者: 作者依赖,
schema_id: str,
base_version: int = Query(ge=1),
chapter_id: str | None = None,
node_id: str | None = None,
):
app = request.app.state.装配
return app.要求规划().读取表单(
创作身份(作者, app.配置),
work_id,
schema_id,
base_version,
chapter_id=chapter_id,
node_id=node_id,
)
@路由.get(
"/works/{work_id}/plans/candidates",
response_model=list[规划候选结果],
operation_id="list_plan_candidates",
)
def 列出规划候选(work_id: str, request: Request, 作者: 作者依赖):
app = request.app.state.装配
return app.要求规划().列出候选(创作身份(作者, app.配置), work_id)

View File

@ -0,0 +1,75 @@
"""人工正文与历史HTTP入口,沿用B05版本服务与S01回执。"""
from fastapi import APIRouter, Query, Request
from muse.接入.http.作者会话 import 作者依赖
from muse.接入.创作操作 import (
人工保存请求,
创作身份,
回退正文请求,
核对路径,
正式回执结果,
正文历史结果,
正文读取结果,
)
路由 = APIRouter(prefix="/api/v1/chapters", tags=["正文写作"])
@路由.get(
"/{chapter_id}/documents/history",
response_model=list[正文历史结果],
operation_id="list_document_history",
)
def 读取历史(
chapter_id: str, request: Request, 作者: 作者依赖, branch_id: str = "main"
) -> list[dict]:
装配 = request.app.state.装配
return 装配.要求正文().列出版本(创作身份(作者, 装配.配置), chapter_id, 分支=branch_id)
@路由.get("/{chapter_id}/documents", response_model=正文读取结果, operation_id="read_document")
def 读取正文(
chapter_id: str,
request: Request,
作者: 作者依赖,
branch_id: str = "main",
revision: int | None = Query(default=None, ge=1),
) -> dict:
装配 = request.app.state.装配
return 装配.要求正文().读取正文(
创作身份(作者, 装配.配置),
chapter_id,
分支=branch_id,
版本=revision,
)
@路由.post("/{chapter_id}/documents", response_model=正式回执结果, operation_id="save_document")
def 保存正文(chapter_id: str, 请求: 人工保存请求, request: Request, 作者: 作者依赖) -> dict:
核对路径(chapter_id, 请求.chapter_id)
装配 = request.app.state.装配
return 装配.要求正文().保存人工(
创作身份(作者, 装配.配置),
请求.command_id,
chapter_id,
请求.expected_revision,
请求.document,
分支=请求.branch_id,
)
@路由.post(
"/{chapter_id}/documents/restore", response_model=正式回执结果, operation_id="restore_document"
)
def 回退正文(chapter_id: str, 请求: 回退正文请求, request: Request, 作者: 作者依赖) -> dict:
核对路径(chapter_id, 请求.chapter_id)
装配 = request.app.state.装配
return 装配.要求正文().回退正文(
创作身份(作者, 装配.配置),
请求.command_id,
chapter_id,
请求.expected_revision,
请求.restore_from,
分支=请求.branch_id,
)

View File

@ -0,0 +1,478 @@
"""人工创作命令的共同外壳;作者身份从接入取得,不接受客户端自报。"""
from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
from muse.作品规划.接口 import (
作品错误,
探索选择,
档案保存,
档案结构,
节点目录保存,
规划保存,
规划定位,
)
from muse.元数据.接口 import 字段变更
from muse.共享.调用身份 import 内容用途, 调用身份
from muse.故事世界.接口 import 事实定位, 事实提案请求, 事实结构, 知情记录
from muse.正文写作.接口 import 候选定位, 正文草稿, 段落
class 创作命令(BaseModel):
model_config = ConfigDict(extra="forbid")
command_id: str = Field(min_length=1)
class 档案保存请求(创作命令):
work_id: str = Field(min_length=1)
expected_revision: int = Field(ge=0)
content: dict = Field(default_factory=dict)
schema_binding: 档案结构
changes: list[字段变更] = Field(default_factory=list)
def 业务请求(self) -> 档案保存:
return 档案保存(
self.work_id,
self.expected_revision,
self.content,
self.schema_binding,
tuple(self.changes),
)
class 新增章节请求(创作命令):
work_id: str = Field(min_length=1)
chapter_id: str = Field(min_length=1)
title: str
expected_directory_revision: int = Field(ge=0)
class 重排章节请求(创作命令):
work_id: str = Field(min_length=1)
chapter_ids: list[str]
expected_directory_revision: int = Field(ge=0)
class 人工保存请求(创作命令):
chapter_id: str = Field(min_length=1)
branch_id: str = "main"
expected_revision: int = Field(ge=0)
document: 正文草稿
class 回退正文请求(创作命令):
chapter_id: str = Field(min_length=1)
branch_id: str = "main"
expected_revision: int = Field(ge=0)
restore_from: int = Field(ge=1)
class 候选修改请求(创作命令):
candidate_id: str = Field(min_length=1)
expected_candidate_revision: int = Field(ge=1)
expected_candidate_hash: str = Field(min_length=1)
document: 正文草稿 | None = None
selected_changes: list[str] = Field(default_factory=list)
class 候选审阅请求(BaseModel):
model_config = ConfigDict(extra="forbid")
candidate_id: str = Field(min_length=1)
candidate_revision: int = Field(ge=1)
candidate_hash: str = Field(min_length=1)
target_ref: str = Field(min_length=1)
expected_revision: int = Field(ge=0)
def 定位(self) -> 候选定位:
return 候选定位(
self.candidate_id,
self.candidate_revision,
self.candidate_hash,
self.target_ref,
self.expected_revision,
)
class 候选决定请求(候选审阅请求, 创作命令):
action: Literal["adopt", "reject", "defer"]
author_review_id: str = Field(min_length=1)
review_hash: str = Field(min_length=1)
approved_changes: list[str] = Field(default_factory=list)
class 候选差异结果(BaseModel):
change_id: str
start: int
end: int
before: list[段落]
after: list[段落]
before_text: str
after_text: str
class 候选检查结果(BaseModel):
format: str
format_version: int
quality: str
class 正文候选结果(BaseModel):
candidate_id: str
revision: int
candidate_hash: str
document_id: str
chapter_id: str
branch_id: str
origin: str
base_revision: int
current_document_revision: int
document: 正文草稿
visible_text: str
checks: 候选检查结果
decision: Literal["undecided", "adopt", "reject", "defer"]
stale: bool
changes: list[候选差异结果]
class 作品表单结果(BaseModel):
work_id: str
revision: int
content: dict[str, Any]
schema_binding: 档案结构
visible_schema: dict[str, Any]
editable_schema: dict[str, Any]
visible_projection_version: str
class 作品概要(BaseModel):
work_id: str
revision: int
content: dict[str, Any]
class 正式回执结果(BaseModel):
receipt_id: str
command_id: str
target_ref: str
action: str
author_review_id: str | None
basis: dict[str, Any]
results: list[dict[str, Any]]
impact_ids: list[str]
class 章节条目(BaseModel):
chapter_id: str
title: str
position: int
class 目录结果(BaseModel):
work_id: str
revision: int
chapters: list[章节条目]
class 节点目录请求(创作命令):
directory: 节点目录保存
class 节点条目(BaseModel):
node_id: str
kind: Literal["volume", "chapter", "scene"]
title: str
parent_id: str | None
position: int
class 节点目录结果(BaseModel):
work_id: str
revision: int
nodes: list[节点条目]
class 正文读取结果(BaseModel):
work_id: str
document_id: str
chapter_id: str
branch_id: str
revision: int
state: Literal["saved", "unsaved"]
document: 正文草稿
visible_text: str
document_hash: str
visible_text_hash: str
class 正文历史结果(BaseModel):
revision: int
document_hash: str
visible_text_hash: str
restored_from: int | None
created_at: datetime
class 候选审阅结果(BaseModel):
review_id: str
review_hash: str
snapshot: dict[str, Any]
candidate: 正文候选结果
before: 正文读取结果
class 候选进度结果(BaseModel):
candidate_id: str
revision: int
decision: str
stale: bool
class 章节进度结果(BaseModel):
chapter_id: str
title: str
position: int
document_revision: int
document_hash: str
visible_characters: int
candidates: list[候选进度结果]
class 任务进度结果(BaseModel):
task_id: str
state: str
event_cursor: int
class 作品进度结果(BaseModel):
work_id: str
directory_revision: int
total_chapters: int
saved_chapters: int
undecided_candidates: int
chapters: list[章节进度结果]
tasks: list[任务进度结果]
def 创作身份(作者: str, 配置) -> 调用身份:
# 人工命令不是S02后台任务,不编造数据库中不存在的task_id。
return 调用身份(作者, None, 配置.运行用途, 内容用途.规划)
def 核对路径(路径身份: str, 请求身份: str) -> None:
if 路径身份 != 请求身份:
raise 作品错误("SCOPE_DENIED", "请求目标与访问路径不一致")
class 事实创建请求(创作命令):
proposal: 事实提案请求
class 事实审阅请求(BaseModel):
model_config = ConfigDict(extra="forbid")
proposal_id: str
proposal_revision: int = Field(ge=1)
proposal_hash: str
target_ref: str
expected_revision: int = Field(ge=0)
def 定位(self) -> 事实定位:
return 事实定位(
self.proposal_id,
self.proposal_revision,
self.proposal_hash,
self.target_ref,
self.expected_revision,
)
class 事实决定请求(事实审阅请求, 创作命令):
action: Literal["adopt", "reject", "defer"]
author_review_id: str
review_hash: str
approved_changes: list[str] = Field(default_factory=list)
class 事实来源版本(BaseModel):
chapter_id: str
branch_id: str
revision: int
document_hash: str
class 世界事实结果(BaseModel):
object_id: str
type_id: str
revision: int
system_revision: int
content: dict[str, Any]
effective_from: str | None
knowledge_mode: str
independent: bool
content_hash: str
projection_version: str
schema_binding: dict[str, Any]
sources: list[事实来源版本]
planned_window: dict[str, str] | None = None
class 事实省略原因(BaseModel):
object_id: str
reason: str
class 世界查询结果(BaseModel):
work_id: str
directory_revision: int
system_revision: int
facts: list[世界事实结果]
beliefs: list[世界事实结果]
plans: list[世界事实结果]
omitted: list[事实省略原因]
class 事实提案结果(BaseModel):
proposal_id: str
revision: int
proposal_hash: str
work_id: str
object_id: str
expected_revision: int
schema_binding: 事实结构 = Field(alias="schema")
content: dict[str, Any]
independent: bool
sources: list[dict[str, Any]]
effective_from: str | None
knowledge_mode: str
known_to: list[知情记录]
planned_window: dict[str, str] | None
decision: str
class 事实审阅结果(BaseModel):
review_id: str
review_hash: str
snapshot: dict[str, Any]
proposal: 事实提案结果
class 探索保存请求(BaseModel):
model_config = ConfigDict(extra="forbid")
draft_id: str
title: str
content: str
expected_hash: str | None
class 探索检查请求(BaseModel):
model_config = ConfigDict(extra="forbid")
expected_hash: str
class 探索稿概要(BaseModel):
draft_id: str
title: str
revision: int
draft_hash: str
class 探索稿结果(探索稿概要):
content: str
class 交付格式问题(BaseModel):
code: str
message: str
class 探索检查结果(BaseModel):
draft_id: str
draft_hash: str
valid: bool
problems: list[交付格式问题]
setting_count: int
contract_version: str
class 探索选定请求(创作命令):
selection: 探索选择
class 已选探索稿(BaseModel):
work_id: str
revision: int
snapshot: 探索稿结果
selected_by: str
selected_at: str
class 探索选定状态(BaseModel):
work_id: str
revision: int
selection: 已选探索稿 | None
class 规划候选请求(创作命令):
request: 规划保存
class 规划编辑请求(创作命令):
candidate: 规划定位
changes: list[字段变更]
class 规划决定请求(创作命令):
candidate: 规划定位
action: Literal["adopt", "reject", "defer"]
author_review_id: str
review_hash: str
approved_changes: list[str]
class 规划候选结果(BaseModel):
candidate_id: str
plan_id: str
revision: int
candidate_hash: str
work_id: str
chapter_id: str | None
node_id: str | None = None
expected_revision: int
schema_binding: 档案结构
content: dict[str, Any]
decision: str
visible_schema: dict[str, Any]
editable_schema: dict[str, Any]
class 规划结果(BaseModel):
plan_id: str
work_id: str
chapter_id: str | None
node_id: str | None = None
type_id: str
revision: int
content: dict[str, Any]
schema_binding: dict[str, Any]
source_dependencies: list[dict[str, str]]
class 规划审阅结果(BaseModel):
review_id: str
review_hash: str
snapshot: dict[str, Any]
candidate: 规划候选结果
before: 规划结果 | None
class 规划表单结果(BaseModel):
work_id: str
chapter_id: str | None
node_id: str | None = None
plan_id: str
revision: int
schema_binding: 档案结构
content: dict[str, Any]
visible_schema: dict[str, Any]
editable_schema: dict[str, Any]

View File

@ -0,0 +1 @@
"""正文工作版本与作者保存;公开接口见接口.py。"""

View File

@ -0,0 +1,82 @@
"""人工写作与回退由作者发起,不取决于模型评分或候选状态。"""
from dataclasses import dataclass
from muse.作品规划.接口 import 读取章节归属
from muse.正式变更.接口 import 依赖引用, 修改影响, 参与操作, 变更错误, 固定哈希, 审阅内容, 提交计划
from muse.正文写作.存储 import 正文存储
from muse.正文写作.模型 import 人工正文保存, 正文草稿
from muse.正文写作.正文格式 import 文稿身份
@dataclass(frozen=True)
class 正文写入:
作者: str
请求: 人工正文保存
草稿: 正文草稿
def 解析新稿(存储: 正文存储, 请求: 人工正文保存) -> 正文草稿:
if 请求.document is not None:
return 请求.document
行 = 存储.读取(文稿身份(请求.chapter_id, 请求.branch_id), 请求.restore_from)
if 行 is None:
raise 变更错误("REVISION_CONFLICT", "指定历史正文版本不存在")
return 存储.恢复草稿(行)
class 正文保存入口:
身份 = "B05.save_manual"
请求类型 = 人工正文保存
def 查看(self, 连, 身份, 命令):
请求 = 命令.业务请求
章节 = 读取章节归属(连, 身份.作者, 请求.chapter_id)
id_ = 文稿身份(请求.chapter_id, 请求.branch_id)
if id_ != 命令.目标:
raise 变更错误("SCOPE_DENIED", "文稿目标与章节分支不一致")
行 = 正文存储(连).读取(id_)
版本 = 行["revision"] if 行 else 0
return 审阅内容(
id_,
版本,
None,
None,
None,
固定哈希([行["document_hash"] if 行 else None, 固定哈希(请求)]),
None,
None,
固定哈希(章节),
("document",),
(
依赖引用(
"B01.directory", "directory:" + 章节["work_id"], str(章节["directory_revision"])
),
依赖引用("B05.document", id_, str(版本)),
),
)
def 准备(self, 连, 身份, 命令, 当前):
请求 = 命令.业务请求
草稿 = 解析新稿(正文存储(连), 请求)
return 提交计划(
(参与操作("B05.write_document", 正文写入(身份.作者, 请求, 草稿)),),
(修改影响(命令.目标, "document_changed", str(当前.数据版本), str(当前.数据版本 + 1)),),
)
class 正文参与者:
身份 = "B05.write_document"
请求类型 = 正文写入
def 提交(self, 连, 操作):
# 作者范围复检仍走B01的同事务公开接口。
读取章节归属(连, 操作.作者, 操作.请求.chapter_id)
return 正文存储(连).写入(操作.作者, 操作.请求, 操作.草稿)
class 文稿依赖:
种类 = "B05.document"
def 锁定当前(self, 连, 对象ID):
return str(正文存储(连).锁文稿(对象ID))

View File

@ -0,0 +1,174 @@
"""固定正文基线上的候选版本与段落差异;作者修改仍先进入候选。"""
from dataclasses import asdict, dataclass
from difflib import SequenceMatcher
from muse.作品规划.接口 import 读取章节归属
from muse.正式变更.接口 import (
依赖引用,
参与操作,
变更错误,
固定哈希,
审阅内容,
提交计划,
读取作者决定,
)
from muse.正文写作.存储 import 正文存储
from muse.正文写作.模型 import 人工候选保存, 正文草稿, 段落
from muse.正文写作.正文格式 import 可见文本, 文稿身份, 正文结构哈希
def 段落差异(基线: 正文草稿, 候选: 正文草稿) -> list[dict]:
前, 后 = 基线.paragraphs, 候选.paragraphs
比较 = SequenceMatcher(a=[固定哈希(p) for p in 前], b=[固定哈希(p) for p in 后], autojunk=False)
return [
{
"change_id": 固定哈希(
[i, j, x, y, [asdict(p) for p in 前[i:j]], [asdict(p) for p in 后[x:y]]]
),
"start": i,
"end": j,
"before": [asdict(p) for p in 前[i:j]],
"after": [asdict(p) for p in 后[x:y]],
"before_text": 可见文本(正文草稿(前[i:j])) if j > i else "",
"after_text": 可见文本(正文草稿(后[x:y])) if y > x else "",
}
for op, i, j, x, y in 比较.get_opcodes()
if op != "equal"
]
def 选择差异(基线: 正文草稿, 候选: 正文草稿, 已选: tuple[str, ...]) -> 正文草稿:
from pydantic import TypeAdapter
差异 = 段落差异(基线, 候选)
if not 已选 or set(已选) - {v["change_id"] for v in 差异}:
raise 变更错误("REVIEW_MISMATCH", "所选差异不属于当前候选版本")
段落集 = list(基线.paragraphs)
for 项 in reversed(差异):
if 项["change_id"] in 已选:
段落集[项["start"] : 项["end"]] = TypeAdapter(list[段落]).validate_python(项["after"])
# 空文稿仍有一个空段落;删除全部内容不会留下无效的零段落文档。
return 正文草稿(tuple(段落集) or (段落(基线.paragraphs[0].paragraph_id, ()),))
def 核对候选归属(连, 身份, candidate_id: str, 版本: int | None = None) -> dict:
if not 身份.允许写正式内容:
raise 变更错误("SCOPE_DENIED", "候选入口需要已认证的生产作者")
行 = 正文存储(连).读取候选(candidate_id, 版本)
if 行 is None:
raise 变更错误("CANDIDATE_NOT_FOUND", "候选或指定版本不存在")
读取章节归属(连, 身份.作者, 行["chapter_id"])
return 行
def 呈现候选(连, 身份, candidate_id: str, 版本: int | None = None) -> dict:
行 = 核对候选归属(连, 身份, candidate_id, 版本)
存储 = 正文存储(连)
文稿ID = 文稿身份(行["chapter_id"], 行["branch_id"])
基线 = 存储.读取草稿(文稿ID, 行["base_revision"])
草稿 = 存储.恢复草稿(行)
当前 = 存储.读取(文稿ID)
return {
"candidate_id": str(行["candidate_id"]),
"revision": 行["revision"],
"candidate_hash": 行["candidate_hash"],
"document_id": 文稿ID,
"chapter_id": 行["chapter_id"],
"branch_id": 行["branch_id"],
"origin": 行["origin"],
"base_revision": 行["base_revision"],
"current_document_revision": 当前["revision"] if 当前 else 0,
"document": asdict(草稿),
"visible_text": 可见文本(草稿),
"checks": 行["checks"],
"decision": 读取作者决定(连, candidate_id, 行["revision"]),
"stale": (当前["revision"] if 当前 else 0) != 行["base_revision"],
"changes": 段落差异(基线, 草稿),
}
@dataclass(frozen=True)
class 候选写入:
作者: str
请求: 人工候选保存
草稿: 正文草稿
基线哈希: str
class 候选保存入口:
身份 = "B05.save_candidate"
请求类型 = 人工候选保存
def 查看(self, 连, 身份, 命令):
请求 = 命令.业务请求
章节 = 读取章节归属(连, 身份.作者, 请求.chapter_id)
文稿ID = 文稿身份(请求.chapter_id, 请求.branch_id)
存储 = 正文存储(连)
行 = 存储.读取候选(请求.candidate_id)
版 = 行["revision"] if 行 else 0
if 行:
核对候选归属(连, 身份, 请求.candidate_id)
if (行["chapter_id"], 行["branch_id"], 行["base_revision"]) != (
请求.chapter_id,
请求.branch_id,
请求.base_revision,
):
raise 变更错误("SCOPE_DENIED", "改稿不能改变候选所属文稿或原基线")
if 请求.expected_candidate_hash != 行["candidate_hash"]:
raise 变更错误("REVISION_CONFLICT", "待编辑候选版本已经改变")
if 读取作者决定(连, 请求.candidate_id, 版) not in {"undecided", "defer"}:
raise 变更错误("REVISION_CONFLICT", "已决定候选保留历史,请从当前正文另建改稿")
正文 = 存储.读取(文稿ID)
文稿版 = 正文["revision"] if 正文 else 0
if 请求.base_revision != 文稿版:
raise 变更错误("SOURCE_STALE", "候选基线正文已经改变,请比较当前正文后另建候选")
return 审阅内容(
请求.candidate_id,
版,
None,
None,
None,
固定哈希(请求),
None,
None,
固定哈希(章节),
("candidate",),
(
依赖引用(
"B01.directory", "directory:" + 章节["work_id"], str(章节["directory_revision"])
),
依赖引用("B05.document", 文稿ID, str(文稿版)),
依赖引用("B05.candidate", 请求.candidate_id, str(版)),
),
)
def 准备(self, 连, 身份, 命令, 当前):
请求 = 命令.业务请求
存储 = 正文存储(连)
基线 = 存储.读取草稿(文稿身份(请求.chapter_id, 请求.branch_id), 请求.base_revision)
草稿 = 请求.document
if 草稿 is None:
行 = 存储.读取候选(请求.candidate_id)
if 行 is None:
raise 变更错误("CANDIDATE_NOT_FOUND", "部分选择需要已有候选")
草稿 = 选择差异(基线, 存储.恢复草稿(行), 请求.selected_changes)
return 提交计划(
(参与操作("B05.write_candidate", 候选写入(身份.作者, 请求, 草稿, 正文结构哈希(基线))),)
)
class 候选参与者:
身份 = "B05.write_candidate"
请求类型 = 候选写入
def 提交(self, 连, 操作):
读取章节归属(连, 操作.作者, 操作.请求.chapter_id)
return 正文存储(连).写入候选(操作.作者, 操作.请求, 操作.草稿, 操作.基线哈希)
class 候选依赖:
种类 = "B05.candidate"
def 锁定当前(self, 连, 对象ID):
return str(正文存储(连).锁候选(对象ID))

View File

@ -0,0 +1,191 @@
"""B05正文与历史版本SQL;不读取作品、事实或S01私表。"""
import hashlib
from dataclasses import asdict
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb
from pydantic import TypeAdapter
from muse.正文写作.模型 import 人工候选保存, 人工正文保存, 正文草稿, 正文错误, 段落
from muse.正文写作.正文格式 import 可见文本哈希, 文稿身份, 正文结构哈希
class 正文存储:
def __init__(self, 连):
self.连 = 连
def 列出版本(self, document_id: str) -> list[dict]:
return (
self.连.cursor(row_factory=dict_row)
.execute(
"SELECT revision,document_hash,visible_text_hash,restored_from,created_at "
"FROM muse_document_version WHERE document_id=%s ORDER BY revision DESC",
(document_id,),
)
.fetchall()
)
def 读取(self, document_id: str, 版本: int | None = None) -> dict | None:
return (
self.连.cursor(row_factory=dict_row)
.execute(
"SELECT d.document_id,d.chapter_id,d.branch_id,v.revision,v.document,"
"v.document_hash,v.visible_text_hash,v.restored_from "
"FROM muse_document d JOIN muse_document_version v ON v.document_id=d.document_id "
"AND v.revision=COALESCE(%s,d.current_revision) WHERE d.document_id=%s",
(版本, document_id),
)
.fetchone()
)
def 读取草稿(self, document_id: str, 版本: int) -> 正文草稿:
if 版本 == 0:
return 正文草稿((段落(document_id + ":p0", ()),))
行 = self.读取(document_id, 版本)
if 行 is None:
raise 正文错误("指定正文版本不存在")
return self.恢复草稿(行)
def 读取候选(self, candidate_id: str, 版本: int | None = None) -> dict | None:
return (
self.连.cursor(row_factory=dict_row)
.execute(
"SELECT c.candidate_id,c.chapter_id,c.branch_id,c.origin,c.current_revision,v.* "
"FROM muse_writing_candidate c JOIN muse_writing_candidate_version v "
"ON v.candidate_id=c.candidate_id AND v.revision=COALESCE(%s,c.current_revision) "
"WHERE c.candidate_id=%s",
(版本, candidate_id),
)
.fetchone()
)
def 列出候选ID(self, chapter_id: str, branch_id: str) -> list[str]:
return [
str(r[0])
for r in self.连.execute(
"SELECT candidate_id FROM muse_writing_candidate "
"WHERE chapter_id=%s AND branch_id=%s "
"ORDER BY candidate_id",
(chapter_id, branch_id),
).fetchall()
]
def 锁候选(self, candidate_id: str) -> int:
key = int.from_bytes(
hashlib.sha256(("B05:candidate:" + candidate_id).encode()).digest()[:8], signed=True
)
self.连.execute("SELECT pg_advisory_xact_lock(%s)", (key,))
行 = self.连.execute(
"SELECT current_revision FROM muse_writing_candidate WHERE candidate_id=%s FOR UPDATE",
(candidate_id,),
).fetchone()
return 行[0] if 行 else 0
def 写入候选(self, 作者: str, 请求: 人工候选保存, 草稿: 正文草稿, 基线哈希: str) -> dict:
from muse.正式变更.接口 import 固定哈希
版 = 请求.expected_candidate_revision + 1
if 请求.expected_candidate_revision == 0:
self.连.execute(
"INSERT INTO muse_writing_candidate"
"(candidate_id,chapter_id,branch_id,origin,current_revision) "
"VALUES (%s,%s,%s,'author',%s)",
(请求.candidate_id, 请求.chapter_id, 请求.branch_id, 版),
)
else:
行 = self.连.execute(
"UPDATE muse_writing_candidate SET current_revision=%s WHERE candidate_id=%s "
"AND current_revision=%s RETURNING candidate_id",
(版, 请求.candidate_id, 请求.expected_candidate_revision),
).fetchone()
if 行 is None:
raise 正文错误("候选已由其他请求修改")
文稿哈希 = 正文结构哈希(草稿)
检查 = {"format": "passed", "format_version": 1, "quality": "not_applicable"}
哈希 = 固定哈希(
[
请求.candidate_id,
版,
请求.chapter_id,
请求.branch_id,
请求.base_revision,
基线哈希,
文稿哈希,
检查,
]
)
self.连.execute(
"INSERT INTO muse_writing_candidate_version"
"(candidate_id,revision,base_revision,base_hash,"
"document,document_hash,visible_text_hash,candidate_hash,checks,author_id) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
(
请求.candidate_id,
版,
请求.base_revision,
基线哈希,
Jsonb(asdict(草稿)),
文稿哈希,
可见文本哈希(草稿),
哈希,
Jsonb(检查),
作者,
),
)
return {"candidate_id": 请求.candidate_id, "revision": 版, "candidate_hash": 哈希}
def 锁文稿(self, document_id: str) -> int:
key = int.from_bytes(
hashlib.sha256(("B05:document:" + document_id).encode()).digest()[:8], signed=True
)
self.连.execute("SELECT pg_advisory_xact_lock(%s)", (key,))
行 = self.连.execute(
"SELECT current_revision FROM muse_document WHERE document_id=%s FOR UPDATE",
(document_id,),
).fetchone()
return 行[0] if 行 else 0
@staticmethod
def 恢复草稿(行: dict) -> 正文草稿:
草稿 = TypeAdapter(正文草稿).validate_python(行["document"])
if (
正文结构哈希(草稿) != 行["document_hash"]
or 可见文本哈希(草稿) != 行["visible_text_hash"]
):
raise 正文错误("已保存正文与其结构或文本哈希不一致")
return 草稿
def 写入(self, 作者: str, 请求: 人工正文保存, 草稿: 正文草稿) -> dict:
id_ = 文稿身份(请求.chapter_id, 请求.branch_id)
版本 = 请求.expected_revision + 1
if 请求.expected_revision == 0:
self.连.execute(
"INSERT INTO muse_document(document_id,chapter_id,branch_id,current_revision) "
"VALUES (%s,%s,%s,%s)",
(id_, 请求.chapter_id, 请求.branch_id, 版本),
)
else:
行 = self.连.execute(
"UPDATE muse_document SET current_revision=%s WHERE document_id=%s "
"AND current_revision=%s RETURNING document_id",
(版本, id_, 请求.expected_revision),
).fetchone()
if 行 is None:
raise 正文错误("正文基线已改变")
结构哈希 = 正文结构哈希(草稿)
文本哈希 = 可见文本哈希(草稿)
self.连.execute(
"INSERT INTO muse_document_version(document_id,revision,document,document_hash,"
"visible_text_hash,author_id,restored_from) VALUES (%s,%s,%s,%s,%s,%s,%s)",
(id_, 版本, Jsonb(asdict(草稿)), 结构哈希, 文本哈希, 作者, 请求.restore_from),
)
return {
"document_id": id_,
"chapter_id": 请求.chapter_id,
"branch_id": 请求.branch_id,
"revision": 版本,
"document_hash": 结构哈希,
"visible_text_hash": 文本哈希,
"restored_from": 请求.restore_from,
}

View File

@ -0,0 +1,261 @@
"""正文写作公开合同;正式版本写入由本模块参与S01事务。"""
from dataclasses import asdict
from uuid import NAMESPACE_URL, uuid5
from muse.作品规划.接口 import 读取章节归属
from muse.基础设施.数据库.连接 import 数据库工厂
from muse.正式变更.接口 import 作者动作, 参与者目录, 变更命令, 变更错误, 正式变更服务
from muse.正文写作.人工保存 import 文稿依赖, 正文保存入口, 正文参与者
from muse.正文写作.候选版本 import 候选依赖, 候选保存入口, 候选参与者, 呈现候选, 核对候选归属
from muse.正文写作.存储 import 正文存储
from muse.正文写作.模型 import (
人工候选保存,
人工正文保存,
候选定位,
换行节点,
文本节点,
正文草稿,
正文错误,
段落,
)
from muse.正文写作.正式提交 import 正文候选决定入口
from muse.正文写作.正文格式 import 可见文本, 可见文本哈希, 文稿身份, 正文结构哈希
def 登记正文参与者(目录: 参与者目录) -> None:
目录.登记入口(正文保存入口())
目录.登记参与者(正文参与者())
目录.登记依赖(文稿依赖())
目录.登记入口(候选保存入口())
目录.登记入口(正文候选决定入口())
目录.登记参与者(候选参与者())
目录.登记依赖(候选依赖())
def 读取正文依据(连, 作者: str, chapter_id: str, revision: int, *, 分支: str = "main") -> dict:
"""事实及其他同事务消费者读取确切正文;同时返回当前指针供来源复检。"""
章节 = 读取章节归属(连, 作者, chapter_id)
id_ = 文稿身份(chapter_id, 分支)
存储 = 正文存储(连)
行 = 存储.读取(id_, revision)
当前 = 存储.读取(id_)
if 行 is None or 当前 is None:
raise 正文错误("事实依据引用的正文版本不存在")
return {
"work_id": 章节["work_id"],
"document_id": id_,
"chapter_id": chapter_id,
"revision": revision,
"current_revision": 当前["revision"],
"document_hash": 行["document_hash"],
"document": 存储.恢复草稿(行),
}
class 正文服务:
def __init__(self, 数据库: 数据库工厂, 正式: 正式变更服务):
self.数据库, self.正式 = 数据库, 正式
def 创建人工候选(
self,
身份,
命令ID: str,
chapter_id: str,
预期版本: int,
草稿: 正文草稿,
*,
分支: str = "main",
) -> dict:
from muse.正式变更.接口 import 固定哈希
id_ = str(uuid5(NAMESPACE_URL, "muse:B05:candidate:" + 固定哈希([身份.作者, 命令ID])))
return self._保存候选(
身份, 命令ID, 人工候选保存(id_, chapter_id, 分支, 预期版本, 0, document=草稿)
)
def 编辑候选(
self,
身份,
命令ID: str,
candidate_id: str,
候选版本: int,
候选哈希: str,
草稿: 正文草稿 | None = None,
*,
选择变更: tuple[str, ...] = (),
) -> dict:
with self.数据库.连接(只读=True) as 连:
行 = 核对候选归属(连, 身份, candidate_id)
return self._保存候选(
身份,
命令ID,
人工候选保存(
candidate_id,
行["chapter_id"],
行["branch_id"],
行["base_revision"],
候选版本,
候选哈希,
草稿,
选择变更,
),
)
def _保存候选(self, 身份, 命令ID: str, 请求: 人工候选保存) -> dict:
return self.正式.提交变更(
身份,
变更命令(
命令ID,
"B05.save_candidate",
请求.candidate_id,
作者动作.人工保存,
请求.expected_candidate_revision,
请求,
),
)
def 读取候选(self, 身份, candidate_id: str, *, 版本: int | None = None) -> dict:
with self.数据库.连接(只读=True) as 连:
return 呈现候选(连, 身份, candidate_id, 版本)
def 列出候选(self, 身份, chapter_id: str, *, 分支: str = "main") -> list[dict]:
if not 身份.允许写正式内容:
raise 变更错误("SCOPE_DENIED", "候选入口需要已认证作者")
with self.数据库.连接(只读=True) as 连:
读取章节归属(连, 身份.作者, chapter_id)
return [呈现候选(连, 身份, id_) for id_ in 正文存储(连).列出候选ID(chapter_id, 分支)]
def 打开候选审阅(self, 身份, 定位: 候选定位) -> dict:
审阅 = self.正式.打开审阅(
身份,
变更命令(
"review",
"B05.decide_candidate",
定位.target_ref,
作者动作.采纳,
定位.expected_revision,
定位,
),
)
候选 = self.读取候选(身份, 定位.candidate_id, 版本=定位.candidate_revision)
return {
**审阅,
"candidate": 候选,
"before": self.读取正文(
身份, 候选["chapter_id"], 分支=候选["branch_id"], 版本=定位.expected_revision
),
}
def 决定候选(
self,
身份,
命令ID: str,
定位: 候选定位,
动作: 作者动作,
审阅ID: str,
审阅哈希: str,
批准变更: tuple[str, ...],
) -> dict:
return self.正式.提交变更(
身份,
变更命令(
命令ID,
"B05.decide_candidate",
定位.target_ref,
动作,
定位.expected_revision,
定位,
审阅ID,
审阅哈希,
批准变更,
),
)
def 列出版本(self, 身份, chapter_id: str, *, 分支: str = "main") -> list[dict]:
if not 身份.允许写正式内容:
raise 变更错误("SCOPE_DENIED", "作者正文入口需要已认证作者")
with self.数据库.连接(只读=True) as 连:
读取章节归属(连, 身份.作者, chapter_id)
return 正文存储(连).列出版本(文稿身份(chapter_id, 分支))
def 保存人工(
self,
身份,
命令ID: str,
chapter_id: str,
预期版本: int,
草稿: 正文草稿,
*,
分支: str = "main",
) -> dict:
return self._提交(身份, 命令ID, 人工正文保存(chapter_id, 分支, 预期版本, document=草稿))
def 回退正文(
self, 身份, 命令ID: str, chapter_id: str, 预期版本: int, 旧版本: int, *, 分支: str = "main"
) -> dict:
return self._提交(
身份, 命令ID, 人工正文保存(chapter_id, 分支, 预期版本, restore_from=旧版本)
)
def _提交(self, 身份, 命令ID: str, 请求: 人工正文保存) -> dict:
return self.正式.提交变更(
身份,
变更命令(
命令ID,
"B05.save_manual",
文稿身份(请求.chapter_id, 请求.branch_id),
作者动作.人工保存,
请求.expected_revision,
请求,
),
)
def 读取正文(
self, 身份, chapter_id: str, *, 分支: str = "main", 版本: int | None = None
) -> dict:
if not 身份.允许写正式内容:
raise 变更错误("SCOPE_DENIED", "作者正文入口需要已认证作者")
with self.数据库.连接(只读=True) as 连:
章节 = 读取章节归属(连, 身份.作者, chapter_id)
id_ = 文稿身份(chapter_id, 分支)
行 = 正文存储(连).读取(id_, 版本)
if 行 is None:
if 版本 not in {None, 0}:
raise 变更错误("REVISION_CONFLICT", "指定正文版本不存在")
草稿 = 正文草稿((段落(id_ + ":p0", ()),))
版 = 0
状态 = "unsaved"
else:
草稿 = 正文存储.恢复草稿(行)
版 = 行["revision"]
状态 = "saved"
return {
"work_id": 章节["work_id"],
"document_id": id_,
"chapter_id": chapter_id,
"branch_id": 分支,
"revision": 版,
"state": 状态,
"document": asdict(草稿),
"visible_text": 可见文本(草稿),
"document_hash": 正文结构哈希(草稿),
"visible_text_hash": 可见文本哈希(草稿),
}
__all__ = [
"文本节点",
"换行节点",
"段落",
"正文草稿",
"正文错误",
"可见文本",
"可见文本哈希",
"正文结构哈希",
"正文服务",
"登记正文参与者",
"人工正文保存",
"候选定位",
"读取正文依据",
]

View File

@ -0,0 +1,124 @@
"""正文协议保护身份与可见文本;题材和故事内容不进入固定代码字段。"""
from typing import Literal
from pydantic import ConfigDict
from pydantic.dataclasses import dataclass
from muse.共享.错误 import Muse错误
class 正文错误(Muse错误):
错误码 = "DOCUMENT_INVALID"
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 文本节点:
text: str
marks: tuple[Literal["strong", "emphasis"], ...] = ()
type: Literal["text"] = "text"
def __post_init__(self):
if not isinstance(self.text, str) or any(c in self.text for c in "\r\n"):
raise 正文错误("段内换行必须使用显式换行节点")
if self.type != "text" or set(self.marks) - {"strong", "emphasis"}:
raise 正文错误("正文只接受登记的文字节点和行内样式")
object.__setattr__(
self, "marks", tuple(m for m in ("strong", "emphasis") if m in self.marks)
)
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 换行节点:
newline: Literal["\n", "\r\n", "\r"] = "\n"
type: Literal["line_break"] = "line_break"
def __post_init__(self):
if self.type != "line_break" or self.newline not in {"\n", "\r\n", "\r"}:
raise 正文错误("显式换行节点无效")
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 段落:
paragraph_id: str
content: tuple[文本节点 | 换行节点, ...]
def __post_init__(self):
if not self.paragraph_id.strip() or any(
not isinstance(v, (文本节点, 换行节点)) for v in self.content
):
raise 正文错误("段落必须有稳定身份与允许的内容节点")
内容: list[文本节点 | 换行节点] = []
for 节点 in self.content:
if isinstance(节点, 文本节点):
if not 节点.text:
continue
前 = 内容[-1] if 内容 else None
if isinstance(前, 文本节点) and 前.marks == 节点.marks:
内容[-1] = 文本节点(前.text + 节点.text, 前.marks)
continue
内容.append(节点)
object.__setattr__(self, "content", tuple(内容))
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 正文草稿:
paragraphs: tuple[段落, ...]
format_version: int = 1
def __post_init__(self):
if self.format_version != 1 or not self.paragraphs:
raise 正文错误("正文需要受支持格式与至少一个段落;空稿使用空段落")
ids = [p.paragraph_id for p in self.paragraphs]
if len(ids) != len(set(ids)):
raise 正文错误("正文段落身份不能重复")
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 人工正文保存:
chapter_id: str
branch_id: str
expected_revision: int
document: 正文草稿 | None = None
restore_from: int | None = None
def __post_init__(self):
if not self.chapter_id or not self.branch_id or self.expected_revision < 0:
raise 正文错误("正文保存需要章节、分支和基线版本")
if (self.document is None) == (self.restore_from is None):
raise 正文错误("正文保存或历史回退必须明确选择一个")
if self.restore_from is not None and self.restore_from < 1:
raise 正文错误("回退需要已保存的历史版本")
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 候选定位:
candidate_id: str
candidate_revision: int
candidate_hash: str
target_ref: str
expected_revision: int
def __post_init__(self):
if self.candidate_revision < 1 or self.expected_revision < 0 or not self.candidate_hash:
raise 正文错误("候选审阅需要明确的候选版本、哈希和正文基线")
@dataclass(frozen=True, slots=True, config=ConfigDict(extra="forbid"))
class 人工候选保存:
candidate_id: str
chapter_id: str
branch_id: str
base_revision: int
expected_candidate_revision: int
expected_candidate_hash: str | None = None
document: 正文草稿 | None = None
selected_changes: tuple[str, ...] = ()
def __post_init__(self):
if self.base_revision < 0 or self.expected_candidate_revision < 0:
raise 正文错误("候选与正文基线不能为负数")
if (self.document is None) == (not self.selected_changes):
raise 正文错误("候选改稿需要全文或明确的差异选择,两者只能选一项")
if len(set(self.selected_changes)) != len(self.selected_changes):
raise 正文错误("候选差异选择不能重复")

View File

@ -0,0 +1,73 @@
"""正文候选的正式决定;复用人工正文参与者和S01共同事务。"""
from muse.作品规划.接口 import 读取章节归属
from muse.正式变更.接口 import (
作者动作,
依赖引用,
修改影响,
参与操作,
变更错误,
固定哈希,
审阅内容,
提交计划,
)
from muse.正文写作.人工保存 import 正文写入
from muse.正文写作.候选版本 import 核对候选归属, 段落差异
from muse.正文写作.存储 import 正文存储
from muse.正文写作.模型 import 人工正文保存, 候选定位
from muse.正文写作.正文格式 import 文稿身份, 正文结构哈希
class 正文候选决定入口:
身份 = "B05.decide_candidate"
请求类型 = 候选定位
def 查看(self, 连, 身份, 命令):
请求 = 命令.业务请求
行 = 核对候选归属(连, 身份, 请求.candidate_id)
if 行["revision"] != 请求.candidate_revision or 行["candidate_hash"] != 请求.candidate_hash:
raise 变更错误("REVISION_CONFLICT", "候选与被审版本不一致")
文稿ID = 文稿身份(行["chapter_id"], 行["branch_id"])
if 请求.target_ref != 文稿ID:
raise 变更错误("SCOPE_DENIED", "候选与目标文稿不一致")
存储 = 正文存储(连)
正文 = 存储.读取(文稿ID)
版 = 正文["revision"] if 正文 else 0
基线 = 存储.读取草稿(文稿ID, 版)
草稿 = 存储.恢复草稿(行)
章节 = 读取章节归属(连, 身份.作者, 行["chapter_id"])
return 审阅内容(
文稿ID,
版,
请求.candidate_id,
行["revision"],
行["candidate_hash"],
固定哈希(段落差异(基线, 草稿)),
固定哈希({"format_version": 1}),
"document-v1",
固定哈希([行["base_revision"], 行["base_hash"], 章节]),
("document",),
(
依赖引用(
"B01.directory", "directory:" + 章节["work_id"], str(章节["directory_revision"])
),
依赖引用("B05.document", 文稿ID, str(版)),
依赖引用("B05.candidate", 请求.candidate_id, str(行["revision"])),
),
)
def 准备(self, 连, 身份, 命令, 当前):
if 命令.动作 in {作者动作.拒绝, 作者动作.暂缓}:
return 提交计划(())
if 命令.动作 is not 作者动作.采纳:
raise 变更错误("INVALID_REQUEST", "候选入口只接受明确作者决定")
行 = 核对候选归属(连, 身份, 命令.业务请求.candidate_id)
基线 = 正文存储(连).读取草稿(当前.目标, 当前.数据版本)
if 行["base_revision"] != 当前.数据版本 or 行["base_hash"] != 正文结构哈希(基线):
raise 变更错误("SOURCE_STALE", "候选依据的正文已改变,不能直接覆盖作者新稿")
草稿 = 正文存储.恢复草稿(行)
请求 = 人工正文保存(行["chapter_id"], 行["branch_id"], 当前.数据版本, document=草稿)
return 提交计划(
(参与操作("B05.write_document", 正文写入(身份.作者, 请求, 草稿)),),
(修改影响(当前.目标, "document_changed", str(当前.数据版本), str(当前.数据版本 + 1)),),
)

View File

@ -0,0 +1,33 @@
"""正文的可见文本与结构序列化;不做Unicode或空白规范化。"""
import hashlib
import json
from dataclasses import asdict
from muse.正文写作.模型 import 文本节点, 正文草稿
def 可见文本(草稿: 正文草稿) -> str:
return "\n".join(
"".join(n.text if isinstance(n, 文本节点) else n.newline for n in p.content)
for p in 草稿.paragraphs
)
def 正文结构哈希(草稿: 正文草稿) -> str:
字节 = json.dumps(
asdict(草稿), ensure_ascii=False, sort_keys=True, separators=(",", ":")
).encode()
return hashlib.sha256(字节).hexdigest()
def 可见文本哈希(草稿: 正文草稿) -> str:
return hashlib.sha256(可见文本(草稿).encode()).hexdigest()
def 文稿身份(chapter_id: str, branch_id: str) -> str:
from uuid import NAMESPACE_URL, uuid5
return str(
uuid5(NAMESPACE_URL, "muse:B05:" + json.dumps([chapter_id, branch_id], ensure_ascii=False))
)

View File

@ -0,0 +1,94 @@
"""正文结构的独立接缝:中文原文、段落身份与结构/可见哈希。"""
import json
from pathlib import Path
from muse.正文写作.接口 import 可见文本, 文本节点, 正文结构哈希, 正文草稿, 段落
def test_正文不擅自改写NFC与换行__7d1f17():
from muse.正文写作.接口 import 可见文本哈希, 换行节点
原文 = "Cafe\u0301\r\n第二行\r第三行"
规范表示 = "Café\n第二行\n第三行"
草稿 = 正文草稿(
(
段落(
"p1",
(
文本节点("Cafe\u0301"),
换行节点("\r\n"),
文本节点("第二行"),
换行节点("\r"),
文本节点("第三行"),
),
),
)
)
对照 = 正文草稿(
(
段落(
"p1",
(
文本节点("Café"),
换行节点("\n"),
文本节点("第二行"),
换行节点("\n"),
文本节点("第三行"),
),
),
)
)
assert 可见文本(草稿) == 原文 and 可见文本(对照) == 规范表示
assert 可见文本哈希(草稿) != 可见文本哈希(对照)
assert 正文结构哈希(草稿) != 正文结构哈希(对照)
def test_中文夹具往返保留空段落与原码点__aa1001():
样本 = json.loads((Path(__file__).parents[1] / "夹具/中文正文.json").read_text())["samples"]
for 项 in 样本:
段 = tuple(段落(f"p-{i}", (文本节点(文本),)) for i, 文本 in enumerate(项["paragraphs"]))
草稿 = 正文草稿(段)
assert 可见文本(草稿) == "\n".join(项["paragraphs"])
for 选区 in 项.get("selections", []):
文本 = 可见文本(正文草稿((段[选区["paragraph_index"]],)))
assert 文本[选区["start"] : 选区["end"]] == 选区["expected_text"]
另名 = 正文草稿(tuple(段落("new-" + p.paragraph_id, p.content) for p in 段))
assert 可见文本(草稿) == 可见文本(另名)
assert 正文结构哈希(草稿) != 正文结构哈希(另名)
def test_正文外壳拒绝额外节点属性__aa1002():
import pytest
from pydantic import TypeAdapter, ValidationError
合同 = TypeAdapter(正文草稿)
输入 = {
"paragraphs": [
{
"paragraph_id": "p1",
"content": [{"type": "text", "text": "文字", "html": "<script/>"}],
}
]
}
with pytest.raises(ValidationError):
合同.validate_python(输入)
def test_编辑器结构规范化不改变正文字符__aa1003():
草稿 = 正文草稿(
(
段落(
"p1",
(
文本节点(""),
文本节点(" e\u0301", ("emphasis", "strong")),
文本节点("🧑‍🏽 ", ("strong", "emphasis", "strong")),
),
),
段落("empty", (文本节点(""),)),
)
)
assert 草稿.paragraphs[0].content == (文本节点(" e\u0301🧑‍🏽 ", ("strong", "emphasis")),)
assert 草稿.paragraphs[1].content == ()
assert 可见文本(草稿) == " e\u0301🧑‍🏽 \n"

View File

@ -0,0 +1,100 @@
"""浏览器将使用的真实作者会话、作品和正文HTTP入口。"""
import pytest
from fastapi.testclient import TestClient
from muse.元数据.接口 import 元数据服务, 导入内置结构
from muse.共享.调用身份 import 用途
from muse.接入.http.应用 import 创建应用
from muse.配置 import 应用配置, 服务配置
pytestmark = pytest.mark.数据库
def test_HTTP创建作品到保存中文正文并拒绝伪身份__aa1301(应用测试库, tmp_path):
with 应用测试库[用途.维护].连接() as 连, 连.transaction():
导入内置结构(元数据服务(连))
口令 = tmp_path / "作者口令.txt"
口令.write_text("synthetic-author-access")
配置 = 应用配置(
应用测试库[用途.生产].引用,
"test",
HTTP=服务配置(
str(口令),
作者ID="author",
公开地址="http://testserver",
允许来源=("http://testserver",),
),
)
with TestClient(创建应用(配置)) as client:
client.headers["Origin"] = "http://testserver"
assert client.get("/api/v1/works").status_code == 401
assert (
client.post("/api/v1/session", json={"password": "synthetic-author-access"}).status_code
== 200
)
form = client.get(
"/api/v1/works/http-work/form", params={"schema_id": "work_core", "base_version": 1}
)
assert form.status_code == 200, form.text
body = {
"command_id": "create-http",
"work_id": "http-work",
"expected_revision": 0,
"schema_binding": form.json()["schema_binding"],
"changes": [{"路径": ["名称"], "动作": "set", "值": "浏览器新故事"}],
}
assert (
client.post(
"/api/v1/works/http-work/profile", json={**body, "author_id": "other"}
).status_code
== 422
)
assert client.post("/api/v1/works/http-work/profile", json=body).status_code == 200
assert client.get("/api/v1/works").json()[0]["work_id"] == "http-work"
chapter = {
"command_id": "new-chapter",
"work_id": "http-work",
"chapter_id": "http-chapter",
"title": "开篇",
"expected_directory_revision": 0,
}
assert client.post("/api/v1/works/http-work/chapters", json=chapter).status_code == 200
draft = {
"format_version": 1,
"paragraphs": [
{
"paragraph_id": "http-p1",
"content": [{"type": "text", "text": " “正文”e\u0301🧑‍🏽", "marks": []}],
}
],
}
save = {
"command_id": "http-save",
"chapter_id": "http-chapter",
"expected_revision": 0,
"document": draft,
}
reply = client.post("/api/v1/chapters/http-chapter/documents", json=save)
assert reply.status_code == 200, reply.text
assert (
client.post("/api/v1/chapters/http-chapter/documents", json=save).json() == reply.json()
)
loaded = client.get("/api/v1/chapters/http-chapter/documents").json()
assert loaded["revision"] == 1 and loaded["visible_text"] == " “正文”e\u0301🧑‍🏽"
history = client.get("/api/v1/chapters/http-chapter/documents/history")
assert history.status_code == 200, history.text
assert [v["revision"] for v in history.json()] == [1]
restored = client.post(
"/api/v1/chapters/http-chapter/documents/restore",
json={
"chapter_id": "http-chapter",
"command_id": "restore-http",
"expected_revision": 1,
"restore_from": 1,
},
)
assert restored.status_code == 200, restored.text
latest = client.get("/api/v1/chapters/http-chapter/documents").json()
assert latest["revision"] == 2 and latest["document"] == loaded["document"]
assert latest["work_id"] == "http-work"

View File

@ -0,0 +1,204 @@
"""人工创作入口使用真实owner、元数据、S01和隔离PG;不需要模型评分。"""
import pytest
from muse.作品规划.接口 import 作品服务, 档案保存, 登记作品参与者
from muse.元数据.接口 import 元数据服务, 导入内置结构
from muse.共享.标识 import 任务标识
from muse.共享.调用身份 import 内容用途, 用途, 调用身份
from muse.正式变更.接口 import 参与者目录, 正式变更服务
pytestmark = pytest.mark.数据库
@pytest.fixture
def 作品环境(应用测试库):
with 应用测试库[用途.维护].连接() as 连, 连.transaction():
导入内置结构(元数据服务(连))
目录 = 参与者目录()
登记作品参与者(目录)
正式 = 正式变更服务(应用测试库[用途.生产], 目录)
服务 = 作品服务(应用测试库[用途.生产], 正式)
作者 = 调用身份("author", 任务标识("manual-create"), 用途.生产, 内容用途.规划)
结构 = 服务.新档案结构(作者, "work-new", "work_core", 1)
输入 = 档案保存(
"work-new", 0, {"名称": "新故事", "题材定位": "未定题材", "主题立意": "关于选择"}, 结构
)
return 服务, 作者, 输入, 应用测试库
def test_创建动态作品档案经S01且同命令不重复__aa1101(作品环境):
服务, 作者, 输入, 应用测试库 = 作品环境
回执 = 服务.保存档案(作者, "create-work", 输入)
assert 服务.保存档案(作者, "create-work", 输入) == 回执
作品 = 服务.读取作品(作者, "work-new")
assert 作品["revision"] == 1
assert 作品["content"] == {"名称": "新故事", "题材定位": "未定题材", "主题立意": "关于选择"}
assert 作品["schema_binding"]["schema_id"] == "work_core"
with 应用测试库[用途.生产].连接() as 连:
assert 连.execute("SELECT count(*) FROM muse_work_version").fetchone()[0] == 1
def test_章节重排只改目录顺序保留稳定身份__aa1102(作品环境):
服务, 作者, 输入, _ = 作品环境
服务.保存档案(作者, "create-work", 输入)
服务.添加章节(作者, "chapter-A", "work-new", "chapter-a", "开篇", 预期目录版本=0)
服务.添加章节(作者, "chapter-B", "work-new", "chapter-b", "随后", 预期目录版本=1)
回执 = 服务.重排章节(作者, "reorder", "work-new", ("chapter-b", "chapter-a"), 预期目录版本=2)
assert (
服务.重排章节(作者, "reorder", "work-new", ("chapter-b", "chapter-a"), 预期目录版本=2)
== 回执
)
目录 = 服务.读取目录(作者, "work-new")
assert 目录["revision"] == 3
assert [(v["chapter_id"], v["title"], v["position"]) for v in 目录["chapters"]] == [
("chapter-b", "随后", 1),
("chapter-a", "开篇", 2),
]
def test_人工正文保存重开与回退保留中文和历史__aa1103(作品环境):
from dataclasses import replace
from muse.正式变更.接口 import 变更错误
from muse.正文写作.接口 import 可见文本, 文本节点, 正文服务, 正文草稿, 段落, 登记正文参与者
作品, 作者, 输入, 库 = 作品环境
登记正文参与者(作品.正式.目录)
作品.保存档案(作者, "create-work", 输入)
作品.添加章节(作者, "chapter-A", "work-new", "chapter-a", "开篇", 预期目录版本=0)
正文 = 正文服务(库[用途.生产], 作品.正式)
草稿 = 正文草稿(
(
段落("p1", (文本节点(" 「中文」 e\u0301 🧑‍🏽 "),)),
段落("p2", ()),
段落("p3", (文本节点("最后一段。"),)),
)
)
首次 = 正文.保存人工(作者, "save-1", "chapter-a", 0, 草稿)
assert 正文.保存人工(作者, "save-1", "chapter-a", 0, 草稿) == 首次
重开 = 正文.读取正文(作者, "chapter-a")
assert 重开["revision"] == 1 and 重开["visible_text"] == 可见文本(草稿)
assert [p["paragraph_id"] for p in 重开["document"]["paragraphs"]] == ["p1", "p2", "p3"]
第二稿 = replace(
草稿, paragraphs=(*草稿.paragraphs[:2], 段落("p3", (文本节点("作者更新的末段。"),)))
)
正文.保存人工(作者, "save-2", "chapter-a", 1, 第二稿)
回退 = 正文.回退正文(作者, "restore-1", "chapter-a", 2, 1)
assert 回退["results"][0]["revision"] == 3
assert 正文.读取正文(作者, "chapter-a")["visible_text"] == 可见文本(草稿)
assert 正文.读取正文(作者, "chapter-a", 版本=2)["visible_text"] == 可见文本(第二稿)
with pytest.raises(变更错误) as 捕获:
正文.保存人工(作者, "stale-save", "chapter-a", 1, 第二稿)
assert 捕获.value.错误码 == "REVISION_CONFLICT"
assert 正文.读取正文(作者, "chapter-a")["revision"] == 3
def test_人工正文CLI使用实际装配并返回相同文稿__aa1104(作品环境, tmp_path):
文稿 = _核对CLI正文(作品环境, tmp_path, " CLI 原文「保持不变」🧑‍🏽\n")
assert 文稿["visible_text"] == " CLI 原文「保持不变」🧑‍🏽\n"
def test_大段中文引号换行与反斜杠经CLI和PG往返__125b0c(作品环境, tmp_path):
big = '中文"引号"\n换行\\反斜杠' * 5000
文稿 = _核对CLI正文(作品环境, tmp_path, big)
assert 文稿["visible_text"] == big
def _核对CLI正文(作品环境, tmp_path, 原文):
import json
import subprocess
import sys
from dataclasses import asdict
from muse.正文写作.接口 import 文本节点, 正文草稿, 段落
作品, 作者, 输入, _ = 作品环境
作品.保存档案(作者, "create-work", 输入)
作品.添加章节(作者, "chapter-A", "work-new", "chapter-a", "开篇", 预期目录版本=0)
配置文件 = tmp_path / "人工.toml"
配置文件.write_text("""["数据库"]
"取值方式"="受控存储"
"位置"="production.txt"
["资源"]
"发布身份"="test"
["HTTP"]
"口令文件"="作者口令.txt"
"作者ID"="author"
""")
草稿 = 正文草稿(
tuple(段落(f"cli-p{i}", (文本节点(行),)) for i, 行 in enumerate(原文.split("\n")))
)
请求文件 = tmp_path / "正文请求.json"
请求文件.write_text(
json.dumps(
{
"command_id": "cli-save",
"chapter_id": "chapter-a",
"expected_revision": 0,
"document": asdict(草稿),
},
ensure_ascii=False,
)
)
base = [sys.executable, "-m", "muse", "作品", str(配置文件)]
def 执行(*args):
p = subprocess.run([*base, *args], cwd="/", capture_output=True, text=True, timeout=15)
assert p.returncode == 0, p.stderr
return json.loads(p.stdout)
回执 = 执行("保存正文", str(请求文件))
assert 执行("保存正文", str(请求文件)) == 回执
文稿 = 执行("正文", "chapter-a")
assert 文稿["revision"] == 1
return 文稿
def test_档案读取当前绑定而不沿用上次写值结构__aa1105(作品环境):
from dataclasses import replace
from muse.作品规划.接口 import 档案保存, 档案结构
from muse.元数据.接口 import 值规则, 启用命令, 字段定义
作品, 作者, 输入, 库 = 作品环境
作品.保存档案(作者, "create-work", 输入)
# 这里只设置已完成的兼容升级;作者升级确认旅程由W11另验。
with 库[用途.维护].连接() as 连, 连.transaction():
元数据 = 元数据服务(连)
旧 = 元数据.读取结构("work_core", 1)
新 = replace(
旧,
schema_version=2,
父版本=1,
字段=(*旧.字段, 字段定义("work_core.读者约定", "读者约定", "读者约定", 值规则("text"))),
)
元数据.登记结构候选(新)
元数据.发布结构("work_core", 2, 新.内容哈希)
有效 = 元数据.读取有效结构("work_core", 2)
元数据.启用绑定(
"work:work-new",
"work-new",
有效,
启用命令(
"synthetic-upgrade",
有效.effective_schema_hash,
1,
"synthetic-reviewed-upgrade",
"work:work-new",
),
)
当前 = 作品.读取作品(作者, "work-new")
assert 当前["revision"] == 1 and 当前["schema_binding"]["base_version"] == 2
frame = 当前["schema_binding"]
结构 = 档案结构(
frame["schema_id"],
frame["base_version"],
frame["effective_schema_hash"],
当前["projection_version"],
frame["extension_version"],
)
作品.保存档案(
作者, "save-new-field", 档案保存("work-new", 1, {"读者约定": "先抛问题,再给答案"}, 结构)
)
assert 作品.读取作品(作者, "work-new")["content"]["读者约定"] == "先抛问题,再给答案"

View File

@ -0,0 +1,310 @@
"""人工候选走真实B05、S01和PG;生成/事实联合采纳另由W15/W16验收。"""
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from threading import Barrier
import pytest
from muse.作品规划.接口 import 档案保存
from muse.元数据.接口 import 元数据服务, 导入内置结构
from muse.共享.调用身份 import 内容用途, 用途, 调用身份
from muse.共享.错误 import Muse错误
from muse.启动 import 构建
from muse.正式变更.接口 import 作者动作, 变更错误
from muse.正文写作.接口 import 候选定位, 文本节点, 正文草稿, 段落
from muse.配置 import 应用配置
pytestmark = pytest.mark.数据库
def 草稿(首句="原句「保留」e\u0301", 尾句="原来的末段"):
return 正文草稿((段落("p1", (文本节点(首句),)), 段落("p2", ()), 段落("p3", (文本节点(尾句),))))
@pytest.fixture
def 候选环境(应用测试库):
with 应用测试库[用途.维护].连接() as 连, 连.transaction():
导入内置结构(元数据服务(连))
装配 = 构建(应用配置(数据库=应用测试库[用途.生产].引用, 资源发布身份="test"))
作者 = 调用身份("author", None, 用途.生产, 内容用途.生成)
作品 = 装配.要求作品()
结构 = 作品.新档案结构(作者, "work-candidate", "work_core", 1)
作品.保存档案(作者, "create", 档案保存("work-candidate", 0, {"名称": "合成候选作品"}, 结构))
作品.添加章节(作者, "chapter", "work-candidate", "chapter-candidate", "改稿", 预期目录版本=0)
正文 = 装配.要求正文()
正文.保存人工(作者, "base", "chapter-candidate", 0, 草稿())
return 正文, 作者, 应用测试库
def 新候选(正文, 作者, 命令="propose"):
回执 = 正文.创建人工候选(
作者, 命令, "chapter-candidate", 1, 草稿("改过的首句。", "改过的末段。")
)
候选 = 正文.读取候选(作者, 回执["results"][0]["candidate_id"])
定位 = 候选定位(
候选["candidate_id"], 候选["revision"], 候选["candidate_hash"], 候选["document_id"], 1
)
return 回执, 候选, 定位
def 决定(正文, 作者, 命令, 定位, 审阅, 动作=作者动作.采纳):
return 正文.决定候选(
作者,
命令,
定位,
动作,
审阅["review_id"],
审阅["review_hash"],
("document",) if 动作 is 作者动作.采纳 else (),
)
def test_候选部分选择追加版本重新审阅且采纳同一回执__aa1201(候选环境):
正文, 作者, 库 = 候选环境
首次, 候选, 定位 = 新候选(正文, 作者)
assert (
正文.创建人工候选(
作者, "propose", "chapter-candidate", 1, 草稿("改过的首句。", "改过的末段。")
)
== 首次
)
assert 正文.读取正文(作者, "chapter-candidate")["revision"] == 1
assert 候选["decision"] == "undecided" and 候选["origin"] == "author"
assert 候选["checks"]["quality"] == "not_applicable"
assert len(候选["changes"]) == 2
旧审阅 = 正文.打开候选审阅(作者, 定位)
编辑 = 正文.编辑候选(
作者,
"pick-first",
候选["candidate_id"],
1,
候选["candidate_hash"],
选择变更=(候选["changes"][0]["change_id"],),
)
assert 编辑["results"][0]["revision"] == 2
当前 = 正文.读取候选(作者, 候选["candidate_id"])
assert 当前["visible_text"] == "改过的首句。\n\n原来的末段"
assert 当前["candidate_hash"] != 候选["candidate_hash"]
assert 正文.读取候选(作者, 候选["candidate_id"], 版本=1)["visible_text"].endswith(
"改过的末段。"
)
with pytest.raises(变更错误) as 旧:
决定(正文, 作者, "old-review", 定位, 旧审阅)
assert 旧.value.错误码 == "REVISION_CONFLICT"
定位 = replace(定位, candidate_revision=2, candidate_hash=当前["candidate_hash"])
审阅 = 正文.打开候选审阅(作者, 定位)
assert 审阅["before"]["revision"] == 1 and 审阅["candidate"]["revision"] == 2
回执 = 决定(正文, 作者, "adopt", 定位, 审阅)
assert 决定(正文, 作者, "adopt", 定位, 审阅) == 回执
assert 正文.读取正文(作者, "chapter-candidate")["visible_text"] == 当前["visible_text"]
assert 正文.读取候选(作者, 候选["candidate_id"])["decision"] == "adopt"
with 库[用途.生产].连接() as 连:
assert 连.execute("SELECT count(*) FROM muse_document_version").fetchone()[0] == 2
assert 连.execute("SELECT count(*) FROM muse_candidate_decision").fetchone()[0] == 1
def test_正文改变阻止采纳但作者仍可拒绝陈旧候选__aa1202(候选环境):
正文, 作者, _ = 候选环境
_, 候选, 定位 = 新候选(正文, 作者)
正文.保存人工(作者, "new-manual", "chapter-candidate", 1, 草稿("作者后来改的原文"))
定位 = replace(定位, expected_revision=2)
审阅 = 正文.打开候选审阅(作者, 定位)
assert 审阅["candidate"]["stale"] is True
with pytest.raises(变更错误) as 冲突:
决定(正文, 作者, "cannot-adopt", 定位, 审阅)
assert 冲突.value.错误码 == "SOURCE_STALE"
for 无权 in (replace(作者, 作者="another-author"), replace(作者, 用途=用途.评测)):
with pytest.raises(Muse错误) as 越权:
正文.读取候选(无权, 候选["candidate_id"])
assert 越权.value.错误码 == "SCOPE_DENIED"
回执 = 决定(正文, 作者, "reject-stale", 定位, 审阅, 作者动作.拒绝)
assert 回执["results"] == []
assert 正文.读取正文(作者, "chapter-candidate")["revision"] == 2
assert 正文.读取候选(作者, 候选["candidate_id"])["decision"] == "reject"
def test_两个真实采纳竞争只产生一个正文与决定__aa1203(候选环境):
正文, 作者, 库 = 候选环境
_, _, 定位 = 新候选(正文, 作者)
审阅 = 正文.打开候选审阅(作者, 定位)
同时 = Barrier(2)
def 提交(序):
同时.wait()
try:
return 决定(正文, 作者, f"race-{序}", 定位, 审阅)
except 变更错误 as exc:
return exc.错误码
with ThreadPoolExecutor(2) as 池:
结果 = list(池.map(提交, (1, 2)))
assert sum(isinstance(v, dict) for v in 结果) == 1
assert any(v in ("REVISION_CONFLICT", "SOURCE_STALE") for v in 结果 if isinstance(v, str))
with 库[用途.生产].连接() as 连:
assert 连.execute("SELECT count(*) FROM muse_document_version").fetchone()[0] == 2
assert 连.execute("SELECT count(*) FROM muse_candidate_decision").fetchone()[0] == 1
def test_决定写入失败整笔正文回滚且候选仍可审阅__aa1204(候选环境):
import psycopg
正文, 作者, 库 = 候选环境
_, 候选, 定位 = 新候选(正文, 作者)
审阅 = 正文.打开候选审阅(作者, 定位)
with 库[用途.维护].连接() as 连:
连.execute("""CREATE FUNCTION reject_decision() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN RAISE EXCEPTION 'synthetic decision failure'; END $$;
CREATE TRIGGER reject_decision BEFORE INSERT ON muse_candidate_decision
FOR EACH ROW EXECUTE FUNCTION reject_decision();""")
with pytest.raises(psycopg.Error):
决定(正文, 作者, "rollback", 定位, 审阅)
assert 正文.读取正文(作者, "chapter-candidate")["revision"] == 1
assert 正文.读取候选(作者, 候选["candidate_id"])["decision"] == "undecided"
with 库[用途.维护].连接() as 连:
连.execute("DROP TRIGGER reject_decision ON muse_candidate_decision")
assert 决定(正文, 作者, "rollback", 定位, 审阅)["results"][0]["revision"] == 2
def test_网页创建审阅暂缓再由CLI按同版本采纳__aa1205(候选环境, tmp_path):
import json
import subprocess
import sys
from dataclasses import asdict
from fastapi.testclient import TestClient
from muse.接入.http.应用 import 创建应用
from muse.配置 import 服务配置
正文, 作者, 库 = 候选环境
口令 = tmp_path / "synthetic-author.txt"
口令.write_text("synthetic-author-access")
口令.chmod(0o600)
配置 = 应用配置(
数据库=库[用途.生产].引用,
资源发布身份="test",
HTTP=服务配置(
str(口令),
作者ID="author",
公开地址="http://testserver",
允许来源=("http://testserver",),
),
)
with TestClient(创建应用(配置), headers={"origin": "http://testserver"}) as 浏览器:
载荷 = {
"command_id": "http-propose",
"chapter_id": "chapter-candidate",
"branch_id": "main",
"expected_revision": 1,
"document": asdict(草稿("另一种开头。")),
}
路径 = "/api/v1/chapters/chapter-candidate/writing-candidates"
assert 浏览器.post(路径, json=载荷).status_code == 401
assert (
浏览器.post("/api/v1/session", json={"password": "synthetic-author-access"}).status_code
== 200
)
回执 = 浏览器.post(路径, json=载荷)
assert 回执.status_code == 200, 回执.text
候选ID = 回执.json()["results"][0]["candidate_id"]
列表 = 浏览器.get(路径).json()
assert len(列表) == 1 and 列表[0]["candidate_id"] == 候选ID
候选 = 浏览器.get(f"/api/v1/writing-candidates/{候选ID}").json()
定位 = {
"candidate_id": 候选ID,
"candidate_revision": 1,
"candidate_hash": 候选["candidate_hash"],
"target_ref": 候选["document_id"],
"expected_revision": 1,
}
审阅 = 浏览器.post("/api/v1/author-reviews", json=定位)
assert 审阅.status_code == 200, 审阅.text
assert 审阅.json()["before"]["visible_text"].startswith("原句")
决定载荷 = {
**定位,
"command_id": "http-defer",
"action": "defer",
"author_review_id": 审阅.json()["review_id"],
"review_hash": 审阅.json()["review_hash"],
"approved_changes": [],
}
决策路径 = f"/api/v1/writing-candidates/{候选ID}/decisions"
assert 浏览器.post(决策路径, json={**决定载荷, "decided_by": "author"}).status_code == 422
暂缓 = 浏览器.post(决策路径, json=决定载荷)
assert 暂缓.status_code == 200 and 暂缓.json()["action"] == "defer"
assert 正文.读取正文(作者, "chapter-candidate")["revision"] == 1
应用文件 = tmp_path / "candidate.toml"
应用文件.write_text(
'["数据库"]\n"取值方式"="受控存储"\n"位置"='
+ json.dumps(库[用途.生产].引用.位置)
+ '\n["资源"]\n"发布身份"="test"\n["HTTP"]\n"作者ID"="author"\n"口令文件"='
+ json.dumps(str(口令))
+ "\n"
)
请求文件 = tmp_path / "adopt.json"
请求文件.write_text(
json.dumps(
{
**决定载荷,
"command_id": "cli-adopt",
"action": "adopt",
"approved_changes": ["document"],
}
)
)
命令 = [sys.executable, "-I", "-m", "muse", "审阅", str(应用文件), "决定", str(请求文件)]
调用 = subprocess.run(命令, cwd=tmp_path, capture_output=True, text=True)
assert 调用.returncode == 0, 调用.stderr
重放 = subprocess.run(命令, cwd=tmp_path, capture_output=True, text=True)
assert 重放.returncode == 0 and json.loads(重放.stdout) == json.loads(调用.stdout)
assert 正文.读取正文(作者, "chapter-candidate")["visible_text"].startswith("另一种开头。")
def test_创作进度只从真实版本候选和任务读取__aa1206(候选环境):
from muse.任务运行.接口 import 任务服务, 任务请求, 步骤处理器, 步骤结果, 步骤计划
from muse.编排.接口 import 流程定义, 流程服务, 流程登记
正文, 作者, 库 = 候选环境
装配 = 构建(应用配置(数据库=库[用途.生产].引用, 资源发布身份="test"))
装配.要求作品().添加章节(
作者, "empty-chapter", "work-candidate", "empty-chapter", "尚未落笔", 预期目录版本=1
)
正文.创建人工候选(作者, "empty-proposal", "empty-chapter", 0, 草稿())
登记 = 流程登记()
登记.登记类型("progress-test", 必需保护=("输入校验",))
登记.登记处理器(
步骤处理器("progress-test", "1", lambda _: 步骤结果({}), "v1", "v1", 保护职责="输入校验")
)
运行 = 任务服务(库[用途.生产], 登记)
流程服务(运行, 登记).发布(
流程定义("progress-test", "1", (步骤计划("inspect", "progress-test", "1"),))
)
任务ID = 运行.创建任务(
任务请求(
"progress-test",
"queued-progress",
"author",
用途.生产,
内容用途.规划,
{},
"v1",
"synthetic",
{
"source_scope": ["work-candidate"],
"schema_versions": {},
"authorization": "synthetic-author",
"budget": {"max_calls": 0},
"stop_conditions": ["no-model-call"],
},
作品ID="work-candidate",
),
"progress-test",
"1",
)
结果 = 装配.要求作品进度().查询(作者, "work-candidate")
assert 结果["directory_revision"] == 2 and 结果["total_chapters"] == 2
assert 结果["saved_chapters"] == 1 and 结果["undecided_candidates"] == 1
assert [c["document_revision"] for c in 结果["chapters"]] == [1, 0]
assert 结果["chapters"][1]["candidates"][0]["revision"] == 1
assert 结果["tasks"] == [{"task_id": 任务ID, "state": "queued", "event_cursor": 1}]

View File

@ -0,0 +1,94 @@
-- B01拥有作品身份、动态档案版本及稳定章节目录。
CREATE TABLE public.muse_work (
work_id text PRIMARY KEY,
author_id text NOT NULL,
current_revision bigint NOT NULL CHECK(current_revision>0),
directory_revision bigint NOT NULL DEFAULT 0 CHECK(directory_revision>=0)
);
CREATE TABLE public.muse_work_version (
work_id text NOT NULL REFERENCES public.muse_work(work_id),
revision bigint NOT NULL CHECK(revision>0),
content jsonb NOT NULL,
schema_binding jsonb NOT NULL,
content_hash text NOT NULL,
created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
PRIMARY KEY(work_id,revision)
);
CREATE TABLE public.muse_chapter (
chapter_id text PRIMARY KEY,
work_id text NOT NULL REFERENCES public.muse_work(work_id),
title text NOT NULL,
position integer NOT NULL CHECK(position>0),
UNIQUE(work_id,position) DEFERRABLE INITIALLY DEFERRED
);
CREATE TABLE public.muse_directory_version (
work_id text NOT NULL REFERENCES public.muse_work(work_id),
revision bigint NOT NULL CHECK(revision>=0),
chapters jsonb NOT NULL,
nodes jsonb NOT NULL DEFAULT '[]',
created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
PRIMARY KEY(work_id,revision)
);
ALTER TABLE public.muse_work ADD CONSTRAINT work_current_version_fk
FOREIGN KEY(work_id,current_revision) REFERENCES public.muse_work_version(work_id,revision)
DEFERRABLE INITIALLY DEFERRED;
ALTER TABLE public.muse_work ADD CONSTRAINT work_current_directory_fk
FOREIGN KEY(work_id,directory_revision) REFERENCES public.muse_directory_version(work_id,revision)
DEFERRABLE INITIALLY DEFERRED;
-- 明确选定的探索来源属于B01;保留不可变快照,不将方向选择当作规划确认。
CREATE TABLE public.muse_exploration_selection (
work_id text PRIMARY KEY REFERENCES public.muse_work(work_id),
current_revision bigint NOT NULL CHECK(current_revision>0)
);
CREATE TABLE public.muse_exploration_selection_version (
work_id text NOT NULL REFERENCES public.muse_work(work_id),
revision bigint NOT NULL CHECK(revision>0),
snapshot jsonb NOT NULL,
selected_by text NOT NULL,
selected_at timestamptz NOT NULL DEFAULT clock_timestamp(),
PRIMARY KEY(work_id,revision)
);
ALTER TABLE public.muse_exploration_selection ADD CONSTRAINT selection_current_version_fk
FOREIGN KEY(work_id,current_revision) REFERENCES public.muse_exploration_selection_version(work_id,revision)
DEFERRABLE INITIALLY DEFERRED;
CREATE TABLE public.muse_plan (
plan_id text PRIMARY KEY,
work_id text NOT NULL REFERENCES public.muse_work(work_id),
type_id text NOT NULL,
chapter_id text REFERENCES public.muse_chapter(chapter_id),
node_id text,
current_revision bigint NOT NULL CHECK(current_revision>0)
);
CREATE TABLE public.muse_plan_version (
plan_id text NOT NULL REFERENCES public.muse_plan(plan_id),
revision bigint NOT NULL CHECK(revision>0),
request jsonb NOT NULL,
content jsonb NOT NULL,
content_hash text NOT NULL,
schema_binding jsonb NOT NULL,
source_dependencies jsonb NOT NULL,
author_id text NOT NULL,
created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
PRIMARY KEY(plan_id,revision)
);
ALTER TABLE public.muse_plan ADD CONSTRAINT plan_current_version_fk
FOREIGN KEY(plan_id,current_revision) REFERENCES public.muse_plan_version(plan_id,revision)
DEFERRABLE INITIALLY DEFERRED;
CREATE TABLE public.muse_plan_candidate (
candidate_id uuid PRIMARY KEY,
work_id text NOT NULL REFERENCES public.muse_work(work_id),
plan_id text NOT NULL,
current_revision bigint NOT NULL CHECK(current_revision>0)
);
CREATE TABLE public.muse_plan_candidate_version (
candidate_id uuid NOT NULL REFERENCES public.muse_plan_candidate(candidate_id),
revision bigint NOT NULL CHECK(revision>0),
request jsonb NOT NULL,
content jsonb NOT NULL,
source_dependencies jsonb NOT NULL,
candidate_hash text NOT NULL,
PRIMARY KEY(candidate_id,revision)
);
ALTER TABLE public.muse_plan_candidate ADD CONSTRAINT plan_candidate_current_version_fk
FOREIGN KEY(candidate_id,current_revision) REFERENCES public.muse_plan_candidate_version(candidate_id,revision)
DEFERRABLE INITIALLY DEFERRED;

View File

@ -0,0 +1,50 @@
-- B05正文版本只追加;段落身份与显式换行保留在受限文档树中。
CREATE TABLE public.muse_document (
document_id uuid PRIMARY KEY,
chapter_id text NOT NULL REFERENCES public.muse_chapter(chapter_id),
branch_id text NOT NULL,
current_revision bigint NOT NULL CHECK(current_revision>0),
UNIQUE(chapter_id,branch_id)
);
CREATE TABLE public.muse_document_version (
document_id uuid NOT NULL REFERENCES public.muse_document(document_id),
revision bigint NOT NULL CHECK(revision>0),
document jsonb NOT NULL,
document_hash text NOT NULL,
visible_text_hash text NOT NULL,
author_id text NOT NULL,
restored_from bigint,
created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
PRIMARY KEY(document_id,revision),
FOREIGN KEY(document_id,restored_from) REFERENCES public.muse_document_version(document_id,revision)
);
ALTER TABLE public.muse_document ADD CONSTRAINT document_current_version_fk
FOREIGN KEY(document_id,current_revision) REFERENCES public.muse_document_version(document_id,revision)
DEFERRABLE INITIALLY DEFERRED;
-- 作者改稿先存候选,决定记录由S01拥有;生成候选的证据绑定由后续写作任务扩展。
CREATE TABLE public.muse_writing_candidate (
candidate_id uuid PRIMARY KEY,
chapter_id text NOT NULL REFERENCES public.muse_chapter(chapter_id),
branch_id text NOT NULL,
origin text NOT NULL CHECK(origin='author'),
current_revision bigint NOT NULL CHECK(current_revision>0)
);
CREATE TABLE public.muse_writing_candidate_version (
candidate_id uuid NOT NULL REFERENCES public.muse_writing_candidate(candidate_id),
revision bigint NOT NULL CHECK(revision>0),
base_revision bigint NOT NULL CHECK(base_revision>=0),
base_hash text NOT NULL,
document jsonb NOT NULL,
document_hash text NOT NULL,
visible_text_hash text NOT NULL,
candidate_hash text NOT NULL,
checks jsonb NOT NULL,
author_id text NOT NULL,
created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
PRIMARY KEY(candidate_id,revision)
);
ALTER TABLE public.muse_writing_candidate ADD CONSTRAINT writing_candidate_version_fk
FOREIGN KEY(candidate_id,current_revision)
REFERENCES public.muse_writing_candidate_version(candidate_id,revision)
DEFERRABLE INITIALLY DEFERRED;