guanlan-src/app_ontology/app_ontology_guanlan/ontology/store.py

# -*- coding: utf-8 -*-
"""对象库存取: JSON 单文件 (轻量自建裁决; 写入必过 validate 闸).

2026-08-28 修: F8 Decision 禁静默覆盖 (实质差异默认 raise, force=True 时旧版本进 props.history,
逻辑时钟 when 由调用方显式传 — 如 pandas 数据窗末; 禁 datetime.now 之外的编造时间);
F12 save 原子写 (tmp + os.replace, 防写一半崩掉留下截断 json)."""
import json, os, pathlib
from .schema import validate, ConstraintError


class Store:
    def __init__(self, path):
        self.path = pathlib.Path(path)
        self.objects = {}
        if self.path.exists():
            # encoding 必须显式给: Python 文本 I/O 默认用**系统 locale 编码**, 中文 Windows 上是 cp936(GBK)。
            # 对象库是 UTF-8 的 JSON, 不写 encoding 会在中文 Windows 上读出乱码/解码失败
            # (开发机是 UTF-8 locale 所以一直没暴露; 2026-09-11 从零重算时逮到)。
            self.objects = json.loads(self.path.read_text(encoding='utf-8'))

    _DECISION_KINDS = {'验收树', '情景沙盘', '用户裁决', '验收状态机', '滚动元数据', '排程计划'}

    @staticmethod
    def _decision_kind(props):
        """F7 (Critic 逮): Decision 装 5 种异构形状, 消费方只能猜 props 有什么键 → 写入时按形状显式分型."""
        if 'branches' in props: return '验收树'
        if 'scenarios' in props: return '情景沙盘'
        if 'slots' in props or 'repair_done' in props: return '验收状态机'
        if 'decided_by' in props: return '用户裁决'
        if 'data_month' in props: return '滚动元数据'
        return '排程计划'

    def put(self, obj, force=False, when=None):
        if obj.get('type') == 'Decision':
            pr = obj.setdefault('props', {})
            pr.setdefault('decision_kind', self._decision_kind(pr))
        """写入 (过 validate 闸). Decision 类型 F8 保护:
        id 已存在且 props 实质差异 (history 除外) → 默认 raise;
        force=True 才覆盖, 且必须显式传 when (逻辑时钟), 旧版本追加进 props['history']."""
        validate(obj)                      # 不过闸不落库, 响亮 raise
        oid = obj['id']
        if obj.get('type') == 'Decision' and oid in self.objects:
            old = self.objects[oid]
            if obj is old:
                raise ConstraintError(f'[{oid}] F8: 禁以库内同一引用就地改后 put (差异检测失效); '
                                      f'先 copy.deepcopy(store.get(id)) 再改再 put')
            old_core = {k: v for k, v in old.get('props', {}).items() if k != 'history'}
            new_core = {k: v for k, v in obj.get('props', {}).items() if k != 'history'}
            if old_core != new_core:
                if not force:
                    raise ConstraintError(
                        f'[{oid}] F8: Decision 已存在且 props 实质差异 — 禁静默覆盖决策记录; '
                        f'确需更新用 put(obj, force=True, when=<逻辑时钟>), 旧版本会进 props.history')
                if when in (None, ''):
                    raise ConstraintError(
                        f'[{oid}] F8: force 覆盖 Decision 必须显式传 when (逻辑时钟, 由调用方给 — '
                        f'如 pandas 数据窗末), 禁编造时间')
                hist = list(old.get('props', {}).get('history', []))
                hist.append(dict(superseded_when=str(when), props=old_core))
                obj.setdefault('props', {})['history'] = hist
        self.objects[oid] = obj
        return oid

    def get(self, oid):
        if oid not in self.objects:
            raise KeyError(f'对象不存在: {oid}')
        return self.objects[oid]

    def by_type(self, t):
        return [o for o in self.objects.values() if o['type'] == t]

    def linked(self, oid, rel):
        return [self.get(x) for x in self.get(oid).get('links', {}).get(rel, [])]

    def save(self):
        # 落盘前全库再过一遍闸 (防手改文件绕闸); registry 传入 → C2 引用完整性/C3 声明面=证据面 全库校验
        for o in self.objects.values():
            validate(o, registry=self.objects)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        # F12 原子写: tmp + os.replace (同目录保证同文件系统, replace 原子)
        tmp = self.path.with_suffix(self.path.suffix + '.tmp')
        tmp.write_text(json.dumps(self.objects, ensure_ascii=False, indent=1), encoding='utf-8')
        os.replace(tmp, self.path)
        return str(self.path)