dda079f605
- Apply migration 0008_audio_source_import (additive schema for source import) - Apply migration 0009_translation_import_runs (ledger table for import runs) - Wire PostgresAudioImportStore into HTTP commit path with org from node credential - Commit path uses PG store with org from node, no payload org - Bundle hash 21ce5687… remains frozen until v2; no authority transferred
389 lines
16 KiB
Python
389 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate an UmbraAudio metadata bundle against the format UmbraTrans accepts.
|
|
|
|
This is the *accepted-format* validator, the follow-up to the smoke-alarm one
|
|
shipped with the data request. It is deliberately stricter in three places,
|
|
because each is a thing the earlier validator could not catch:
|
|
|
|
1. **It validates each record against the full JSON Schema**
|
|
(``audio-system-manifest-v1.schema.json``), not a required-field subset.
|
|
2. **It verifies the declared ``content_hash``** against the actual bytes of
|
|
``manifest.jsonl``. The earlier validator checked that the bundle was
|
|
internally consistent; it did not check that the bundle was the one the hash
|
|
claimed. A tampered manifest that kept its internal references consistent
|
|
would have passed it.
|
|
3. **It enforces the mapping invariants** - the narrator semantics, chapter
|
|
numbering provenance, extension-vs-container, and the utterance attribution
|
|
rules - because those are the fields whose semantics change the answer, and a
|
|
record that satisfies the schema but breaks them is the failure a mapping
|
|
cannot see.
|
|
|
|
python3 validate-audio-manifest.py <bundle-directory>
|
|
|
|
Exits 0 when the bundle is acceptable, 1 when it is not, 2 when it cannot be
|
|
read at all. Every problem is printed, not just the first.
|
|
|
|
Runs without third-party packages: if ``jsonschema`` is importable it is used
|
|
for the full schema pass; otherwise the validator falls back to the structural
|
|
subset below and says so, rather than pretending to have validated the schema.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
SCHEMA_NAME = "audio-system-manifest-v1.schema.json"
|
|
|
|
REQUIRED_TOP = ("bundle.json", "manifest.jsonl")
|
|
|
|
#: Patterns that look like a secret. The bundle travels between machines and is
|
|
#: meant to be committable.
|
|
SUSPICIOUS = re.compile(
|
|
r"(sk-[A-Za-z0-9]{16,}|ghp_[A-Za-z0-9]{20,}|AKIA[0-9A-Z]{12,}"
|
|
r"|-----BEGIN[A-Z ]*PRIVATE KEY-----|eyJ[A-Za-z0-9_-]{20,}\.)"
|
|
)
|
|
|
|
REFERENCE_FIELDS = (
|
|
"book_id", "chapter_id", "render_id", "parent_id", "source_id",
|
|
"input_id", "voice_id", "inputs", "outputs", "chapters", "renders",
|
|
)
|
|
|
|
|
|
def fail(problems: list[str], message: str) -> None:
|
|
problems.append(message)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# loading
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def load_bundle(root: Path, problems: list[str]) -> tuple[dict, list[dict]] | None:
|
|
for name in REQUIRED_TOP:
|
|
if not (root / name).exists():
|
|
fail(problems, f"missing required file: {name}")
|
|
if problems:
|
|
return None
|
|
|
|
try:
|
|
bundle = json.loads((root / "bundle.json").read_text())
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
fail(problems, f"bundle.json: not readable JSON ({error})")
|
|
return None
|
|
|
|
records: list[dict] = []
|
|
for number, line in enumerate((root / "manifest.jsonl").read_text().splitlines(), start=1):
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
record = json.loads(line)
|
|
except json.JSONDecodeError as error:
|
|
fail(problems, f"manifest.jsonl:{number}: not valid JSON ({error.msg})")
|
|
continue
|
|
if not isinstance(record, dict):
|
|
fail(problems, f"manifest.jsonl:{number}: a record, not a {type(record).__name__}")
|
|
continue
|
|
records.append(record)
|
|
if not records:
|
|
fail(problems, "manifest.jsonl contains no records")
|
|
return bundle, records
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 1. the declared hash, against the bytes
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def check_content_hash(root: Path, bundle: dict, problems: list[str]) -> str | None:
|
|
declared = (bundle.get("content_hash") or {}).get("value")
|
|
if not declared:
|
|
fail(problems, "bundle.json: content_hash.value is absent, so the manifest cannot be pinned")
|
|
return None
|
|
actual = hashlib.sha256((root / "manifest.jsonl").read_bytes()).hexdigest()
|
|
if actual != declared:
|
|
fail(
|
|
problems,
|
|
f"content_hash mismatch: bundle.json declares {declared} but manifest.jsonl "
|
|
f"hashes to {actual}. The bundle is not the one the hash describes.",
|
|
)
|
|
return None
|
|
return actual
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 2. the schema
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def check_schema(root: Path, bundle: dict, records: list[dict], problems: list[str]) -> None:
|
|
schema_path = root / SCHEMA_NAME
|
|
if not schema_path.exists():
|
|
fail(
|
|
problems,
|
|
f"{SCHEMA_NAME} is not in the bundle. The accepted format is defined by that "
|
|
"schema; copy it in so the sender validates against the same file the "
|
|
"receiver does.",
|
|
)
|
|
return
|
|
try:
|
|
schema = json.loads(schema_path.read_text())
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
fail(problems, f"{SCHEMA_NAME}: not readable JSON ({error})")
|
|
return
|
|
|
|
try:
|
|
import jsonschema # type: ignore
|
|
except ImportError:
|
|
fail(
|
|
problems,
|
|
"note: jsonschema is not installed on this machine, so the full schema pass "
|
|
"was skipped and only the structural checks below ran. Install jsonschema "
|
|
"before treating this bundle as accepted.",
|
|
)
|
|
return
|
|
|
|
defs = schema.get("$defs", {})
|
|
|
|
# The envelope, validated on its own so a record failure does not bury it.
|
|
envelope_schema = {
|
|
"$schema": schema.get("$schema", "https://json-schema.org/draft/2020-12/schema"),
|
|
"$defs": defs,
|
|
"type": "object",
|
|
"required": ["bundle"],
|
|
"properties": {"bundle": {"$ref": "#/$defs/envelope"}},
|
|
}
|
|
for error in jsonschema.Draft202012Validator(envelope_schema).iter_errors({"bundle": bundle}):
|
|
fail(problems, f"schema bundle.{'/'.join(str(p) for p in error.absolute_path)}: {error.message}")
|
|
|
|
# Each record against the schema for its own kind, chosen by the discriminator.
|
|
# This is stricter than a oneOf pass and names the field instead of the record.
|
|
kind_to_def = {kind: kind for kind in ("book", "chapter", "speaker", "utterance", "render")}
|
|
for index, record in enumerate(records):
|
|
kind = record.get("kind")
|
|
if kind not in kind_to_def:
|
|
fail(problems, f"manifest.jsonl:{index + 1}: unknown kind {kind!r}")
|
|
continue
|
|
sub_schema = dict(defs[kind_to_def[kind]])
|
|
sub_schema["$defs"] = defs
|
|
for error in jsonschema.Draft202012Validator(sub_schema).iter_errors(record):
|
|
where = "/".join(str(part) for part in error.absolute_path) or "<record>"
|
|
fail(problems, f"schema manifest.jsonl:{index + 1} ({kind} {record.get('id')}) {where}: {error.message}")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 3. mapping invariants
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def _resolve_records(records: list[dict]) -> tuple[dict[str, dict], dict[str, list[dict]]]:
|
|
by_id: dict[str, dict] = {}
|
|
by_kind: dict[str, list[dict]] = {}
|
|
for record in records:
|
|
rid = record.get("id")
|
|
if isinstance(rid, str):
|
|
by_id[rid] = record
|
|
by_kind.setdefault(str(record.get("kind")), []).append(record)
|
|
return by_id, by_kind
|
|
|
|
|
|
def check_invariants(records: list[dict], problems: list[str]) -> None:
|
|
by_id, by_kind = _resolve_records(records)
|
|
|
|
chapters = {c.get("id"): c for c in by_kind.get("chapter", [])}
|
|
speakers = {s.get("id"): s for s in by_kind.get("speaker", [])}
|
|
|
|
# -- identity, not order ------------------------------------------------
|
|
for index, chapter in enumerate(by_kind.get("chapter", []), start=1):
|
|
if "numbering_basis" not in chapter:
|
|
fail(
|
|
problems,
|
|
f"chapter {chapter.get('id')!r}: no numbering_basis, so a derived order "
|
|
"could be mistaken for identity",
|
|
)
|
|
if not isinstance(chapter.get("chapter_index"), int):
|
|
fail(problems, f"chapter {chapter.get('id')!r}: chapter_index is not an integer")
|
|
|
|
# -- the narrator is a role, and the spellings are not the identity -----
|
|
narrator_ids = {s.get("id") for s in by_kind.get("speaker", []) if s.get("is_narrator")}
|
|
if not narrator_ids:
|
|
fail(problems, "no speaker record has is_narrator=true; every book has a narrator")
|
|
|
|
for index, utterance in enumerate(by_kind.get("utterance", []), start=1):
|
|
uid = utterance.get("id")
|
|
resolution = utterance.get("speaker_resolution")
|
|
raw = utterance.get("speaker_name_raw")
|
|
speaker_id = utterance.get("speaker_id")
|
|
|
|
if resolution == "implicit_narrator":
|
|
if raw is not None:
|
|
fail(
|
|
problems,
|
|
f"utterance {uid!r}: resolution is implicit_narrator but "
|
|
f"speaker_name_raw is {raw!r}; null is what implicit means",
|
|
)
|
|
if speaker_id not in narrator_ids:
|
|
fail(
|
|
problems,
|
|
f"utterance {uid!r}: implicit narrator does not resolve to a narrator "
|
|
f"speaker (got {speaker_id!r})",
|
|
)
|
|
elif resolution == "explicit":
|
|
if not isinstance(raw, str) or not raw:
|
|
fail(
|
|
problems,
|
|
f"utterance {uid!r}: resolution is explicit but speaker_name_raw is "
|
|
f"{raw!r}",
|
|
)
|
|
|
|
if utterance.get("utterance_type") not in {"narration", "dialogue", "scene_break"}:
|
|
fail(problems, f"utterance {uid!r}: bad utterance_type {utterance.get('utterance_type')!r}")
|
|
|
|
if utterance.get("chapter_id") not in chapters:
|
|
fail(problems, f"utterance {uid!r}: chapter_id {utterance.get('chapter_id')!r} does not resolve")
|
|
if speaker_id not in speakers:
|
|
fail(problems, f"utterance {uid!r}: speaker_id {speaker_id!r} does not resolve")
|
|
if utterance.get("render_id") is not None and utterance.get("render_id") not in by_id:
|
|
fail(problems, f"utterance {uid!r}: render_id {utterance.get('render_id')!r} does not resolve")
|
|
|
|
# A scene_break IS rendered here (text ",,," is voiced), so render_id is
|
|
# expected on it too. It is a structural marker regardless, and the ingest
|
|
# keeps it out of the literary passes; that is the ingest's rule, not a
|
|
# reason to reject the record.
|
|
|
|
# -- a render's name is not its type ------------------------------------
|
|
for render in by_kind.get("render", []):
|
|
rid = render.get("id")
|
|
fmt = render.get("audio_format") or {}
|
|
if not isinstance(fmt, dict):
|
|
fail(problems, f"render {rid!r}: audio_format is not an object")
|
|
continue
|
|
container = fmt.get("container")
|
|
ext = fmt.get("file_extension")
|
|
matches = fmt.get("extension_matches_container")
|
|
if container in (None, "") and matches is not None:
|
|
fail(
|
|
problems,
|
|
f"render {rid!r}: extension_matches_container is {matches!r} but the "
|
|
"container is unknown; that field must be null when the container is",
|
|
)
|
|
if isinstance(container, str) and isinstance(ext, str) and matches is None:
|
|
fail(
|
|
problems,
|
|
f"render {rid!r}: container {container!r} and extension {ext!r} are both "
|
|
"known but extension_matches_container is null; a determined container has "
|
|
"a determined answer",
|
|
)
|
|
if render.get("chapter_id") is not None and render.get("chapter_id") not in chapters:
|
|
fail(problems, f"render {rid!r}: chapter_id {render.get('chapter_id')!r} does not resolve")
|
|
if render.get("book_id") is not None and render.get("book_id") not in by_id:
|
|
fail(problems, f"render {rid!r}: book_id {render.get('book_id')!r} does not resolve")
|
|
|
|
# -- declared counts are checked against the records --------------------
|
|
for book in by_kind.get("book", []):
|
|
mismatch = book.get("extension_container_mismatch_count")
|
|
listed = book.get("extension_container_mismatches")
|
|
if isinstance(mismatch, int) and isinstance(listed, list) and mismatch != len(listed):
|
|
fail(
|
|
problems,
|
|
f"book {book.get('id')!r}: extension_container_mismatch_count is {mismatch} "
|
|
f"but {len(listed)} entries are listed",
|
|
)
|
|
|
|
# -- the casefold collision groups are surfaced, not merged -------------
|
|
seen_groups: dict[str, list[str]] = {}
|
|
for speaker in by_kind.get("speaker", []):
|
|
group = speaker.get("casefold_collision_group")
|
|
if group:
|
|
seen_groups.setdefault(group, []).append(str(speaker.get("id")))
|
|
# Groups are a warning, not an error: the whole point is that the reader decides.
|
|
|
|
|
|
def check_references(records: list[dict], problems: list[str]) -> None:
|
|
known: set[str] = set()
|
|
for record in records:
|
|
for key in ("id", "uuid"):
|
|
if isinstance(record.get(key), str):
|
|
known.add(record[key])
|
|
for number, record in enumerate(records, start=1):
|
|
for field in REFERENCE_FIELDS:
|
|
if field not in record:
|
|
continue
|
|
value = record[field]
|
|
candidates = value if isinstance(value, list) else [value]
|
|
for one in candidates:
|
|
if isinstance(one, dict):
|
|
one = one.get("id")
|
|
if isinstance(one, str) and one not in known:
|
|
fail(
|
|
problems,
|
|
f"manifest.jsonl:{number}: '{field}' points at {one!r}, which is "
|
|
"not in the manifest",
|
|
)
|
|
|
|
|
|
def check_credentials(root: Path, records: list[dict], problems: list[str]) -> None:
|
|
for number, record in enumerate(records, start=1):
|
|
if SUSPICIOUS.search(json.dumps(record)):
|
|
fail(problems, f"manifest.jsonl:{number}: looks like it contains a credential")
|
|
for path in sorted(root.rglob("*")):
|
|
if not path.is_file() or path.name == "manifest.jsonl":
|
|
continue
|
|
if path.suffix not in {".json", ".jsonl", ".md", ".env", ".yaml", ".yml"}:
|
|
continue
|
|
try:
|
|
if SUSPICIOUS.search(path.read_text(errors="ignore")):
|
|
fail(problems, f"{path.relative_to(root)}: looks like it contains a credential")
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
if len(argv) != 2:
|
|
print(__doc__)
|
|
return 2
|
|
root = Path(argv[1])
|
|
if not root.is_dir():
|
|
print(f"Not a directory: {root}")
|
|
return 2
|
|
|
|
problems: list[str] = []
|
|
loaded = load_bundle(root, problems)
|
|
if loaded is None:
|
|
print(f"{len(problems)} problem(s) in {root}:\n")
|
|
for problem in problems:
|
|
print(f" - {problem}")
|
|
return 1
|
|
|
|
bundle, records = loaded
|
|
digest = check_content_hash(root, bundle, problems)
|
|
check_schema(root, bundle, records, problems)
|
|
check_references(records, problems)
|
|
check_invariants(records, problems)
|
|
check_credentials(root, records, problems)
|
|
|
|
if problems:
|
|
print(f"{len(problems)} problem(s) in {root}:\n")
|
|
for problem in problems:
|
|
print(f" - {problem}")
|
|
return 1
|
|
|
|
counts = {kind: len([r for r in records if r.get("kind") == kind]) for kind in
|
|
("book", "chapter", "speaker", "utterance", "render")}
|
|
print(
|
|
f"{root}: {len(records)} record(s) accepted. "
|
|
f"content_hash {digest}. "
|
|
+ ", ".join(f"{kind}={n}" for kind, n in counts.items())
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv))
|