masi
398 lines
19 KiB
Python
398 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""One-off story-knowledge cleanup:
|
|
./maintain-job.sh <novel> (runs as a Job; deploys do not interrupt it)
|
|
[--dry-run] [--duplicates-only | --prune-only | --looks-only | --terms-only | --aliases-only]
|
|
|
|
1. Fold one person recorded under several spellings ("Choi Jong Bom" / "Jong-bum Choi") into a
|
|
single character, so transliteration variants do not fill the cast with the same face.
|
|
2. Prune entries that are not named individuals (groups, "bald tattooed cultivator",
|
|
translators). Deterministic rules decide the obvious cases; one batched MiniMax M2.7 call judges
|
|
the rest.
|
|
3. Fold inflated look lists (power levels, injuries, disguises recorded as looks) into a
|
|
few real looks: one MiniMax M2.7 call per character with more than MAX_LOOKS looks.
|
|
4. Drop glossary "terms" that are ordinary words ("Ring", "Return"): a script test on the
|
|
chapter texts, no model call. A proper noun keeps its capitals; a common word shows up
|
|
lowercase mid-sentence. Mentions are then recounted from the text.
|
|
5. Drop aliases that link mentions to the wrong person: shared titles and kinship words,
|
|
pasted lists, and aliases two characters both have (store.alias_problem).
|
|
6. Point the timeline back at the looks that still exist, so a chapter that names a look whose
|
|
record was lost is illustrated as its owner again (store.repair_looks)."""
|
|
import json
|
|
import difflib
|
|
import re
|
|
import sys
|
|
|
|
import kavita
|
|
import llm
|
|
import store
|
|
|
|
MAX_LOOKS = 3
|
|
KEEP_CHAPTERS = 5 # an entry in this many chapters, or ever drawn, is a character
|
|
BATCH = 150
|
|
# Words that mark a group rather than one individual.
|
|
GROUP = re.compile(r"\b(cultivators|elders|disciples|members|guards|soldiers|people|trio|duo|group|crowd|"
|
|
r"shadows|villagers|sorcerers|warriors|men|women|children|monks|beasts|demons)\b", re.I)
|
|
def translators(novel):
|
|
"""Names from credit lines ("Translator: X", "Editors: A, B") are never characters."""
|
|
names = set()
|
|
with store.connect() as conn:
|
|
for c in store.known_characters(conn, novel):
|
|
if re.search(r"\b(translat|editor|proofread)", c["role"] or "", re.I):
|
|
names.add(c["name"])
|
|
return names
|
|
|
|
|
|
def prune(novel, dry_run):
|
|
"""Groups and translators always go. Anything else MiniMax judges "not one individual"
|
|
goes only if it is rare and was never drawn: recurring beings known by an epithet
|
|
("the fox", "[She]") are real characters in many novels."""
|
|
with store.connect() as conn:
|
|
known = store.known_characters(conn, novel)
|
|
counts = store.mention_counts(conn, novel)
|
|
drawn = {r["name"] for r in conn.execute(
|
|
"SELECT DISTINCT name FROM looks WHERE novel = ? AND portrait != ''", (novel,))}
|
|
rule = {c["name"] for c in known if GROUP.search(c["name"])} | translators(novel)
|
|
# "Geomchi16" and "Geomchi-3" are members of a group named Geomchi, not people with names:
|
|
# thirty of them were recorded, each with the same dojo-disciple description. A member that
|
|
# was already drawn in a scene stays, so the picture still matches its own cast.
|
|
rule |= {c["name"] for c in known if c["name"] not in drawn and store.variant_of(known, c["name"])}
|
|
keep = {c["name"] for c in known if counts.get(("c", c["name"]), 0) >= KEEP_CHAPTERS or c["name"] in drawn}
|
|
rest = [c for c in known if c["name"] not in rule and c["name"] not in keep]
|
|
judged = set()
|
|
for i in range(0, len(rest), BATCH):
|
|
batch = rest[i:i + BATCH]
|
|
listing = "\n".join(f"- {c['name']}: {c['role'] or c['description'][:80]}" for c in batch)
|
|
wanted = {c["name"] for c in batch}
|
|
judged |= {n for n in llm.chat(llm.PRUNE_SYSTEM, listing).get("remove") or [] if n in wanted}
|
|
print(f"prune: {len(rule)} by rule, {len(judged)} by MiniMax, of {len(known)} "
|
|
f"({len(keep)} recurring or drawn characters were not questioned)")
|
|
for name in sorted(rule | judged):
|
|
print(f" - {name}")
|
|
if not dry_run:
|
|
store.remove_characters(novel, rule | judged)
|
|
|
|
|
|
DUP_BATCH = 50 # a longer listing times out on the fallback provider
|
|
|
|
|
|
def _same_person_groups(entries, size=DUP_BATCH):
|
|
"""The groups of entries the model says are one person. A batch that times out is asked
|
|
again in two halves, so one oversized listing cannot lose the whole run."""
|
|
out = []
|
|
for i in range(0, len(entries), size):
|
|
batch = entries[i:i + size]
|
|
listing = "\n".join(f"- {name} ({n} chapters, first in chapter {first + 1}): {text}"
|
|
for name, first, n, text in batch)
|
|
try:
|
|
answer = llm.chat(llm.DUPLICATE_SYSTEM, listing)
|
|
except Exception as e:
|
|
if len(batch) > 10:
|
|
print(f"duplicates: {len(batch)}-name batch at {i} halved ({e})", file=sys.stderr)
|
|
out += _same_person_groups(batch, len(batch) // 2)
|
|
else:
|
|
print(f"duplicates: batch at {i} skipped ({e})", file=sys.stderr)
|
|
continue
|
|
for group in answer.get("groups") or []:
|
|
if not isinstance(group, list):
|
|
continue
|
|
members, seen = [], set()
|
|
for name in group:
|
|
# A group or a title is not a person to merge into anybody.
|
|
if not isinstance(name, str) or name in seen or GROUP.search(name) or name not in _names(batch):
|
|
continue
|
|
seen.add(name)
|
|
members.append(name)
|
|
if len(members) > 1:
|
|
out += spelling_variants(members)
|
|
return out
|
|
|
|
|
|
def _names(batch):
|
|
return {entry[0] for entry in batch}
|
|
|
|
|
|
NAME_SIMILAR = 0.78 # name similarity at which two spellings are the same person's
|
|
|
|
|
|
def _plain(name):
|
|
return re.sub(r"[^a-z]", "", name.casefold())
|
|
|
|
|
|
def _close(a, b):
|
|
"""How alike two spellings are, 0 to 1, by the letters they share in order."""
|
|
return difflib.SequenceMatcher(None, _plain(a), _plain(b)).ratio()
|
|
|
|
|
|
def spelling_variants(members):
|
|
"""The sub-groups of a proposed group that are really one name written two ways. Asked
|
|
which cast members are the same person, a model handed fifty names will happily report a
|
|
chain — Gross, Heiller, Pandal, Porom, Vindel "the same individual" — and merging that
|
|
chain folds five strangers into one face. Transliteration variants ("Jang Yoon-soo" /
|
|
"Jang Yoon Soo", "Kalis" / "Kallis") look alike on the page, so that is the test; so does
|
|
a name whose words are the same words in another order ("Choi Jong Bom" / "Jong-bum Choi")."""
|
|
groups, rest = [], list(dict.fromkeys(members))
|
|
while rest:
|
|
cluster = [rest.pop(0)]
|
|
grew = True
|
|
while grew:
|
|
grew = False
|
|
for name in list(rest):
|
|
if any(_close(name, other) >= NAME_SIMILAR or shares_a_name(name, other)
|
|
for other in cluster):
|
|
cluster.append(name)
|
|
rest.remove(name)
|
|
grew = True
|
|
if len(cluster) > 1:
|
|
groups.append(cluster)
|
|
return groups
|
|
|
|
|
|
SAME_PERSON = 0.8 # description overlap at which two records are one person
|
|
NO_DETAIL = re.compile(r"\bnot (?:further |physically |otherwise )?described\b"
|
|
r"|\bno (?:physical |other |further )?details?\b", re.I)
|
|
|
|
|
|
def _name_tokens(name):
|
|
return {t for t in re.split(r"[^A-Za-z]+", name) if len(t) >= 3}
|
|
|
|
|
|
def shares_a_name(one, other):
|
|
"""Whether two names are largely the same words: "Choi Jong Bom" and "Jong-bum Choi" are
|
|
the same name backwards, which is how a transliteration ends up recorded twice. One word
|
|
in common is not enough — Abalone Porridge and Cockles Porridge are two guilds, and they
|
|
would pass on "porridge" alone."""
|
|
first, second = _name_tokens(one), _name_tokens(other)
|
|
return bool(first & second) and len(first & second) / len(first | second) >= 0.5
|
|
|
|
|
|
def described_twice(known):
|
|
"""The pairs of characters whose descriptions are the same person's, by content words.
|
|
"Choi Jong Bom" and "Jong Bom" were both written from the same paragraph, and a model
|
|
asked to spot spelling variants will happily agree they are two people.
|
|
|
|
Two things keep this from folding strangers together: a description that says the text
|
|
gave no detail is a placeholder (the three bullies who corner Kim Yo in one corridor are
|
|
all written "Teenage bully; not further described."), and the two names have to share a
|
|
word, because the cast is full of people written from one template — every one of the
|
|
Instructor's disciples reads "Adult male disciple of the dojang: hard-trained martial
|
|
artist's build, short-cropped black hair", and they are thirty different men."""
|
|
words = {c["name"]: set() if NO_DETAIL.search(c["description"] or "")
|
|
else store.content(c["description"] or "") for c in known}
|
|
pairs = []
|
|
for i, a in enumerate(known):
|
|
for b in known[i + 1:]:
|
|
first, second = words[a["name"]], words[b["name"]]
|
|
if len(first) < 5 or len(second) < 5 or not shares_a_name(a["name"], b["name"]):
|
|
continue
|
|
shared = len(first & second) / min(len(first), len(second))
|
|
if shared > SAME_PERSON:
|
|
pairs.append((a["name"], b["name"]))
|
|
return pairs
|
|
|
|
|
|
def duplicates(novel, dry_run):
|
|
"""Fold one person recorded under several spellings into a single character, so the cast
|
|
does not fill up with transliteration variants of the same face. MiniMax groups the cast
|
|
in batches ordered by the chapter each name first appears in, so the variants of one
|
|
person are compared together; the best-attested name survives and the rest become its
|
|
aliases. Records with the same description are folded first, without asking anyone."""
|
|
with store.connect() as conn:
|
|
known = store.known_characters(conn, novel)
|
|
counts = store.mention_counts(conn, novel)
|
|
firsts = {r["name"]: r["p"] for r in conn.execute(
|
|
"SELECT name, MIN(position) AS p FROM mentions WHERE novel = ? AND kind = 'c' GROUP BY name", (novel,))}
|
|
drawn = {r["name"] for r in conn.execute(
|
|
"SELECT DISTINCT name FROM looks WHERE novel = ? AND portrait != ''", (novel,))}
|
|
groups = [[a, b] for a, b in described_twice(known)]
|
|
# Ordered by where the name first turns up: two spellings of one person were written
|
|
# down chapters apart at most, so they land in the same batch.
|
|
entries = sorted((c["name"], firsts.get(c["name"], 0), counts.get(("c", c["name"]), 0),
|
|
" ".join((c["description"] or c["role"] or "").split()[:8])) for c in known)
|
|
groups += [g for g in _same_person_groups(entries) if not set(g) & {n for pair in groups for n in pair}]
|
|
merged = 0
|
|
for members in groups:
|
|
# The best-attested name keeps the record; a drawn one keeps it over a busier rival.
|
|
keep = max(members, key=lambda n: (counts.get(("c", n), 0), n in drawn, n))
|
|
for name in members:
|
|
if name == keep:
|
|
continue
|
|
print(f" {name} -> {keep}")
|
|
if not dry_run:
|
|
store.merge_characters(novel, keep, name)
|
|
merged += 1
|
|
print(f"duplicates: {merged} duplicate names of {len(groups)} groups, of {len(known)} characters")
|
|
|
|
|
|
def yield_to_reader():
|
|
"""MiniMax calls share one Token Plan; wait while the reader's illustrations are queued
|
|
(a busy cleanup made every scene call take minutes)."""
|
|
import time
|
|
while any(r["kind"] == "illustration" for r in store.queue_pending()):
|
|
time.sleep(30)
|
|
|
|
|
|
def consolidate(novel, dry_run):
|
|
with store.connect() as conn:
|
|
crowded = [(c["name"], store.looks_of(conn, novel, c["name"])) for c in store.known_characters(conn, novel)]
|
|
crowded = [(name, looks) for name, looks in crowded if len(looks) > MAX_LOOKS]
|
|
firsts = {name: {l["label"]: store.first_seen(conn, novel, name, l["label"]) for l in looks}
|
|
for name, looks in crowded}
|
|
failed = []
|
|
for name, looks in crowded:
|
|
# Short descriptions keep the prompt small (a 100-look list timed out otherwise).
|
|
listing = "\n".join(f"- \"{l['label']}\" (from chapter {(firsts[name][l['label']] or 0) + 1}): "
|
|
f"{' '.join(l['description'].split()[:25])}" for l in looks)
|
|
yield_to_reader()
|
|
try:
|
|
groups = llm.chat(llm.CONSOLIDATE_SYSTEM, f"Character: {name}\nLooks:\n{listing}")["looks"]
|
|
except Exception as e: # one slow or odd answer must not stop the rest
|
|
failed.append(name)
|
|
print(f"{name}: skipped ({e})")
|
|
continue
|
|
# Anything MiniMax forgot to place keeps its own group rather than being lost.
|
|
placed = {m for g in groups for m in g.get("members", [])}
|
|
groups = [g for g in groups if g.get("members")]
|
|
groups += [{"label": l["label"], "description": l["description"], "members": [l["label"]]}
|
|
for l in looks if l["label"] not in placed]
|
|
seen = set()
|
|
for g in groups:
|
|
g["members"] = [m for m in g["members"] if m not in seen and m in firsts[name] and not seen.add(m)]
|
|
groups = [g for g in groups if g["members"]]
|
|
labels = set()
|
|
for g in groups: # distinct labels, as consolidate_looks requires
|
|
base, n = (g.get("label") or g["members"][0]).strip()[:55], 2
|
|
g["label"] = base
|
|
while g["label"] in labels:
|
|
g["label"], n = f"{base} {n}", n + 1
|
|
labels.add(g["label"])
|
|
print(f"{name}: {len(looks)} looks -> {len(groups)}: " + "; ".join(g["label"] for g in groups))
|
|
if not dry_run:
|
|
store.consolidate_looks(novel, name, groups)
|
|
if failed:
|
|
print(f"looks: {len(failed)} characters skipped, rerun to retry: {', '.join(failed)}")
|
|
|
|
|
|
TERM_SAMPLE = 25 # chapters sampled per term
|
|
|
|
|
|
def timeline(novel, dry_run):
|
|
"""Point the chapters back at the looks that still exist.
|
|
|
|
Folding, renaming and merging looks all move the timeline with them, but a look record
|
|
can go missing on its own — restoring `characters` and `looks` from a snapshot leaves a
|
|
newer timeline naming labels the snapshot never had — and every chapter from then on is
|
|
illustrated from the "initial" look instead. On the Moonlight Sculptor that is how Weed,
|
|
a lean twenty-year-old labourer, ended up drawn as a two-metre Chaos Warrior through the
|
|
first eighty chapters of his own story.
|
|
"""
|
|
with store.connect() as conn:
|
|
stale = store.stale_looks(conn, novel)
|
|
if not stale:
|
|
print("timeline: every look the chapters name is on file")
|
|
return
|
|
|
|
def choose(label, listing):
|
|
try:
|
|
return llm.chat(llm.LOOK_SYSTEM,
|
|
f"Look the chapters name but no record survives: {label}\nLooks:\n{listing}")["look"]
|
|
except Exception as e: # one slow or odd answer must not stop the rest
|
|
print(f" {label}: skipped ({e})", file=sys.stderr)
|
|
return None
|
|
|
|
moved = store.repair_looks(novel, choose, dry_run)
|
|
print(f"timeline: {len(moved)} of {len(stale)} lost looks found a record "
|
|
f"({sum(n for _, _, _, n in moved)} chapters)")
|
|
for name, label, target, n in moved:
|
|
print(f" {name}: {label} ({n} chapters) -> {target}")
|
|
if not dry_run and moved:
|
|
print(" the chapters that were illustrated as somebody else keep their pictures; "
|
|
"say so and they are repainted")
|
|
|
|
|
|
|
|
|
|
def clean_aliases(novel, dry_run):
|
|
"""Drop aliases that link mentions to the wrong person (store.alias_problem): descriptive
|
|
ones, shared titles and kinship words, pasted lists, and any alias another character also
|
|
has, which is dropped from all of them. Mentions are then recounted from the text."""
|
|
dropped = {}
|
|
with store.connect() as conn:
|
|
known = store.known_characters(conn, novel)
|
|
for c in known:
|
|
bad = {a: store.alias_problem(a, c["name"], known) for a in c["aliases"]}
|
|
bad.update({a: "own name" for a in c["aliases"] if a.casefold() == c["name"].casefold()})
|
|
bad = {a: why for a, why in bad.items() if why}
|
|
if bad:
|
|
dropped[c["name"]] = bad
|
|
if not dry_run:
|
|
for c in known:
|
|
if c["name"] in dropped:
|
|
conn.execute("UPDATE characters SET aliases = ? WHERE novel = ? AND name = ?",
|
|
(json.dumps([a for a in c["aliases"] if a not in dropped[c["name"]]]),
|
|
novel, c["name"]))
|
|
print(f"aliases: dropped from {len(dropped)} characters ({sum(map(len, dropped.values()))} aliases)")
|
|
for name, bad in dropped.items():
|
|
print(f" {name}: " + "; ".join(f"{a} ({why})" for a, why in bad.items()))
|
|
if dropped and not dry_run:
|
|
store.reset_mentions(novel)
|
|
|
|
|
|
def generic_terms(novel, dry_run):
|
|
state = store.index_status(novel)
|
|
with store.connect() as conn:
|
|
book = conn.execute("SELECT book FROM index_state WHERE novel = ?", (novel,)).fetchone()
|
|
terms = [r["term"] for r in conn.execute("SELECT term FROM terms WHERE novel = ?", (novel,))]
|
|
where = {}
|
|
for r in conn.execute("SELECT name, position FROM mentions WHERE novel = ? AND kind = 't'", (novel,)):
|
|
where.setdefault(r["name"], []).append(r["position"])
|
|
if not state or not book:
|
|
print("terms: not pre-read from Kavita; skipped")
|
|
return
|
|
texts = {}
|
|
|
|
def text(position):
|
|
if position not in texts:
|
|
got = kavita.chapter(book["book"], position)
|
|
texts[position] = got[1] if got else ""
|
|
return texts[position]
|
|
|
|
generic = []
|
|
for term in terms:
|
|
proper = common = 0
|
|
lower = term.lower()
|
|
for position in sorted(where.get(term, []))[:TERM_SAMPLE]:
|
|
body = text(position)
|
|
# Capitalised uses that are not just sentence starts, vs lowercase uses.
|
|
proper += len(re.findall(rf"(?<![.!?\"'“‘]\s)(?<![\w-]){re.escape(term)}(?![\w-])", body))
|
|
common += len(re.findall(rf"(?<![\w-]){re.escape(lower)}(?![\w-])", body)) if lower != term else 0
|
|
if common > proper:
|
|
generic.append(term)
|
|
print(f"terms: {len(generic)} generic of {len(terms)}: " + ", ".join(sorted(generic)[:60]))
|
|
if not dry_run:
|
|
store.remove_terms(novel, generic)
|
|
store.reset_mentions(novel)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
sys.exit(__doc__)
|
|
novel, dry_run = sys.argv[1], "--dry-run" in sys.argv
|
|
store.migrate()
|
|
only = {f for f in ("--prune-only", "--looks-only", "--terms-only", "--aliases-only",
|
|
"--duplicates-only", "--timeline-only") if f in sys.argv}
|
|
# duplicates first: a variant name judged rare on its own is pruned before it can be folded in
|
|
if not only or "--duplicates-only" in only:
|
|
duplicates(novel, dry_run)
|
|
if not only or "--prune-only" in only:
|
|
prune(novel, dry_run)
|
|
if not only or "--looks-only" in only:
|
|
consolidate(novel, dry_run)
|
|
# last of the look work: the steps above are what leave a chapter pointing at a look that
|
|
# is no longer there, and the answer to that is whatever record survived them.
|
|
if not only or "--looks-only" in only or "--timeline-only" in only:
|
|
timeline(novel, dry_run)
|
|
if not only or only & {"--terms-only", "--aliases-only"}:
|
|
clean_aliases(novel, dry_run)
|
|
if not only or "--terms-only" in only:
|
|
generic_terms(novel, dry_run)
|