guanlan-src/scripts/ontology_release_build.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""本体发布层 r1/r2 的**观澜自算**生成端（源代码化清单第 ① 件）。

网关的 `/release/` 只读暴露两层（`scripts/guanlan_gateway.py` 的 `dirs = {'r1': ont/release_r1, 'r2': ont/release_r2}`）：
`/release/` 回两层的 manifest，`/release/<r>/manifest.json` 回该层清单，`/release/<r>/objects.json` 回该层对象。
这两层此前是**随包件**（全库 0 处写入方）⇒ 清过产物、只从 data/raw 重算的机器上 `/release/` 两件必然 503。

本器把两层由**观澜自算**出来（同一份本体对象库 → 契约层 / 分层覆盖层），并如实写清口径与来历：

    r1（契约层）: objects.json（对象库快照）+ manifest.json（计数、类型分布、源件 sha16、口径）
    r2（分层覆盖层）: layering.json（按类型/字段的覆盖声明）+ objects.json（同 r1 快照, 供只读对比）

★口径纪律: 这里**不新造判断**, 只是把 `ontology/objects.json` 按发布层的形态投影出来;
   manifest 里写明 `source.sha16`, 谁都能核出这两层是同一份对象库的投影。

用法: python scripts/ontology_release_build.py [--status]
"""
from __future__ import annotations

import argparse
import hashlib
import json
import pathlib
import sys
import time
from collections import Counter

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / 'scripts'))
from src import paths as P                                              # noqa: E402

LAYERS = ('release_r1', 'release_r2')


def status(farm: str | None = None) -> int:
    ont = P.ont(farm)
    out = {}
    for L in LAYERS:
        m = ont / L / 'manifest.json'
        out[L] = dict(exists=m.is_file(),
                      built=(json.loads(m.read_text(encoding='utf-8')).get('built') if m.is_file() else None))
    print(json.dumps(dict(root=P.rel(ont), layers=out,
                          note='不在位 ⇒ 网关 /release/ 两件回 503(如实); 跑本器可由本体对象库生成'),
                     ensure_ascii=False))
    return 0


def build(farm: str | None = None) -> int:
    ont = P.ont(farm)
    src = ont / 'objects.json'
    if not src.is_file():
        print(f'[X] 缺 {P.rel(src)} —— 发布层是它的投影, 先跑本体链（⑦ kb_ingest → populate）')
        return 2
    raw = src.read_bytes()
    objs = json.loads(raw.decode('utf-8'))
    by_type = Counter((v or {}).get('type') or '?' for v in objs.values())
    sha16 = hashlib.sha256(raw).hexdigest()[:16]
    built = time.strftime('%Y-%m-%d %H:%M:%S')
    common = dict(source=dict(path=P.rel(src), sha16=sha16, n_objects=len(objs)),
                  counts=dict(by_type=dict(by_type.most_common()), total=len(objs)),
                  built=built, generated_by='scripts/ontology_release_build.py (观澜自算)',
                  note='只读发布层: 由本体对象库**投影**得到, 不含新判断; source.sha16 可核同一份对象库')
    for L in LAYERS:
        d = ont / L
        d.mkdir(parents=True, exist_ok=True)
        (d / 'objects.json').write_bytes(raw)                    # 投影: 同源同 sha
        man = dict(common, release=L.replace('release_', ''))
        if L == 'release_r2':
            # r2 = 分层覆盖层: 声明"哪些类型允许被上层覆盖", 供只读对比（不合并主库 —— E9/E10 裁定）
            lay = {t: dict(overlay_fields=['props'], merge='层内覆盖', basis='类型分层')
                   for t, _n in by_type.most_common()}
            (d / 'layering.json').write_text(json.dumps(dict(
                schema='guanlan-ontology-layering/v1', built=built, source_sha16=sha16,
                note='分层覆盖声明: 仅描述允许覆盖的类型与字段; 运行期不合并主库(E9/E10 裁定)',
                layers=lay), ensure_ascii=False, indent=1), encoding='utf-8')
            man['files'] = ['manifest.json', 'objects.json', 'layering.json']
        else:
            man['files'] = ['manifest.json', 'objects.json']
        (d / 'manifest.json').write_text(json.dumps(man, ensure_ascii=False, indent=1), encoding='utf-8')
        print(f'已写 {P.rel(d)}: 对象 {len(objs)} 件 · 类型 {len(by_type)} 种 · 源 sha16 {sha16}')
    try:
        from src import derived_manifest as DM
        DM.record(P.out_root(farm), {
            f'ontology/release_r1/manifest.json': 'scripts/ontology_release_build.py (对象库投影)',
            f'ontology/release_r1/objects.json': 'scripts/ontology_release_build.py (对象库投影, 同 sha)',
            f'ontology/release_r2/manifest.json': 'scripts/ontology_release_build.py (对象库投影)',
            f'ontology/release_r2/objects.json': 'scripts/ontology_release_build.py (对象库投影, 同 sha)',
            f'ontology/release_r2/layering.json': 'scripts/ontology_release_build.py (分层覆盖声明)',
        }, by='ontology_release_build')
        print('   已自登记 → _derived_manifest.json')
    except Exception as e:
        print(f'   [i] 自登记跳过: {type(e).__name__}: {e}')
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description='本体发布层 r1/r2 的观澜自算生成端')
    ap.add_argument('--farm', default=None)
    ap.add_argument('--status', action='store_true')
    a = ap.parse_args()
    return status(a.farm) if a.status else build(a.farm)


if __name__ == '__main__':
    for _s in (sys.stdout, sys.stderr):
        try:
            _s.reconfigure(errors='replace')
        except Exception:
            pass
    sys.exit(main())