#!/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())