a4c88f8bde
Reviewer report: pressing Approve left the "needs check" badge in place and
appeared to navigate to a new page.
Root cause was a split brain, not a single bug. Approve writes the
voice_status.json ledger, but speaker_confidence read the voice_status key
embedded in speakers.json. That inline copy is stale - the web layer
overwrites it from the ledger on read (_load_speakers), so the page showed
the right value while the assessment computed from the old one. In Last
Dwarf, "Advisor" was 'pending' in speakers.json and 'approved' in the
ledger, so the reviewer's click had saved and still left it flagged.
- assess_speakers() takes the ledger as an explicit argument, authoritative
over the inline key; load_speaker_assessments() reads voice_status.json.
- New SpeakerAssessment.reviewed flag, surfaced in the serialized view.
- Approval clears the badge but keeps the reasons visible: the underlying
facts (role-only name, one utterance) stay inspectable on hover rather
than being silently discarded.
- Corrupt/ non-dict ledgers degrade to {} - this module never raises.
- Approve swapped #speakers-list with innerHTML, replacing the whole
reviewer list with a single <tr>; now swaps only its own row via
outerHTML on #speaker-row-N.
- The review panel explained only rename/merge and never mentioned Approve,
so the process was unclear; it now lists the three outcomes.
16 new tests. Suite: 681 passed, 8 pre-existing failures unchanged.
350 lines
14 KiB
Python
350 lines
14 KiB
Python
"""Speaker identification confidence (spec 17 §3.1, axis A).
|
||
|
||
Answers "how sure are we that this speaker exists, and that we have them
|
||
complete?" — a *speaker-level* question, distinct from per-utterance attribution
|
||
confidence. The result drives the designation shown on the speakers page:
|
||
|
||
- ``clear`` — auto-accepted, no human input needed
|
||
- ``needs_review`` — a human should look at this speaker
|
||
|
||
Design bias (per the reviewer): **prefer a redundant speaker over a missing one**.
|
||
Anything suspicious is surfaced for review rather than silently dropped, because a
|
||
missing speaker is far more damaging downstream (wrong voice, unresolvable name in
|
||
the translation export) than a duplicate that can be merged in one click.
|
||
|
||
Signals (all deterministic, no LLM):
|
||
- ``unidentified`` flag from discovery
|
||
- generic / role-ish or polluted name (narration verb glued on, scene suffix)
|
||
- zero attributed dialogue, or a very thin share
|
||
- share of that speaker's utterances flagged ``needs_review``
|
||
- near-duplicate name against another speaker (possible unmerged double)
|
||
- positive: real dialogue volume, alias evidence, an approved voice
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import difflib
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
# Below this, a speaker is surfaced for review.
|
||
CLEAR_THRESHOLD = 0.75
|
||
|
||
# Narration verbs/participles that must never be part of a speaker's name.
|
||
# "Walking tallest" is "the tallest" with a stray participle.
|
||
_NARRATION_VERB_PREFIX = re.compile(
|
||
r"^(walking|nodding|running|standing|turning|looking|speaking|shouting|"
|
||
r"crying|laughing|smiling|leaning|glancing|crossing|stepping|rising|"
|
||
r"falling|stopping|gesturing|putting|taking|shifting|approaching|"
|
||
r"reaching|pointing|watching|waiting|following|leading|riding|"
|
||
r"pulling|pushing|holding|wearing|carrying|standing|bowing|nodding)"
|
||
r"\s+",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# "(Scene 2)", "(Scene 10)" — discovery leaked an internal label into the name.
|
||
_SCENE_SUFFIX = re.compile(r"\s*\(\s*Scene\s+\d+\s*\)\s*$", re.IGNORECASE)
|
||
|
||
# Role/identity words: a speaker named only by one of these is an unnamed
|
||
# character, not an identified one.
|
||
_GENERIC_NAME_WORDS = frozenset(
|
||
{
|
||
"guard", "guards", "officer", "captain", "advisor", "soldier", "man",
|
||
"woman", "girl", "boy", "kid", "king", "queen", "doctor", "master",
|
||
"commander", "leader", "pilot", "crew", "sergeant", "lieutenant",
|
||
"general", "nurse", "teacher", "student", "engineer", "merchant",
|
||
"trader", "hunter", "pirate", "scum", "voice", "voices", "speaker",
|
||
"stranger", "friend", "child", "baby", "group", "crowd", "one",
|
||
"other", "someone", "somebody", "nobody", "everyone", "tallest",
|
||
"shortest", "youngest", "oldest", "biggest", "smallest", "eldest",
|
||
"younger", "older", "men", "women", "people",
|
||
}
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class SpeakerAssessment:
|
||
"""Identification confidence + designation for one speaker."""
|
||
|
||
name: str
|
||
confidence: float = 1.0
|
||
status: str = "clear" # "clear" | "needs_review"
|
||
reasons: list[str] = field(default_factory=list)
|
||
# True once a human has accepted this speaker's voice. The heuristic can
|
||
# still list reasons; approval is what lifts the badge.
|
||
reviewed: bool = False
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"name": self.name,
|
||
"identification_confidence": round(self.confidence, 3),
|
||
"review_status": self.status,
|
||
"review_reasons": self.reasons,
|
||
"reviewed": self.reviewed,
|
||
}
|
||
|
||
|
||
def _words(name: str) -> set[str]:
|
||
return {w for w in re.findall(r"[a-z]+", (name or "").lower()) if len(w) > 1}
|
||
|
||
|
||
def name_quality_issues(name: str) -> list[str]:
|
||
"""Mechanical problems with a speaker's name itself."""
|
||
issues: list[str] = []
|
||
raw = (name or "").strip()
|
||
if _NARRATION_VERB_PREFIX.match(raw):
|
||
verb = raw.split()[0]
|
||
issues.append(f"name includes the narration verb “{verb}” (e.g. “{raw}”)")
|
||
if _SCENE_SUFFIX.search(raw):
|
||
issues.append("name ends with an internal “(Scene N)” label")
|
||
lowered = raw.lower()
|
||
if raw and raw != raw.strip():
|
||
issues.append("name has leading/trailing whitespace")
|
||
words = {w for w in _words(raw) if w not in ("the", "a", "an")}
|
||
if words and words <= _GENERIC_NAME_WORDS:
|
||
issues.append("named only by a role/comparative description, not a real name")
|
||
if lowered and not raw[0].isupper():
|
||
issues.append("name is not capitalised")
|
||
return issues
|
||
|
||
|
||
def _has_proper_word(name: str, exclude: set[str]) -> bool:
|
||
"""True if ``name`` adds at least one word that looks like a proper name."""
|
||
for w in re.findall(r"[A-Za-z][A-Za-z'’-]*", name or ""):
|
||
if w.lower() in exclude:
|
||
continue
|
||
if w[0].isupper() and w.lower() not in _GENERIC_NAME_WORDS:
|
||
# "Lieutenant Stonesplitter" adds a surname → likely a different person
|
||
return True
|
||
return False
|
||
|
||
|
||
def _near_duplicate(name: str, others: list[str]) -> str | None:
|
||
"""Return the other speaker most likely to be the same person."""
|
||
best: str | None = None
|
||
best_score = 0.0
|
||
nw = _words(name)
|
||
for other in others:
|
||
if other.lower() == name.lower():
|
||
continue
|
||
ow = _words(other)
|
||
if not ow:
|
||
continue
|
||
# A generic name is a subset of a longer one ("The Girl" ⊆ "The Girl
|
||
# helping") — but only a duplicate if the extra words are NOT a proper
|
||
# name. "Lieutenant" ⊆ "Lieutenant Stonesplitter" is two different people.
|
||
if nw and ow and (nw == ow or nw < ow or ow < nw):
|
||
longer = name if len(nw) > len(ow) else other
|
||
shared = nw & ow
|
||
score = 0.0 if _has_proper_word(longer, shared) else 0.95
|
||
else:
|
||
# difflib is noisy on same-length similar strings ("Walking oldest"
|
||
# vs "Walking Tallest" are DIFFERENT people in the same group), so
|
||
# demand a much tighter bar here than for exact-subset matches.
|
||
score = difflib.SequenceMatcher(
|
||
None, name.lower(), other.lower()
|
||
).ratio()
|
||
if score < 0.92:
|
||
score = 0.0
|
||
if score > best_score:
|
||
best_score = score
|
||
best = other
|
||
if best is not None and best_score >= 0.82:
|
||
return best
|
||
return None
|
||
|
||
|
||
def assess_speakers(
|
||
speakers: list[dict[str, Any]],
|
||
utterance_stats: dict[str, dict[str, Any]] | None = None,
|
||
approvals: dict[str, Any] | None = None,
|
||
) -> list[SpeakerAssessment]:
|
||
"""Compute an identification confidence + designation for every speaker.
|
||
|
||
``utterance_stats`` maps a speaker's primary name to its attribution health:
|
||
``{"utterances": int, "needs_review": int, "mean_confidence": float|None}``.
|
||
When omitted, only name/profile signals are used.
|
||
|
||
``approvals`` is the ``voice_status.json`` ledger - the map the Approve
|
||
button actually writes. It is authoritative over any ``voice_status`` key
|
||
already sitting in ``speakers.json``: the speaker records carry a stale
|
||
inline copy that the web layer overwrites on read, so trusting it here made
|
||
an approved speaker keep its warning badge forever.
|
||
|
||
Some defects are *categorical* rather than a matter of degree: a name with a
|
||
narration verb glued on ("Walking oldest"), an internal "(Scene 2)" label, an
|
||
uncapitalised name, a probable unmerged duplicate, or a speaker with no
|
||
dialogue at all. Those force ``needs_review`` no matter how healthy the rest
|
||
of the profile looks, because they are the defects a human must actually
|
||
decide on (rename / merge / confirm).
|
||
"""
|
||
stats = utterance_stats or {}
|
||
approvals = approvals or {}
|
||
names = [s.get("primary_name", "") for s in speakers if s.get("primary_name")]
|
||
out: list[SpeakerAssessment] = []
|
||
|
||
for spk in speakers:
|
||
name = spk.get("primary_name", "")
|
||
if not name:
|
||
continue
|
||
reasons: list[str] = []
|
||
blocking = False
|
||
confidence = 1.0
|
||
|
||
# --- profile-level signals -------------------------------------------
|
||
if spk.get("unidentified"):
|
||
confidence -= 0.45
|
||
blocking = True
|
||
reason = (spk.get("unidentified_reason") or "").strip()
|
||
reasons.append(
|
||
"discovery marked this speaker unidentified"
|
||
+ (f": {reason[:90]}" if reason else "")
|
||
)
|
||
|
||
for issue in name_quality_issues(name):
|
||
confidence -= 0.18
|
||
# every current name-quality defect needs a human decision
|
||
blocking = True
|
||
reasons.append(issue)
|
||
|
||
if not spk.get("aliases"):
|
||
confidence -= 0.05
|
||
elif not spk.get("alias_evidence"):
|
||
confidence -= 0.05
|
||
reasons.append("has aliases but no evidence for them")
|
||
|
||
demo = spk.get("demographics") or {}
|
||
if not demo.get("gender") or demo.get("gender") == "Unknown":
|
||
confidence -= 0.10
|
||
reasons.append("demographics unknown (no gender)")
|
||
if not spk.get("vocal_traits") and not spk.get("description"):
|
||
confidence -= 0.10
|
||
reasons.append("no vocal traits and no description")
|
||
|
||
# --- attribution health ----------------------------------------------
|
||
st = stats.get(name) or {}
|
||
n_utts = int(st.get("utterances") or 0)
|
||
n_flagged = int(st.get("needs_review") or 0)
|
||
if n_utts == 0:
|
||
confidence -= 0.30
|
||
blocking = True
|
||
reasons.append("no dialogue attributed to this speaker anywhere in the book")
|
||
else:
|
||
if n_utts <= 2:
|
||
confidence -= 0.15
|
||
reasons.append(f"only {n_utts} utterance(s) attributed")
|
||
share = n_flagged / n_utts
|
||
if share >= 0.5:
|
||
confidence -= 0.25
|
||
reasons.append(
|
||
f"{n_flagged}/{n_utts} of their utterances are flagged for review"
|
||
)
|
||
elif share > 0.25:
|
||
confidence -= 0.12
|
||
reasons.append(
|
||
f"{n_flagged}/{n_utts} of their utterances are flagged for review"
|
||
)
|
||
|
||
# --- duplicate risk ----------------------------------------------------
|
||
dup = _near_duplicate(name, names)
|
||
if dup:
|
||
confidence -= 0.20
|
||
blocking = True
|
||
reasons.append(f"possibly the same person as “{dup}” — consider merging")
|
||
|
||
confidence = max(0.0, min(1.0, confidence))
|
||
status = "clear"
|
||
if blocking or confidence < CLEAR_THRESHOLD:
|
||
status = "needs_review"
|
||
|
||
# A human decision outranks the heuristic. The docstring already claimed
|
||
# an approved voice was a positive signal, but nothing read voice_status -
|
||
# so clicking Approve could never clear the badge, which made the button
|
||
# look broken. Approval records "a person looked at this speaker and
|
||
# accepted it"; the reasons are kept on the assessment so the underlying
|
||
# facts (one utterance, no alias evidence) remain visible rather than
|
||
# being silently discarded.
|
||
raw = approvals.get(name, spk.get("voice_status"))
|
||
approved = str(raw or "").strip().lower() == "approved"
|
||
if approved and status == "needs_review":
|
||
status = "clear"
|
||
|
||
out.append(
|
||
SpeakerAssessment(
|
||
name=name,
|
||
confidence=confidence,
|
||
status=status,
|
||
# de-dupe, preserve order
|
||
reasons=list(dict.fromkeys(reasons)),
|
||
reviewed=approved,
|
||
)
|
||
)
|
||
|
||
out.sort(key=lambda a: (a.status != "needs_review", -a.confidence, a.name.lower()))
|
||
return out
|
||
|
||
|
||
def _utterance_stats(db_path: Path) -> dict[str, dict[str, Any]]:
|
||
"""Per-speaker attribution health from the project DB (no DB -> {})."""
|
||
import sqlite3
|
||
|
||
if not db_path.exists():
|
||
return {}
|
||
try:
|
||
conn = sqlite3.connect(str(db_path))
|
||
conn.row_factory = sqlite3.Row
|
||
rows = conn.execute(
|
||
"SELECT speaker_name, COUNT(*) AS n, "
|
||
"SUM(COALESCE(needs_review, 0)) AS nr "
|
||
"FROM utterances WHERE type = 'dialogue' GROUP BY speaker_name"
|
||
).fetchall()
|
||
conn.close()
|
||
except Exception:
|
||
return {}
|
||
return {
|
||
r["speaker_name"]: {"utterances": r["n"], "needs_review": r["nr"] or 0}
|
||
for r in rows
|
||
}
|
||
|
||
|
||
def _load_voice_ledger(data_dir: Path) -> dict[str, Any]:
|
||
"""The ``voice_status.json`` approval ledger written by the Approve button.
|
||
|
||
Never raises: a missing or corrupt ledger yields ``{}`` so the speakers page
|
||
still renders, matching this module's contract everywhere else.
|
||
"""
|
||
import json
|
||
|
||
path = data_dir / "voice_status.json"
|
||
if not path.exists():
|
||
return {}
|
||
try:
|
||
with open(path) as f:
|
||
data = json.load(f)
|
||
except Exception:
|
||
return {}
|
||
return data if isinstance(data, dict) else {}
|
||
|
||
|
||
def load_speaker_assessments(data_dir: Path) -> dict[str, SpeakerAssessment]:
|
||
"""Assess every speaker in a project; returns a name -> assessment map.
|
||
|
||
Never raises: a missing/broken speakers.json or DB yields an empty map so
|
||
the speakers page still renders.
|
||
"""
|
||
import json
|
||
|
||
speakers_path = data_dir / "speakers.json"
|
||
if not speakers_path.exists():
|
||
return {}
|
||
try:
|
||
with open(speakers_path) as f:
|
||
speakers = json.load(f).get("speakers", [])
|
||
except Exception:
|
||
return {}
|
||
stats = _utterance_stats(data_dir / "project.db")
|
||
approvals = _load_voice_ledger(data_dir)
|
||
return {a.name: a for a in assess_speakers(speakers, stats, approvals)}
|