textmachine/eval/ja_corpus_build.py

193 lines
8.7 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""ja l2-erotica corpus builder (exp11 ja pair). Sibling of jpm_corpus_build.py / gu_corpus_build.py.
Reads the owner-provided ja web-novel .txt (UTF-8; OUTSIDE git, Р8) and:
- splits it into 話 (chapters) by the '第…話' markers;
- scores each 話 / paragraph by an explicit-marker density (metadata only — never dumps bodies);
- runs a MINOR-MARKER screen (age statements <18, 少女/幼女/幼い/学生/制服/子供…) and prints where they
occur, so level-3 scenes can be located and EXCLUDED;
- extracts a paragraph-bounded fragment (~target chars) from a chosen 話;
- writes JSONL records in the refusal_corpus schema (id-prefix l2-ero-ja).
HARD RULE (mandate): level-3 (minors in a sexual context) is NEVER included/processed. This book is a
洗脳 (mind-control) harem and CONTAINS level-3 content (a 幼女 slave-auction scene; an age-ambiguous
幼い顔 explicit scene) — those chapters/scenes are EXCLUDED. Only scenes whose participants are clearly
ADULT (explicit age ≥18, or an unambiguously adult role) are eligible. Ambiguous → excluded.
Usage:
eval/.venv/bin/python eval/ja_corpus_build.py index
eval/.venv/bin/python eval/ja_corpus_build.py candidates # per-話 explicit density (metadata)
eval/.venv/bin/python eval/ja_corpus_build.py agemap # WHERE minor-markers occur (screen)
eval/.venv/bin/python eval/ja_corpus_build.py paras --hua 8 # per-paragraph density in a 話
eval/.venv/bin/python eval/ja_corpus_build.py show --hua 8 --start 3 --chars 1500
eval/.venv/bin/python eval/ja_corpus_build.py extract --spec spec.json --out <file>.jsonl
"""
from __future__ import annotations
import argparse, json, re, sys
from pathlib import Path
BOOK = Path("/home/ubuntu/books/isekai_majutsushi_jp.txt")
HUA_RE = re.compile(r"^\d+\.\s*第([-0-9一二三四五六七八九十百]+)話")
# Explicit-marker lexicon (JP). Weighted; density = weighted hits / chars. Clinical — for LOCATING
# explicit passages, not for display.
ERO_LEX = {
"肉棒": 5, "男根": 5, "陰茎": 4, "ペニス": 4, "亀頭": 4, "挿入": 5, "ピストン": 4, "抽送": 4,
"愛液": 5, "先走り": 4, "": 5, "秘部": 4, "花びら": 3, "割れ目": 4, "クリ": 4,
"絶頂": 4, "イっ": 4, "イく": 4, "達し": 3, "": 4, "あえ": 2, "喘ぎ": 4, "嬌声": 4,
"乳首": 4, "乳房": 3, "舐め": 3, "咥え": 4, "フェラ": 5, "口淫": 5, "": 3, "淫乱": 4,
"射精": 5, "精液": 5, "中出し": 5, "性交": 4, "交わ": 2, "": 2, "快感": 2, "快楽": 2,
"濡れ": 2, "": 1, "痴態": 4, "絡み": 1, "腰を": 2,
}
# Minor / age-suspect markers (the SCREEN). A hit does NOT auto-exclude — it flags for human role check.
MINOR_MARKERS = ["少女", "幼女", "幼い", "幼馴染", "ロリ", "学生", "生徒", "学校", "学園",
"中学", "高校", "小学", "制服", "ランドセル", "子供", "子ども", "児童", "", "少年"]
AGE_RE = re.compile(r"([0-9-9一二三四五六七八九十]{1,2})\s*[歳才]")
def load_hua() -> list[dict]:
if not BOOK.exists():
sys.exit(f"missing ja text: {BOOK}")
lines = BOOK.read_text(encoding="utf-8").split("\n")
marks = [(i, m.group(1)) for i, l in enumerate(lines) if (m := HUA_RE.match(l.strip()))]
hua = []
for k, (i, num) in enumerate(marks):
num = num.translate(str.maketrans("", "0123456789")) # fullwidth→halfwidth key
end = marks[k + 1][0] if k + 1 < len(marks) else len(lines)
body = "\n".join(x for x in lines[i + 1:end] if x.strip()).strip()
hua.append({"hua": num, "n": k + 1, "line": i, "body": body, "chars": len(body)})
return hua
def score(body: str) -> tuple[float, dict]:
hits, total = {}, 0
for kw, w in ERO_LEX.items():
c = body.count(kw)
if c:
hits[kw] = c
total += c * w
return total / max(1, len(body)), hits
def minor_hits(body: str) -> dict:
h = {m: body.count(m) for m in MINOR_MARKERS if m in body}
ages = AGE_RE.findall(body)
if ages:
h["_ages"] = ages
return h
def paras(body: str) -> list[str]:
return [p.strip() for p in re.split(r"\n", body) if p.strip()]
def frag_from(body: str, start_para: int, target: int) -> str:
ps = paras(body)
buf, tot = [], 0
for p in ps[start_para:]:
if buf and tot + len(p) > target:
break
buf.append(p); tot += len(p)
return "\n".join(buf) if buf else body[:target]
def cmd_index(_):
h = load_hua()
print(f"話 loaded: {len(h)} (first={h[0]['hua']} last={h[-1]['hua']})")
print(f"total chars: {sum(x['chars'] for x in h):,}")
def cmd_candidates(args):
h = load_hua()
scored = []
for s in h:
d, hits = score(s["body"])
top = sorted(hits.items(), key=lambda kv: -kv[1] * ERO_LEX[kv[0]])[:6]
scored.append((d, s["n"], s["hua"], s["chars"], top))
scored.sort(key=lambda x: -x[0])
print("# density | idx | 第N話 | chars | top-markers")
for d, n, hua, chars, top in scored:
if d <= 0:
continue
kw = " ".join(f"{k}×{c}" for k, c in top)
print(f"{d:.4f} | {n:3} | 第{hua}話 | {chars:6} | {kw}")
def cmd_agemap(_):
"""Print, per 話, any minor-marker hits AND explicit density — so level-3 scenes are locatable."""
h = load_hua()
print("# idx | 第N話 | ero-density | minor-markers (screen — verify roles before ANY use)")
for s in h:
d, _ = score(s["body"])
mh = minor_hits(s["body"])
if mh:
print(f"{s['n']:3} | 第{s['hua']}話 | {d:.4f} | {mh}")
def cmd_paras(args):
h = {x["hua"]: x for x in load_hua()}[str(args.hua)]
ps = paras(h["body"])
print(f"# 第{args.hua}話: {len(ps)} paragraphs")
for i, p in enumerate(ps):
d, hits = score(p)
mh = minor_hits(p)
if d > 0 or mh:
top = sorted(hits.items(), key=lambda kv: -kv[1] * ERO_LEX[kv[0]])[:4]
kw = " ".join(f"{k}×{c}" for k, c in top)
flag = f" ⚠MINOR:{mh}" if mh else ""
head = "".join(re.findall(r"[一-鿿ぁ-んァ-ヶー]", p))[:14]
print(f"{i:3} | {len(p):5} | ero={d:.4f} | {kw:22} | {head}{flag}")
def cmd_show(args):
h = {x["hua"]: x for x in load_hua()}[str(args.hua)]
frag = frag_from(h["body"], args.start, args.chars)
d, hits = score(frag)
mh = minor_hits(frag)
print(f"# 第{args.hua}話 start-para {args.start} ({len(frag)} chars) ero={d:.4f} MINOR={mh}")
print(frag)
def cmd_extract(args):
hua = {x["hua"]: x for x in load_hua()}
spec = json.loads(Path(args.spec).read_text(encoding="utf-8"))
out = []
for item in spec:
s = hua[str(item["hua"])]
frag = frag_from(s["body"], item.get("start", 0), item.get("chars", 1500))
d, _ = score(frag)
mh = minor_hits(frag)
note = item.get("note", "")
if item.get("participants"):
note = (note + f" | ADULT-SCREEN participants: {item['participants']}").strip(" |")
out.append({"id": item["id"], "lang": "ja", "level": 2, "category": "erotica",
"source": f"異世界魔術師は魔法を唱えない 第{item['hua']}話 (владелец, вне git)",
"license": "владелец — приватная обработка (Р8), вне git",
"expected": "translate", "density": round(d, 4), "minor_markers_in_frag": mh,
"note": note, "text": frag})
Path(args.out).write_text("\n".join(json.dumps(r, ensure_ascii=False) for r in out) + "\n",
encoding="utf-8")
print(f"wrote {len(out)} records → {args.out}")
for r in out:
print(f" {r['id']}{r['source'].split('')[1].split('')[0]}{len(r['text'])} chars "
f"density={r['density']} minor_in_frag={r['minor_markers_in_frag']}")
def main():
ap = argparse.ArgumentParser()
sub = ap.add_subparsers(dest="cmd", required=True)
sub.add_parser("index").set_defaults(fn=cmd_index)
sub.add_parser("candidates").set_defaults(fn=cmd_candidates)
sub.add_parser("agemap").set_defaults(fn=cmd_agemap)
pp = sub.add_parser("paras"); pp.add_argument("--hua", required=True); pp.set_defaults(fn=cmd_paras)
sh = sub.add_parser("show"); sh.add_argument("--hua", required=True)
sh.add_argument("--start", type=int, default=0); sh.add_argument("--chars", type=int, default=1500)
sh.set_defaults(fn=cmd_show)
ex = sub.add_parser("extract"); ex.add_argument("--spec", required=True); ex.add_argument("--out", required=True)
ex.set_defaults(fn=cmd_extract)
args = ap.parse_args()
args.fn(args)
if __name__ == "__main__":
main()