cf0f4fb985
接续 88dd570，保存 W20–W24 已实现的共享接口、业务入口、迁移、工作台、测试与文档。 W20/W22/W23 保持 in_progress，W21/W24 保持 verified；此提交不宣称方法或规则正式启用、多轮返修、真实角色评测完成。 W25 新增实验、标定、逐调用交付与角色执行及其迁移/测试/索引留在实施工作树，原有私人和旧实现保留项不纳入。 验证：离线 571、前端 33 通过；PG 469 项通过、2 项浏览器未启用，2 项误带入的 W25 用例已移出本提交；最终任务与交付边界 37 项通过。make 检查、最终类型、84 项资源及 diff 检查通过。未重跑浏览器或 Pi 宿主，不以合成调用认证外部模型效果。 独立整体审查四维通过；证据保存在 R2-20260909/提交W20-W24。
339 lines
16 KiB
Python
339 lines
16 KiB
Python
"""B02/B04提案与B05正文候选的类型化转换；不确认，不伪造研究或模型任务。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
from dataclasses import asdict
|
||
|
||
from pydantic import TypeAdapter
|
||
|
||
from muse.元数据.接口 import 元数据服务
|
||
from muse.共享.调用身份 import 调用身份
|
||
from muse.启动 import 应用装配
|
||
from muse.故事世界.接口 import 事实提案请求, 正文依据
|
||
from muse.正文写作.接口 import 可见文本哈希, 正文结构哈希, 正文草稿
|
||
from muse.知识方法.接口 import 方法提案请求, 方法结构, 证据引用
|
||
from muse.资料研究.接口 import 参考分析请求
|
||
from 关联已确认 import 冻结模式, 构建关联冻结
|
||
from 建立映射 import 稳定JSON, 旧快照, 映射配置, 迁移错误
|
||
from 映射台账 import 原回执产物
|
||
from 知识记录分流 import 分流结果
|
||
from 转换正文 import 转换文件正文
|
||
from 转换结构 import 转换目标字段
|
||
|
||
|
||
def _正版本(值) -> int:
|
||
if isinstance(值, bool) or not isinstance(值, (str, int)):
|
||
raise 迁移错误("target_version_invalid", "目标版本必须是已登记的正整数")
|
||
try:
|
||
版本 = int(值)
|
||
except ValueError:
|
||
raise 迁移错误("target_version_invalid", "目标版本不是已登记的正整数") from None
|
||
if 版本 < 1:
|
||
raise 迁移错误("target_version_invalid", "目标版本必须大于0")
|
||
return 版本
|
||
|
||
|
||
def 准备目标请求(
|
||
装配: 应用装配, 身份: 调用身份, 快照: 旧快照, 结果: 分流结果, 映射: 映射配置
|
||
) -> dict:
|
||
if not 身份.允许写正式内容 or not 身份.作者:
|
||
raise 迁移错误("target_denied", "目标提案要求已认证作者")
|
||
if 快照.source.system == "file":
|
||
return _准备正文请求(装配, 身份, 快照, 结果, 映射)
|
||
作品 = 映射.作品(快照.source, 快照.原行.get("work_id"), 映射.选择用途(快照))
|
||
if 作品["author_id"] != 身份.作者:
|
||
raise 迁移错误("target_denied", "映射中的作者与本地已认证身份不同")
|
||
if 结果.reason_code == "linked_existing_confirmation" and 结果.owner == "B02":
|
||
世界 = 装配.要求故事世界()
|
||
return 构建关联冻结(
|
||
世界.数据库,
|
||
身份,
|
||
世界.正式,
|
||
世界,
|
||
快照,
|
||
结果.导出(),
|
||
hashlib.sha256(稳定JSON(映射.值).encode()).hexdigest(),
|
||
)
|
||
if 结果.disposition != "待转换" or 结果.state != "pending":
|
||
raise 迁移错误("legacy_target_unavailable", "本入口不把旧确认声明转成新批准或活跃目标")
|
||
if 结果.owner not in {"B02", "B03", "B04"}:
|
||
raise 迁移错误("target_adapter_unavailable", "该目标域尚无旧记录公开导入接缝")
|
||
绑定 = 映射.值.get("target_schemas", {}).get(结果.owner + ":" + 结果.type_id)
|
||
if not isinstance(绑定, dict) or not isinstance(绑定.get("schema_id"), str):
|
||
raise 迁移错误("target_schema_missing", "必须明确目标S04结构与字段映射")
|
||
版本 = _正版本(绑定.get("base_version"))
|
||
with 装配.要求知识方法().数据库.连接(只读=True) as 连:
|
||
目标结构 = 元数据服务(连).读取结构(绑定["schema_id"], 版本)
|
||
if 目标结构.type_id != 结果.type_id:
|
||
raise 迁移错误("target_type_mismatch", "目标S04结构不属于已分流类型")
|
||
if 结果.type_id == "character_relation":
|
||
字段映射 = 绑定.get("fields")
|
||
if not isinstance(字段映射, dict) or any(
|
||
字段映射.get(k) != v
|
||
for k, v in (("甲方", "source"), ("乙方", "target"), ("关系类型", "relation_type"))
|
||
):
|
||
raise 迁移错误("target_relation_mapping", "关系必须使用已核对的归一端点和关系类型")
|
||
内容 = 转换目标字段(结果.content, 绑定.get("fields"))
|
||
if 结果.owner == "B03":
|
||
if 绑定.get("extension_version") is not None:
|
||
raise 迁移错误("target_schema_invalid", "参考分析不使用作品扩展")
|
||
引用 = [r for r in 结果.references if "target_ref" in r]
|
||
if not 引用 or any(r["target_ref"] != 结果.scope for r in 引用):
|
||
raise 迁移错误("target_scope_mismatch", "参考卡必须指向作品用途映射的真实资料源")
|
||
源版本 = _正版本(引用[0].get("revision"))
|
||
if any(_正版本(r.get("revision")) != 源版本 for r in 引用):
|
||
raise 迁移错误("target_source_mismatch", "同卡引用版本不一致")
|
||
with 装配.要求数据库().连接(只读=True) as 连:
|
||
源 = 装配.要求资料().读取版本依据(连, 结果.scope, 源版本)
|
||
请求 = 参考分析请求(
|
||
analysis_id=结果.target_ref,
|
||
source_id=结果.scope,
|
||
source_revision=源版本,
|
||
source_content_hash=源["content_hash"],
|
||
type_id=结果.type_id,
|
||
schema_id=绑定["schema_id"],
|
||
base_version=版本,
|
||
content=内容,
|
||
title=结果.content["name"],
|
||
brief=结果.content.get("brief", ""),
|
||
references=引用,
|
||
legacy_source={"source_key": 快照.source.记录键, "source_hash": 快照.源哈希},
|
||
)
|
||
elif 结果.owner == "B04":
|
||
引用 = tuple(
|
||
证据引用(
|
||
kind="source",
|
||
source_id=r["target_ref"],
|
||
revision=_正版本(r.get("revision")),
|
||
定位=r.get("locator", ""),
|
||
)
|
||
for r in 结果.references
|
||
if "target_ref" in r
|
||
)
|
||
if not 引用:
|
||
raise 迁移错误("target_source_missing", "方法提案必须钉住真实B03来源版本")
|
||
请求 = 方法提案请求(
|
||
method_id=结果.target_ref,
|
||
owner=身份.作者,
|
||
type_id=结果.type_id,
|
||
title=结果.content["name"],
|
||
schema=方法结构(绑定["schema_id"], 版本),
|
||
content=内容,
|
||
证据=引用,
|
||
)
|
||
else:
|
||
依据行 = [
|
||
r
|
||
for r in 映射.值.get("body_bindings", [])
|
||
if r.get("source_key") == 快照.source.记录键 and r.get("source_hash") == 快照.源哈希
|
||
]
|
||
if len(依据行) != 1 or not 依据行[0].get("sources"):
|
||
raise 迁移错误(
|
||
"target_source_missing", "有源事实缺少明确正文版本及引文，不能降级独立设定"
|
||
)
|
||
原依据 = 依据行[0]["sources"]
|
||
章级 = 快照.原行.get("source_type") == "chapter_extract"
|
||
if 章级 and (
|
||
not isinstance(原依据, list)
|
||
or any(
|
||
not isinstance(s, dict) or type(s.get("revision")) not in (str, int) for s in 原依据
|
||
)
|
||
):
|
||
raise 迁移错误(
|
||
"target_source_mismatch", "正文依据必须为对象数组，版本为明确字符串或整数"
|
||
)
|
||
依据 = tuple(TypeAdapter(正文依据).validate_python(s) for s in 原依据)
|
||
if 章级:
|
||
_核对章级来源(结果, 依据)
|
||
家族 = "relation" if 结果.type_id == "character_relation" else "entity"
|
||
结构 = 装配.要求故事世界().新事实结构(
|
||
身份,
|
||
结果.scope,
|
||
结果.target_ref,
|
||
绑定["schema_id"],
|
||
版本,
|
||
家族=家族,
|
||
扩展版本=绑定.get("extension_version"),
|
||
)
|
||
请求 = 事实提案请求(
|
||
work_id=结果.scope,
|
||
object_id=结果.target_ref,
|
||
expected_revision=0,
|
||
schema=结构,
|
||
content=内容,
|
||
sources=依据,
|
||
effective_from=依据行[0].get("effective_from"),
|
||
instance_family=家族,
|
||
independent=False,
|
||
knowledge_mode="fact",
|
||
)
|
||
return {
|
||
"owner": 结果.owner,
|
||
"scope": 结果.scope,
|
||
"subject_ref": 结果.target_ref,
|
||
"payload": asdict(请求),
|
||
}
|
||
|
||
|
||
def _核对章级来源(结果: 分流结果, 依据: tuple[正文依据, ...]) -> None:
|
||
if not 结果.references:
|
||
raise 迁移错误("target_source_mismatch", "旧章来源缺失")
|
||
原章 = 结果.references[0]
|
||
匹配 = [
|
||
s
|
||
for s in 依据
|
||
if s.chapter_id == 原章.get("target_ref") and str(s.revision) == str(原章.get("revision"))
|
||
]
|
||
引文 = 结果.content.get("source_hint", {}).get("evidence")
|
||
if not 匹配 or (
|
||
引文 is not None
|
||
and 引文 != ""
|
||
and not any(e.quote == 引文 for s in 匹配 for e in s.evidence)
|
||
):
|
||
raise 迁移错误("target_source_mismatch", "旧章声明与正文依据或原引文不一致")
|
||
|
||
|
||
def _准备正文请求(装配, 身份, 快照, 结果, 映射):
|
||
路由 = 映射.文件路由(快照)
|
||
if (
|
||
路由["route"] != "body_candidate"
|
||
or 结果.owner != "B05"
|
||
or 结果.state != "pending"
|
||
or 结果.disposition != "待转换"
|
||
):
|
||
raise 迁移错误("target_adapter_unavailable", "文件不属于未决正文候选")
|
||
作品 = 映射.作品(快照.source, 路由["legacy_work_id"], "author")
|
||
if 作品["author_id"] != 身份.作者:
|
||
raise 迁移错误("target_denied", "文件作品映射不属于当前作者")
|
||
t = 路由["target"]
|
||
当前 = 装配.要求正文().读取正文(身份, t["chapter_id"], 分支=t["branch_id"])
|
||
if 当前["work_id"] != 作品["target_ref"] or 结果.scope != 当前["work_id"]:
|
||
raise 迁移错误("target_scope_mismatch", "文件目标章不属于映射的作品")
|
||
if 当前["revision"] != t["base_revision"]:
|
||
raise 迁移错误("target_base_stale", "正文基线已变化，不自动更新映射")
|
||
return {
|
||
"owner": "B05",
|
||
"scope": 当前["work_id"],
|
||
"subject_ref": t["chapter_id"],
|
||
"payload": {
|
||
"source_key": 快照.source.记录键,
|
||
"source_hash": 快照.源哈希,
|
||
"mapping_hash": hashlib.sha256(稳定JSON(映射.值).encode()).hexdigest(),
|
||
"file_route": 路由,
|
||
"kind": "candidate",
|
||
**t,
|
||
"base_document_hash": 当前["document_hash"],
|
||
"draft": 转换文件正文(快照, 路由),
|
||
},
|
||
}
|
||
|
||
|
||
def _冻结正文草稿(冻结: dict) -> 正文草稿:
|
||
p = 冻结["payload"]
|
||
if (
|
||
p["kind"] != "candidate"
|
||
or p["chapter_id"] != 冻结["subject_ref"]
|
||
or any(
|
||
p[k] != p["file_route"]["target"][k]
|
||
for k in ("chapter_id", "branch_id", "base_revision")
|
||
)
|
||
or any(p[k] != p["file_route"][k] for k in ("source_key", "source_hash"))
|
||
):
|
||
raise 迁移错误("target_request_invalid", "冻结正文身份或基线与路由不一致")
|
||
草稿 = TypeAdapter(正文草稿).validate_python(p["draft"]["document"])
|
||
if (
|
||
正文结构哈希(草稿) != p["draft"]["document_hash"]
|
||
or 可见文本哈希(草稿) != p["draft"]["visible_text_hash"]
|
||
):
|
||
raise 迁移错误("target_request_invalid", "冻结正文内容与其哈希不一致")
|
||
return 草稿
|
||
|
||
|
||
def 提交目标请求(装配: 应用装配, 身份: 调用身份, 命令ID: str, 冻结: dict) -> dict:
|
||
if 冻结模式(冻结) == "link_confirmed":
|
||
raise 迁移错误("target_request_invalid", "历史关联不执行任何业务提交")
|
||
if 冻结["owner"] == "B05":
|
||
草稿 = _冻结正文草稿(冻结)
|
||
p = 冻结["payload"]
|
||
return 装配.要求正文().创建人工候选(
|
||
身份, 命令ID, p["chapter_id"], p["base_revision"], 草稿, 分支=p["branch_id"]
|
||
)
|
||
if 冻结["owner"] == "B03":
|
||
请求 = TypeAdapter(参考分析请求).validate_python(冻结["payload"])
|
||
if 请求.analysis_id != 冻结["subject_ref"] or 请求.source_id != 冻结["scope"]:
|
||
raise 迁移错误("target_request_invalid", "参考分析冻结的身份或来源已改变")
|
||
return 装配.要求资料().提出参考分析(身份, 命令ID, 请求)
|
||
if 冻结["owner"] == "B04":
|
||
请求 = TypeAdapter(方法提案请求).validate_python(冻结["payload"])
|
||
if 请求.method_id != 冻结["subject_ref"] or 请求.owner != 身份.作者:
|
||
raise 迁移错误("target_request_invalid", "方法冻结请求的身份已改变")
|
||
return 装配.要求知识方法().提出方法(身份, 命令ID, 请求)
|
||
if 冻结["owner"] == "B02":
|
||
请求 = TypeAdapter(事实提案请求).validate_python(冻结["payload"])
|
||
if (
|
||
请求.independent
|
||
or 请求.object_id != 冻结["subject_ref"]
|
||
or 请求.work_id != 冻结["scope"]
|
||
):
|
||
raise 迁移错误("target_request_invalid", "事实冻结请求的作用域或正文依赖已改变")
|
||
return 装配.要求故事世界().提出事实(身份, 命令ID, 请求)
|
||
raise 迁移错误("target_adapter_unavailable", "未登记的目标提交入口")
|
||
|
||
|
||
def 核对目标产物(装配: 应用装配, 身份: 调用身份, 冻结: dict, 回执: dict) -> list[dict]:
|
||
if not 身份.作者 or not 身份.允许写正式内容:
|
||
raise 迁移错误("target_denied", "迁移对账需要本地已认证作者")
|
||
if 冻结模式(冻结) == "link_confirmed":
|
||
raise 迁移错误("target_request_invalid", "历史关联必须同时核对台账正式源")
|
||
目标 = 原回执产物(冻结, 回执)
|
||
提案ID, 原版本 = 目标["id"], 目标["revision"]
|
||
if 冻结["owner"] == "B05":
|
||
_冻结正文草稿(冻结)
|
||
p = 冻结["payload"]
|
||
正文 = 装配.要求正文()
|
||
当前 = 正文.读取候选(身份, 提案ID, 版本=原版本)
|
||
基线 = 正文.读取正文(身份, p["chapter_id"], 分支=p["branch_id"], 版本=p["base_revision"])
|
||
草稿 = TypeAdapter(正文草稿).validate_python(当前["document"])
|
||
if any(当前[k] != p[k] for k in ("chapter_id", "branch_id", "base_revision")) or (
|
||
当前["origin"] != "author"
|
||
or 当前["revision"] != 原版本
|
||
or 当前["candidate_hash"] != 目标["candidate_hash"]
|
||
or 正文结构哈希(草稿) != p["draft"]["document_hash"]
|
||
or 可见文本哈希(草稿) != p["draft"]["visible_text_hash"]
|
||
or 基线["work_id"] != 冻结["scope"]
|
||
or 基线["document_hash"] != p["base_document_hash"]
|
||
):
|
||
raise 迁移错误("target_mismatch", "实际正文候选、基线或回执与冻结请求不同")
|
||
return [目标]
|
||
if 冻结["owner"] == "B03":
|
||
请求 = TypeAdapter(参考分析请求).validate_python(冻结["payload"])
|
||
with 装配.要求数据库().连接(只读=True) as 连:
|
||
当前 = 装配.要求资料().读取分析(连, 身份.作者, 提案ID)
|
||
源 = 装配.要求资料().读取版本依据(连, 请求.source_id, 请求.source_revision)
|
||
if (
|
||
当前["analysis_id"] != 冻结["subject_ref"]
|
||
or 当前["source_id"] != 冻结["scope"]
|
||
or 当前["origin"] != "legacy_import"
|
||
or 当前["revision"] != 原版本
|
||
or 当前["payload"]["request"] != asdict(请求)
|
||
or 当前["content_hash"] != 目标["content_hash"]
|
||
or 当前["source_content_hash"] != 源["content_hash"]
|
||
):
|
||
raise 迁移错误("target_mismatch", "实际分析版本或原来源与冻结请求不一致")
|
||
return [目标]
|
||
if 冻结["owner"] == "B04":
|
||
当前 = 装配.要求知识方法().读取方法提案(身份, 提案ID)
|
||
if 当前["method_id"] != 冻结["subject_ref"] or 当前["owner"] != 身份.作者:
|
||
raise 迁移错误("target_mismatch", "实际方法提案不属于原身份")
|
||
elif 冻结["owner"] == "B02":
|
||
当前 = 装配.要求故事世界().读取提案(身份, 提案ID)
|
||
if 当前["object_id"] != 冻结["subject_ref"] or 当前["work_id"] != 冻结["scope"]:
|
||
raise 迁移错误("target_mismatch", "实际世界提案不属于原作用域")
|
||
else:
|
||
raise 迁移错误("target_adapter_unavailable", "未登记的目标回查入口")
|
||
if 当前["revision"] < 原版本:
|
||
raise 迁移错误("target_mismatch", "实际候选版本低于原回执")
|
||
# 导入版本取S01原回执，不以作者后续编辑后的当前版本冒充导入结果。
|
||
return [目标]
|