DazedTL/util/wolfdawn/db_classify.py
2026-07-08 16:17:34 -05:00

489 lines
16 KiB
Python

"""WolfDawn database sheet classification for the WOLF translation workflow.
Database JSON (``kind: "db"``) varies widely across games: standard RPG sheets
(items, skills, states) vs custom narrative sheets (event dialogue, profiles).
This module classifies each ``groups[]`` sheet into tiers so translators can
translate foundational content before narrative-heavy custom tables.
Tiers:
foundation - standard RPG sheets (items, skills, weapons, enemies, etc.)
system - UI / system messages in database fields
narrative - custom dialogue / profile sheets (defer until foundation is done)
unknown - unclassified sheets (included in foundation runs by default)
"""
from __future__ import annotations
import json
import re
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from util.wolfdawn.names import _LABEL_SEP
TIER_FOUNDATION = "foundation"
TIER_SYSTEM = "system"
TIER_NARRATIVE = "narrative"
TIER_UNKNOWN = "unknown"
FOUNDATION_TIERS = frozenset({TIER_FOUNDATION, TIER_SYSTEM, TIER_UNKNOWN})
NARRATIVE_TIERS = frozenset({TIER_NARRATIVE})
DB_PROFILE_NAME = "db_profile.json"
# Japanese halves of standard WolfDawn bilingual typeName labels.
STANDARD_SHEETS_JP = frozenset({
"武器",
"防具",
"技能",
"アイテム",
"状態設定",
"用語設定",
"戦闘コマンド",
"システム設定",
"属性名の設定",
"主人公ステータス",
"敵グループ",
"敵キャラ個体データ",
})
STANDARD_FIELDS_RE = re.compile(
r"説明|Description|商品説明|セーブ|ロード|使用時文章|発生時の文章|回復時の文章",
re.IGNORECASE,
)
SYSTEM_FIELDS_RE = re.compile(
r"メッセージ|Message|タイトル|用語|Terms?",
re.IGNORECASE,
)
CUSTOM_SHEET_RE = re.compile(
r"^[■├■]|イベント|セリフ|プロフィール|MOB|感想|所持アイテム|CG",
)
DIALOGUE_FIELDS_RE = re.compile(
r"セリフ|コメント|相手|行為|台詞|会話|メッセージ_",
)
ARCHETYPE_CLASSIC = "classic_rpg"
ARCHETYPE_DB_HEAVY = "simulation_db_heavy"
ARCHETYPE_HYBRID = "hybrid"
ARCHETYPE_UNKNOWN = "unknown"
@dataclass
class DbGroupInfo:
"""One classified database sheet (a ``groups[]`` entry)."""
json_file: str
type_name: str
tier: str
line_count: int
sample_fields: list[str] = field(default_factory=list)
default_checked: bool = True
@property
def key(self) -> str:
return group_key(self.json_file, self.type_name)
def group_key(json_file: str, type_name: str) -> str:
"""Stable id for a sheet: ``DataBase.project.json|Skill · 技能``."""
return f"{json_file}|{type_name}"
def _japanese_sheet_name(type_name: str) -> str:
if _LABEL_SEP in type_name:
_, _, jp = type_name.partition(_LABEL_SEP)
return jp.strip()
return type_name.strip()
def json_file_from_doc(doc: dict[str, Any], path: Path | None = None) -> str:
if path is not None:
return path.name
file_field = str(doc.get("file") or "")
if file_field:
return f"{file_field}.json" if not file_field.endswith(".json") else file_field
return "unknown.project.json"
def classify_group_tier(type_name: str, lines: list[dict[str, Any]]) -> str:
"""Return the tier for one database sheet."""
jp = _japanese_sheet_name(type_name)
if CUSTOM_SHEET_RE.search(type_name) or CUSTOM_SHEET_RE.search(jp):
return TIER_NARRATIVE
if jp in STANDARD_SHEETS_JP:
return TIER_FOUNDATION
field_names = [str(line.get("fieldName") or "") for line in lines]
dialogue_hits = sum(1 for fn in field_names if DIALOGUE_FIELDS_RE.search(fn))
standard_hits = sum(1 for fn in field_names if STANDARD_FIELDS_RE.search(fn))
system_hits = sum(1 for fn in field_names if SYSTEM_FIELDS_RE.search(fn))
if lines and dialogue_hits > max(standard_hits, system_hits):
return TIER_NARRATIVE
if standard_hits > 0:
return TIER_FOUNDATION
if system_hits > 0:
return TIER_SYSTEM
if not lines:
return TIER_UNKNOWN
return TIER_UNKNOWN
def classify_db_document(
doc: dict[str, Any],
*,
json_file: str | None = None,
path: Path | None = None,
) -> list[DbGroupInfo]:
"""Classify every group in one ``kind: db`` document."""
if doc.get("kind") != "db":
return []
jf = json_file or json_file_from_doc(doc, path)
groups: list[DbGroupInfo] = []
for group in doc.get("groups") or []:
type_name = str(group.get("typeName") or "")
lines = group.get("lines") or []
tier = classify_group_tier(type_name, lines)
fields: Counter[str] = Counter()
for line in lines:
fn = str(line.get("fieldName") or "").strip()
if fn:
fields[fn] += 1
sample = [fn for fn, _ in fields.most_common(3)]
default_checked = tier in FOUNDATION_TIERS
groups.append(
DbGroupInfo(
json_file=jf,
type_name=type_name,
tier=tier,
line_count=len(lines),
sample_fields=sample,
default_checked=default_checked,
)
)
return groups
def _count_json_lines(doc: dict[str, Any]) -> int:
kind = doc.get("kind")
if kind in ("map", "common"):
return sum(len(scene.get("lines") or []) for scene in doc.get("scenes") or [])
if kind == "gamedat":
return len(doc.get("lines") or [])
if kind == "txt-dir":
return sum(
len(f.get("lines") or [])
for f in doc.get("files") or []
)
if kind == "txt":
return len(doc.get("lines") or [])
return 0
def _load_json(path: Path) -> dict[str, Any] | None:
try:
return json.loads(path.read_text(encoding="utf-8-sig"))
except Exception:
return None
@dataclass
class ContentDistribution:
"""Text distribution across WolfDawn extraction kinds."""
db_lines: int = 0
db_foundation_lines: int = 0
db_narrative_lines: int = 0
db_system_lines: int = 0
db_unknown_lines: int = 0
map_lines: int = 0
map_files: int = 0
common_lines: int = 0
gamedat_lines: int = 0
evtext_lines: int = 0
groups: list[DbGroupInfo] = field(default_factory=list)
archetype: str = ARCHETYPE_UNKNOWN
@property
def total_event_lines(self) -> int:
return self.map_lines + self.common_lines + self.gamedat_lines + self.evtext_lines
@property
def db_narrative_pct(self) -> float:
if self.db_lines <= 0:
return 0.0
return 100.0 * self.db_narrative_lines / self.db_lines
def analyze_content_distribution(files_dir: str | Path) -> ContentDistribution:
"""Scan staged ``files/`` JSON and classify database sheets."""
base = Path(files_dir)
dist = ContentDistribution()
if not base.is_dir():
return dist
for path in sorted(base.glob("*.json")):
doc = _load_json(path)
if not doc:
continue
kind = doc.get("kind")
if kind == "db":
groups = classify_db_document(doc, path=path)
dist.groups.extend(groups)
for g in groups:
dist.db_lines += g.line_count
if g.tier == TIER_FOUNDATION:
dist.db_foundation_lines += g.line_count
elif g.tier == TIER_NARRATIVE:
dist.db_narrative_lines += g.line_count
elif g.tier == TIER_SYSTEM:
dist.db_system_lines += g.line_count
else:
dist.db_unknown_lines += g.line_count
elif kind == "map":
dist.map_files += 1
dist.map_lines += _count_json_lines(doc)
elif kind == "common":
dist.common_lines += _count_json_lines(doc)
elif kind == "gamedat":
dist.gamedat_lines += _count_json_lines(doc)
elif kind == "txt-dir":
dist.evtext_lines += _count_json_lines(doc)
dist.archetype = _infer_archetype(dist)
return dist
def _infer_archetype(dist: ContentDistribution) -> str:
total = dist.db_lines + dist.total_event_lines
if total <= 0:
return ARCHETYPE_UNKNOWN
db_share = dist.db_lines / total
narrative_share = dist.db_narrative_lines / max(dist.db_lines, 1)
if db_share > 0.5 and narrative_share > 0.4:
return ARCHETYPE_DB_HEAVY
if dist.total_event_lines > dist.db_lines * 2 and dist.db_narrative_lines < dist.db_lines * 0.2:
return ARCHETYPE_CLASSIC
return ARCHETYPE_HYBRID
def format_discovery_summary(dist: ContentDistribution) -> str:
"""Multi-line discovery report for the Database workflow step."""
lines = ["This game's text distribution:"]
if dist.db_lines:
pct_narr = f"{dist.db_narrative_pct:.0f}% narrative sheets"
lines.append(
f" Database: {dist.db_lines:,} lines "
f"({dist.db_foundation_lines + dist.db_system_lines:,} foundation/system, "
f"{dist.db_narrative_lines:,} narrative, {pct_narr})"
)
else:
lines.append(" Database: (no DB JSON in files/)")
lines.append(f" Maps: {dist.map_files} file(s), {dist.map_lines:,} dialogue lines")
lines.append(f" CommonEvent: {dist.common_lines:,} lines")
if dist.gamedat_lines:
lines.append(f" Game.dat: {dist.gamedat_lines:,} lines")
if dist.evtext_lines:
lines.append(f" Evtext: {dist.evtext_lines:,} lines")
lines.append("")
lines.append(archetype_guidance(dist))
return "\n".join(lines)
def archetype_guidance(dist: ContentDistribution) -> str:
foundation = dist.db_foundation_lines + dist.db_system_lines + dist.db_unknown_lines
if dist.archetype == ARCHETYPE_DB_HEAVY:
return (
"Recommended order: Names → DB foundation "
f"({foundation:,} lines) → DB narrative "
f"({dist.db_narrative_lines:,} lines) → Maps/events.\n"
"This game stores most story dialogue in custom database sheets, not maps."
)
if dist.archetype == ARCHETYPE_CLASSIC:
return (
"Recommended order: Names → DB foundation "
f"({foundation:,} lines) → Maps/events "
f"({dist.total_event_lines:,} lines).\n"
"Classic layout: most story dialogue is in maps and common events. "
"You can skip DB narrative if there are no custom sheets."
)
if dist.db_narrative_lines == 0:
return (
"Recommended order: Names → DB foundation "
f"({foundation:,} lines) → Maps/events.\n"
"No custom narrative DB sheets detected — skip narrative DB translation."
)
return (
"Recommended order: Names → DB foundation "
f"({foundation:,} lines) → DB narrative "
f"({dist.db_narrative_lines:,} lines) → Maps/events.\n"
"Hybrid layout: significant text in both database sheets and maps."
)
def build_ai_audit_prompt(dist: ContentDistribution) -> str:
"""Compact prompt for AI-assisted database structure analysis."""
group_lines = []
for g in sorted(dist.groups, key=lambda x: -x.line_count):
fields = ", ".join(g.sample_fields[:2]) if g.sample_fields else "(none)"
group_lines.append(
f" - {g.json_file} | {g.type_name}: {g.line_count} lines, "
f"tier={g.tier}, fields: {fields}"
)
groups_text = "\n".join(group_lines) if group_lines else " (no database groups found)"
return (
"You are an expert Japanese WOLF RPG Editor game analyst.\n"
"\n"
"<task>\n"
"Classify this game's database sheets for translation workflow. "
"Return ONLY valid JSON (no markdown fences) matching this schema:\n"
"{\n"
' "archetype": "classic_rpg" | "simulation_db_heavy" | "hybrid",\n'
' "foundation_groups": ["DataBase.project.json|Skill · 技能", ...],\n'
' "narrative_groups": ["DataBase.project.json|■イベント(名前)", ...],\n'
' "defer_groups": [],\n'
' "notes": "one sentence about where this game stores dialogue"\n'
"}\n"
"\n"
"foundation_groups = items, skills, descriptions, system UI (translate first).\n"
"narrative_groups = custom dialogue/profile sheets (translate after foundation).\n"
"defer_groups = sheets to skip entirely (rare).\n"
"Use the exact group keys shown below (json_file|typeName).\n"
"</task>\n"
"\n"
"<auto_summary>\n"
f"{format_discovery_summary(dist)}\n"
"</auto_summary>\n"
"\n"
"<database_groups>\n"
f"{groups_text}\n"
"</database_groups>\n"
)
def db_profile_path(work_dir: str | Path) -> Path:
return Path(work_dir) / DB_PROFILE_NAME
def load_db_profile(work_dir: str | Path) -> dict[str, Any]:
path = db_profile_path(work_dir)
if not path.is_file():
return {}
try:
data = json.loads(path.read_text(encoding="utf-8"))
return data if isinstance(data, dict) else {}
except Exception:
return {}
def save_db_profile(work_dir: str | Path, profile: dict[str, Any]) -> None:
path = db_profile_path(work_dir)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(profile, ensure_ascii=False, indent=4) + "\n", encoding="utf-8")
def merge_profile_with_groups(
profile: dict[str, Any],
groups: list[DbGroupInfo],
) -> dict[str, bool]:
"""Return ``{group_key: checked}`` merging AI profile with heuristic defaults."""
selected: set[str] = set()
for key in ("foundation_groups", "narrative_groups"):
for item in profile.get(key) or []:
if isinstance(item, str) and item.strip():
selected.add(item.strip())
defer = {str(x) for x in (profile.get("defer_groups") or []) if str(x).strip()}
result: dict[str, bool] = {}
for g in groups:
if g.key in defer:
result[g.key] = False
elif g.key in selected:
result[g.key] = True
elif profile.get("foundation_groups") or profile.get("narrative_groups"):
# Profile exists but group not listed: use tier default.
result[g.key] = g.default_checked
else:
result[g.key] = g.default_checked
return result
def selected_groups_for_tiers(
groups: list[DbGroupInfo],
tiers: frozenset[str],
) -> list[str]:
return [g.key for g in groups if g.tier in tiers]
def parse_db_filter_tiers(raw: str) -> frozenset[str]:
text = (raw or "").strip()
if not text:
return frozenset()
try:
parsed = json.loads(text)
if isinstance(parsed, list):
return frozenset(str(x).strip() for x in parsed if str(x).strip())
except json.JSONDecodeError:
pass
return frozenset(part.strip() for part in text.split(",") if part.strip())
def parse_db_filter_groups(raw: str) -> frozenset[str]:
text = (raw or "").strip()
if not text:
return frozenset()
try:
parsed = json.loads(text)
if isinstance(parsed, list):
return frozenset(str(x).strip() for x in parsed if str(x).strip())
except json.JSONDecodeError:
pass
return frozenset(part.strip() for part in text.split(",") if part.strip())
def load_db_filter_config() -> tuple[frozenset[str], frozenset[str]]:
"""Read ``wolfDbIncludeTiers`` / ``wolfDbIncludeGroups`` from ``.env``."""
import os
return (
parse_db_filter_tiers(os.getenv("wolfDbIncludeTiers", "")),
parse_db_filter_groups(os.getenv("wolfDbIncludeGroups", "")),
)
def group_matches_filter(
json_file: str,
type_name: str,
include_tiers: frozenset[str],
include_groups: frozenset[str],
) -> bool:
"""True when a DB sheet should be translated under the active filter."""
if not include_tiers and not include_groups:
return True
key = group_key(json_file, type_name)
if include_groups:
return key in include_groups
# Tier-only filter: need to classify on the fly without line data.
tier = classify_group_tier(type_name, [])
return tier in include_tiers
def import_ai_profile(raw: str) -> dict[str, Any]:
"""Parse AI-returned JSON profile text (strips markdown fences if present)."""
text = raw.strip()
if text.startswith("```"):
lines = text.splitlines()
if lines and lines[0].startswith("```"):
lines = lines[1:]
if lines and lines[-1].strip() == "```":
lines = lines[:-1]
text = "\n".join(lines)
data = json.loads(text)
if not isinstance(data, dict):
raise ValueError("Profile must be a JSON object")
return data