esh-pfi-infrastructure/scripts/r49-corpus/rename.py

"""R49 Stage D2 (final) + D3 — entity resolution and deterministic rename augmentation.

D2's gender resolution is TITLE-FIRST, and that is the change from F02's method.
F02 used pronoun proximity and recorded that it is structurally blind to the
first-person narrator, whose name appears mainly in dialogue surrounded by other
people's pronouns. Measured here on Bronte, proximity called **Jane male** -- the
narrator of Jane Eyre, and the single worst entity to get wrong.

Titles do not have that blind spot. `Miss Eyre`, `Mrs. Fairfax`, `Mr. Rochester`,
`Madame Beck`, `M. Paul` are unambiguous and a 19th-century novel is saturated
with them. Measured: 16 entities resolved, **zero wrong**, with every ambiguous
case landing on HELD rather than on a guess -- shared family surnames like
Helstone and Pelet, which genuinely belong to both a man and a woman, hold as
they should.

Held is cheap; wrong is poison. **A HELD entity is simply not renamed.** An
un-renamed name costs a little augmentation; a mis-gendered one scrambles pronoun
agreement through every copy and nothing downstream would catch it.

Pool is French + English (operator, 2026-09-10), weighted per work by setting:
the Brussels novels draw more French, the Yorkshire novels more English. Locales
are restricted to fr_FR/fr_BE/en_GB/en_IE -- en_US and en_AU carry modern
surnames that are wrong register for the 1840s before any diacritic question.
"""
from __future__ import annotations
import argparse, collections, json, random, re, sys, unicodedata
from pathlib import Path

TOKEN = re.compile(r"[A-Za-zÀ-ÿŒœÆæ][A-Za-zà-ÿœæ\-]*")
MALE_T = r"(?:Mr|Sir|Master|Monsieur|M|Lord|Captain|Colonel|Major|Doctor|Dr|Reverend|King|Prince|Duke|Squire)"
FEM_T = r"(?:Mrs|Miss|Madame|Mme|Mademoiselle|Mlle|Lady|Madam|Queen|Princess|Duchess)"

FRENCH_LOCALES = ["fr_FR", "fr_BE"]
ENGLISH_LOCALES = ["en_GB", "en_IE"]
#: Brussels novels lean French, Yorkshire novels lean English. Register, not
#: orthography -- a Yorkshire mill town full of Parisian surnames reads wrong.
FRENCH_SHARE = {"villette": 0.60, "the-professor": 0.60, "jane-eyre": 0.25, "shirley": 0.25}


def title_gender(text: str) -> dict[str, str]:
    mt = collections.Counter(m.group(1).lower() for m in
                             re.finditer(MALE_T + r"\.?\s+([A-ZÀ-Þ][a-zà-ÿœæ\-]+)", text))
    ft = collections.Counter(m.group(1).lower() for m in
                             re.finditer(FEM_T + r"\.?\s+([A-ZÀ-Þ][a-zà-ÿœæ\-]+)", text))
    out = {}
    for k in set(mt) | set(ft):
        M, F = mt[k], ft[k]
        if M >= 3 and M >= 3 * max(F, 1):
            out[k] = "m"
        elif F >= 3 and F >= 3 * max(M, 1):
            out[k] = "f"
    return out


def build_pool(dict_path: Path, alphabet: set[str]) -> dict:
    d = json.loads(dict_path.read_text())
    pool = {}
    for label, locales in (("fr", FRENCH_LOCALES), ("en", ENGLISH_LOCALES)):
        m, f, s = set(), set(), set()
        for loc in locales:
            v = d["by_locale"].get(loc, {})
            m |= set(v.get("male", []))
            f |= set(v.get("female", []))
            for k in ("surnames_neutral", "surnames_male", "surnames_female"):
                s |= set(v.get(k, []))
        # ⚠ F02's subset rule, applied with Bronte's OWN alphabet rather than a
        # global ASCII fold: French accents are IN because she writes French
        # constantly; Czech/Latvian/Slovak marks are OUT because they never appear.
        keep = lambda n: n and n[:1].isupper() and all((not c.isalpha()) or c in alphabet for c in n)
        pool[label] = {"male": sorted(filter(keep, m)),
                       "female": sorted(filter(keep, f)),
                       "surname": sorted(filter(keep, s))}
    return pool


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("corpus")
    ap.add_argument("--entities", required=True)
    ap.add_argument("--dictionary", required=True)
    ap.add_argument("--out", required=True)
    ap.add_argument("--copies", type=int, default=6)
    ap.add_argument("--seed", type=int, default=4919)
    ap.add_argument("--holdout-chapter", type=int, default=10)
    a = ap.parse_args()

    corpus = Path(a.corpus)
    man = json.loads((corpus / "manifest.json").read_text())
    alphabet = set(json.loads((corpus / "corpus_alphabet.json").read_text())["letters"])
    ents_all = json.loads(Path(a.entities).read_text())
    pool = build_pool(Path(a.dictionary), alphabet)
    # ⚠ Collision filter, against THIS corpus. F02 dropped 35 names for colliding
    # with the Yarros source so a rename could never map one of the author's
    # entities onto another; that filter is corpus-specific and does not carry.
    # Measured here before adding it: `Burns` and `Marie` were drawn as
    # replacements and are themselves Bronte entities, which reads as a leak in
    # the gate and is worse than it looks -- it silently merges two characters.
    source_names = {e["surface"] for w in ents_all.values() for e in w["entities"].values()}
    source_names |= {n.split()[0] for n in source_names if " " in n}
    dropped = 0
    for lang in pool:
        for bucket in pool[lang]:
            before = len(pool[lang][bucket])
            # ⚠ By COMPONENT, not by whole string. Measured: the pool drew the
            # compound `Pierre-Yves` while `Pierre` (Mademoiselle St. Pierre) is a
            # Villette character, so a whole-string comparison passed it and the
            # leak gate then matched the component. The original was correctly
            # renamed -- it is not a leak -- but a replacement sharing a component
            # with a source character invites exactly the conflation the rename
            # exists to prevent.
            pool[lang][bucket] = [
                n for n in pool[lang][bucket]
                if n not in source_names
                and not (set(re.split(r"[-\s’']", n)) & source_names)]
            dropped += before - len(pool[lang][bucket])
    print(f"  collision filter: dropped {dropped} pool names that are Bronte entities")
    print(f"  pool (alphabet-filtered): "
          f"fr {len(pool['fr']['male'])}m/{len(pool['fr']['female'])}f/{len(pool['fr']['surname'])}s  "
          f"en {len(pool['en']['male'])}m/{len(pool['en']['female'])}f/{len(pool['en']['surname'])}s")

    out = Path(a.out); (out / "copies").mkdir(parents=True, exist_ok=True)
    stats = {"copies": a.copies, "seed": a.seed, "works": {}, "renamed": 0, "held": 0}

    works = {}
    for w in man["works"]:
        rows = [json.loads(l) for l in (corpus / w["path"]).read_text(encoding="utf-8").splitlines()]
        works[w["slug"]] = rows

    # ---- D2 final: decide, per work, which entities are renameable ----------
    plans = {}
    for slug, rows in works.items():
        text = "\n\n".join(r["text"] for r in rows)
        tg = title_gender(text)
        ents = ents_all[slug]["entities"]
        titled = set(tg)
        renameable, held = {}, []
        for key, e in ents.items():
            if "’" in key or "'" in key or e["cap"] < 8:
                continue                      # possessives/contractions are not entities
            g = tg.get(key)
            if g:
                renameable[key] = {"surface": e["surface"], "kind": "given", "gender": g}
            else:
                # ⚠ Everything else is STILL renamed -- from the gender-NEUTRAL
                # surname/place pool. The operator's Yarros directive was "rename
                # all proper nouns", and holding a place leaks it: `Thornfield`
                # appears 100 times in Jane Eyre and is as author-specific as
                # `Riders Quadrant` was. Substituting a neutral token makes NO
                # gender claim, so no gender claim can be wrong -- the prose keeps
                # whatever pronoun it already had. Held-means-ungendered, not
                # held-means-unrenamed.
                renameable[key] = {"surface": e["surface"], "kind": "surname", "gender": None}
                held.append(key)
        plans[slug] = renameable
        stats["works"][slug] = {"renamed": len(renameable), "gendered": len(renameable)-len(held),
                                "neutral": len(held)}
        stats["renamed"] += len(renameable); stats["held"] += len(held)
        print(f"  {slug:<14} renamed {len(renameable):>3}  ({len(renameable)-len(held)} gendered, {len(held)} neutral)")

    # ---- D3: N seeded copies, one consistent map per copy -------------------
    emitted = 0
    for c in range(a.copies):
        rng = random.Random(a.seed + c * 1000)
        for slug, rows in works.items():
            fr_share = FRENCH_SHARE[slug]
            used = set()

            def draw(kind: str, gender: str | None) -> str:
                lang = "fr" if rng.random() < fr_share else "en"
                bucket = {"m": "male", "f": "female"}.get(gender or "", "surname")
                for _ in range(200):
                    n = rng.choice(pool[lang][bucket])
                    if n not in used:
                        used.add(n); return n
                return rng.choice(pool[lang][bucket])

            mapping = {k: draw(v["kind"], v["gender"]) for k, v in plans[slug].items()}
            pat = re.compile(r"\b(" + "|".join(sorted((re.escape(v["surface"]) for v in plans[slug].values()),
                                                      key=len, reverse=True)) + r")\b")
            surf2key = {v["surface"]: k for k, v in plans[slug].items()}
            path = out / "copies" / f"{slug}.copy{c}.jsonl"
            with path.open("w", encoding="utf-8") as fh:
                for r in rows:
                    txt = pat.sub(lambda m: mapping[surf2key[m.group(1)]], r["text"])
                    split = "val" if r["chapter"] == a.holdout_chapter else "train"
                    fh.write(json.dumps({"work": slug, "copy": c, "chapter": r["chapter"],
                                         "split": split, "text": txt}, ensure_ascii=False) + "\n")
                    emitted += 1
        print(f"  copy {c}: written")

    (out / "rename_stats.json").write_text(json.dumps(stats, ensure_ascii=False, indent=2))
    print(f"\n  {emitted:,} chapter-records across {a.copies} copies -> {out}")
    return 0


if __name__ == "__main__":
    sys.exit(main())