ツール出力10万文字の境界で消える一行 — 実測から組み立てた出力封筒の設計

# 誤: バイト数。日本語混じりの出力では境界より大きく出る
grep -rn 'claude' content/articles/ja --include='*.mdx' | wc -c   # 1218722
# 正: 文字数。退避の境界と同じ単位
grep -rn 'claude' content/articles/ja --include='*.mdx' | wc -m   #  945213

import json, pathlib
 
raw = pathlib.Path("src/data/articles.json").read_text()
print(f"bytes={len(raw)}  ({len(raw)/100_000:.2f}x)")   # bytes=822582  (8.23x)
 
cut = raw[:100_000]
try:
    json.loads(cut)
    print("先頭100Kのみ: パース成功")
except json.JSONDecodeError as e:
    print(f"先頭100Kのみ: パース失敗 -> {e}")
    print(f"カット位置の直前40文字: ...{cut[-40:]!r}")

# 現在値と、境界に到達するまでの残り本数を出す
now=$(ls -R content | wc -c)
ja=$(find content/articles/ja -name '*.mdx' | wc -l)
python3 -c "
n, a = $now, $ja
per = n / a
need = (100_000 - n) / per
print(f'現在 {n} bytes / JA {a} 本 / 1本あたり {per:.1f} bytes')
print(f'到達まで残り約 {need:.0f} 本（日英2本/日なら約 {need/2:.0f} 日）')"

#!/usr/bin/env python3
"""outbox.py — ツール出力を、判断材料が先頭に来る封筒に詰め直す。
 
  使い方:  <command> | python3 outbox.py --label build --tail 20
  標準入力を受け取り、ヘッダ・シグナル行・末尾行・先頭抜粋を stdout へ出す。
  全文は必ずファイルへ書き、そのパスをヘッダに載せる。
"""
import sys, os, re, json, hashlib, pathlib, argparse
 
SPILL_LIMIT = 100_000     # 全文がファイルへ退避される閾値
HEADER_BUDGET = 2_000     # 判断用ヘッダに割く上限（文字）
EXCERPT_BUDGET = 12_000   # 先頭抜粋に割く上限（文字）
 
# 既定のシグナル。テストランナーやビルドツールの語彙に合わせて --signal で差し替える
ERROR_RE = re.compile(r"\b(error|failed|failure|fatal|panic|exception)\b", re.I)
 
 
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--label", default="tool-output")
    ap.add_argument("--dir", default=os.environ.get("OUTBOX_DIR", "/tmp/outbox"))
    ap.add_argument("--signal", default=None, help="重要行を拾う正規表現（既定はエラー語）")
    ap.add_argument("--tail", type=int, default=20, help="末尾から常に残す行数")
    args = ap.parse_args()
 
    raw = sys.stdin.read()
    lines = raw.splitlines()
 
    # 全文を先に保存する。ここを後回しにすると、途中で落ちたときに何も残らない
    digest = hashlib.sha256(raw.encode("utf-8", "replace")).hexdigest()[:12]
    outdir = pathlib.Path(args.dir)
    outdir.mkdir(parents=True, exist_ok=True)
    full = outdir / f"{args.label}-{digest}.txt"
    full.write_text(raw, encoding="utf-8", errors="replace")
 
    pat = re.compile(args.signal, re.I) if args.signal else ERROR_RE
    hits = [(i, l) for i, l in enumerate(lines, 1) if pat.search(l)]
 
    header = {
        "label": args.label,
        "bytes": len(raw),
        "lines": len(lines),
        "spill_ratio": round(len(raw) / SPILL_LIMIT, 2),
        "signal_hits": len(hits),
        "full_output": str(full),          # 全文が要るときはここを読ませる
        "tail_lines_included": min(args.tail, len(lines)),
    }
 
    parts = ["<<<OUTBOX-HEADER",
             json.dumps(header, ensure_ascii=False, indent=2),
             "OUTBOX-HEADER"]
 
    if hits:
        parts.append(f"--- signal lines (first 40 of {len(hits)}) ---")
        parts += [f"{i}: {l[:300]}" for i, l in hits[:40]]
 
    # CLI が結論を置く場所。ここを落とすと封筒の意味がなくなる
    parts.append(f"--- tail {header['tail_lines_included']} lines ---")
    parts += lines[-args.tail:]
 
    body = "\n".join(parts)
    if len(body) > HEADER_BUDGET + EXCERPT_BUDGET:
        body = body[: HEADER_BUDGET + EXCERPT_BUDGET] + "\n--- excerpt truncated; read full_output ---"
 
    sys.stdout.write("\n".join([
        body,
        f"--- head excerpt ({EXCERPT_BUDGET} chars max) ---",
        raw[:EXCERPT_BUDGET],
    ]))
 
 
if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""outbox2.py — 末尾ブロックを先に確保してから、残りを配る封筒。"""
import sys, os, re, json, hashlib, pathlib, argparse
 
SPILL_LIMIT  = 100_000
TOTAL_BUDGET = 14_000     # 先頭抜粋まで含めた出力全体の上限
TAIL_BUDGET  = 3_000      # 末尾ブロックに先に取り置く分
 
 
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--label", default="tool-output")
    ap.add_argument("--dir", default=os.environ.get("OUTBOX_DIR", "/tmp/outbox"))
    ap.add_argument("--signal", default=None)
    ap.add_argument("--tail", type=int, default=20)
    args = ap.parse_args()
 
    raw = sys.stdin.read()
    lines = raw.splitlines()
 
    digest = hashlib.sha256(raw.encode("utf-8", "replace")).hexdigest()[:12]
    outdir = pathlib.Path(args.dir)
    outdir.mkdir(parents=True, exist_ok=True)
    full = outdir / f"{args.label}-{digest}.txt"
    full.write_text(raw, encoding="utf-8", errors="replace")
 
    pat = re.compile(args.signal, re.I) if args.signal else re.compile(
        r"\b(error|failed|failure|fatal|panic|exception)\b", re.I)
    hits = [(i, l) for i, l in enumerate(lines, 1) if pat.search(l)]
 
    # 1. 末尾ブロックを先に確保する。ここが削られたら封筒の意味がない
    tail_lines, used = [], 0
    for l in reversed(lines[-args.tail:]):
        if used + len(l) + 1 > TAIL_BUDGET:
            break
        tail_lines.insert(0, l)
        used += len(l) + 1
    tail_block = [f"--- tail {len(tail_lines)}/{min(args.tail, len(lines))} lines ---"] + tail_lines
 
    header = {
        "label": args.label,
        "chars": len(raw),                       # 境界と同じ単位。比較にはこちらを使う
        "bytes": len(raw.encode("utf-8")),
        "lines": len(lines),
        "spill_ratio_chars": round(len(raw) / SPILL_LIMIT, 2),
        "signal_hits": len(hits),
        "full_output": str(full),
        "tail_lines_included": len(tail_lines),
        "tail_lines_dropped": min(args.tail, len(lines)) - len(tail_lines),
    }
    head_block = ["<<<OUTBOX-HEADER",
                  json.dumps(header, ensure_ascii=False, indent=2),
                  "OUTBOX-HEADER"]
 
    # 2. 残った予算をシグナル行へ。入りきらない分は件数で伝える
    fixed = len("\n".join(head_block)) + len("\n".join(tail_block)) + 2
    sig_budget = max(0, TOTAL_BUDGET - fixed)
    sig, used, shown = [], 0, 0
    for i, l in hits:
        e = f"{i}: {l[:300]}"
        if used + len(e) + 1 > sig_budget:
            break
        sig.append(e)
        used += len(e) + 1
        shown += 1
    sig_block = ([f"--- signal lines ({shown} of {len(hits)} shown) ---"] + sig) if hits else []
    if hits and shown < len(hits):
        sig_block.append(f"--- {len(hits) - shown} more signal lines omitted; read full_output ---")
 
    # 3. 先頭抜粋は最後。余りがあるときだけ載せる
    body = "\n".join(head_block + sig_block + tail_block)
    room = max(0, TOTAL_BUDGET - len(body))
    out = [body]
    if room > 500:
        out += [f"--- head excerpt ({room} chars) ---", raw[:room]]
    sys.stdout.write("\n".join(out))
 
 
if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""json_outbox.py — 大きなJSONを、妥当なJSONのまま畳む封筒。"""
import json, sys, hashlib, pathlib, collections
 
 
def envelope(path, keep=8, outdir="/tmp/outbox"):
    raw = pathlib.Path(path).read_text()
    data = json.loads(raw)
 
    # トップレベルが配列でなくても、最初に見つかった配列を対象にする
    items = data if isinstance(data, list) else next(
        (v for v in data.values() if isinstance(v, list)), [])
 
    digest = hashlib.sha256(raw.encode()).hexdigest()[:12]
    outdir = pathlib.Path(outdir)
    outdir.mkdir(parents=True, exist_ok=True)
    full = outdir / f"json-{digest}.json"
    full.write_text(raw)
 
    # 全件走査すると大きなファイルで無駄に遅い。形の把握には先頭2000件で十分だった
    keys = collections.Counter()
    for it in items[:2000]:
        if isinstance(it, dict):
            keys.update(it.keys())
 
    body = {
        "header": {
            "source": str(path),
            "bytes": len(raw),
            "item_count": len(items),
            "keys": [k for k, _ in keys.most_common()],
            "full_output": str(full),
            "sample_policy": f"first {keep} items, full fidelity",
        },
        "sample": items[:keep],
    }
    return json.dumps(body, ensure_ascii=False, indent=1), len(raw)
 
 
if __name__ == "__main__":
    out, n = envelope(sys.argv[1])
    json.loads(out)   # 自己検証: 封筒の出力は必ずパースできること
    print(f"raw={n} envelope={len(out)} ratio={len(out)/n*100:.2f}% parse=OK")

for c in "grep -rn TODO ." "find . -type f -exec wc -c {} +" "git log --stat -50"; do
  n=$(eval "$c" 2>/dev/null | wc -m)   # -c ではなく -m。境界は文字数で引かれている
  printf "%-40s %10d  %5.2fx\n" "${c:0:38}" "$n" "$(python3 -c "print($n/100000)")"
done