"""多粒度写作任务合同:形成目标、范围、前置和预算;不调用模型、不采纳候选。""" from __future__ import annotations from dataclasses import dataclass, field from typing import Any from uuid import NAMESPACE_URL, uuid5 from muse.作品规划.接口 import 读取章节归属 from muse.共享.错误 import Muse错误 from muse.正文写作.存储 import 正文存储 from muse.正文写作.模型 import 正文错误 from muse.正文写作.正文格式 import 可见文本, 文稿身份 class 写作任务错误(Muse错误): 错误码 = "WRITING_TASK_INVALID" def 写作粒度集() -> tuple[str, str, str]: """多粒度写作的稳定取值:新章、扩写、选段修改。""" return ("new_chapter", "expand_scene", "edit_selection") @dataclass(frozen=True, slots=True) class 字符选区: """候选可见文本上的字符区间;半开区间 [起点, 终点)。""" 起点: int 终点: int def __post_init__(self) -> None: if ( isinstance(self.起点, bool) or isinstance(self.终点, bool) or not (0 <= self.起点 < self.终点) ): raise 写作任务错误("字符选区必须是 0 ≤ 起点 < 终点的整数区间") @dataclass(frozen=True, slots=True) class 写作任务合同: """一次多粒度写作的业务目标;身份由作者与命令确定性派生,重放同身份。""" 任务ID: str 命令ID: str 作者: str 粒度: str 用途说明: str work_id: str chapter_id: str branch_id: str 基线版本: int 选区: tuple[字符选区, ...] = () 保护区: tuple[字符选区, ...] = () 预算: dict[str, Any] = field(default_factory=dict) 停止条件: tuple[str, ...] = () def __post_init__(self) -> None: if self.粒度 not in 写作粒度集(): raise 写作任务错误(f"未知写作粒度:{self.粒度}") if self.粒度 == "edit_selection" and not self.选区: raise 写作任务错误("选段修改必须携带明确选区") if self.基线版本 < 0 or not self.任务ID or not self.命令ID or not self.作者: raise 写作任务错误("写作任务缺少身份或基线版本不合法") for 甲 in self.选区: for 乙 in self.保护区: if 甲.起点 < 乙.终点 and 乙.起点 < 甲.终点: raise 写作任务错误("选区与保护区不能重叠") def 创建写作任务( 连, 身份, 命令ID: str, 粒度: str, *, work_id: str, chapter_id: str, 分支: str = "main", 用途说明: str = "", 选区: tuple[字符选区, ...] = (), 保护区: tuple[字符选区, ...] = (), 预算: dict[str, Any] | None = None, 停止条件: tuple[str, ...] = (), ) -> 写作任务合同: """按当前章位冻结写作任务;章位归属与选区边界在创建时机械校验。""" if not 命令ID: raise 写作任务错误("写作任务需要明确命令身份") 章节 = 读取章节归属(连, 身份.作者, chapter_id) if 章节["work_id"] != work_id: raise 写作任务错误("章节不属于目标作品") 存储 = 正文存储(连) 文档ID = 文稿身份(chapter_id, 分支) 正文 = 存储.读取(文档ID) 基线版本 = 正文["revision"] if 正文 else 0 任务ID = str(uuid5(NAMESPACE_URL, "muse:B05:writing_task:" + _身份哈希(身份.作者, 命令ID))) 任务 = 写作任务合同( 任务ID, 命令ID, 身份.作者, 粒度, 用途说明, work_id, chapter_id, 分支, 基线版本, 选区, 保护区, 预算 or {}, 停止条件, ) if 基线版本 == 0: if 任务.粒度 != "new_chapter": raise 写作任务错误("没有既有正文,只允许新章粒度") return 任务 可见长度 = len(可见文本(存储.读取草稿(文档ID, 基线版本))) for 区 in 任务.选区 + 任务.保护区: if 区.终点 > 可见长度: raise 写作任务错误(f"选区越界:终点 {区.终点} 超出当前正文长度 {可见长度}") return 任务 def _身份哈希(作者: str, 命令ID: str) -> str: from muse.正式变更.接口 import 固定哈希 return 固定哈希([作者, 命令ID]) __all__ = [ "写作任务合同", "写作任务错误", "写作粒度集", "字符选区", "创建写作任务", "正文错误", ] _写作输出合同 = { "type": "object", "properties": { "paragraphs": { "description": ( "正文段落数组;每项 text 是完整段落文本,不能包含 LF 或 CR 换行符," "需要另起一段时另建一个段落对象;系统不改写或规范化段内换行。" ), "type": "array", "minItems": 1, "items": { "type": "object", "properties": { "text": {"type": "string", "minLength": 1, "not": {"pattern": "[\\r\\n]"}} }, "required": ["text"], "additionalProperties": False, }, }, "blocked": { "description": ( "写作死结出口:仅当细纲与前置事实存在不可化解冲突" "(任何写法都必然虚构冲突事实)时输出,reason 一句话说明。" ), "type": "object", "properties": {"reason": {"type": "string", "minLength": 1}}, "required": ["reason"], "additionalProperties": False, }, }, "oneOf": [{"required": ["paragraphs"]}, {"required": ["blocked"]}], "additionalProperties": False, } def 写作输出合同() -> dict: """生产与评测共享同一无工具写作输出外壳;返回独立副本。""" import copy return copy.deepcopy(_写作输出合同) def 生成正文模板() -> tuple[str, str]: """读取发布包登记的写作模板及实际字节哈希。 角色身份、权限、预算及输出合同由调用方装配,不作为写作指令发给模型。 """ import hashlib from muse.资源加载 import 打开资源 data = 打开资源("正文/生成正文.md") return data.decode("utf-8"), hashlib.sha256(data).hexdigest()