UmbraAudio/scripts/build_scale_totals.py

#!/usr/bin/env python3
"""Measure totals across every project on this machine.

Asked for as "books, hours, bytes, so the catalogue is something a person
maintains or something that has to be generated". Handing over a number typed
into a chat message would be exactly the estimate the request refuses, so this
measures instead.

Two separate quantities, because conflating them is how a catalogue ends up
wrong:

* **Project data** — chapters, utterances, speakers. Cheap, exact, from SQLite.
* **Audio on disk** — files, bytes, duration. Expensive to measure exactly, so
  duration is measured from headers and reported with the caveat that it is a
  header read and not a decode.

    python3 scripts/build_scale_totals.py --out docs/tts/scale-totals.json

Read-only. Opens each project database with ``mode=ro`` and walks directories.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import sqlite3
import wave
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

REPO = Path(__file__).resolve().parent.parent

#: Directories that hold intermediate work rather than product. Counted
#: separately, because a catalogue that sweeps scratch into the product is a
#: catalogue nobody can answer a question against.
SCRATCH_DIRS = ("cache", "archives", "pronunciation_audio", "voice_tests",
                "samples", "debug", "tmp")


def schema_fingerprint(conn: sqlite3.Connection) -> str:
    """Fingerprint of the table/column *shape*, not the data.

    Two projects with different books can share a fingerprint. That is the
    point: it answers "is this the same schema", which cannot be confused with
    "is this the same data". Counting one table's presence is a proxy for that
    and was replaced by the real thing.
    """

    parts: list[str] = []
    tables = [row[0] for row in conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' "
        "AND name NOT LIKE 'sqlite_%' ORDER BY name")]
    for table in tables:
        for row in conn.execute(f"PRAGMA table_info('{table}')"):
            parts.append(f"{table}.{row[1]}:{row[2]}")
    return hashlib.sha256("\n".join(sorted(parts)).encode()).hexdigest()[:16]


def measure_project(directory: Path) -> dict[str, Any] | None:
    umbra = directory / ".umbra"
    db = umbra / "project.db"
    if not db.exists() or db.stat().st_size == 0:
        return None

    data: dict[str, Any] = {"project": directory.name}
    try:
        conn = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
        for label, sql in (
            ("chapters", "SELECT count(*) FROM chapters"),
            ("utterances", "SELECT count(*) FROM utterances"),
            ("speakers_named", "SELECT count(DISTINCT trim(speaker_name)) FROM utterances "
                               "WHERE speaker_name IS NOT NULL"),
            ("words", "SELECT sum(word_count) FROM chapters"),
        ):
            try:
                data[label] = conn.execute(sql).fetchone()[0]
            except sqlite3.Error:
                data[label] = None
        data["schema_fingerprint"] = schema_fingerprint(conn)
        conn.close()
    except sqlite3.Error as error:
        data["error"] = str(error)[:160]
        return data

    # --- audio on disk ------------------------------------------------
    chapter_audio = umbra / "chapter_audio"
    audio_files = marker_files = 0
    audio_bytes = marker_bytes = scratch_bytes = 0
    scratch = 0
    duration_ms = 0
    unreadable = 0
    if chapter_audio.is_dir():
        for path in chapter_audio.rglob("*"):
            if not path.is_file():
                continue
            relative = path.relative_to(chapter_audio)
            if any(part in SCRATCH_DIRS for part in relative.parts):
                scratch += 1
                scratch_bytes += path.stat().st_size
                continue
            # A render and its staleness marker are different things and are
            # counted separately: one is audio, the other is a JSON sidecar,
            # and lumping them makes "files" mean neither.
            if path.name.endswith(".wav.txt"):
                marker_files += 1
                marker_bytes += path.stat().st_size
                continue
            audio_files += 1
            audio_bytes += path.stat().st_size
            if path.suffix.lower() == ".wav":
                try:
                    with wave.open(str(path)) as handle:
                        rate = handle.getframerate()
                        if rate:
                            duration_ms += round(handle.getnframes() / rate * 1000)
                except (OSError, wave.Error, EOFError):
                    unreadable += 1

    data["audio"] = {
        "audio_files": audio_files,
        "audio_bytes": audio_bytes,
        "audio_duration_ms": duration_ms,
        "audio_duration_hours": round(duration_ms / 3_600_000, 2),
        "files_with_unreadable_header": unreadable,
        "marker_files_excluded_from_audio": marker_files,
        "marker_bytes": marker_bytes,
        "scratch_files_excluded": scratch,
        "scratch_bytes_excluded": scratch_bytes,
    }
    for label, directory_name in (("voices", "voices"), ("export", "export")):
        target = umbra / directory_name
        if target.is_dir():
            data[label] = {
                "files": sum(1 for p in target.rglob("*") if p.is_file()),
                "bytes": sum(p.stat().st_size for p in target.rglob("*") if p.is_file()),
            }
    return data


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--root", default=str(REPO))
    parser.add_argument("--out", default=str(REPO / "docs" / "tts" / "scale-totals.json"))
    args = parser.parse_args()

    root = Path(args.root)
    projects = []
    for directory in sorted(root.iterdir()):
        if not directory.is_dir() or directory.name.startswith("."):
            continue
        measured = measure_project(directory)
        if measured is not None:
            projects.append(measured)

    chapters = sum(p.get("chapters") or 0 for p in projects)
    utterances = sum(p.get("utterances") or 0 for p in projects)
    files = sum((p.get("audio") or {}).get("audio_files") or 0 for p in projects)
    total_bytes = sum((p.get("audio") or {}).get("audio_bytes") or 0 for p in projects)
    duration = sum((p.get("audio") or {}).get("audio_duration_ms") or 0 for p in projects)
    markers = sum((p.get("audio") or {}).get("marker_files_excluded_from_audio") or 0
                  for p in projects)

    document: dict[str, Any] = {
        "document_version": 1,
        "generated_at": datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
        "generated_by": "scripts/build_scale_totals.py",
        "root": "<REPO>",
        "projects": projects,
        "totals": {
            "projects": len(projects),
            "chapters": chapters,
            "utterances": utterances,
            "audio_files": files,
            "audio_bytes": total_bytes,
            "audio_gigabytes": round(total_bytes / 1e9, 2),
            "audio_duration_hours": round(duration / 3_600_000, 2),
            "marker_files": markers,
            "distinct_schema_fingerprints": len(
                {p.get("schema_fingerprint") for p in projects if p.get("schema_fingerprint")}),
        },
        "caveats": [
            "Duration is read from WAV headers, not decoded. It is exact for "
            "PCM WAV and absent for anything else, which is why the file count "
            "and the duration are reported together rather than one from the other.",
            "Audio totals cover chapter_audio only, and exclude the *.wav.txt "
            "staleness markers, which are counted separately. Voice references "
            "and exports are counted separately per project.",
            "Scratch and intermediate output is excluded and counted "
            "separately, not silently dropped.",
            "Projects without a populated database are not listed. Absence here "
            "means 'not measured', not 'does not exist'.",
        ],
    }
    hashed = {k: v for k, v in document.items() if k != "generated_at"}
    payload = json.dumps(hashed, sort_keys=True, separators=(",", ":"))
    document["content_hash"] = {
        "algorithm": "sha256",
        "of": "this document with generated_at omitted, canonical JSON",
        "value": hashlib.sha256(payload.encode()).hexdigest(),
    }

    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n")

    totals = document["totals"]
    print(f"projects: {totals['projects']}  chapters: {totals['chapters']}  "
          f"utterances: {totals['utterances']}")
    print(f"audio:    {totals['audio_files']} files, "
          f"{totals['audio_gigabytes']} GB, "
          f"{totals['audio_duration_hours']} h "
          f"(+{totals['marker_files']} markers, "
          f"{totals['distinct_schema_fingerprints']} schema shapes)")
    print(f"hash:     {document['content_hash']['value']}")
    print(f"written:  {out}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())