tts-voice-creator-clone-and-design-2/scripts/audit_character_records.py

#!/usr/bin/env python3
"""Read-only audit of saved character records for extraction/merge damage.

Usage:
    python3 scripts/audit_character_records.py [DB_PATH] [--book SUBSTRING] [--json] [--details]

DB_PATH defaults to config/tts_creator.db. The database is opened with
SQLite's read-only URI mode (``mode=ro``) and ``PRAGMA query_only``; nothing is
written. Run it against a copy when the app is busy.

Reports, per book:
  duplicated_sheets      two records with the same long field text / the same
                         sources list (one sheet copied into two records)
  alias_collisions       an alias or first/last name that is another cast
                         member's record name in the same book
  generic_alias_links    two records linked only by a generic alias such as
                         "Zwerg" / "der Zwerg" / "Mann" (never identity proof)
  thinking_leakage       reasoning text in a field ("Warte:", "Text sagt", ...)
  foreign_sources        source quotes that name another cast member but not
                         this character
  scene_quote_other      the story-moment quote the image prompt would pick
                         names another cast member and not this character
See architecture/character-extraction.md.
"""
from __future__ import annotations

import argparse
import collections
import json
import os
import sqlite3
import sys
import tempfile
from pathlib import Path

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

from core import character_records as cr  # noqa: E402

DUP_FIELDS = ("race_species", "relationships", "physical", "backstory", "motivation", "clothing",
              "profession", "age_estimate", "skills", "capabilities", "fears", "mannerisms", "secret")
DUP_MIN_CHARS = 25


def load_records(db_path: str, book_filter: str = "") -> list[dict]:
    uri = Path(db_path).resolve().as_uri() + "?mode=ro"
    con = sqlite3.connect(uri, uri=True)
    try:
        con.execute("PRAGMA query_only = ON")
        rows = con.execute("SELECT id, book, name, sheet, created FROM characters").fetchall()
    finally:
        con.close()
    out = []
    for rid, book, name, sheet, created in rows:
        if book_filter and book_filter.lower() not in str(book).lower():
            continue
        try:
            data = json.loads(sheet or "{}")
        except json.JSONDecodeError:
            data = {}
        out.append({"id": rid, "book": book or "", "name": name or "", "sheet": data if isinstance(data, dict) else {},
                    "created": created or ""})
    return out


def _picker():
    """The story-moment picker the image prompts use (current code)."""
    os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-audit-config-"))
    try:
        import routes.conversation as conv  # noqa: WPS433
    except Exception as exc:  # pragma: no cover - environment without app deps
        print(f"[audit] scene-quote check skipped: {exc}", file=sys.stderr)
        return None

    def pick(name: str, sheet: dict) -> str:
        sheet = dict(sheet)
        sheet.setdefault("name", name)
        sheet.pop("visual_scene", None)   # audit the extracted evidence, not a manual override
        try:
            return conv._character_scene_evidence(sheet, name)
        except TypeError:                 # older signature without the name argument
            return conv._character_scene_evidence(sheet)
    return pick


def audit(records: list[dict], with_scene: bool = True) -> dict:
    by_book: dict[str, list[dict]] = collections.defaultdict(list)
    for rec in records:
        by_book[rec["book"]].append(rec)
    issues: dict[str, list] = {k: [] for k in (
        "duplicated_sheets", "alias_collisions", "generic_alias_links",
        "thinking_leakage", "foreign_sources", "scene_quote_other")}
    pick = _picker() if with_scene else None
    for book, recs in by_book.items():
        cast = [(r["name"], r["sheet"]) for r in recs]
        names_lower = {cr.normalize_identity_token(r["name"]): r["name"] for r in recs}
        for i, a in enumerate(recs):
            for b in recs[i + 1:]:
                same = [f for f in DUP_FIELDS
                        if len(str(a["sheet"].get(f) or "")) >= DUP_MIN_CHARS and a["sheet"].get(f) == b["sheet"].get(f)]
                same_sources = bool(a["sheet"].get("sources")) and a["sheet"].get("sources") == b["sheet"].get("sources")
                if len(same) >= 2 or same_sources or "race_species" in same:
                    issues["duplicated_sheets"].append({
                        "book": book, "a": a["name"], "b": b["name"], "fields": same, "same_sources": same_sources,
                        "a_created": a["created"], "b_created": b["created"]})
                ta = cr.identity_tokens(a["name"], a["sheet"], include_generic=True)
                tb = cr.identity_tokens(b["name"], b["sheet"], include_generic=True)
                shared = ta & tb
                if shared and not cr.shared_identity_evidence(ta, tb):
                    issues["generic_alias_links"].append({"book": book, "a": a["name"], "b": b["name"],
                                                          "shared": sorted(shared)})
        for rec in recs:
            name, sheet = rec["name"], rec["sheet"]
            own = cr.normalize_identity_token(name)
            for alias in cr.split_identity_tokens(sheet.get("aliases")):
                other = names_lower.get(cr.normalize_identity_token(alias))
                if other and cr.normalize_identity_token(other) != own:
                    issues["alias_collisions"].append({"book": book, "name": name, "field": "aliases", "value": alias})
            for field in ("first_name", "last_name", "full_name"):
                other = names_lower.get(cr.normalize_identity_token(sheet.get(field) or ""))
                if other and cr.normalize_identity_token(other) != own:
                    issues["alias_collisions"].append({"book": book, "name": name, "field": field,
                                                       "value": sheet.get(field)})
            leaked = [f for f in cr.TEXT_FIELDS if cr.has_thinking_leakage(sheet.get(f))]
            if leaked:
                issues["thinking_leakage"].append({"book": book, "name": name, "fields": leaked})
            for src in sheet.get("sources") or []:
                quote = str((src or {}).get("quote") or "") if isinstance(src, dict) else ""
                if not quote or cr.quote_mentions_identity(quote, name, sheet):
                    continue
                others = cr.quote_mentions_other_cast(quote, name, sheet, cast)
                if others:
                    issues["foreign_sources"].append({"book": book, "name": name, "others": others,
                                                      "quote": quote[:160]})
            if pick is not None:
                quote = pick(name, sheet)
                source_quotes = {str((x or {}).get("quote") or "").strip()[:700]
                                 for x in sheet.get("sources") or [] if isinstance(x, dict)}
                # Only a picked SOURCE quote counts; the fallback is the
                # character's own mannerisms/capabilities text.
                if quote and quote.strip() in source_quotes and not cr.quote_mentions_identity(quote, name, sheet):
                    others = cr.quote_mentions_other_cast(quote, name, sheet, cast)
                    if others:
                        issues["scene_quote_other"].append({"book": book, "name": name, "others": others,
                                                            "quote": quote[:160]})
    summary = {
        "records": len(records),
        "books": len(by_book),
        "duplicated_sheet_pairs": len(issues["duplicated_sheets"]),
        "alias_collisions": len(issues["alias_collisions"]),
        "records_with_alias_collisions": len({(x["book"], x["name"]) for x in issues["alias_collisions"]}),
        "generic_alias_links": len(issues["generic_alias_links"]),
        "records_with_thinking_leakage": len(issues["thinking_leakage"]),
        "foreign_source_quotes": len(issues["foreign_sources"]),
        "records_with_foreign_sources": len({(x["book"], x["name"]) for x in issues["foreign_sources"]}),
        "scene_quote_about_other": len(issues["scene_quote_other"]),
    }
    return {"summary": summary, "issues": issues}


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("db", nargs="?", default=str(ROOT / "config" / "tts_creator.db"))
    parser.add_argument("--book", default="", help="only books whose title contains this text")
    parser.add_argument("--json", action="store_true", help="print the full result as JSON")
    parser.add_argument("--details", action="store_true", help="list every finding")
    parser.add_argument("--no-scene", action="store_true", help="skip the image-prompt scene-quote check")
    args = parser.parse_args(argv)
    if not Path(args.db).is_file():
        print(f"database not found: {args.db}", file=sys.stderr)
        return 2
    result = audit(load_records(args.db, args.book), with_scene=not args.no_scene)
    if args.json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0
    for key, value in result["summary"].items():
        print(f"{key:32} {value}")
    if args.details:
        for kind, items in result["issues"].items():
            print(f"\n## {kind} ({len(items)})")
            for item in items:
                print("  " + json.dumps(item, ensure_ascii=False))
    return 0


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