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

#!/usr/bin/env python3
"""Dry-run-first cleanup of extraction damage in saved character records.

Usage:
    python3 scripts/clean_character_records.py [DB_PATH] [--book SUBSTRING] [--out DIR]
    python3 scripts/clean_character_records.py [DB_PATH] --apply --out DIR   # owner approval only

Without ``--apply`` nothing is written to the database: the database is opened
read-only and the planned changes are written as a report (report.json +
report.md) into ``--out`` (default: a new temp directory).

With ``--apply`` the script first writes ``backup.json`` (every affected row,
unchanged) and the report into ``--out``, then updates only those rows in one
transaction. Restore = write the backed-up ``sheet`` values back (see
architecture/character-extraction.md, "Cleanup and rollback").

Deterministic, evidence-based fixes only:
  1. reasoning leakage ("Warte: ...", "Der Text sagt ...") cut from text fields
  2. aliases / first / last / full names that are ANOTHER cast member's record
     name in the same book removed
  3. only with --drop-foreign-sources: source quotes that name another cast
     member but not this character removed (lossy, off by default)
Duplicated sheets (one sheet copied into two records) are only REPORTED: which
record is genuine needs the owner's judgement.
"""
from __future__ import annotations

import argparse
import copy
import datetime as _dt
import json
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
from scripts.audit_character_records import audit, load_records  # noqa: E402


def plan_changes(records: list[dict], drop_foreign_sources: bool = False) -> list[dict]:
    """Return one entry per record that would change: {id, book, name, changes, before, after}."""
    by_book: dict[str, list[dict]] = {}
    for rec in records:
        by_book.setdefault(rec["book"], []).append(rec)
    # Records that are copies of one another (same sources / long fields) are
    # probably one person cast twice: their names in each other's aliases are
    # not "foreign". That duplication is reported for the owner instead.
    copies: dict[tuple[str, str], set[str]] = {}
    for dup in audit(records, with_scene=False)["issues"]["duplicated_sheets"]:
        copies.setdefault((dup["book"], dup["a"]), set()).add(dup["b"])
        copies.setdefault((dup["book"], dup["b"]), set()).add(dup["a"])
    plans = []
    for book, recs in by_book.items():
        cast = [(r["name"], r["sheet"]) for r in recs]
        for rec in recs:
            twins = copies.get((book, rec["name"]), set())
            cast_names = [r["name"] for r in recs if r["name"] not in twins]
            before = rec["sheet"]
            after = copy.deepcopy(before)
            changes: list[dict] = []
            for field, (old, new) in cr.sanitize_sheet_text_fields(after).items():
                changes.append({"kind": "thinking_leakage", "field": field, "old": old, "new": new})
            # Only where the record contradicts itself: its own relationships
            # name that person as somebody else (Kolon "dient ... Sharraz
            # Garthai" yet alias "Sharraz"). Other collisions may be one person
            # cast twice (Promos / Alchimist): listed for review, never applied.
            probe = copy.deepcopy(after)
            relationships = str(after.get("relationships") or "")
            real_name = f"{after.get('first_name') or ''} {after.get('last_name') or ''}".strip()
            for removed in cr.drop_foreign_identity(rec["name"], probe, cast_names):
                field, _, value = removed.partition("=")
                probe_text = real_name if field in ("first_name", "last_name") else value
                related = [n for n in cast_names if cr.normalize_identity_token(n) != cr.normalize_identity_token(rec["name"])
                           and not cr.is_generic_identity_token(n) and cr.text_mentions_token(probe_text, n)
                           and cr.text_mentions_token(relationships, n)]
                kind = "foreign_identity" if related else "review_identity"
                changes.append({"kind": kind, "field": field, "removed": value,
                                **({"evidence": f"relationships names {', '.join(related)}"} if related else {})})
            contradicted = {c["removed"] for c in changes if c["kind"] == "foreign_identity" and c["field"] == "aliases"}
            if contradicted:
                after["aliases"] = ", ".join(a for a in cr.split_identity_tokens(after.get("aliases")) if a not in contradicted)
            for c in changes:
                if c["kind"] == "foreign_identity" and c["field"] in ("first_name", "last_name", "full_name"):
                    after[c["field"]] = ""
            sources = after.get("sources") if isinstance(after.get("sources"), list) else []
            kept = []
            for src in sources:
                quote = str(src.get("quote") or "") if isinstance(src, dict) else ""
                if drop_foreign_sources and quote and not cr.quote_mentions_identity(quote, rec["name"], after):
                    others = cr.quote_mentions_other_cast(quote, rec["name"], after, cast)
                    if others:
                        changes.append({"kind": "foreign_source", "field": "sources", "removed": quote[:240],
                                        "names": others})
                        continue
                kept.append(src)
            if len(kept) != len(sources):
                after["sources"] = kept
            if changes:
                plans.append({"id": rec["id"], "book": book, "name": rec["name"], "changes": changes,
                              "before": before, "after": after})
    return plans


def write_report(out: Path, plans: list[dict], audit_result: dict, applied: bool) -> None:
    out.mkdir(parents=True, exist_ok=True)
    slim = [{k: v for k, v in p.items() if k not in ("before", "after")} for p in plans]
    (out / "report.json").write_text(json.dumps({
        "applied": applied, "records_changed": len(plans), "plans": slim,
        "duplicated_sheets_report_only": audit_result["issues"]["duplicated_sheets"],
        "audit_summary": audit_result["summary"],
    }, ensure_ascii=False, indent=2), encoding="utf-8")
    counts: dict[str, int] = {}
    for p in plans:
        for c in p["changes"]:
            counts[c["kind"]] = counts.get(c["kind"], 0) + 1
    lines = [f"# Character record cleanup ({'APPLIED' if applied else 'dry run, nothing written'})", "",
             f"Records that would change: {sum(1 for p in plans if p['after'] != p['before'])}", ""]
    lines += [f"- {kind}: {n}" for kind, n in sorted(counts.items())]
    lines += ["", f"Duplicated sheets (report only, owner decides): {len(audit_result['issues']['duplicated_sheets'])}", ""]
    for p in plans:
        lines.append(f"## {p['book']} | {p['name']}")
        for c in p["changes"]:
            if c["kind"] == "thinking_leakage":
                lines.append(f"- {c['field']}: cut to {c['new']!r} (was {len(c['old'])} chars)")
            elif c["kind"] == "foreign_identity":
                lines.append(f"- {c['field']}: remove {c['removed']!r} ({c['evidence']})")
            elif c["kind"] == "review_identity":
                lines.append(f"- REVIEW ONLY (not changed): {c['field']} {c['removed']!r} is another record's name")
            else:
                lines.append(f"- source quote naming {', '.join(c['names'])}: remove {c['removed'][:100]!r}")
        lines.append("")
    (out / "report.md").write_text("\n".join(lines), encoding="utf-8")


def apply_plans(db_path: str, plans: list[dict], out: Path) -> None:
    out.mkdir(parents=True, exist_ok=True)
    plans = [p for p in plans if p["after"] != p["before"]]   # review-only entries are never written
    con = sqlite3.connect(db_path)
    try:
        ids = [p["id"] for p in plans]
        rows = []
        for rid in ids:
            row = con.execute("SELECT id, book, name, sheet, updated FROM characters WHERE id = ?", (rid,)).fetchone()
            if row:
                rows.append(dict(zip(("id", "book", "name", "sheet", "updated"), row)))
        backup = out / "backup.json"
        backup.write_text(json.dumps(rows, ensure_ascii=False, indent=1), encoding="utf-8")
        if len(rows) != len(ids):
            raise SystemExit("records changed since planning; nothing written")
        now = _dt.datetime.now(_dt.timezone.utc).isoformat().replace("+00:00", "Z")
        with con:
            for p in plans:
                con.execute("UPDATE characters SET sheet = ?, updated = ? WHERE id = ?",
                            (json.dumps(p["after"], ensure_ascii=False), now, p["id"]))
    finally:
        con.close()


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="")
    parser.add_argument("--out", default="", help="report/backup directory (default: new temp dir)")
    parser.add_argument("--drop-foreign-sources", action="store_true",
                        help="also remove source quotes that name another cast member but not this one "
                             "(lossy: some are about this character by pronoun/role; image prompts already skip them)")
    parser.add_argument("--apply", action="store_true",
                        help="write the changes (backup first). Only with the owner's approval.")
    args = parser.parse_args(argv)
    if not Path(args.db).is_file():
        print(f"database not found: {args.db}", file=sys.stderr)
        return 2
    out = Path(args.out) if args.out else Path(tempfile.mkdtemp(prefix="ttvc-character-cleanup-"))
    records = load_records(args.db, args.book)          # read-only connection
    plans = plan_changes(records, drop_foreign_sources=args.drop_foreign_sources)
    audit_result = audit(records, with_scene=False)
    if args.apply:
        if not args.out:
            print("--apply needs an explicit --out directory for the backup", file=sys.stderr)
            return 2
        apply_plans(args.db, plans, out)
    write_report(out, plans, audit_result, applied=args.apply)
    kinds: dict[str, int] = {}
    for p in plans:
        for c in p["changes"]:
            kinds[c["kind"]] = kinds.get(c["kind"], 0) + 1
    changed = sum(1 for p in plans if p["after"] != p["before"])
    print(f"{'applied' if args.apply else 'dry run'}: {changed} records would change {kinds}; report: {out}")
    return 0


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