UmbraAudio/scripts/scan_name_tokens.py

#!/usr/bin/env python3
# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Scan the manuscript for orphan name tokens (unresolved speaker identities).

Walks every chapter's utterances, collects proper-noun tokens used as speaker
evidence (unresolved name speech-tags and un-profiled ``speaker_name`` values),
groups them book-wide, and writes them to the ``name_tokens`` table. Human-set
resolutions (assigned/created/dismissed) are preserved across re-runs.

Usage:
    python scripts/scan_name_tokens.py --project "Last Dwarf"
    python scripts/scan_name_tokens.py --project "Last Dwarf" --json
"""
from __future__ import annotations

import argparse
import json
import sqlite3
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

from umbraaudio.analysis.name_tokens import (
    build_alias_map,
    ensure_name_tokens_table,
    persist_name_tokens,
    scan_project,
)
from umbraaudio.config import load_config


def _build_alias_map(speakers_path: Path) -> dict[str, str]:
    if not speakers_path.exists():
        return {}
    with open(speakers_path) as f:
        data = json.load(f)
    return build_alias_map(data.get("speakers", []))


def _project_id(db: sqlite3.Connection) -> int:
    row = db.execute("SELECT id FROM project LIMIT 1").fetchone()
    if row is None:
        raise SystemExit("No project row found in DB.")
    return row["id"]


def _run_action(args, cfg, db: sqlite3.Connection) -> int:
    """Handle a one-shot resolution action and exit."""
    from umbraaudio.analysis.orphan_resolution import (
        ResolutionError,
        assign_token_to_speaker,
        create_speaker_from_token,
        dismiss_token,
        merge_speakers,
    )

    pid = _project_id(db)
    try:
        if args.assign:
            result = assign_token_to_speaker(
                db, cfg.data_dir, pid, args.token, args.speaker
            )
        elif args.create:
            result = create_speaker_from_token(
                db, cfg.data_dir, pid, args.token, description=args.reason or ""
            )
        elif args.dismiss:
            result = dismiss_token(db, pid, args.token, reason=args.reason or "")
        else:  # merge
            result = merge_speakers(db, cfg.data_dir, args.token, args.speaker)
    except ResolutionError as e:
        print(f"Error: {e}")
        return 1
    print(json.dumps(result, indent=2))
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--project", default=".", help="Project name or path")
    parser.add_argument(
        "--chapters", type=str, default=None, help="Comma-separated chapter IDs"
    )
    parser.add_argument(
        "--write", action="store_true", help="Persist results to the name_tokens table"
    )
    parser.add_argument("--json", action="store_true", help="Emit JSON")
    # Resolution actions (mutating; run instead of a scan)
    parser.add_argument("--token", default=None, help="Orphan token to act on")
    parser.add_argument(
        "--speaker", default=None, help="Target speaker primary name (--assign/--merge)"
    )
    parser.add_argument(
        "--assign", action="store_true", help="Assign --token as an alias of --speaker"
    )
    parser.add_argument(
        "--create", action="store_true", help="Create a new speaker named --token"
    )
    parser.add_argument(
        "--dismiss", action="store_true", help="Dismiss --token (not a character)"
    )
    parser.add_argument(
        "--merge", action="store_true", help="Merge speaker --token into --speaker"
    )
    parser.add_argument("--reason", default="", help="Optional note/reason")
    args = parser.parse_args()

    cfg = load_config(args.project)
    db_path = cfg.db_path
    if not db_path.exists():
        print(f"Database not found: {db_path}")
        return 1

    db = sqlite3.connect(str(db_path))
    db.row_factory = sqlite3.Row

    if args.assign or args.create or args.dismiss or args.merge:
        if not args.token:
            print("--token is required for resolution actions")
            return 1
        if (args.assign or args.merge) and not args.speaker:
            print("--speaker is required for --assign/--merge")
            return 1
        try:
            return _run_action(args, cfg, db)
        finally:
            # Close explicitly so any committed action is flushed and no
            # dangling transaction is left open.
            db.close()

    ids = (
        [int(x.strip()) for x in args.chapters.split(",")] if args.chapters else None
    )

    tokens = scan_project(db, cfg.data_dir, ids)

    if args.write:
        pid_row = db.execute("SELECT id FROM project LIMIT 1").fetchone()
        if pid_row is None:
            print("No project row found in DB; cannot write name_tokens.")
            return 1
        ensure_name_tokens_table(db)
        inserted, updated = persist_name_tokens(db, pid_row["id"], tokens)
        print(f"Wrote name_tokens: {inserted} inserted, {updated} updated.")

    if args.json:
        print(json.dumps([t.to_dict() for t in tokens], indent=2))
    else:
        print(f"Orphan name tokens: {len(tokens)}\n")
        for t in tokens:
            sug = (
                f"  [suggest: {t.suggestion} {t.suggestion_score:.2f}]"
                if t.suggestion
                else ""
            )
            d = t.to_dict()
            print(
                f"  {t.occurrences:3d}x  [{t.kind}]  {t.token!r}  "
                f"chapters={d['chapters']}{sug}"
            )
            for sample in d["sample_texts"][:1]:
                print(f"          e.g. {sample[:100]!r}")

    db.close()
    return 0


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