216 lines
11 KiB
Python
216 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""规划落库 —— 把 planner 产出的 schema 驱动结构化内容落 example_planning_section（shadow→confirmed）。
|
||
|
||
落库设计 §2.6：规划/细纲/设定的正式权威在库（planning_section），works/<书名>/ 文件退为人类可读留痕。
|
||
- write：落一条 shadow 规划（planner 产出结构化 payload，主会话调本脚本落库；planner 不碰库/git）。
|
||
- confirm：shadow→confirmed（用户确认后翻态，confirmed_at 落时间）——确认后的规划才进生成上下文。
|
||
- 已确认细纲取用见 planning SKILL.md 的 SQL。
|
||
|
||
payload = 对应 section_type 的 schema 字段结构化内容（JSON），由 planner/主会话按 meta/schemas 产出。
|
||
--dry-run 试跑：插入/翻态后回滚，校验但不落库。
|
||
"""
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import yaml
|
||
|
||
from muse_db import connect
|
||
|
||
CREATOR = "planning"
|
||
SECTION_TYPES = ("setting", "outline", "state", "assembly", "fine_outline")
|
||
# 书级 section 归 制定作品规划；章级 fine_outline 写入必须显式声明 owner=规划下一章（防双入口）。
|
||
BOOK_LEVEL_SECTION_TYPES = ("setting", "outline", "state", "assembly")
|
||
FINE_OUTLINE_OWNER = "规划下一章"
|
||
DEFAULT_OWNER_SKILL = "制定作品规划"
|
||
# 字段权威（git 侧设计稿）：落库字段覆盖门禁按 schema_type 读这里的 <型>.yaml。
|
||
PROJECT_ROOT = next(
|
||
parent
|
||
for parent in (Path(__file__).resolve().parent, *Path(__file__).resolve().parents)
|
||
if (parent / "AGENTS.md").is_file() and (parent / ".git").exists()
|
||
)
|
||
SCHEMAS_DIR = PROJECT_ROOT / "muse" / "content" / "meta" / "schemas"
|
||
|
||
|
||
def assert_section_owner(section_type: str, owner_skill: str) -> None:
|
||
"""写入侧归属硬门：fine_outline 只能由 规划下一章 声明写入。"""
|
||
if section_type == "fine_outline" and owner_skill != FINE_OUTLINE_OWNER:
|
||
raise ValueError(
|
||
"FINE_OUTLINE_OWNER: section_type=fine_outline 只允许 "
|
||
f"--owner-skill {FINE_OUTLINE_OWNER}；书级设定/大纲请用 制定作品规划，"
|
||
"章级细纲走 规划下一章"
|
||
)
|
||
if section_type in BOOK_LEVEL_SECTION_TYPES and owner_skill == FINE_OUTLINE_OWNER:
|
||
raise ValueError(
|
||
f"BOOK_LEVEL_OWNER: section_type={section_type} 归 制定作品规划，"
|
||
f"不得用 owner_skill={FINE_OUTLINE_OWNER} 写入"
|
||
)
|
||
|
||
|
||
def _load_schema(schema_type):
|
||
"""从 meta/schemas/<schema_type>.yaml 读字段清单、必填集与门禁档位。
|
||
|
||
返回 (fields, required, enforce)：
|
||
fields = {字段名: [别名...]}（特有字段全集，含推荐）；
|
||
required= 必填字段名列表（只有这些参与失败关闭）；
|
||
enforce = 该型是否「字段覆盖门禁: 强制」。
|
||
schema 文件不存在或非法 → (None, [], False)，门禁跳过（该型暂未纳入覆盖校验，
|
||
如中文键的老 schema 与英文 payload 尚未对齐，不溯及强拦）。
|
||
"""
|
||
if not schema_type:
|
||
return None, [], False
|
||
path = SCHEMAS_DIR / f"{schema_type}.yaml"
|
||
if not path.exists():
|
||
return None, [], False
|
||
try:
|
||
doc = yaml.safe_load(path.read_text(encoding="utf-8"))
|
||
except yaml.YAMLError:
|
||
return None, [], False
|
||
if not isinstance(doc, dict):
|
||
return None, [], False
|
||
fields = {}
|
||
for item in doc.get("特有字段", []) or []:
|
||
if not isinstance(item, dict) or not item:
|
||
continue
|
||
if "key" in item: # 流式 - { key: X, 说明:..., aliases:[...] }
|
||
name = str(item["key"])
|
||
aliases = [str(a) for a in (item.get("aliases", []) or [])]
|
||
if "alias" in item:
|
||
aliases.append(str(item["alias"]))
|
||
elif len(item) == 1: # 块式 - X:\n 说明:... （单键映射，键即字段名）
|
||
name = next(iter(item))
|
||
sub = item[name]
|
||
aliases = [str(a) for a in (sub.get("aliases", []) or [])] if isinstance(sub, dict) else []
|
||
if isinstance(sub, dict) and "alias" in sub:
|
||
aliases.append(str(sub["alias"]))
|
||
else:
|
||
continue
|
||
fields[name] = aliases
|
||
required = [str(x) for x in (doc.get("必填字段", []) or [])]
|
||
enforce = str(doc.get("字段覆盖门禁", "")).strip() == "强制"
|
||
return fields, required, enforce
|
||
|
||
|
||
def check_field_coverage(schema_type, payload):
|
||
"""机械校验 payload 对 schema 字段的覆盖（「字段全覆盖」纪律的门禁化）。
|
||
|
||
每个字段算「已覆盖」当且仅当：字段名或某别名出现在 payload，或在 payload['字段存疑']
|
||
里被显式标注了非空原因（逃生口：确无依据的字段可声明存疑而非静默缺省）。
|
||
返回 {schema_type, enforced, required, covered, missing_required, missing_recommended, excused}。
|
||
enforced=True 且 missing_required 非空时，由 write_section 失败关闭。
|
||
"""
|
||
fields, required, enforce = _load_schema(schema_type)
|
||
if fields is None:
|
||
return {"schema_type": schema_type, "enforced": False, "required": [],
|
||
"covered": [], "missing_required": [], "missing_recommended": [], "excused": {}}
|
||
excused_raw = payload.get("字段存疑", {})
|
||
excused = excused_raw if isinstance(excused_raw, dict) else {}
|
||
|
||
def covered(name):
|
||
if any(key in payload for key in [name, *fields.get(name, [])]):
|
||
return True
|
||
return name in excused and str(excused[name]).strip() != ""
|
||
|
||
required_set = set(required)
|
||
missing_required = [name for name in required if not covered(name)]
|
||
missing_recommended = [name for name in fields
|
||
if name not in required_set and not covered(name)]
|
||
return {"schema_type": schema_type, "enforced": enforce, "required": required,
|
||
"covered": [name for name in fields if covered(name)],
|
||
"missing_required": missing_required,
|
||
"missing_recommended": missing_recommended,
|
||
"excused": dict(excused)}
|
||
|
||
|
||
def write_section(work_id, *, section_type, payload, target_chapter=None, schema_type=None,
|
||
version=1, state="shadow", creator=CREATOR, owner_skill=DEFAULT_OWNER_SKILL,
|
||
dry_run=False):
|
||
"""落一条规划（默认 shadow）。返回 {status, id, state}。"""
|
||
if section_type not in SECTION_TYPES:
|
||
raise ValueError(f"section_type 非法：{section_type}（应为 {SECTION_TYPES}）")
|
||
assert_section_owner(section_type, owner_skill)
|
||
if section_type == "fine_outline" and target_chapter is None:
|
||
raise ValueError("fine_outline 是章级 section，必须给 target_chapter")
|
||
if not isinstance(payload, dict) or not payload:
|
||
raise ValueError("payload 必须是非空 JSON 对象（schema 字段结构化内容）")
|
||
# 字段覆盖门禁：声明「强制」的 schema（如 fine_outline）缺必填字段即失败关闭；
|
||
# 推荐字段缺失只随返回值报告，不拦。未纳入校验的型（无 schema 或未标强制）跳过。
|
||
coverage = check_field_coverage(schema_type, payload)
|
||
if coverage["enforced"] and coverage["missing_required"]:
|
||
raise ValueError(
|
||
f"字段覆盖门禁：{schema_type} 缺必填字段 {coverage['missing_required']}；"
|
||
f"补齐后再落库，确无依据的字段在 payload['字段存疑'] 里逐条标注原因")
|
||
with connect() as conn:
|
||
try:
|
||
row = conn.execute(
|
||
"INSERT INTO example_planning_section(work_id, target_chapter, section_type, schema_type, "
|
||
"payload, version, state, creator) VALUES (%s,%s,%s,%s,%s::jsonb,%s,%s,%s) RETURNING id, state",
|
||
(work_id, target_chapter, section_type, schema_type,
|
||
json.dumps(payload, ensure_ascii=False), version, state, creator)).fetchone()
|
||
if dry_run:
|
||
conn.rollback()
|
||
return {"status": "dry_run_ok", "id": row[0], "state": row[1],
|
||
"note": "试跑已回滚，未落库", "fieldCoverage": coverage,
|
||
"ownerSkill": owner_skill}
|
||
conn.commit()
|
||
return {"status": "written", "id": row[0], "state": row[1],
|
||
"fieldCoverage": coverage, "ownerSkill": owner_skill}
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
|
||
|
||
def confirm_section(section_id, *, updater=CREATOR, dry_run=False):
|
||
"""确认规划：shadow→confirmed。返回 {status, id, state}。"""
|
||
with connect() as conn:
|
||
try:
|
||
row = conn.execute(
|
||
"UPDATE example_planning_section SET state='confirmed', confirmed_at=CURRENT_TIMESTAMP, "
|
||
"updater=%s WHERE id=%s AND state='shadow' AND deleted=false RETURNING id, state",
|
||
(updater, section_id)).fetchone()
|
||
if not row:
|
||
cur = conn.execute("SELECT state FROM example_planning_section WHERE id=%s", (section_id,)).fetchone()
|
||
raise ValueError(f"规划 {section_id} 不可确认（不存在或 state={cur[0] if cur else '无'}，只有 shadow 可确认）")
|
||
if dry_run:
|
||
conn.rollback()
|
||
return {"status": "dry_run_ok", "id": row[0], "state": row[1], "note": "试跑已回滚，未落库"}
|
||
conn.commit()
|
||
return {"status": "confirmed", "id": row[0], "state": row[1]}
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
ap = argparse.ArgumentParser(description="规划落库（shadow→confirmed）")
|
||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
pw = sub.add_parser("write", help="落一条规划（默认 shadow）")
|
||
pw.add_argument("payload_json", help="payload JSON 文件路径（schema 字段结构化内容）")
|
||
pw.add_argument("--work-id", type=int, required=True)
|
||
pw.add_argument("--section-type", required=True, choices=SECTION_TYPES)
|
||
pw.add_argument("--target-chapter", type=int, default=None)
|
||
pw.add_argument("--schema-type", default=None)
|
||
pw.add_argument("--version", type=int, default=1)
|
||
pw.add_argument("--creator", default=CREATOR)
|
||
pw.add_argument(
|
||
"--owner-skill",
|
||
default=DEFAULT_OWNER_SKILL,
|
||
choices=(DEFAULT_OWNER_SKILL, FINE_OUTLINE_OWNER),
|
||
help="写入归属：fine_outline 必须 规划下一章；书级 section 用 制定作品规划",
|
||
)
|
||
pw.add_argument("--dry-run", action="store_true")
|
||
pc = sub.add_parser("confirm", help="确认规划（shadow→confirmed）")
|
||
pc.add_argument("section_id", type=int)
|
||
pc.add_argument("--updater", default=CREATOR)
|
||
pc.add_argument("--dry-run", action="store_true")
|
||
args = ap.parse_args()
|
||
if args.cmd == "write":
|
||
payload = json.loads(Path(args.payload_json).read_text(encoding="utf-8"))
|
||
out = write_section(args.work_id, section_type=args.section_type, payload=payload,
|
||
target_chapter=args.target_chapter, schema_type=args.schema_type,
|
||
version=args.version, creator=args.creator,
|
||
owner_skill=args.owner_skill, dry_run=args.dry_run)
|
||
else:
|
||
out = confirm_section(args.section_id, updater=args.updater, dry_run=args.dry_run)
|
||
print(json.dumps(out, ensure_ascii=False))
|