Files
ncmr-songs-videos/scripts/bom_gen.py
T
irene 2c6e227c82 feat(bom): add --hash, and fix the BOM marker breaking YAML
`feat/bom-generator` carried one commit that never landed on master: file
content hashes in the BOM ("v2 with hashes"). The Python rewrite dropped that
capability, so port it forward rather than losing it when the branch goes.

- `--hash` records a short BLAKE2b digest per file. Opt-in, because it reads
  every asset in full and that is wasted work for a plain inventory.
- Hashes are streamed in chunks with `hashlib.blake2b`, not `md5` via haveged
  imports, and not whole-file reads — these assets run to hundreds of megabytes.
- YAML gains `hash:` after `size:`; Markdown gains a hash column only when
  hashes were requested, so the common table stays readable.

Fix a pre-existing bug found while testing: the BOM marker was written as
`; generated by ...`, but `;` is not a YAML comment, so every generated BOM
failed to parse with "mapping values are not allowed here". Now `#`. Verified by
re-parsing output from both the PyYAML path and the built-in fallback writer.

Sizes alone cannot tell you whether an asset changed; a same-length re-render
keeps the byte count identical. That is the gap the hash closes.
2026-09-10 18:37:18 +09:00

517 lines
17 KiB
Python

"""Generate a bill of materials for a song's untracked binary assets.
The repo's guideline says every song keeps a ``BOM.md`` (or ``BOM.yaml``)
listing the untracked input assets and explaining what each file is. This walks
the conventional asset directories, prompts for a description per file, and
writes the result.
Stdlib only: the YAML is emitted by hand. If PyYAML happens to be installed it is
used to *validate* the output, but it is never required.
Usage::
python3 scripts/bom_gen.py # walk ./assets, write BOM.yaml
python3 scripts/bom_gen.py path/to/song # walk a song directory
python3 scripts/bom_gen.py --format md # write BOM.md instead
python3 scripts/bom_gen.py --descriptions d.toml # non-interactive
python3 scripts/bom_gen.py --describe-one audio/song.flac
"""
from __future__ import annotations
import argparse
import hashlib
import os
import re
import sys
from dataclasses import dataclass, field
if __package__ in (None, ""):
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from _common import ( # type: ignore[import-not-found]
Die,
Report,
add_common_arguments,
human_size,
main_wrapper,
text_table,
)
else: # pragma: no cover
from ._common import (
Die,
Report,
add_common_arguments,
human_size,
main_wrapper,
text_table,
)
#: Directories that hold untracked input assets, with the role we assume.
ASSET_DIRS: tuple[tuple[str, str], ...] = (
("audio", "Song audio: final mix, instrumental, stems"),
("images", "Cover art, stills, background plates"),
("fonts", "Typefaces used by the subtitles or overlays"),
("videos", "Material video clips"),
)
#: File extensions that are worth flagging as (probably) not-yet-described.
_DESCRIPTION_PLACEHOLDER = "<describe this file>"
_BOM_MARKER = "# generated by scripts/bom_gen.py"
# --------------------------------------------------------------------------- #
# YAML emission (hand-rolled, stdlib only)
# --------------------------------------------------------------------------- #
#: Characters that force a value to be quoted in YAML.
_NEEDS_QUOTES = re.compile(r"""[:\{\}\[\],&*#?|<>=!%@`"']|^\s|\s$|^$|^[-?]""")
def yaml_scalar(value: str) -> str:
"""Quote *value* for YAML only when it needs it."""
if value.isdigit():
return f'"{value}"' # keep hashes/filenames that look numeric as strings
if _NEEDS_QUOTES.search(value):
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
return f'"{escaped}"'
return value
def yaml_key(value: str) -> str:
"""Quote a key that would otherwise be ambiguous (spaces, colons, leading -)."""
if _NEEDS_QUOTES.search(value) or value.startswith("-") or value != value.strip():
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
return f'"{escaped}"'
return value
# --------------------------------------------------------------------------- #
# Asset discovery
# --------------------------------------------------------------------------- #
@dataclass
class Asset:
path: str # relative to the BOM root, POSIX separators
size: int
description: str = ""
digest: str = "" # short content hash, empty when --hash was not requested
extra: dict[str, str] = field(default_factory=dict)
@property
def directory(self) -> str:
return os.path.dirname(self.path).split("/")[0] or "."
def file_digest(full: str, *, length: int = 12, chunk: int = 1 << 20) -> str:
"""Short BLAKE2b content hash, streamed so large assets do not land in RAM.
Assets here are audio and video files that can run to hundreds of megabytes,
so this reads in chunks rather than slurping the whole file. BLAKE2b is in
the standard library and is faster than md5 with no collision worries.
``length`` bytes of digest = 3 * length hex characters.
"""
h = hashlib.blake2b(digest_size=length)
with open(full, "rb") as fh:
while bytes_read := fh.read(chunk):
h.update(bytes_read)
return h.hexdigest()
def walk_assets(
root: str,
dirs: tuple[tuple[str, str], ...] = ASSET_DIRS,
*,
hashes: bool = False,
) -> list[Asset]:
"""Collect files under each asset directory, sorted by path.
Hashes are opt-in: they cost a full read of every asset, which is wasted
work when you only want the inventory.
"""
assets: list[Asset] = []
for name, _role in dirs:
base = os.path.join(root, name)
if not os.path.isdir(base):
continue
for dirpath, dirnames, filenames in os.walk(base):
dirnames.sort()
for filename in sorted(filenames):
if filename.startswith("."):
continue
full = os.path.join(dirpath, filename)
if not os.path.isfile(full):
continue
rel = os.path.relpath(full, root).replace(os.sep, "/")
assets.append(
Asset(
path=rel,
size=os.path.getsize(full),
digest=file_digest(full) if hashes else "",
)
)
return assets
def load_descriptions(path: str) -> dict[str, str]:
"""Read a descriptions file: simple ``path = description`` or ``path<tab>desc``.
A deliberately forgiving format so descriptions can also be pasted in from a
spreadsheet. ``#`` starts a comment.
"""
try:
with open(path, "r", encoding="utf-8") as handle:
raw = handle.read()
except OSError as exc:
raise Die(f"cannot read --descriptions {path}: {exc.strerror}") from exc
mapping: dict[str, str] = {}
for line in raw.splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, _, value = line.partition("=")
elif "\t" in line:
key, _, value = line.partition("\t")
else:
continue
mapping[key.strip().lstrip("./")] = value.strip()
return mapping
# --------------------------------------------------------------------------- #
# Rendering
# --------------------------------------------------------------------------- #
def render_yaml(root: str, assets: list[Asset], *, title: str | None, by_dir: bool) -> str:
total = sum(a.size for a in assets)
lines = [
"---",
_BOM_MARKER,
f"root: {yaml_scalar(os.path.basename(os.path.abspath(root)) or root)}",
]
if title:
lines.append(f"title: {yaml_scalar(title)}")
lines.append(f"total_files: {len(assets)}")
lines.append(f"total_size: {total}")
lines.append("")
if by_dir:
for dirname, role in ASSET_DIRS:
grouping = [a for a in assets if a.directory == dirname]
if not grouping:
continue
subtotal = sum(a.size for a in grouping)
lines.append(f"{yaml_key(dirname)}:")
lines.append(f" role: {yaml_scalar(role)}")
lines.append(f" count: {len(grouping)}")
lines.append(f" size: {subtotal}")
lines.append(" files:")
for asset in grouping:
lines.append(f" {yaml_key(asset.path)}:")
lines.append(f" size: {asset.size}")
if asset.digest:
lines.append(f" hash: {asset.digest}")
lines.append(f" description: {yaml_scalar(asset.description)}")
lines.append("")
else:
lines.append("files:")
for asset in assets:
lines.append(f" {yaml_key(asset.path)}:")
lines.append(f" size: {asset.size}")
if asset.digest:
lines.append(f" hash: {asset.digest}")
lines.append(f" description: {yaml_scalar(asset.description)}")
# Anything under an unrecognised top-level directory still deserves a mention.
others = [a for a in assets if a.directory not in {d for d, _ in ASSET_DIRS}]
if others:
lines.append("")
lines.append("unclassified:")
for asset in others:
lines.append(f" {yaml_key(asset.path)}:")
lines.append(f" size: {asset.size}")
if asset.digest:
lines.append(f" hash: {asset.digest}")
lines.append(f" description: {yaml_scalar(asset.description)}")
lines.append("...")
return "\n".join(lines) + "\n"
def render_markdown(root: str, assets: list[Asset], *, title: str | None, by_dir: bool) -> str:
total = sum(a.size for a in assets)
heading = title or os.path.basename(os.path.abspath(root)) or "Untracked assets"
lines = [
f"# Bill of materials — {heading}",
"",
f"{len(assets)} file(s), {human_size(total)}. Generated by `scripts/bom_gen.py`.",
"",
]
groups: list[tuple[str, list[Asset]]] = []
if by_dir:
for dirname, _role in ASSET_DIRS:
grouping = [a for a in assets if a.directory == dirname]
if grouping:
groups.append((dirname, grouping))
others = [a for a in assets if a.directory not in {d for d, _ in ASSET_DIRS}]
if others:
groups.append(("unclassified", others))
else:
groups.append(("files", assets))
for group_name, grouping in groups:
lines.append(f"## {group_name}")
lines.append("")
# Only carry a hash column when hashes were asked for, so the common
# inventory stays narrow enough to read.
with_hash = any(asset.digest for asset in grouping)
rows = [
(
f"`{asset.path}`",
human_size(asset.size),
*( (f"`{asset.digest}`",) if with_hash else () ),
asset.description or f"_{_DESCRIPTION_PLACEHOLDER}_",
)
for asset in grouping
]
headers = ("file", "size", "hash", "description") if with_hash else ("file", "size", "description")
lines.append(text_table(headers, rows))
lines.append("")
return "\n".join(lines).rstrip() + "\n"
def validate_yaml(document: str, report: Report) -> None:
"""If PyYAML is importable, use it to confirm the emitted document parses."""
try:
import yaml # type: ignore[import-not-found]
except ImportError:
return
try:
yaml.safe_load(document)
except Exception as exc: # noqa: BLE001 - report whatever PyYAML says
report.warn(f"PyYAML could not parse the generated document: {exc}")
# --------------------------------------------------------------------------- #
# Interactive prompts
# --------------------------------------------------------------------------- #
def prompt_for_descriptions(assets: list[Asset], report: Report) -> None:
"""Ask for a description per asset. EOF ends prompting and keeps what we have."""
if not sys.stdin.isatty():
report.warn(
"stdin is not a terminal — skipping prompts; "
"descriptions left blank (use --descriptions FILE to supply them)"
)
return
total = len(assets)
report.info(f"Describing {total} file(s). Blank line accepts the current value.\n")
report.info("Press Ctrl-C or Ctrl-D to stop early; the BOM is still written.\n")
try:
for i, asset in enumerate(assets, start=1):
report.info(f"[{i}/{total}] {asset.path} ({human_size(asset.size)})")
try:
answer = input(" description: ").strip()
except EOFError:
report.info("")
report.warn("end of input — remaining descriptions left blank")
return
if answer:
asset.description = answer
except KeyboardInterrupt:
report.info("")
report.warn("interrupted — writing the BOM with the descriptions gathered so far")
# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="bom_gen",
description="Generate a bill of materials for a song's untracked binary assets.",
)
add_common_arguments(parser)
parser.add_argument(
"root",
nargs="?",
default=".",
help="directory holding audio/, images/, fonts/, videos/ (default: .)",
)
parser.add_argument(
"-o",
"--output",
help="output path (default: BOM.yaml or BOM.md in the root directory)",
)
parser.add_argument(
"--format",
default="yaml",
choices=("yaml", "md"),
help="output format (default: yaml)",
)
parser.add_argument(
"--descriptions",
metavar="FILE",
help="read 'path = description' lines from FILE instead of prompting",
)
parser.add_argument(
"--no-prompt",
action="store_true",
help="never prompt; write the inventory with blank descriptions",
)
parser.add_argument(
"--flat",
action="store_true",
help="list files without grouping them by asset directory",
)
parser.add_argument(
"--hash",
dest="hashes",
action="store_true",
help="record a short content hash per file (reads every asset; slower)",
)
parser.add_argument(
"--title",
help="song title, recorded in the output header",
)
parser.add_argument(
"--describe-one",
metavar="PATH",
help="update the description of a single file in an existing BOM",
)
return parser
@main_wrapper
def main() -> int:
args = build_parser().parse_args()
report = Report(quiet=args.quiet)
root = os.path.abspath(args.root)
if not os.path.isdir(root):
raise Die(f"not a directory: {args.root}")
if args.describe_one:
return describe_one(root, args, report)
assets = walk_assets(root, hashes=args.hashes)
if not assets:
expected = ", ".join(f"{name}/" for name, _ in ASSET_DIRS)
report.warn(f"no assets found under {root} (looked in: {expected})")
return 1
if args.descriptions:
mapping = load_descriptions(args.descriptions)
applied = 0
for asset in assets:
if asset.path in mapping:
asset.description = mapping[asset.path]
applied += 1
missing = [a.path for a in assets if not a.description]
report.verbose(f"applied {applied} description(s) from {args.descriptions}")
if missing:
report.warn(f"{len(missing)} file(s) have no description in {args.descriptions}")
elif not args.no_prompt:
prompt_for_descriptions(assets, report)
renderer = render_markdown if args.format == "md" else render_yaml
document = renderer(root, assets, title=args.title, by_dir=not args.flat)
if args.format == "yaml":
validate_yaml(document, report)
default_name = "BOM.md" if args.format == "md" else "BOM.yaml"
output = args.output or os.path.join(root, default_name)
if args.dry_run:
report.info(document)
report.info(f"[dry-run] would write {output}")
return 0
with open(output, "w", encoding="utf-8", newline="\n") as handle:
handle.write(document)
total = sum(a.size for a in assets)
described = sum(1 for a in assets if a.description)
report.ok(f"{output} — {len(assets)} file(s), {human_size(total)}, {described} described")
if described < len(assets):
report.info(f" {len(assets) - described} file(s) still need a description")
return 0
def describe_one(root: str, args: argparse.Namespace, report: Report) -> int:
"""Set the description for a single path inside an existing BOM."""
target = args.describe_one.lstrip("./")
if not os.path.isfile(os.path.join(root, target)):
raise Die(f"no such asset: {target}")
output = args.output or os.path.join(root, "BOM.yaml" if args.format == "yaml" else "BOM.md")
description = args.title or "" # reused as the description payload for this mode
if not description:
if not sys.stdin.isatty():
raise Die("--describe-one needs a description; pass --title \"...\"")
description = input(f"description for {target}: ").strip()
if not os.path.isfile(output):
raise Die(f"no existing BOM at {output}; run without --describe-one first")
with open(output, "r", encoding="utf-8") as handle:
content = handle.read()
pattern = re.compile(
rf"(^[ \t]*{re.escape(target)}:[ \t]*\n"
rf"(?:[ \t]+\w+:.*\n)*?"
rf"[ \t]+description:[ \t]*)"
rf".*$",
re.MULTILINE,
)
updated, count = pattern.subn(rf"\g<1>{yaml_scalar(description)}", content)
if count == 0:
raise Die(f"could not find an entry for {target} in {output}")
if args.dry_run:
report.info(f"[dry-run] would set {target} description in {output}")
return 0
with open(output, "w", encoding="utf-8", newline="\n") as handle:
handle.write(updated)
report.ok(f"{target} described in {output}")
return 0
if __name__ == "__main__":
raise SystemExit(main())