UmbraAudio/scripts/speaker_review_audit.py

# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Speaker-review audit: is the attribution data fit to hand to the next stage?

Answers the two questions asked after a speaker review completes:

  1. Are there utterances with no speaker?
  2. Now that speakers have been defined, renamed, merged or deleted, are
     there utterances that should change?

Everything here is mechanical and read-only. It never edits the database -
these are candidate lists for a human to accept or dismiss, and the report
records what was *measured*, not what is assumed.

A null speaker on a NARRATION row is not a defect: narration is not spoken by
a character. Only dialogue without a speaker counts as missing, and a
narration row carrying a character's name is reported separately because it
usually means an inner monologue that may want the character's voice.

Usage:
    python scripts/speaker_review_audit.py --project DealHunterShip_4.0e_whole-to-audio
    python scripts/speaker_review_audit.py --project Last Dwarf --json
"""
from __future__ import annotations

import argparse
import csv
import json
import sqlite3
import sys
from collections import defaultdict
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from umbraaudio.analysis.attribution_qa import (  # noqa: E402
    build_alias_map,
    check_attribution,
)
from umbraaudio.config import load_config  # noqa: E402

# Issues that are safe to act on mechanically: the narration immediately
# before the line names a different speaker.
HIGH_CONFIDENCE_ISSUES = {"speech_tag_mismatch"}
# Advisory only - pronoun coreference and role guessing are judgement calls.
#
# narrator_dialogue is advisory rather than an action item on purpose. In a
# text like this most of these rows are quoted labels - a console reading
# "operations", a legend saying "Reactor." - which are typed as dialogue but
# are correctly read by the narrator. Treating them as defects would bury
# the one row that is a real segmentation error under nine false positives.
ADVISORY_ISSUES = {
    "role_tag_guessed",
    "address_mismatch",
    "narration_speaker_mismatch",
    "long_monologue",
}

PLACEHOLDER_SPEAKERS = {"", "__unassigned__", "unassigned", "unknown", "none", "null"}


@dataclass
class Finding:
    """One reviewable row. ``chapter``/``utterance`` are what a human needs."""

    check: str
    severity: str  # "action" | "advisory" | "info"
    chapter: int
    utterance: int
    utterance_id: int
    speaker: str
    expected: str
    detail: str
    text: str


@dataclass
class Audit:
    project: str
    generated: str
    findings: list[Finding] = field(default_factory=list)
    stats: dict[str, Any] = field(default_factory=dict)

    def add(self, **kw: Any) -> None:
        self.findings.append(Finding(**kw))

    def by_check(self) -> dict[str, list[Finding]]:
        out: dict[str, list[Finding]] = defaultdict(list)
        for f in self.findings:
            out[f.check].append(f)
        return out


def _chapters(conn: sqlite3.Connection) -> list[int]:
    return [
        r[0]
        for r in conn.execute("SELECT DISTINCT chapter_id FROM utterances ORDER BY chapter_id")
    ]


def _chapter_utterances(conn: sqlite3.Connection, chapter_id: int) -> list[dict[str, Any]]:
    rows = conn.execute(
        "SELECT id, utterance_index, speaker_name, type, text FROM utterances "
        "WHERE chapter_id = ? ORDER BY utterance_index",
        (chapter_id,),
    ).fetchall()
    return [
        {
            "id": r[0],
            "utterance_index": r[1],
            "speaker_name": r[2],
            "type": r[3],
            "text": r[4] or "",
        }
        for r in rows
    ]


def run_audit(project: str | None, as_json: bool) -> Audit:
    cfg = load_config(project) if project else load_config()
    data_dir = cfg.data_dir
    conn = sqlite3.connect(f"file:{data_dir / 'project.db'}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
    from datetime import UTC, datetime

    audit = Audit(
        project=cfg.project_dir.name,
        generated=datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%SZ"),
    )

    speakers_path = data_dir / "speakers.json"
    speakers = []
    if speakers_path.exists():
        speakers = json.loads(speakers_path.read_text()).get("speakers", [])
    alias_map = build_alias_map(speakers_path)
    known_lower = set(alias_map.keys())

    # ---------------------------------------------------------- 1. no speaker
    missing = conn.execute(
        "SELECT id, chapter_id, utterance_index, type, speaker_name, text FROM utterances "
        "WHERE type = 'dialogue' AND ("
        "  speaker_name IS NULL OR TRIM(speaker_name) = '' "
        "  OR LOWER(TRIM(speaker_name)) IN (" + ",".join("?" * len(PLACEHOLDER_SPEAKERS)) + "))",
        tuple(PLACEHOLDER_SPEAKERS),
    ).fetchall()
    for r in missing:
        audit.add(
            check="dialogue_missing_speaker",
            severity="action",
            chapter=r[1],
            utterance=r[2],
            utterance_id=r[0],
            speaker=r[4] or "",
            expected="",
            detail="dialogue with no speaker",
            text=(r[5] or "")[:110],
        )

    # Narration with no speaker is normal; count it so the report is honest
    # about what was excluded rather than quietly dropping it.
    audit.stats["narration_without_speaker"] = conn.execute(
        "SELECT COUNT(*) FROM utterances WHERE type IN ('narration','scene_break') "
        "AND (speaker_name IS NULL OR TRIM(speaker_name) = '')"
    ).fetchone()[0]

    # ------------------------------------------------- 2. unresolvable speaker
    used = {
        r[0]
        for r in conn.execute(
            "SELECT DISTINCT speaker_name FROM utterances "
            "WHERE speaker_name IS NOT NULL AND TRIM(speaker_name) <> ''"
        )
    }
    for name in sorted(used):
        if name.casefold() in known_lower:
            continue
        rows = conn.execute(
            "SELECT id, chapter_id, utterance_index, text FROM utterances WHERE speaker_name = ?",
            (name,),
        ).fetchall()
        for r in rows:
            audit.add(
                check="speaker_not_resolvable",
                severity="action",
                chapter=r[1],
                utterance=r[2],
                utterance_id=r[0],
                speaker=name,
                expected="",
                detail="attributed to a speaker that no longer exists",
                text=(r[3] or "")[:110],
            )

    # ------------------------------------- 3. identity / case fragmentation
    # The database stores whatever string was written at attribution time, so
    # one speaker can appear under several spellings ("narrator" and "Narrator",
    # for instance). Both resolve through the alias map so the audio is correct
    # today, but anything that groups by the raw column will split them - which
    # is exactly what a translation export would do. Reported here so it is a
    # decision rather than a surprise downstream.
    canonical: dict[str, set[str]] = defaultdict(set)
    for name in used:
        target = alias_map.get(name.casefold())
        if target:
            canonical[target].add(name)
    for target, variants in canonical.items():
        if len(variants) < 2:
            continue
        for variant in sorted(variants):
            count = conn.execute(
                "SELECT COUNT(*) FROM utterances WHERE speaker_name = ?", (variant,)
            ).fetchone()[0]
            if not count:
                continue
            audit.add(
                check="speaker_name_variants",
                severity="action",
                chapter=0,
                utterance=0,
                utterance_id=0,
                speaker=variant,
                expected=target,
                detail=(
                    f"{count} utterances stored under this spelling; the same "
                    "speaker also appears as "
                    f"{sorted(variants - {variant})}. They resolve to one voice, "
                    "but a per-speaker export would split them."
                ),
                text="",
            )

    # ------------------------------------- 4. new speaker vs stale attribution
    # This is the "a speaker has since been defined" question: the narration
    # right before the line names speaker X, and the line is not attributed
    # to X.
    for chapter_id in _chapters(conn):
        utts = _chapter_utterances(conn, chapter_id)
        result = check_attribution(utts, chapter_id, alias_map)
        for issue in result.issues:
            idx = issue.utterance_index
            row = next((u for u in utts if u["utterance_index"] == idx), None)
            if row is None:
                continue
            severity = (
                "action"
                if issue.issue_type in HIGH_CONFIDENCE_ISSUES
                else "advisory"
            )
            audit.add(
                check=issue.issue_type,
                severity=severity,
                chapter=chapter_id,
                utterance=idx,
                utterance_id=row["id"],
                speaker=row["speaker_name"] or "",
                expected=issue.expected or "",
                detail=issue.evidence or "",
                text=(row["text"] or "")[:110],
            )

    audit.stats.update(
        utterances=conn.execute("SELECT COUNT(*) FROM utterances").fetchone()[0],
        dialogue=conn.execute(
            "SELECT COUNT(*) FROM utterances WHERE type='dialogue'"
        ).fetchone()[0],
        speakers=len(speakers),
        names_and_aliases=len(known_lower),
    )
    return audit


def write_reports(audit: Audit, out_dir: Path) -> tuple[Path, Path]:
    out_dir.mkdir(parents=True, exist_ok=True)
    md = out_dir / "speaker_review_audit.md"
    csv_path = out_dir / "speaker_review_audit.csv"

    grouped = audit.by_check()
    order = [
        ("dialogue_missing_speaker", "Dialogue with no speaker"),
        ("speaker_not_resolvable", "Attributed to a speaker that no longer exists"),
        ("narrator_dialogue", "Dialogue still voiced by the narrator"),
        ("speech_tag_mismatch", "Narration names a different speaker"),
        ("speaker_name_variants", "One speaker, several spellings in the database"),
        ("role_tag_guessed", "Role tag guessed as a named character"),
        ("narration_speaker_mismatch", "Narration speaker disagrees (pronoun)"),
        ("address_mismatch", "Addressed to the speaker themselves"),
        ("long_monologue", "Long monologue"),
    ]
    action = [f for f in audit.findings if f.severity == "action"]

    lines = [
        f"# Speaker review audit - {audit.project}",
        "",
        f"Generated {audit.generated}. Read-only: nothing was modified.",
        "",
        "## Summary",
        "",
        f"- Utterances: **{audit.stats.get('utterances')}** "
        f"({audit.stats.get('dialogue')} dialogue)",
        f"- Speakers: **{audit.stats.get('speakers')}** "
        f"({audit.stats.get('names_and_aliases')} distinct names + aliases)",
        f"- Narration rows with no speaker: "
        f"**{audit.stats.get('narration_without_speaker')}** - expected, narration is "
        "not spoken by a character",
        f"- Findings needing a decision: **{len(action)}**",
        "",
    ]

    if not audit.findings:
        lines += ["No findings. The attribution data is clean.", ""]

    for check, title in order:
        rows = grouped.get(check)
        if not rows:
            continue
        lines += [f"## {title} ({len(rows)})", ""]
        lines += ["| Chapter | Utt | Severity | Speaker | Expected | Detail |",
                  "|---|---|---|---|---|---|"]
        for f in sorted(rows, key=lambda r: (r.chapter, r.utterance)):
            detail = (f.detail or "").replace("|", "\\|")[:70]
            # Book-level findings have no single utterance; "-" beats a
            # misleading chapter 0 / utterance 0.
            where = f"{f.chapter} | {f.utterance}" if f.chapter else "- | -"
            lines.append(
                f"| {where} | {f.severity} | {f.speaker} | "
                f"{f.expected} | {detail} |"
            )
        lines.append("")

    md.write_text("\n".join(lines))

    with open(csv_path, "w", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=list(asdict(audit.findings[0]).keys()))
        w.writeheader()
        for f in audit.findings:
            w.writerow(asdict(f))
    return md, csv_path


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--project")
    ap.add_argument("--json", action="store_true", help="print JSON to stdout")
    ap.add_argument("--out", help="output directory (default: <project>/.umbra/reports)")
    args = ap.parse_args()

    audit = run_audit(args.project, args.json)
    cfg = load_config(args.project) if args.project else load_config()
    out_dir = Path(args.out) if args.out else cfg.data_dir / "reports"
    md, csv_path = write_reports(audit, out_dir)

    if args.json:
        print(json.dumps({
            "project": audit.project,
            "generated": audit.generated,
            "stats": audit.stats,
            "findings": [asdict(f) for f in audit.findings],
        }, indent=2))
    else:
        action = sum(1 for f in audit.findings if f.severity == "action")
        print(f"  {audit.project}: {len(audit.findings)} findings "
              f"({action} need a decision)")
        for f in audit.findings:
            if f.severity != "action":
                continue
            print(f"    [{f.check}] ch{f.chapter} utt{f.utterance} "
                  f"speaker={f.speaker!r} expected={f.expected!r}")
        print(f"  written: {md}")
        print(f"  written: {csv_path}")
    return 0


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