"""PG作品依赖与逐块共享候选：只经B01/B03/B05公开入口提交、精确读回。""" from dataclasses import asdict from pydantic import TypeAdapter from muse.作品规划.接口 import 档案保存 from muse.正式变更.接口 import 固定哈希 from muse.正文写作.接口 import 可见文本, 可见文本哈希, 正文结构哈希, 正文草稿 from muse.资料研究.接口 import 导入请求 from PG作品映射 import PG目标ID, 核对PG组, 指针 from 建立映射 import 稳定JSON, 迁移错误 from 转换PG正文 import 转换PG正文 def PG分流(源, 映射, 已知源): # 分流身份只取固定清单；依赖是否到齐不改变原路由，分批期间留reserved。 from PG作品映射 import PG源身份, PG绑定 组, 章 = PG绑定(映射, 源) 作者 = 映射.值.get("pg_author_id") if not isinstance(作者, str) or not 作者: raise 迁移错误("pg_mapping_invalid", "PG映射必须绑定应用配置作者，不从旧owner/tenant推导") 作品身份 = PG源身份(组["work"]) if 组["use"] == "reference": owner, kind, scope, target = "B03", "reference", 作品身份.对象键, 作品身份.对象键 else: scope = PG目标ID(作品身份, 作者, "work") if 章 is None: owner, kind, target = "B01", "profile", scope elif 指针(源) == 章["chapter"]: owner, kind, target = "B01", "directory", PG目标ID(源, 作者, "chapter") else: owner, kind, target = "B05", "body", PG目标ID(PG源身份(章["chapter"]), 作者, "chapter") return { "owner": owner, "type_id": "pg_" + kind, "scope": scope, "target_ref": target, "state": "pending", "content": {"pg_kind": kind, "use": 组["use"], "work_source": 组["work"]}, } def 准备PG请求(装配, 身份, 源, 判定, 映射, 已知源): if 映射.值.get("pg_author_id") != 身份.作者 or not 身份.允许写正式内容: raise 迁移错误("target_denied", "PG目标作者必须来自当前应用身份") 当前分流 = PG分流(源, 映射, 已知源) if any(当前分流[k] != getattr(判定, k) for k in ("owner", "scope", "target_ref", "type_id")): raise 迁移错误("pg_mapping_invalid", "PG分流身份发生变化") 组, 作品, 章节 = 核对PG组(映射, 源, 已知源) kind = 当前分流["content"]["pg_kind"] command_source = 指针(源) dependencies = [指针(作品)] if kind == "profile": work_id = 判定.scope 结构 = 装配.要求作品().新档案结构(身份, work_id, "work_core", 1) payload = asdict(档案保存(work_id, 0, {"名称": 作品.原行["title"]}, 结构)) elif kind == "directory": i = next(i for i, (c, _) in enumerate(章节) if 指针(c) == 指针(源)) 目录 = 装配.要求作品().读取目录(身份, 判定.scope) 前章 = [PG目标ID(c, 身份.作者, "chapter") for c, _ in 章节[:i]] if 目录["revision"] != i or [c["chapter_id"] for c in 目录["chapters"]] != 前章: raise 迁移错误("pg_directory_stale", "目录必须由本清单的真实前序章节构成") dependencies.extend(指针(c) for c, _ in 章节[: i + 1]) payload = { "work_id": 判定.scope, "chapter_id": 判定.target_ref, "title": 源.原行["title"], "expected_revision": i, "chapter_ids": [*前章, 判定.target_ref], } elif kind == "body": 章, 块组 = next((c, b) for c, b in 章节 if any(指针(源) == 指针(x) for x in b)) 当前 = 装配.要求正文().读取正文(身份, 判定.target_ref, 分支="main") if 当前["work_id"] != 判定.scope or 当前["revision"] != 0: raise 迁移错误("target_base_stale", "PG候选只面向真实本作品章的空正式基线") dependencies.extend([指针(章), *(指针(b) for b in 块组)]) command_source = 指针(块组[0]) payload = { "chapter_id": 判定.target_ref, "branch_id": "main", "base_revision": 0, "base_document_hash": 当前["document_hash"], "draft": 转换PG正文(块组), } else: for 章, 块组 in 章节: dependencies.extend([指针(章), *(指针(b) for b in 块组)]) 文本, 对照 = 组装PG参考(章节) command_source = 指针(作品) 请求 = 导入请求( kind="reference", title=作品.原行["title"], origin="legacy-pg:" + 稳定JSON([身份.作者, 作品.source.对象键]), content=文本, authorized_uses=(), ) payload = {"request": asdict(请求), "chapter_map": 对照} return { "owner": 判定.owner, "scope": 判定.scope, "subject_ref": 判定.target_ref, "pg_kind": kind, "legacy_source": 指针(源), "command_source": command_source, "dependencies": dependencies, "mapping_hash": 固定映射哈希(映射), "payload": payload, } def 组装PG参考(章节): 文本, 对照, 起 = [], [], 0 for 章, 块组 in 章节: draft = 转换PG正文(块组) if 块组 else None text = 可见文本(TypeAdapter(正文草稿).validate_python(draft["document"])) if draft else "" 对照.append( { **指针(章), "order_no": 章.原行["order_no"], "source_start": 起, "source_end": 起 + len(text), "draft": draft, } ) 起 += len(text) + 1 文本.append(text) return "\n".join(文本), 对照 def 固定映射哈希(映射): import hashlib return hashlib.sha256(稳定JSON(映射.值).encode()).hexdigest() def PG命令ID(身份, 冻结): from 映射台账 import 迁移命令ID return 迁移命令ID(身份.作者, 冻结["command_source"]["source_key"]) def 提交PG请求(装配, 身份, 冻结): p, kind = 冻结["payload"], 冻结["pg_kind"] cmd = PG命令ID(身份, 冻结) if kind == "profile": return 装配.要求作品().保存档案(身份, cmd, TypeAdapter(档案保存).validate_python(p)) if kind == "directory": return 装配.要求作品().添加章节( 身份, cmd, p["work_id"], p["chapter_id"], p["title"], 预期目录版本=p["expected_revision"], ) if kind == "body": draft = TypeAdapter(正文草稿).validate_python(p["draft"]["document"]) return 装配.要求正文().创建人工候选( 身份, cmd, p["chapter_id"], p["base_revision"], draft, 分支=p["branch_id"] ) if kind == "reference": with 装配.要求数据库().连接() as 连, 连.transaction(): 结果 = 装配.要求资料().导入( 连, 身份.作者, TypeAdapter(导入请求).validate_python(p["request"]) ) # 重复位只是本次调用情况，不冒充另一个不可变导入回执。 # 参考导入零授权用途，原文须当前用途才能返回，迁移只回读回执与存储哈希。 版本 = 装配.要求资料().读取版本回执(连, 结果.source_id, 结果.revision) return { "source_id": 结果.source_id, "revision": 结果.revision, "content_hash": 结果.content_hash, "import_result": 版本["import_result"], } raise 迁移错误("target_request_invalid", "未知PG目标种类") def 核对PG产物(装配, 身份, 冻结, 回执): p, kind = 冻结["payload"], 冻结["pg_kind"] if not 身份.允许写正式内容 or not 身份.作者: raise 迁移错误("target_denied", "PG核对需要应用作者身份") if kind == "reference": with 装配.要求数据库().连接(只读=True) as 连: v = 装配.要求资料().读取版本回执(连, 回执["source_id"], 回执["revision"]) # 参考导入零授权用途，原文受当前用途门禁保护，迁移不自开用途； # 核对冻结请求重算哈希、库内存储哈希与不可变回执三者一致。 import hashlib h = hashlib.sha256(p["request"]["content"].encode()).hexdigest() if ( v["content_hash"] != h or 回执["content_hash"] != h or v["import_result"] != 回执["import_result"] or v["import_result"].get("导入者") != 身份.作者 ): raise 迁移错误("target_mismatch", "参考原文版本与冻结来源或作者不同") 来源 = [ s for s in 装配.要求资料().列出来源(连) if str(s["source_id"]) == 回执["source_id"] ] if ( len(来源) != 1 or 来源[0]["origin"] != p["request"]["origin"] or 来源[0]["kind"] != "reference" ): raise 迁移错误("target_mismatch", "参考源身份不同") return [ { "id": 回执["source_id"], "revision": 回执["revision"], "kind": "source", "owner": "B03", "scope": 冻结["scope"], "subject_ref": 冻结["subject_ref"], "content_hash": h, } ] 真实 = 装配.要求作品().正式.读取回执(身份, PG命令ID(身份, 冻结)) if 稳定JSON(真实) != 稳定JSON(回执) or 回执.get("action") != "manual_save": raise 迁移错误("target_receipt_invalid", "PG依赖/候选必须来自真实人工保存回执，不接受确认") if kind == "profile": rows = [ r for r in 回执["results"] if r.get("work_id") == p["work_id"] and r.get("revision") == 1 ] v = 装配.要求作品().读取档案版本依据(身份, p["work_id"], 1) if ( len(rows) != 1 or v["content_hash"] != 固定哈希(p["content"]) or v["schema_binding"] != rows[0]["schema_binding"] ): raise 迁移错误("target_mismatch", "PG作品档案版本与真实回执/冻结不同") id_, revision = p["work_id"], 1 elif kind == "directory": revision = p["expected_revision"] + 1 rows = [ r for r in 回执["results"] if r.get("work_id") == p["work_id"] and r.get("revision") == revision and r.get("chapter_ids") == p["chapter_ids"] ] v = 装配.要求作品().读取节点目录(身份, p["work_id"], 版本=revision) chapters = [c for c in v["nodes"] if c["kind"] == "chapter"] if ( len(rows) != 1 or [c["node_id"] for c in chapters] != p["chapter_ids"] or chapters[-1]["title"] != p["title"] ): raise 迁移错误("target_mismatch", "PG目录版本与真实回执/原章不同") id_ = p["chapter_id"] elif kind == "body": id_ = 回执["target_ref"] rows = [ r for r in 回执["results"] if r.get("candidate_id") == id_ and r.get("revision") == 1 ] v = 装配.要求正文().读取候选(身份, id_, 版本=1) doc = TypeAdapter(正文草稿).validate_python(v["document"]) base = 装配.要求正文().读取正文( 身份, p["chapter_id"], 分支=p["branch_id"], 版本=p["base_revision"] ) if ( len(rows) != 1 or v["origin"] != "author" or any(v[k] != p[k] for k in ("chapter_id", "branch_id", "base_revision")) or v["candidate_hash"] != rows[0]["candidate_hash"] or 正文结构哈希(doc) != p["draft"]["document_hash"] or 可见文本哈希(doc) != p["draft"]["visible_text_hash"] or base["work_id"] != 冻结["scope"] or base["document_hash"] != p["base_document_hash"] ): raise 迁移错误("target_mismatch", "PG正文候选版本、原文或基线不同") revision = 1 else: raise 迁移错误("target_request_invalid", "未知PG目标种类") return [ { "id": id_, "revision": revision, "kind": {"profile": "work", "directory": "chapter", "body": "candidate"}[kind], "owner": 冻结["owner"], "scope": 冻结["scope"], "subject_ref": 冻结["subject_ref"], } ]