2c6e227c82
`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.
517 lines
17 KiB
Python
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())
|