140 lines
4.5 KiB
Python

"""多粒度写作任务合同:形成目标、范围、前置和预算;不调用模型、不采纳候选。"""
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__ = [
"写作任务合同",
"写作任务错误",
"写作粒度集",
"字符选区",
"创建写作任务",
"正文错误",
]