esh-pfi-infrastructure/scripts/yarros-corpus/build_corpus_yarros.py

"""D1 for BabyYarros: build the corpus from the licensed Kvasir masters.

Deliberately emits the SAME record schema as the Brontë builder
({work, chapter, heading, words, text} per work file, plus corpus_alphabet.json and
manifest.json), so entities.py, rename.py and train_voice_lora.py all run unchanged.
Matching an existing schema beats teaching three downstream tools a new one.

Differences from the Brontë build, and each is a property of the source rather than
a preference:

  * NO Gutenberg boilerplate strip and NO download -- Kvasir already extracted and
    cleaned these, and the catalog records the cleaner and its version.
  * NO unwrap step. The Brontë corpus came hard-wrapped at ~70 characters and the
    adapter learned the line breaks; these masters are already flowing paragraphs
    (median non-blank line 102 chars), so the defect does not exist here.
  * Chapters are "Chapter One" style words, not roman numerals, and are followed by
    a POV name and often a location on their own lines -- first-person contemporary
    romance with rotating narrators. Those header lines are KEPT: they are part of
    the form the voice lives in, and dropping them would teach the model that
    chapters begin mid-scene.
  * ASCII alphabet, confirmed on the text rather than inherited from F02: 2
    non-ASCII letters across this corpus. Under the F02 rule (a rename pool's
    character inventory must be a SUBSET of the corpus's) that means an ASCII-only
    pool -- the opposite of Brontë, who needed French accents kept.
"""
from __future__ import annotations
import argparse, collections, json, os, re, sqlite3, sys
from pathlib import Path

CATALOG = "/home/lkraven/development/kvasir/data/library/catalog.sqlite"
KVASIR = "/home/lkraven/development/kvasir"

SLUGS = {
    "Fourth Wing (Exclusive Holiday Edition)": "fourth-wing",
    "Iron Flame": "iron-flame",
    "Wilder (The Renegades)": "wilder",
    "Nova (The Renegades #2)": "nova",
    "Rebel (The Renegades)": "rebel",
}
# "Chapter One" / "Chapter Twenty-Three" / "Chapter 12" / "Prologue" / "Epilogue".
CHAPTER = re.compile(
    r"^[ \t]*((?:Chapter|CHAPTER)[ \t]+(?:[A-Za-z-]+|\d+)|Prologue|PROLOGUE|Epilogue|EPILOGUE)"
    r"[ \t]*\.?[ \t]*$", re.M)


def masters():
    c = sqlite3.connect(CATALOG)
    rows = c.execute("select title, master_path, rights, normalized_text_sha256 "
                     "from masters where lower(author) like '%yarros%'").fetchall()
    out = []
    for title, path, rights, sha in rows:
        p = Path(path if os.path.isabs(path) else os.path.join(KVASIR, path))
        if not p.exists():
            print(f"  ⚠ MISSING master for {title}: {p}", file=sys.stderr)
            continue
        out.append({"title": title, "slug": SLUGS.get(title, re.sub(r"\W+", "-", title.lower()).strip("-")),
                    "path": p, "rights": rights, "sha256": sha})
    return sorted(out, key=lambda w: w["slug"])


def split_chapters(text: str):
    """Return [(heading, body)]. Everything before the first heading is front matter."""
    marks = [(m.start(), m.group(1).strip()) for m in CHAPTER.finditer(text)]
    if not marks:
        return [("(whole)", text.strip())]
    out = []
    for i, (pos, head) in enumerate(marks):
        end = marks[i + 1][0] if i + 1 < len(marks) else len(text)
        body = text[pos:end].strip()
        if len(body.split()) >= 150:      # skip a bare heading with no chapter behind it
            out.append((head, body))
    return out


ap = argparse.ArgumentParser()
ap.add_argument("--out", required=True)
ap.add_argument("--survey", action="store_true", help="report and write nothing")
a = ap.parse_args()

works = masters()
if not works:
    raise SystemExit("REFUSING: no Yarros masters resolved from the catalog")

out = Path(a.out)
alphabet = collections.Counter()
total_words = total_chaps = 0
manifest = {"corpus": "BabyYarros", "author": "Rebecca Yarros",
            "source": "kvasir data/library masters (licensed, rights=gated)",
            "built_at": __import__("datetime").date.today().isoformat(), "works": []}

for w in works:
    text = w["path"].read_text(encoding="utf-8", errors="replace")
    chaps = split_chapters(text)
    alphabet.update(ch for ch in text if ch.isalpha())
    words = sum(len(b.split()) for _, b in chaps)
    total_words += words; total_chaps += len(chaps)
    print(f"  {w['slug']:14} {len(chaps):>3} chapters  {words:>7,} words   rights={w['rights']}")
    manifest["works"].append({"slug": w["slug"], "title": w["title"], "rights": w["rights"],
                              "master_sha256": w["sha256"], "chapters": len(chaps), "words": words,
                              "path": f"works/{w['slug']}.jsonl"})
    if not a.survey:
        (out / "works").mkdir(parents=True, exist_ok=True)
        with (out / "works" / f"{w['slug']}.jsonl").open("w", encoding="utf-8") as fh:
            for i, (head, body) in enumerate(chaps, 1):
                fh.write(json.dumps({"work": w["slug"], "chapter": i, "heading": head,
                                     "words": len(body.split()), "text": body},
                                    ensure_ascii=False) + "\n")

non_ascii = {c: n for c, n in alphabet.items() if ord(c) > 127}
print(f"\n  TOTAL {total_chaps} chapters · {total_words:,} words · {len(alphabet)} distinct letters")
print(f"  non-ASCII letters: {sum(non_ascii.values())} across {len(non_ascii)} forms {non_ascii or ''}")
manifest["totals"] = {"chapters": total_chaps, "words": total_words,
                      "distinct_letters": len(alphabet), "non_ascii_letters": sum(non_ascii.values())}
manifest["total_words"] = total_words
manifest["total_chapters"] = total_chaps

if not a.survey:
    (out / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
    (out / "corpus_alphabet.json").write_text(json.dumps({
        "derived_from": "Rebecca Yarros, 5 novels, Kvasir licensed library",
        "derived_at": manifest["built_at"],
        "note": ("R49 F02 rule: a rename pool's character inventory must be a SUBSET of this. "
                 f"Measured on the built text: {sum(non_ascii.values())} non-ASCII letters, so the "
                 "pool is ASCII-only -- the opposite of the Brontë corpus, which needed French "
                 "accents kept."),
        "letters": sorted(alphabet), "non_ascii": {c: n for c, n in sorted(non_ascii.items())},
    }, indent=2, ensure_ascii=False), encoding="utf-8")
    print(f"  wrote {out}")