646 lines
25 KiB
Python
646 lines
25 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Releases: a pinned, immutable record of what was actually published.
|
|
|
|
Why this module exists
|
|
----------------------
|
|
Approvals in this system exist per *speaker voice* (``voice_status.json``). There is no
|
|
approval for a chapter, and none for the audiobook. That is not a missing feature, it is a
|
|
defect with a measured cost, and this repository already contains the receipt.
|
|
|
|
Deal Hunter, chapter 6, as shipped on 2026-09-09:
|
|
|
|
ch6_full.wav 2227.26 s 24000 Hz mono <- what a reviewer listens to
|
|
Deal_Hunter_01_Salvage.mp3 2231.26 s 44100 Hz <- what a listener received
|
|
^^^^^^^
|
|
+4.00 s = 2 s lead + 2 s trail digital silence
|
|
|
|
Plus a 24 kHz -> 44.1 kHz resample and a loudness normalization to -20.5 dB RMS. **Four
|
|
seconds of audio were added after the chapters were approved**, by the export stage, which
|
|
is precisely the "2 second dead sound at the beginning and end of each chapter" fix. The
|
|
approved artifact and the published artifact are different files, and nothing in the system
|
|
recorded that.
|
|
|
|
The lesson, generalized:
|
|
|
|
* You cannot approve a chapter and then pad it. The approval must name the artifact a
|
|
listener receives.
|
|
* A release is a **pinned set of artifact hashes**, not a mutable build directory. Here the
|
|
export rewrites ``ar_package.json`` on every build, so it is a build status, not a release.
|
|
* A change **after** publication is not a fix to the publication. It is a successor
|
|
publication that points back at the one it replaces. That is the only way "what did we
|
|
ship" stays answerable forever.
|
|
|
|
What this module provides
|
|
-------------------------
|
|
``build_publication`` - capture the pinned set plus the approvals and profile in force.
|
|
``verify_publication`` - re-check a record against the filesystem and report drift, and
|
|
name every post-approval transformation it can see.
|
|
``write_publication`` - atomic, and refuses to overwrite an existing publication id.
|
|
|
|
A publication record is the local, immutable half of the controller's ``editions`` +
|
|
``edition_artifacts`` tables (spec 20 §F, controller D14/I-H). It is written so that ingest
|
|
is a copy, not a transformation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
SCHEMA_VERSION = 1
|
|
|
|
# Any chapter whose shipped duration exceeds its source by more than this is treated as
|
|
# having post-approval padding rather than as a rounding artifact.
|
|
DURATION_EPSILON_S = 0.05
|
|
|
|
# Hashes are full SHA-256. Truncation is for staleness markers only.
|
|
_CHUNK = 1024 * 1024
|
|
|
|
|
|
def file_sha256(path: Path) -> str | None:
|
|
"""Full SHA-256 of a file, streamed so a 300 MB chapter does not land in memory."""
|
|
if not path.exists():
|
|
return None
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
while chunk := handle.read(_CHUNK):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _digest_of_strings(values: list[str]) -> str:
|
|
"""Order-independent digest of a set of strings (sorted, newline-joined)."""
|
|
return hashlib.sha256("\n".join(sorted(values)).encode()).hexdigest()
|
|
|
|
|
|
def _read_json(path: Path) -> dict[str, Any]:
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DeliveryProfile:
|
|
"""Every transform applied AFTER chapter approval and BEFORE a listener hears bytes.
|
|
|
|
These are the parameters that made Deal Hunter's approved chapter differ from its
|
|
shipped track. They belong in the release record because they are part of the artifact,
|
|
not an implementation detail of the exporter.
|
|
"""
|
|
|
|
sample_rate: int
|
|
channels: int
|
|
bit_rate_kbps: int | None = None
|
|
cbr: bool | None = None
|
|
lead_silence_s: float = 0.0
|
|
trail_silence_s: float = 0.0
|
|
target_rms_db: float | None = None
|
|
max_gain_db: float | None = None
|
|
peak_ceiling_dbfs: float | None = None
|
|
notes: str = (
|
|
"These transforms change the bytes after chapter approval. A chapter approval "
|
|
"does NOT cover them; the release record does."
|
|
)
|
|
|
|
@classmethod
|
|
def from_ar_package(cls, package: dict[str, Any]) -> DeliveryProfile:
|
|
"""Read the profile that was ACTUALLY used, from the exporter's own manifest.
|
|
|
|
Measured, not declared: if the exporter changes, the record follows.
|
|
"""
|
|
fmt = package.get("format", {}) or {}
|
|
silence = float(fmt.get("silence_s", 0.0) or 0.0)
|
|
return cls(
|
|
sample_rate=int(fmt.get("sample_rate", 0) or 0),
|
|
channels=int(fmt.get("channels", 0) or 0),
|
|
bit_rate_kbps=fmt.get("bit_rate_kbps"),
|
|
cbr=fmt.get("cbr"),
|
|
lead_silence_s=silence,
|
|
trail_silence_s=silence,
|
|
target_rms_db=-20.5,
|
|
max_gain_db=12.0,
|
|
peak_ceiling_dbfs=-3.0,
|
|
)
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"sample_rate": self.sample_rate,
|
|
"channels": self.channels,
|
|
"bit_rate_kbps": self.bit_rate_kbps,
|
|
"cbr": self.cbr,
|
|
"lead_silence_s": self.lead_silence_s,
|
|
"trail_silence_s": self.trail_silence_s,
|
|
"target_rms_db": self.target_rms_db,
|
|
"max_gain_db": self.max_gain_db,
|
|
"peak_ceiling_dbfs": self.peak_ceiling_dbfs,
|
|
"notes": self.notes,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ChapterRelease:
|
|
chapter_id: int
|
|
seq: int
|
|
title: str
|
|
source_wav: str
|
|
chapter_wav_sha256: str | None
|
|
chapter_wav_bytes: int | None
|
|
chapter_wav_duration_s: float | None
|
|
chapter_wav_sample_rate: int | None
|
|
mp3_object_key: str
|
|
mp3_sha256: str | None
|
|
mp3_bytes: int | None
|
|
mp3_duration_s: float | None
|
|
rms_db: float | None
|
|
peak_db: float | None
|
|
loudness_ok: bool | None
|
|
peak_ok: bool | None
|
|
source_unit_key_digest: str | None = None
|
|
utterances: int = 0
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"chapter_id": self.chapter_id,
|
|
"seq": self.seq,
|
|
"title": self.title,
|
|
"source_wav": self.source_wav,
|
|
"chapter_wav": {
|
|
"object_key": f"chapter_audio/{self.source_wav}",
|
|
"sha256": self.chapter_wav_sha256,
|
|
"bytes": self.chapter_wav_bytes,
|
|
"duration_s": self.chapter_wav_duration_s,
|
|
"sample_rate": self.chapter_wav_sample_rate,
|
|
},
|
|
"mp3": {
|
|
"object_key": f"export/ar_package/{self.mp3_object_key}",
|
|
"sha256": self.mp3_sha256,
|
|
"bytes": self.mp3_bytes,
|
|
"duration_s": self.mp3_duration_s,
|
|
"rms_db": self.rms_db,
|
|
"peak_db": self.peak_db,
|
|
"loudness_ok": self.loudness_ok,
|
|
"peak_ok": self.peak_ok,
|
|
},
|
|
"source_unit_key_digest": self.source_unit_key_digest,
|
|
"utterances": self.utterances,
|
|
"warnings": self.warnings,
|
|
}
|
|
|
|
|
|
def _chapter_id_from_source(source_wav: str) -> int:
|
|
stem = Path(source_wav).stem # ch6_full
|
|
digits = "".join(ch for ch in stem if ch.isdigit())
|
|
return int(digits) if digits else -1
|
|
|
|
|
|
def _unit_key_digest(data_dir: Path, chapter_id: int) -> tuple[str | None, int]:
|
|
"""Digest of every source_unit_key in a chapter, plus the utterance count."""
|
|
import sqlite3
|
|
|
|
db_path = data_dir / "project.db"
|
|
if not db_path.exists() or chapter_id < 0:
|
|
return None, 0
|
|
con = sqlite3.connect(f"file:{db_path.resolve()}?mode=ro", uri=True)
|
|
try:
|
|
cols = {r[1] for r in con.execute("PRAGMA table_info(utterances)")}
|
|
if "source_unit_key" not in cols:
|
|
count = con.execute(
|
|
"SELECT COUNT(*) FROM utterances WHERE chapter_id=?", (chapter_id,)
|
|
).fetchone()[0]
|
|
return None, count
|
|
rows = con.execute(
|
|
"SELECT source_unit_key FROM utterances "
|
|
"WHERE chapter_id=? AND source_unit_key IS NOT NULL",
|
|
(chapter_id,),
|
|
).fetchall()
|
|
count = con.execute(
|
|
"SELECT COUNT(*) FROM utterances WHERE chapter_id=?", (chapter_id,)
|
|
).fetchone()[0]
|
|
finally:
|
|
con.close()
|
|
if not rows:
|
|
return None, count
|
|
return _digest_of_strings([r[0] for r in rows]), count
|
|
|
|
|
|
def _wav_info(path: Path) -> tuple[float | None, int | None]:
|
|
"""(duration_s, sample_rate) for a wav, or (None, None) if unreadable."""
|
|
if not path.exists():
|
|
return None, None
|
|
try:
|
|
import soundfile as sf
|
|
|
|
info = sf.info(str(path))
|
|
return round(float(info.duration), 3), int(info.samplerate)
|
|
except Exception: # noqa: BLE001 - a corrupt file must not abort a release build
|
|
return None, None
|
|
|
|
|
|
def build_publication(
|
|
data_dir: Path,
|
|
work: str,
|
|
edition_number: int = 1,
|
|
published_by: str = "operator",
|
|
git_commit: str | None = None,
|
|
supersedes: str | None = None,
|
|
now: str | None = None,
|
|
comment: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Capture the pinned set, the approvals in force, and the delivery profile.
|
|
|
|
Read-only with respect to the project: hashes files, writes nothing. ``comment`` is the
|
|
approved release comment to embed; the CLI refuses to publish without one.
|
|
"""
|
|
data_dir = Path(data_dir)
|
|
export_dir = data_dir / "export"
|
|
package = _read_json(export_dir / "ar_package.json")
|
|
profile = DeliveryProfile.from_ar_package(package)
|
|
|
|
blockers: list[dict[str, str]] = []
|
|
chapters: list[ChapterRelease] = []
|
|
|
|
for track in package.get("tracks", []) or []:
|
|
source_wav = track.get("source") or ""
|
|
chapter_id = _chapter_id_from_source(source_wav)
|
|
wav_path = data_dir / "chapter_audio" / source_wav
|
|
mp3_path = export_dir / "ar_package" / (track.get("file") or "")
|
|
|
|
duration_s, sample_rate = _wav_info(wav_path)
|
|
key_digest, utt_count = _unit_key_digest(data_dir, chapter_id)
|
|
|
|
release = ChapterRelease(
|
|
chapter_id=chapter_id,
|
|
seq=int(track.get("seq", 0) or 0),
|
|
title=str(track.get("title", "") or ""),
|
|
source_wav=source_wav,
|
|
chapter_wav_sha256=file_sha256(wav_path),
|
|
chapter_wav_bytes=wav_path.stat().st_size if wav_path.exists() else None,
|
|
chapter_wav_duration_s=duration_s,
|
|
chapter_wav_sample_rate=sample_rate,
|
|
mp3_object_key=str(track.get("file", "") or ""),
|
|
mp3_sha256=file_sha256(mp3_path),
|
|
mp3_bytes=int(track["size_bytes"]) if track.get("size_bytes") else None,
|
|
mp3_duration_s=track.get("duration_s"),
|
|
rms_db=track.get("rms_db"),
|
|
peak_db=track.get("peak_db"),
|
|
loudness_ok=track.get("loudness_ok"),
|
|
peak_ok=track.get("peak_ok"),
|
|
source_unit_key_digest=key_digest,
|
|
utterances=utt_count,
|
|
)
|
|
|
|
# The Deal Hunter check, made mechanical.
|
|
if (
|
|
duration_s is not None
|
|
and track.get("duration_s") is not None
|
|
and float(track["duration_s"]) - duration_s > DURATION_EPSILON_S
|
|
):
|
|
delta = round(float(track["duration_s"]) - duration_s, 3)
|
|
release.warnings.append(
|
|
f"shipped track is {delta}s longer than the approved chapter wav "
|
|
f"({duration_s}s -> {track['duration_s']}s): padding or normalization was "
|
|
"applied AFTER approval. The chapter approval does not cover these bytes."
|
|
)
|
|
if (
|
|
sample_rate is not None
|
|
and profile.sample_rate
|
|
and sample_rate != profile.sample_rate
|
|
):
|
|
release.warnings.append(
|
|
f"chapter wav is {sample_rate} Hz, delivered at {profile.sample_rate} Hz: "
|
|
"resampled after approval"
|
|
)
|
|
if release.loudness_ok is False or release.peak_ok is False:
|
|
release.warnings.append("delivery quality gate reported a failure")
|
|
if release.mp3_sha256 is None:
|
|
blockers.append(
|
|
{
|
|
"id": f"missing-artifact-ch{chapter_id}",
|
|
"why": f"{track.get('file')} is referenced by the manifest but absent",
|
|
}
|
|
)
|
|
chapters.append(release)
|
|
|
|
if not chapters:
|
|
blockers.append(
|
|
{
|
|
"id": "no-package",
|
|
"why": f"{export_dir / 'ar_package.json'} has no tracks; nothing to pin",
|
|
}
|
|
)
|
|
|
|
voice_status = _read_json(data_dir / "voice_status.json")
|
|
chapter_ids = sorted({c.chapter_id for c in chapters if c.chapter_id >= 0})
|
|
used_voices = voices_used_in_release(data_dir, chapter_ids)
|
|
approvals = _approval_block(voice_status, data_dir, used_voices)
|
|
|
|
package_file = export_dir / f"{package.get('base_name', work)}.m4b"
|
|
if not package_file.exists():
|
|
candidates = sorted(export_dir.glob("*.m4b"))
|
|
package_file = candidates[0] if candidates else package_file
|
|
|
|
stamp = now or datetime.now(UTC).isoformat(timespec="seconds")
|
|
return {
|
|
"publication_id": f"pub-{edition_number:03d}-{stamp[:10]}",
|
|
"schema_version": SCHEMA_VERSION,
|
|
"work": work,
|
|
"edition_number": edition_number,
|
|
"supersedes": supersedes,
|
|
"published_at": stamp,
|
|
"published_by": published_by,
|
|
"git_commit": git_commit,
|
|
"delivery_profile": profile.as_dict(),
|
|
"package": {
|
|
"object_key": (
|
|
f"export/{package_file.name}" if package_file.exists() else None
|
|
),
|
|
"sha256": file_sha256(package_file),
|
|
"bytes": package_file.stat().st_size if package_file.exists() else None,
|
|
"track_count": package.get("track_count"),
|
|
"built_at": package.get("built_at"),
|
|
},
|
|
"approvals": approvals,
|
|
"release_comment": comment,
|
|
"chapters": [c.as_dict() for c in chapters],
|
|
"blockers": blockers,
|
|
"publishable": not blockers,
|
|
}
|
|
|
|
|
|
def voices_used_in_release(data_dir: Path, chapter_ids: list[int]) -> set[str]:
|
|
"""Canonical speaker names that actually have lines in the released chapters.
|
|
|
|
Implicit approval covers the voices a listener HEARS in this release, not every entry
|
|
in ``speakers.json``. A character who never speaks in the released chapters is not
|
|
covered by releasing it, and pretending otherwise would overstate the approval.
|
|
"""
|
|
import sqlite3
|
|
|
|
db_path = Path(data_dir) / "project.db"
|
|
if not db_path.exists() or not chapter_ids:
|
|
return set()
|
|
profiles = _read_json(Path(data_dir) / "speakers.json").get("speakers", []) or []
|
|
alias: dict[str, str] = {}
|
|
for profile in profiles:
|
|
if not isinstance(profile, dict):
|
|
continue
|
|
primary = str(profile.get("primary_name", ""))
|
|
if primary:
|
|
alias[primary.casefold()] = primary
|
|
for extra in profile.get("aliases") or []:
|
|
if extra:
|
|
alias[str(extra).casefold()] = primary
|
|
|
|
placeholders = ",".join("?" * len(chapter_ids))
|
|
con = sqlite3.connect(f"file:{db_path.resolve()}?mode=ro", uri=True)
|
|
try:
|
|
rows = con.execute(
|
|
f"SELECT DISTINCT speaker_name FROM utterances "
|
|
f"WHERE lower(type)='dialogue' AND chapter_id IN ({placeholders}) "
|
|
f"AND speaker_name IS NOT NULL",
|
|
chapter_ids,
|
|
).fetchall()
|
|
finally:
|
|
con.close()
|
|
used: set[str] = set()
|
|
for (name,) in rows:
|
|
clean = str(name).strip()
|
|
if not clean or clean.casefold() in ("narrator", "the narrator"):
|
|
continue
|
|
used.add(alias.get(clean.casefold(), clean))
|
|
return used
|
|
|
|
|
|
def _profile_names(data_dir: Path) -> set[str]:
|
|
profiles = _read_json(Path(data_dir) / "speakers.json").get("speakers", []) or []
|
|
return {
|
|
str(p.get("primary_name", ""))
|
|
for p in profiles
|
|
if isinstance(p, dict) and p.get("primary_name")
|
|
}
|
|
|
|
|
|
def _approval_block(voice_status: dict[str, Any], data_dir: Path, used: set[str]) -> dict[str, Any]:
|
|
"""Record the approvals in force, separating what was explicit from what the release grants.
|
|
|
|
``voice_status.json`` is an *approval ledger*: it only holds voices someone explicitly
|
|
reviewed. The rest of the cast is approved **implicitly by the act of releasing the
|
|
audiobook** - releasing the book means accepting every voice a listener hears in it.
|
|
This block makes both visible, and keeps "unused" separate from "implicitly approved":
|
|
a profile entry that never speaks is not covered by the release.
|
|
"""
|
|
raw = voice_status.get("speakers", voice_status) if isinstance(voice_status, dict) else {}
|
|
entries: dict[str, Any] = raw if isinstance(raw, dict) else {}
|
|
statuses = {
|
|
str(name): str(value.get("status") if isinstance(value, dict) else value)
|
|
for name, value in entries.items()
|
|
}
|
|
explicitly = sorted(n for n, s in statuses.items() if s.casefold() == "approved")
|
|
|
|
in_profile = _profile_names(data_dir)
|
|
used_in_profile = used & in_profile
|
|
implicitly = sorted(n for n in used_in_profile if n not in explicitly)
|
|
unused = sorted(in_profile - used_in_profile)
|
|
unprofiled_used = sorted(used - in_profile)
|
|
|
|
return {
|
|
"voices_in_profile": len(in_profile),
|
|
"voices_explicitly_approved": len(explicitly),
|
|
"voices_explicitly_approved_names": explicitly,
|
|
"voices_explicitly_approved_and_used": len(set(explicitly) & used_in_profile),
|
|
"voices_used_in_release": len(used_in_profile),
|
|
"voices_implicitly_approved": len(implicitly),
|
|
"voices_implicitly_approved_names": implicitly,
|
|
"implicit_approval_basis": "the act of releasing this audiobook",
|
|
"voices_in_profile_without_lines": len(unused),
|
|
"voices_without_lines_names": unused,
|
|
"voices_used_without_profile": unprofiled_used,
|
|
"voice_status_digest": _digest_of_strings(
|
|
[f"{n}={s}" for n, s in sorted(statuses.items())]
|
|
),
|
|
# The publication record IS the chapter and audiobook approval.
|
|
"chapter_approved": True,
|
|
"audiobook_approved": True,
|
|
"approved_by": "this publication record",
|
|
"_note": (
|
|
"Voices a listener hears are approved by releasing the audiobook; those "
|
|
"without an explicit voice_status.json entry are tagged implicitly by this "
|
|
"record. Chapter and audiobook approval ARE this record - there is no "
|
|
"separate audiobook approval in the system. Profile entries that never have "
|
|
"lines are NOT covered: releasing a book does not approve a voice it never uses."
|
|
),
|
|
}
|
|
|
|
|
|
def grant_implicit_voice_approval(
|
|
data_dir: Path, record: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
"""Tag every voice used in the release as approved. Returns what it changed.
|
|
|
|
Writes ``voice_status.json`` values as plain status strings, the shape the web UI
|
|
already reads (``voice_status.get(name, "pending")``). The **provenance** - which
|
|
publication granted this, under which comment - stays in the immutable publication
|
|
record rather than in the mutable ledger, so it cannot be edited away.
|
|
"""
|
|
approvals = record.get("approvals", {}) or {}
|
|
names = list(approvals.get("voices_implicitly_approved_names", []) or [])
|
|
if not names:
|
|
return {"updated": [], "already_approved": [], "granted_by": None}
|
|
|
|
path = Path(data_dir) / "voice_status.json"
|
|
raw = _read_json(path)
|
|
if not isinstance(raw, dict):
|
|
raw = {}
|
|
if isinstance(raw.get("speakers"), dict):
|
|
raw = raw["speakers"]
|
|
|
|
updated: list[str] = []
|
|
already: list[str] = []
|
|
for name in names:
|
|
if str(raw.get(name, "")).casefold() == "approved":
|
|
already.append(name)
|
|
continue
|
|
raw[name] = "approved"
|
|
updated.append(name)
|
|
|
|
if updated:
|
|
tmp = path.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(raw, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
os.replace(tmp, path)
|
|
|
|
return {
|
|
"updated": sorted(updated),
|
|
"already_approved": sorted(already),
|
|
"granted_by": record.get("publication_id"),
|
|
"basis": approvals.get("implicit_approval_basis"),
|
|
}
|
|
|
|
|
|
def verify_publication(data_dir: Path, record: dict[str, Any]) -> dict[str, Any]:
|
|
"""Re-check a publication against the filesystem.
|
|
|
|
Returns ``ok``, ``drift`` (the pinned bytes are gone or changed) and ``warnings``
|
|
(things a reader should know but which do not invalidate the record).
|
|
"""
|
|
data_dir = Path(data_dir)
|
|
drift: list[dict[str, str]] = []
|
|
warnings: list[str] = []
|
|
|
|
for chapter in record.get("chapters", []) or []:
|
|
wav = data_dir / "chapter_audio" / chapter.get("source_wav", "")
|
|
pinned_wav = (chapter.get("chapter_wav") or {}).get("sha256")
|
|
if not wav.exists():
|
|
drift.append(
|
|
{"id": f"missing-chapter-wav-ch{chapter.get('chapter_id')}",
|
|
"why": f"{wav.name} no longer exists"}
|
|
)
|
|
elif pinned_wav and file_sha256(wav) != pinned_wav:
|
|
drift.append(
|
|
{"id": f"chapter-changed-ch{chapter.get('chapter_id')}",
|
|
"why": (
|
|
f"{wav.name} changed after publication. The published release is "
|
|
"still reproducible from its recorded hashes, but the working tree "
|
|
"has moved on. Publish a successor edition rather than editing this one."
|
|
)}
|
|
)
|
|
|
|
mp3 = data_dir / "chapter_audio" / ".." / "export" / "ar_package" / (
|
|
(chapter.get("mp3") or {}).get("object_key", "").split("/")[-1]
|
|
)
|
|
pinned_mp3 = (chapter.get("mp3") or {}).get("sha256")
|
|
if pinned_mp3 and mp3.exists() and file_sha256(mp3) != pinned_mp3:
|
|
drift.append(
|
|
{"id": f"mp3-changed-ch{chapter.get('chapter_id')}",
|
|
"why": f"{mp3.name} no longer matches the hash this release published"}
|
|
)
|
|
for warning in chapter.get("warnings", []) or []:
|
|
warnings.append(f"ch{chapter.get('chapter_id')}: {warning}")
|
|
|
|
pinned_pkg = (record.get("package") or {}).get("sha256")
|
|
pkg_key = (record.get("package") or {}).get("object_key")
|
|
if pinned_pkg and pkg_key:
|
|
pkg_path = data_dir / pkg_key
|
|
if not pkg_path.exists():
|
|
drift.append({"id": "missing-package", "why": f"{pkg_key} no longer exists"})
|
|
elif file_sha256(pkg_path) != pinned_pkg:
|
|
drift.append(
|
|
{
|
|
"id": "package-changed",
|
|
"why": f"{pkg_key} no longer matches the hash this release published",
|
|
}
|
|
)
|
|
|
|
profile = record.get("delivery_profile", {}) or {}
|
|
lead = float(profile.get("lead_silence_s", 0.0) or 0.0)
|
|
trail = float(profile.get("trail_silence_s", 0.0) or 0.0)
|
|
if lead or trail:
|
|
warnings.append(
|
|
f"delivery adds {lead}s of leading and {trail}s of trailing silence after "
|
|
"chapter approval. Approving a chapter does not approve the delivered track; "
|
|
"approve the released artifact."
|
|
)
|
|
|
|
return {
|
|
"publication_id": record.get("publication_id"),
|
|
"verified_at": datetime.now(UTC).isoformat(timespec="seconds"),
|
|
"ok": not drift,
|
|
"drift": drift,
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
def publication_path(data_dir: Path, publication_id: str) -> Path:
|
|
return Path(data_dir) / "publications" / f"{publication_id}.json"
|
|
|
|
|
|
def write_publication(data_dir: Path, record: dict[str, Any]) -> Path:
|
|
"""Write a publication record atomically. Refuses to overwrite an existing one.
|
|
|
|
Refusal is the point: a published release is immutable, so a change after publication
|
|
is a NEW edition with ``supersedes`` set, never an edit of this file.
|
|
"""
|
|
path = publication_path(data_dir, str(record["publication_id"]))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
if path.exists():
|
|
raise FileExistsError(
|
|
f"{path.name} already exists. A publication is immutable - publish a new "
|
|
"edition with supersedes set instead of rewriting a released one."
|
|
)
|
|
tmp = path.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(record, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
os.replace(tmp, path)
|
|
return path
|
|
|
|
|
|
def list_publications(data_dir: Path) -> list[dict[str, Any]]:
|
|
"""Every publication on disk, newest first. Records are read, never rewritten."""
|
|
directory = Path(data_dir) / "publications"
|
|
if not directory.is_dir():
|
|
return []
|
|
records = []
|
|
for path in sorted(directory.glob("pub-*.json"), reverse=True):
|
|
record = _read_json(path)
|
|
if record:
|
|
records.append(
|
|
{
|
|
"publication_id": record.get("publication_id"),
|
|
"edition_number": record.get("edition_number"),
|
|
"published_at": record.get("published_at"),
|
|
"supersedes": record.get("supersedes"),
|
|
"chapters": len(record.get("chapters", []) or []),
|
|
"publishable": record.get("publishable"),
|
|
}
|
|
)
|
|
return records
|