348 lines
13 KiB
Python
348 lines
13 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Release comments: the explicit step between "ready" and "published".
|
|
|
|
A release comment is the human-authored note that accompanies a publish, in the spirit of a
|
|
git commit message: a short summary line, an optional body, an author, and a designator you
|
|
can quote later when something needs to go back to something.
|
|
|
|
Two properties make it useful rather than decorative:
|
|
|
|
1. **The designator is date + time + a per-date sequence.** ``2026-10-03 #2 14:32:05``. If a
|
|
problem is found and corrected the same day, the sequence makes it unambiguous which
|
|
attempt was last. Without it, two releases on one date are indistinguishable in a
|
|
conversation.
|
|
|
|
2. **It is a gate, not a caption.** A publication cannot be written without an *approved*
|
|
comment. An approved comment is frozen: editing one requires a new comment and therefore
|
|
a new publication, which is the same rule the publication itself follows.
|
|
|
|
Lifecycle::
|
|
|
|
draft ──approve──▶ approved ──publish──▶ frozen inside the publication
|
|
▲
|
|
└── edit freely while it is still a draft
|
|
|
|
Storage is one JSON file per comment under ``.umbra/release_comments/``, named
|
|
``release-<YYYYMMDD>-<HHMMSS>-<NN>.json``. The publication record holds a *copy* of the
|
|
approved comment, so a publication stays readable even if the comment directory is ever
|
|
tidied.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
SCHEMA_VERSION = 1
|
|
|
|
STATUS_DRAFT = "draft"
|
|
STATUS_APPROVED = "approved"
|
|
|
|
COMMENT_DIR = "release_comments"
|
|
|
|
_SLUG_RE = re.compile(r"^release-(\d{8})-(\d{6})-(\d{2,})$")
|
|
|
|
|
|
class CommentError(RuntimeError):
|
|
"""Raised when a comment operation would violate the release rules."""
|
|
|
|
|
|
@dataclass
|
|
class ReleaseComment:
|
|
"""One authored release note."""
|
|
|
|
slug: str
|
|
designator: str
|
|
date: str
|
|
time: str
|
|
sequence: int
|
|
summary: str
|
|
body: str
|
|
author: str
|
|
authored_at: str
|
|
status: str = STATUS_DRAFT
|
|
approved_at: str | None = None
|
|
facts: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> ReleaseComment:
|
|
return cls(
|
|
slug=data["slug"],
|
|
designator=data["designator"],
|
|
date=data["date"],
|
|
time=data["time"],
|
|
sequence=int(data["sequence"]),
|
|
summary=data.get("summary", ""),
|
|
body=data.get("body", ""),
|
|
author=data.get("author", "operator"),
|
|
authored_at=data.get("authored_at", ""),
|
|
status=data.get("status", STATUS_DRAFT),
|
|
approved_at=data.get("approved_at"),
|
|
facts=data.get("facts", {}) or {},
|
|
)
|
|
|
|
def render(self) -> str:
|
|
"""The block a human reads, edits and approves."""
|
|
lines = [
|
|
f" designator {self.designator}",
|
|
f" slug {self.slug}",
|
|
f" author {self.author}",
|
|
f" authored {self.authored_at}",
|
|
f" status {self.status.upper()}",
|
|
]
|
|
if self.approved_at:
|
|
lines.append(f" approved {self.approved_at}")
|
|
lines.append("")
|
|
lines.append(f" {self.summary}")
|
|
if self.body:
|
|
lines.append("")
|
|
for para in self.body.splitlines():
|
|
lines.append(f" {para}" if para else "")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def comments_dir(data_dir: Path) -> Path:
|
|
return Path(data_dir) / COMMENT_DIR
|
|
|
|
|
|
def comment_path(data_dir: Path, slug: str) -> Path:
|
|
return comments_dir(data_dir) / f"{slug}.json"
|
|
|
|
|
|
def parse_comment(text: str) -> tuple[str, str]:
|
|
"""Split authored text into (summary, body).
|
|
|
|
The first non-empty line is the summary, like a commit subject. Everything after it is
|
|
the body. A first line longer than 72 characters is accepted but reported as a weak
|
|
summary rather than silently truncated, because truncating someone's note is worse than
|
|
telling them it is long.
|
|
"""
|
|
lines = text.strip().splitlines()
|
|
if not lines:
|
|
raise CommentError("a release comment needs at least a summary line")
|
|
summary = lines[0].strip()
|
|
body = "\n".join(lines[1:]).strip()
|
|
if not summary:
|
|
raise CommentError("the first line of a release comment is the summary and cannot be empty")
|
|
return summary, body
|
|
|
|
|
|
def _existing_sequences(data_dir: Path, date_compact: str) -> list[int]:
|
|
directory = comments_dir(data_dir)
|
|
if not directory.is_dir():
|
|
return []
|
|
found: list[int] = []
|
|
for path in directory.glob(f"release-{date_compact}-*.json"):
|
|
match = _SLUG_RE.match(path.stem)
|
|
if match:
|
|
found.append(int(match.group(3)))
|
|
return found
|
|
|
|
|
|
def next_sequence(data_dir: Path, when: datetime) -> int:
|
|
"""1-based sequence for this date. Two releases on one day are never ambiguous."""
|
|
return max(_existing_sequences(data_dir, when.strftime("%Y%m%d")), default=0) + 1
|
|
|
|
|
|
def build_slug(when: datetime, sequence: int) -> str:
|
|
return f"release-{when.strftime('%Y%m%d')}-{when.strftime('%H%M%S')}-{sequence:02d}"
|
|
|
|
|
|
def build_designator(when: datetime, sequence: int) -> str:
|
|
"""``2026-10-03 #2 14:32:05`` - readable, and quotable when going back to something."""
|
|
return f"{when.strftime('%Y-%m-%d')} #{sequence} {when.strftime('%H:%M:%S')}"
|
|
|
|
|
|
def create_comment(
|
|
data_dir: Path,
|
|
text: str,
|
|
author: str = "operator",
|
|
now: datetime | None = None,
|
|
status: str = STATUS_DRAFT,
|
|
facts: dict[str, Any] | None = None,
|
|
) -> ReleaseComment:
|
|
"""Author a new comment. Refuses an empty note."""
|
|
summary, body = parse_comment(text)
|
|
when = now or datetime.now(UTC)
|
|
sequence = next_sequence(data_dir, when)
|
|
comment = ReleaseComment(
|
|
slug=build_slug(when, sequence),
|
|
designator=build_designator(when, sequence),
|
|
date=when.strftime("%Y-%m-%d"),
|
|
time=when.strftime("%H:%M:%S"),
|
|
sequence=sequence,
|
|
summary=summary,
|
|
body=body,
|
|
author=author,
|
|
authored_at=when.isoformat(timespec="seconds"),
|
|
status=status,
|
|
approved_at=when.isoformat(timespec="seconds") if status == STATUS_APPROVED else None,
|
|
facts=facts or {},
|
|
)
|
|
path = comment_path(data_dir, comment.slug)
|
|
if path.exists():
|
|
# Same second, same sequence. Never silently overwrite someone's note.
|
|
raise CommentError(
|
|
f"{comment.slug} already exists. Wait a second or edit the existing comment."
|
|
)
|
|
_write_json(path, comment.as_dict())
|
|
return comment
|
|
|
|
|
|
def load_comment(data_dir: Path, slug: str) -> ReleaseComment | None:
|
|
path = comment_path(data_dir, slug)
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
return ReleaseComment.from_dict(json.loads(path.read_text(encoding="utf-8")))
|
|
except (OSError, ValueError, KeyError):
|
|
return None
|
|
|
|
|
|
def list_comments(data_dir: Path) -> list[dict[str, Any]]:
|
|
"""Every comment, newest first. Read-only."""
|
|
directory = comments_dir(data_dir)
|
|
if not directory.is_dir():
|
|
return []
|
|
out: list[dict[str, Any]] = []
|
|
for path in sorted(directory.glob("release-*.json"), reverse=True):
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
continue
|
|
out.append(
|
|
{
|
|
"slug": data.get("slug"),
|
|
"designator": data.get("designator"),
|
|
"date": data.get("date"),
|
|
"sequence": data.get("sequence"),
|
|
"author": data.get("author"),
|
|
"status": data.get("status"),
|
|
"summary": data.get("summary"),
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
def update_comment(
|
|
data_dir: Path,
|
|
slug: str,
|
|
text: str | None = None,
|
|
status: str | None = None,
|
|
now: datetime | None = None,
|
|
) -> ReleaseComment:
|
|
"""Edit a draft, or approve it.
|
|
|
|
An **approved** comment is frozen. Changing an approved note is not editing - it is a
|
|
different claim about a different state, so it needs a new comment and a new
|
|
publication. This is the same rule the publication itself follows, deliberately.
|
|
"""
|
|
comment = load_comment(data_dir, slug)
|
|
if comment is None:
|
|
raise CommentError(f"no such release comment: {slug}")
|
|
|
|
when = now or datetime.now(UTC)
|
|
|
|
if status == STATUS_APPROVED:
|
|
if text is not None:
|
|
summary, body = parse_comment(text)
|
|
if (summary, body) != (comment.summary, comment.body):
|
|
raise CommentError(
|
|
"cannot change the text of a comment while approving it. Edit the "
|
|
"draft first, then approve."
|
|
)
|
|
if comment.status != STATUS_APPROVED:
|
|
comment.status = STATUS_APPROVED
|
|
comment.approved_at = when.isoformat(timespec="seconds")
|
|
elif text is not None:
|
|
if comment.status == STATUS_APPROVED:
|
|
raise CommentError(
|
|
f"{comment.slug} is approved and frozen. Write a new comment: a changed "
|
|
"note is a new release, not an edit of this one."
|
|
)
|
|
comment.summary, comment.body = parse_comment(text)
|
|
|
|
_write_json(comment_path(data_dir, slug), comment.as_dict())
|
|
return comment
|
|
|
|
|
|
def require_approved(data_dir: Path, slug: str) -> ReleaseComment:
|
|
"""The publication gate. Raises if the comment is missing or still a draft."""
|
|
comment = load_comment(data_dir, slug)
|
|
if comment is None:
|
|
raise CommentError(
|
|
f"no release comment {slug}. A release needs an approved comment: author one "
|
|
"with --comment, then --approve-comment."
|
|
)
|
|
if comment.status != STATUS_APPROVED:
|
|
raise CommentError(
|
|
f"release comment {comment.designator} is a DRAFT. Read it, change it if it "
|
|
"is wrong, then --approve-comment."
|
|
)
|
|
return comment
|
|
|
|
|
|
def _write_json(path: Path, payload: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
os.replace(tmp, path)
|
|
|
|
|
|
def draft_comment_text(record: dict[str, Any], author: str) -> str:
|
|
"""Compose a factual release comment from what was actually measured.
|
|
|
|
This is the Deal Hunter case: the comment is derived from the real build rather than
|
|
written from memory, so it says what shipped, under which profile, with which gates,
|
|
and which approvals were granted implicitly by the act of releasing.
|
|
"""
|
|
profile = record.get("delivery_profile", {}) or {}
|
|
approvals = record.get("approvals", {}) or {}
|
|
chapters = record.get("chapters", []) or []
|
|
deltas = [c for c in chapters if c.get("warnings")]
|
|
|
|
summary = (
|
|
f"{record.get('work')} edition {record.get('edition_number')}: "
|
|
f"{len(chapters)} chapters pinned"
|
|
+ (f", {len(deltas)} with post-approval delivery deltas" if deltas else "")
|
|
)
|
|
|
|
loud_ok = sum(1 for c in chapters if (c.get("mp3") or {}).get("loudness_ok"))
|
|
peak_ok = sum(1 for c in chapters if (c.get("mp3") or {}).get("peak_ok"))
|
|
lines = [
|
|
summary,
|
|
"",
|
|
f"package {(record.get('package') or {}).get('object_key')} "
|
|
f"({(record.get('package') or {}).get('bytes')} bytes, built "
|
|
f"{(record.get('package') or {}).get('built_at')})",
|
|
f"delivery {profile.get('sample_rate')} Hz / {profile.get('channels')} ch"
|
|
+ (f" / {profile.get('bit_rate_kbps')} kbps" if profile.get("bit_rate_kbps") else "")
|
|
+ f", +{profile.get('lead_silence_s')}s lead and +{profile.get('trail_silence_s')}s "
|
|
f"trail silence, {profile.get('target_rms_db')} dB RMS target",
|
|
f"voices {approvals.get('voices_explicitly_approved', 0)} explicitly approved; "
|
|
f"{approvals.get('voices_implicitly_approved', 0)} approved implicitly by this "
|
|
f"release (they speak in it); "
|
|
f"{approvals.get('voices_in_profile_without_lines', 0)} in the profile are not "
|
|
"covered because they do not speak",
|
|
f"gates loudness_ok on {loud_ok}/{len(chapters)}, "
|
|
f"peak_ok on {peak_ok}/{len(chapters)}",
|
|
]
|
|
if deltas:
|
|
lines += [
|
|
"",
|
|
f"NOTE: {len(deltas)} chapters shipped longer or resampled than the reviewed "
|
|
"chapter wav. The release pins the DELIVERED bytes; the per-chapter review "
|
|
"covered the pre-delivery artifact. See delivery_profile.",
|
|
]
|
|
return "\n".join(lines) + f"\n\nAuthored by {author} from the measured build."
|