位置で読む API 応答を変えるとき、出荷済みの旧版を起動時に落とさないための門番を置きました

#!/usr/bin/env python3
"""出荷済み版の応答(baseline)と、これから上げる応答(candidate)を比べ、
「位置で読む」クライアントを壊す変更だけを落とす。
使い方: append_only_check.py BASELINE.json CANDIDATE.json [--mutable 3,5]
--mutable: 出荷済みの行でも値が変わってよい列の位置(例: 閲覧数)。それ以外は値まで一致を求める
終了コード: 0=安全 / 1=壊す変更あり / 2=入力不正
"""
import json, sys
from collections import Counter
 
def kind(v):
    # 値でなく「型」を見る。壁紙が増えて値が変わるのは正常だから
    if isinstance(v, bool): return "bool"
    if isinstance(v, int): return "int"
    if isinstance(v, float): return "float"
    if isinstance(v, str): return "str"
    if v is None: return "null"
    return type(v).__name__
 
def column_kinds(rows):
    """列ごとの型の集合(null は「未設定」なので型の判定から外す)"""
    cols = {}
    for r in rows:
        for i, v in enumerate(r):
            if v is not None:
                cols.setdefault(i, set()).add(kind(v))
    return cols
 
def main(base_path, cand_path, mutable=frozenset()):
    try:
        base = json.load(open(base_path, encoding="utf-8"))
        cand = json.load(open(cand_path, encoding="utf-8"))
        assert all(isinstance(r, list) for r in base + cand)
    except Exception as e:
        print(f"入力不正: {e}"); return 2
 
    problems = []
    # 1) 行ごとの列数: 出荷済みの行より短くしない
    base_w = Counter(len(r) for r in base)
    cand_w = Counter(len(r) for r in cand)
    min_base, min_cand = min(base_w), min(cand_w)
    if min_cand < min_base:
        problems.append(f"最短の行が {min_base} 列 → {min_cand} 列に短くなっています")
 
    # 2) 既存の列の型が変わっていないか(位置 i の意味が入れ替わっていないか)
    bk, ck = column_kinds(base), column_kinds(cand)
    for i in sorted(bk):
        if i not in ck:
            problems.append(f"列 {i} が候補から消えています(型 {sorted(bk[i])})")
        elif not ck[i] <= bk[i]:
            problems.append(f"列 {i} の型が {sorted(bk[i])} → {sorted(ck[i])} に変わっています")
 
    # 3) 先頭列を主キーとみなし、出荷済みの行が消えていないか・並びが入れ替わっていないか
    base_ids = [r[0] for r in base]
    cand_ids = [r[0] for r in cand]
    if cand_ids[:len(base_ids)] != base_ids:
        gone = [x for x in base_ids if x not in set(cand_ids)]
        problems.append(f"出荷済みの行の先頭一致が崩れています(消えた行 {len(gone)} 件)"
                        if gone else "出荷済みの行の並びが入れ替わっています")
 
    # 4) 同じ型の列どうしの入れ替えは型だけでは見えない。出荷済みの行は値まで一致を見る
    by_id = {r[0]: r for r in cand}
    diffs = Counter()
    for r in base:
        c = by_id.get(r[0])
        if c is None: continue
        for i, v in enumerate(r):
            if i in mutable: continue
            if i >= len(c) or c[i] != v: diffs[i] += 1
    for i, n in sorted(diffs.items()):
        problems.append(f"列 {i}: 出荷済みの行 {n} 件で値が変わっています(入れ替えの疑い)")
 
    print(f"baseline: {len(base)} 行 / 列数 {dict(sorted(base_w.items()))}")
    print(f"candidate: {len(cand)} 行 / 列数 {dict(sorted(cand_w.items()))}")
    if problems:
        for p in problems: print("NG  " + p)
        return 1
    added_cols = max(cand_w) - max(base_w)
    print(f"OK  末尾追加のみ(追加行 {len(cand)-len(base)} 件・追加列 {max(added_cols,0)} 本)")
    return 0
 
if __name__ == "__main__":
    args = [a for a in sys.argv[1:]]
    mutable = frozenset()
    if "--mutable" in args:
        k = args.index("--mutable")
        mutable = frozenset(int(x) for x in args[k + 1].split(",") if x)
        del args[k:k + 2]
    if len(args) != 2:
        print(__doc__); sys.exit(2)
    sys.exit(main(args[0], args[1], mutable))

--- ok（行を1件足し、各行の末尾に1列足した）
baseline: 3 行 / 列数 {4: 3}
candidate: 4 行 / 列数 {5: 4}
OK  末尾追加のみ(追加行 1 件・追加列 1 本)
exit=0
 
--- drop（3列目を丸ごと消した）
baseline: 3 行 / 列数 {4: 3}
candidate: 3 行 / 列数 {3: 3}
NG  最短の行が 4 列 → 3 列に短くなっています
NG  列 2 の型が ['str'] → ['int'] に変わっています
NG  列 3 が候補から消えています(型 ['int'])
NG  列 2: 出荷済みの行 3 件で値が変わっています(入れ替えの疑い)
exit=1
 
--- rowgone（出荷済みの2行目を消した）
NG  出荷済みの行の先頭一致が崩れています(消えた行 1 件)
exit=1