221 lines
7.8 KiB
Python
221 lines
7.8 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Per-utterance QA box: issue detection, recheck/fix loop, and accept state.
|
|
|
|
These cover the helpers behind ``/api/utterance/{id}/recheck`` and
|
|
``/api/utterance/{id}/qa-accept`` without spinning up the web app.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import types
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
from umbraaudio.web.routes import (
|
|
SCHEMA_SQL,
|
|
_ensure_utterance_columns,
|
|
_load_speaker_alias_map,
|
|
_qa_accept_utterance,
|
|
_qa_accepted_for,
|
|
_qa_accepted_sig,
|
|
_qa_issues_for_chapter,
|
|
_recheck_utterance_qa,
|
|
)
|
|
|
|
|
|
def _make_db(temp_dir: Path) -> tuple[sqlite3.Connection, types.SimpleNamespace]:
|
|
db = sqlite3.connect(str(temp_dir / "project.db"))
|
|
db.row_factory = sqlite3.Row
|
|
db.executescript(SCHEMA_SQL)
|
|
db.execute(
|
|
"INSERT INTO project (id, name, source_path, source_format) VALUES (1, 'p', '', 'txt')"
|
|
)
|
|
db.execute(
|
|
"INSERT INTO chapters (id, project_id, chapter_index, title, source_text) "
|
|
"VALUES (1, 1, 1, 'Slate', '')"
|
|
)
|
|
db.commit()
|
|
return db, types.SimpleNamespace(data_dir=temp_dir)
|
|
|
|
|
|
def _write_speakers(temp_dir: Path) -> None:
|
|
(temp_dir / "speakers.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"speakers": [
|
|
{"primary_name": "Slate", "aliases": []},
|
|
{"primary_name": "Xylia", "aliases": []},
|
|
]
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
def _insert_utt(
|
|
db: sqlite3.Connection, utt_id: int, idx: int, text: str, utt_type: str, speaker: str | None
|
|
) -> None:
|
|
db.execute(
|
|
"INSERT INTO utterances (id, chapter_id, utterance_index, text, type, speaker_name) "
|
|
"VALUES (?, 1, ?, ?, ?, ?)",
|
|
(utt_id, idx, text, utt_type, speaker),
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def test_qa_accepted_sig_responds_to_speaker_and_text() -> None:
|
|
sig = _qa_accepted_sig("Slate", "Move now.")
|
|
assert sig == _qa_accepted_sig("Slate", "Move now.")
|
|
assert sig != _qa_accepted_sig("Xylia", "Move now.")
|
|
assert sig != _qa_accepted_sig("Slate", "Move now")
|
|
|
|
|
|
def test_qa_accepted_for_staleness() -> None:
|
|
sig = _qa_accepted_sig("Slate", "Move now.")
|
|
assert _qa_accepted_for(1, sig, "Slate", "Move now.") is True
|
|
# Speaker or text changed -> acceptance must not hide a fresh issue.
|
|
assert _qa_accepted_for(1, sig, "Xylia", "Move now.") is False
|
|
assert _qa_accepted_for(1, sig, "Slate", "Different") is False
|
|
assert _qa_accepted_for(0, sig, "Slate", "Move now.") is False
|
|
|
|
|
|
def test_load_speaker_alias_map_includes_aliases_and_first_words(temp_dir: Path) -> None:
|
|
_write_speakers(temp_dir)
|
|
db, _ = _make_db(temp_dir)
|
|
db.close()
|
|
(temp_dir / "speakers.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"speakers": [
|
|
{
|
|
"primary_name": "Captain Slate Ironbeard",
|
|
"aliases": ["Slate", "The Captain"],
|
|
},
|
|
{"primary_name": "Xylia", "aliases": []},
|
|
]
|
|
}
|
|
)
|
|
)
|
|
alias_map = _load_speaker_alias_map(types.SimpleNamespace(data_dir=temp_dir))
|
|
assert alias_map["slate"] == "Captain Slate Ironbeard"
|
|
assert alias_map["the captain"] == "Captain Slate Ironbeard"
|
|
assert alias_map["captain"] == "Captain Slate Ironbeard"
|
|
assert alias_map["xylia"] == "Xylia"
|
|
assert alias_map.get("another") is None
|
|
|
|
|
|
def test_qa_issues_for_chapter_flags_mismatch_and_narrator_dialogue(temp_dir: Path) -> None:
|
|
_write_speakers(temp_dir)
|
|
db, stub = _make_db(temp_dir)
|
|
_insert_utt(db, 1, 0, 'Slate says, "We have to hurry."', "narration", None)
|
|
_insert_utt(db, 2, 1, "Fine, let's go.", "dialogue", "Xylia")
|
|
_insert_utt(db, 3, 2, "Great, we're ready.", "dialogue", None)
|
|
|
|
issues = _qa_issues_for_chapter(db, 1, _load_speaker_alias_map(stub))
|
|
db.close()
|
|
|
|
assert {i["type"] for i in issues[1]} == {"speech_tag_mismatch"}
|
|
tag = issues[1][0]
|
|
assert tag["expected"] == "Slate"
|
|
assert tag["confidence"] == 0.9
|
|
|
|
assert {i["type"] for i in issues[2]} == {"narrator_dialogue"}
|
|
assert issues[2][0]["confidence"] == 0.95
|
|
assert 0 not in issues
|
|
|
|
|
|
def test_recheck_auto_fixes_resolved_mismatch(temp_dir: Path) -> None:
|
|
_write_speakers(temp_dir)
|
|
db, stub = _make_db(temp_dir)
|
|
_insert_utt(db, 1, 0, 'Slate says, "We have to hurry."', "narration", None)
|
|
_insert_utt(db, 2, 1, "Fine, let's go.", "dialogue", "Xylia")
|
|
|
|
payload = _recheck_utterance_qa(db, stub, 2)
|
|
assert payload["clean"] is True
|
|
assert payload["issues"] == []
|
|
assert payload["fixed"] == [{"utt_id": 2, "utt_index": 1, "from": "Xylia", "to": "Slate"}]
|
|
row = db.execute("SELECT speaker_name FROM utterances WHERE id = 2").fetchone()
|
|
assert row["speaker_name"] == "Slate"
|
|
db.close()
|
|
|
|
|
|
def test_recheck_leaves_unresolved_role_tag_flagged(temp_dir: Path) -> None:
|
|
_write_speakers(temp_dir)
|
|
db, stub = _make_db(temp_dir)
|
|
_insert_utt(db, 1, 0, 'Another says, "Someone must go."', "narration", None)
|
|
_insert_utt(db, 2, 1, "I will not stay.", "dialogue", "Slate")
|
|
|
|
payload = _recheck_utterance_qa(db, stub, 2)
|
|
assert payload["clean"] is False
|
|
assert payload["fixed"] == []
|
|
assert payload["issues"][0]["type"] == "speech_tag_mismatch"
|
|
assert payload["issues"][0]["confidence"] == 0.7
|
|
row = db.execute("SELECT speaker_name FROM utterances WHERE id = 2").fetchone()
|
|
assert row["speaker_name"] == "Slate"
|
|
db.close()
|
|
|
|
|
|
def test_recheck_unknown_utterance_returns_error(temp_dir: Path) -> None:
|
|
db, stub = _make_db(temp_dir)
|
|
payload = _recheck_utterance_qa(db, stub, 999)
|
|
db.close()
|
|
assert payload["error"] == "utterance not found"
|
|
|
|
|
|
def test_accept_records_across_speaker_change(temp_dir: Path) -> None:
|
|
_write_speakers(temp_dir)
|
|
db, stub = _make_db(temp_dir)
|
|
_insert_utt(db, 1, 0, 'Slate says, "We have to hurry."', "narration", None)
|
|
_insert_utt(db, 2, 1, "Fine, let's go.", "dialogue", "Xylia")
|
|
|
|
payload = _qa_accept_utterance(db, 2)
|
|
assert payload == {"utterance_id": 2, "accepted": True}
|
|
|
|
row = db.execute("SELECT qa_accepted, qa_accepted_sig FROM utterances WHERE id = 2").fetchone()
|
|
assert row["qa_accepted"] == 1
|
|
assert _qa_accepted_for(row["qa_accepted"], row["qa_accepted_sig"], "Xylia", "Fine, let's go.")
|
|
|
|
# The recheck loop re-assigns the speaker; the old acceptance must lapse.
|
|
db.execute("UPDATE utterances SET speaker_name = 'Slate' WHERE id = 2")
|
|
db.commit()
|
|
stale = db.execute(
|
|
"SELECT qa_accepted, qa_accepted_sig FROM utterances WHERE id = 2"
|
|
).fetchone()
|
|
assert (
|
|
_qa_accepted_for(stale["qa_accepted"], stale["qa_accepted_sig"], "Slate", "Fine, let's go.")
|
|
is False
|
|
)
|
|
db.close()
|
|
|
|
|
|
def test_accept_unknown_utterance_returns_error(temp_dir: Path) -> None:
|
|
db, _ = _make_db(temp_dir)
|
|
payload = _qa_accept_utterance(db, 999)
|
|
db.close()
|
|
assert payload["error"] == "utterance not found"
|
|
|
|
|
|
def test_ensure_utterance_columns_migrates_legacy_db(temp_dir: Path) -> None:
|
|
"""A project DB predating the qa_accepted/user_edited columns must upgrade.
|
|
|
|
Raw-connection routes (chapter audio, speaker/text/emotion edits, add
|
|
utterance) read and write these columns, so a missing migration 500s the
|
|
whole page.
|
|
"""
|
|
db = sqlite3.connect(str(temp_dir / "project.db"))
|
|
db.execute("CREATE TABLE utterances (id INTEGER PRIMARY KEY, chapter_id INTEGER, text TEXT)")
|
|
db.commit()
|
|
|
|
_ensure_utterance_columns(db)
|
|
cols = {r[1] for r in db.execute("PRAGMA table_info(utterances)")}
|
|
assert {"qa_accepted", "qa_accepted_at", "qa_accepted_sig", "user_edited"} <= cols
|
|
|
|
# Idempotent: a second call must not raise (columns already present).
|
|
_ensure_utterance_columns(db)
|
|
db.close()
|