105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Run character discovery to a SEPARATE output file (never overwrites speakers.json).
|
|
|
|
Discovery is slow (4 LLM passes/chapter). This runner lets you validate the
|
|
prompt on a few chapters before committing to a full-book run, and writes
|
|
results somewhere safe so the existing curated cast can be merged deliberately.
|
|
|
|
Usage:
|
|
python scripts/run_discovery.py --project "Last Dwarf" --chapters 6 --out /tmp/disc_ch6.json
|
|
python scripts/run_discovery.py --project "Last Dwarf" --all --out /tmp/disc_full.json
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from umbraaudio.analysis.character_discovery import CharacterDiscoverer
|
|
from umbraaudio.config import load_config
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def main() -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument("--project", default=".")
|
|
ap.add_argument("--chapters", default=None, help="Comma-separated chapter IDs")
|
|
ap.add_argument("--all", action="store_true", help="All chapters")
|
|
ap.add_argument("--out", required=True, help="Output JSON path (speakers list)")
|
|
ap.add_argument("--model", default=None, help="Override LLM model id")
|
|
args = ap.parse_args()
|
|
|
|
cfg = load_config(args.project)
|
|
db_path = cfg.db_path
|
|
if not db_path.exists():
|
|
print(f"DB not found: {db_path}")
|
|
return 1
|
|
|
|
db = sqlite3.connect(str(db_path))
|
|
db.row_factory = sqlite3.Row
|
|
if args.chapters:
|
|
ids = [int(x.strip()) for x in args.chapters.split(",")]
|
|
ph = ",".join("?" * len(ids))
|
|
rows = db.execute(
|
|
f"SELECT id, chapter_index, title, source_text FROM chapters "
|
|
f"WHERE id IN ({ph}) ORDER BY chapter_index",
|
|
ids,
|
|
).fetchall()
|
|
else:
|
|
rows = db.execute(
|
|
"SELECT id, chapter_index, title, source_text FROM chapters "
|
|
"ORDER BY chapter_index"
|
|
).fetchall()
|
|
db.close()
|
|
|
|
chapters = [
|
|
{
|
|
"id": r["id"],
|
|
"chapter_index": r["chapter_index"],
|
|
"title": r["title"],
|
|
"source_text": r["source_text"] or "",
|
|
}
|
|
for r in rows
|
|
if (r["source_text"] or "").strip()
|
|
]
|
|
if not chapters:
|
|
print("No chapters with text selected.")
|
|
return 1
|
|
logger.info("Running discovery over %d chapter(s)", len(chapters))
|
|
|
|
disc = CharacterDiscoverer(cfg.llm, data_dir=cfg.data_dir)
|
|
if args.model:
|
|
disc._active_model = args.model
|
|
result = await disc.discover_speakers(
|
|
chapters=chapters,
|
|
book_title=cfg.project_dir.name,
|
|
)
|
|
|
|
out = Path(args.out)
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(out, "w") as f:
|
|
json.dump(
|
|
{"speakers": [s.to_dict() for s in result.speakers]}, f, indent=2
|
|
)
|
|
|
|
print(f"\n=== DISCOVERY COMPLETE: {len(result.speakers)} speakers -> {out} ===")
|
|
for s in result.speakers:
|
|
un = " [UNIDENTIFIED]" if s.unidentified else ""
|
|
print(f" {s.primary_name}{un}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(asyncio.run(main()))
|