Files
UmbraAudio/scripts/backfill_source_unit_keys.py
2026-10-18 19:01:04 -07:00

91 lines
3.0 KiB
Python

#!/usr/bin/env python3
# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0
"""Backfill content-addressed ``source_unit_key`` values (spec 21 §9.1 M4-04).
The key is derived from the chapter's source text and the utterance's own text, so
it can be computed for existing rows at any time. It is what lets the controller
name a source unit by content instead of by ``utterance_index``.
Read-only by default. ``--write`` is required to change anything, and ``--dry-run``
reports exactly what would change.
Usage:
python scripts/backfill_source_unit_keys.py # report only
python scripts/backfill_source_unit_keys.py --write # apply
python scripts/backfill_source_unit_keys.py --project "Last Dwarf"
python scripts/backfill_source_unit_keys.py --json # evidence
"""
from __future__ import annotations
import argparse
import json
import sqlite3
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from umbraaudio.config import load_config
from umbraaudio.source_units import backfill, key_report
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--project", default=None, help="project name or path")
parser.add_argument("--chapters", default=None, help="comma-separated chapter ids")
parser.add_argument(
"--write",
action="store_true",
help="actually write keys (default is a read-only report)",
)
parser.add_argument("--json", action="store_true", help="machine-readable output")
args = parser.parse_args()
cfg = load_config(args.project)
db_path = cfg.db_path
if not db_path.exists():
print(f"Database not found: {db_path}", file=sys.stderr)
return 1
chapter_ids = None
if args.chapters:
chapter_ids = [int(x.strip()) for x in args.chapters.split(",") if x.strip()]
# Read-only connection for the report so a plain run cannot write.
ro = sqlite3.connect(f"file:{db_path.resolve()}?mode=ro", uri=True)
try:
before = key_report(ro)
finally:
ro.close()
result: dict[str, object] = {"project": cfg.project_dir.name, "before": before}
if not args.write:
result["verdict"] = "report_only"
if args.json:
print(json.dumps(result, indent=2))
else:
print(f"project: {cfg.project_dir.name}")
print(f"before: {before}")
print("DRY RUN - nothing written. Re-run with --write to apply.")
return 0
rw = sqlite3.connect(str(db_path))
try:
result["after"] = backfill(rw, chapter_ids)
finally:
rw.close()
result["verdict"] = "written"
if args.json:
print(json.dumps(result, indent=2))
else:
print(f"project: {cfg.project_dir.name}")
print(f"before: {before}")
print(f"after: {result['after']}")
return 0
if __name__ == "__main__":
raise SystemExit(main())