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