377 lines
16 KiB
Python
377 lines
16 KiB
Python
"""Full conversion: uploaded archive → blobs + version manifest."""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import time
|
|
from collections.abc import Callable
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from dataclasses import dataclass
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any
|
|
|
|
from app.blobs import put_bytes, put_file
|
|
from app.config import get_settings
|
|
from app.converter import assets, locales
|
|
from app.converter.archive import extract, find_game_dir
|
|
from app.converter.rpa import RpaError, extract_rpa
|
|
from app.converter.rpy_lexer import ParseError
|
|
from app.converter.rpy_parser import RpyParser, ScriptData
|
|
from app.converter.rpyc import decompile_missing
|
|
from app.converter.source import read_source
|
|
from app.converter.compatibility import CompatibilityError, Issue, audit_ir, detect, require_supported_scripts
|
|
|
|
log = logging.getLogger(__name__)
|
|
FORMAT = "pgvn/1"
|
|
|
|
Progress = Callable[[str], None]
|
|
|
|
|
|
@dataclass
|
|
class Result:
|
|
manifest: dict[str, Any]
|
|
report: dict[str, Any]
|
|
title: str | None
|
|
cover_source: str | None # blob id of an image suitable as a cover
|
|
size: int
|
|
|
|
|
|
def convert_archive(archive: Path, filename: str, work: Path, progress: Progress) -> Result:
|
|
"""Convert an uploaded archive, or an already unpacked game directory (left untouched)."""
|
|
started = time.monotonic()
|
|
if work.exists():
|
|
shutil.rmtree(work)
|
|
src_root = work / "src"
|
|
out_dir = work / "out"
|
|
out_dir.mkdir(parents=True)
|
|
data = ScriptData()
|
|
|
|
if archive.is_dir():
|
|
detection = detect(archive)
|
|
if detection.engine == "naninovel":
|
|
from app.converter.naninovel import convert_naninovel
|
|
return convert_naninovel(detection.root, work, progress)
|
|
# Inspect scripts before copying/converting multi-gigabyte media collections.
|
|
from app.converter.preflight import inspect_directory
|
|
inspect_directory(detection.root)
|
|
progress("Копирование файлов игры")
|
|
game_dir = src_root / "game"
|
|
shutil.copytree(find_game_dir(archive), game_dir)
|
|
else:
|
|
from app.converter.preflight import inspect_archive
|
|
progress("Проверка движка и сценариев до распаковки медиа")
|
|
inspect_archive(archive, filename)
|
|
progress("Распаковка архива")
|
|
extract(archive, filename, src_root)
|
|
detection = detect(src_root)
|
|
if detection.engine == "naninovel":
|
|
from app.converter.naninovel import convert_naninovel
|
|
return convert_naninovel(detection.root, work, progress)
|
|
game_dir = find_game_dir(src_root)
|
|
|
|
rpas = sorted(game_dir.rglob("*.rpa"))
|
|
for i, rpa in enumerate(rpas, 1):
|
|
progress(f"Распаковка {rpa.name} ({i}/{len(rpas)})")
|
|
try:
|
|
extract_rpa(rpa, game_dir)
|
|
except RpaError as e:
|
|
data.warn(str(e))
|
|
rpa.unlink()
|
|
|
|
progress("Декомпиляция .rpyc")
|
|
decompiled = decompile_missing(game_dir, data.warn)
|
|
|
|
progress("Разбор скриптов")
|
|
rpy_files = sorted((p for p in game_dir.rglob("*.rpy") if not _is_engine_file(game_dir, p)),
|
|
key=lambda p: p.relative_to(game_dir).as_posix().lower())
|
|
for p in rpy_files:
|
|
rel = p.relative_to(game_dir).as_posix()
|
|
try:
|
|
RpyParser(data, rel).parse(read_source(p))
|
|
except ParseError as e:
|
|
data.warn(f"Ошибка разбора файла: {e.msg}", f"{e.filename}:{e.lineno}")
|
|
except RecursionError:
|
|
data.warn("Слишком глубокая вложенность", rel)
|
|
if not any(lbl["name"] == "start" for labels in data.modules.values() for lbl in labels):
|
|
raise RuntimeError("В скриптах не найдена метка start — нечего запускать")
|
|
audit_ir(data)
|
|
require_supported_scripts(data)
|
|
|
|
files: dict[str, dict[str, Any]] = {}
|
|
aliases: dict[str, str] = {}
|
|
media = [p for p in game_dir.rglob("*")
|
|
if p.is_file() and assets.kind_of(p) and not _is_engine_file(game_dir, p)]
|
|
counts = {"image": 0, "audio": 0, "video": 0, "font": 0}
|
|
failed: list[str] = []
|
|
done = 0
|
|
|
|
def work_one(p: Path) -> tuple[str, str, str, int]:
|
|
out, ext = assets.convert(p, out_dir)
|
|
blob_id, size = put_file(out, ext, move=True)
|
|
rel = p.relative_to(game_dir).as_posix()
|
|
logical = str(PurePosixPath(rel).with_suffix(ext))
|
|
return rel, logical, blob_id, size
|
|
|
|
with ThreadPoolExecutor(max_workers=get_settings().worker_threads) as pool:
|
|
futures = {pool.submit(work_one, p): p for p in media}
|
|
for fut in as_completed(futures):
|
|
p = futures[fut]
|
|
done += 1
|
|
if done % 20 == 0 or done == len(media):
|
|
progress(f"Конвертация медиа {done}/{len(media)}")
|
|
try:
|
|
rel, logical, blob_id, size = fut.result()
|
|
except Exception as e: # noqa: BLE001 — one broken file must not fail the game
|
|
failed.append(p.name)
|
|
data.warn("Не удалось сконвертировать файл", f"{p.name}: {e}")
|
|
continue
|
|
files[logical] = {"hash": blob_id, "size": size}
|
|
aliases[rel.lower()] = logical
|
|
counts[assets.kind_of(p)] += 1
|
|
|
|
if failed:
|
|
raise CompatibilityError("renpy", [Issue("media_conversion", "Не удалось конвертировать обязательное медиа", name)
|
|
for name in failed])
|
|
|
|
progress("Сборка формата PGVN")
|
|
_auto_images(data, aliases)
|
|
missing: set[str] = set()
|
|
images = {name: _resolve_image(img, aliases, missing) for name, img in data.images.items()}
|
|
audio = {name: _resolve(path, aliases) or path for name, path in data.audio.items()}
|
|
|
|
progress("Извлечение текстов и переводов")
|
|
by_line = locales.assign_ids(data.modules)
|
|
translations = locales.import_tl(game_dir, data.modules, by_line, data.warn)
|
|
locales.strip_temp(data.modules)
|
|
language = data.config.get("language")
|
|
source_lang = (locales.lang_code(language) if isinstance(language, str)
|
|
else locales.detect_language(data.modules)) or "und"
|
|
|
|
scripts: list[str] = []
|
|
for rel, labels in data.modules.items():
|
|
logical = "script/" + str(PurePosixPath(rel).with_suffix(".json"))
|
|
payload = json.dumps({"file": rel, "labels": labels, "anchors": anchors(labels)},
|
|
ensure_ascii=False,
|
|
separators=(",", ":")).encode()
|
|
blob_id, size = put_bytes(payload, ".json")
|
|
files[logical] = {"hash": blob_id, "size": size}
|
|
scripts.append(logical)
|
|
|
|
locale_index = _write_locales(translations, data.modules, files)
|
|
|
|
gui = data.gui
|
|
resolution = gui.get("resolution") or [
|
|
data.config.get("screen_width") or 1280, data.config.get("screen_height") or 720]
|
|
title = data.config.get("name") if isinstance(data.config.get("name"), str) else None
|
|
game = {
|
|
"format": FORMAT,
|
|
"engine": "renpy",
|
|
"title": title,
|
|
"version": data.config.get("version"),
|
|
"resolution": resolution,
|
|
"entry": "start",
|
|
"scripts": scripts,
|
|
"characters": data.characters,
|
|
"images": images,
|
|
"transforms": data.transforms,
|
|
"transitions": data.transitions,
|
|
"audio": audio,
|
|
"channels": data.channels,
|
|
"files": aliases,
|
|
"defaults": data.defaults,
|
|
"init": data.init_code,
|
|
"ui": _ui(gui, aliases),
|
|
"sourceLang": source_lang,
|
|
"locales": locale_index,
|
|
}
|
|
payload = json.dumps(game, ensure_ascii=False, separators=(",", ":")).encode()
|
|
blob_id, size = put_bytes(payload, ".json")
|
|
files["game.json"] = {"hash": blob_id, "size": size}
|
|
|
|
cover = None
|
|
for candidate in ("gui/main_menu.png", "gui/main_menu.jpg", "gui/main_menu.webp",
|
|
"gui/window_icon.png"):
|
|
if (logical := aliases.get(candidate)) and logical in files:
|
|
cover = files[logical]["hash"]
|
|
break
|
|
|
|
total = sum(f["size"] for f in files.values())
|
|
report = {
|
|
"engine": "renpy",
|
|
"compatible": True,
|
|
"converterVersion": "pgvn-tools/2",
|
|
"unrpycRevision": (get_settings().unrpyc_path.parent / "REVISION").read_text().strip()
|
|
if (get_settings().unrpyc_path.parent / "REVISION").is_file() else "unavailable",
|
|
"format": FORMAT,
|
|
"durationSec": round(time.monotonic() - started, 1),
|
|
"scripts": len(rpy_files),
|
|
"decompiled": decompiled,
|
|
"labels": sum(len(v) for v in data.modules.values()),
|
|
"stats": dict(data.stats),
|
|
"media": counts,
|
|
"failedMedia": failed[:50],
|
|
"missingFiles": sorted(missing)[:100],
|
|
"sourceLang": source_lang,
|
|
"locales": {code: {"lines": meta["lines"], "strings": meta["strings_count"],
|
|
"progress": meta["progress"]}
|
|
for code, meta in locale_index.items()},
|
|
"warnings": [
|
|
{"message": msg, "count": n, "example": data.warning_examples.get(msg)}
|
|
for msg, n in data.warnings.most_common()
|
|
],
|
|
"sourceSize": _size(archive),
|
|
"outputSize": total,
|
|
}
|
|
return Result({"format": FORMAT, "files": files}, report, title, cover, total)
|
|
|
|
|
|
def _write_locales(translations: dict[str, "locales.LocaleData"],
|
|
modules: dict[str, list[dict[str, Any]]],
|
|
files: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]:
|
|
"""Store each language as per-module line tables plus one strings table."""
|
|
tid_module: dict[str, str] = {}
|
|
tid_source: dict[str, str] = {}
|
|
for rel, labels in modules.items():
|
|
for label in labels:
|
|
for node in locales.walk(label["body"]):
|
|
if node.get("op") == "say":
|
|
tid_module[node["tid"]] = rel
|
|
tid_source[node["tid"]] = node["what"]
|
|
index: dict[str, dict[str, Any]] = {}
|
|
|
|
def put_json(path: str, value: Any) -> None:
|
|
payload = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()
|
|
blob_id, size = put_bytes(payload, ".json")
|
|
files[path] = {"hash": blob_id, "size": size}
|
|
|
|
for code, loc in sorted(translations.items()):
|
|
grouped: dict[str, dict[str, Any]] = {}
|
|
for tid, text in loc.lines.items():
|
|
grouped.setdefault(tid_module[tid], {})[tid] = {
|
|
"t": text, "src": locales.short_hash(tid_source[tid]), "state": "translated"}
|
|
module_files = {}
|
|
for rel, entries in grouped.items():
|
|
path = f"locales/{code}/" + str(PurePosixPath(rel).with_suffix(".json"))
|
|
put_json(path, entries)
|
|
module_files["script/" + str(PurePosixPath(rel).with_suffix(".json"))] = path
|
|
strings_path = f"locales/{code}/strings.json"
|
|
put_json(strings_path, {old: {"t": new, "state": "translated"}
|
|
for old, new in loc.strings.items()})
|
|
index[code] = {
|
|
"name": locales.LANG_NAMES.get(code, loc.renpy),
|
|
"renpy": loc.renpy,
|
|
"modules": module_files,
|
|
"strings": strings_path,
|
|
"lines": len(loc.lines),
|
|
"strings_count": len(loc.strings),
|
|
"progress": round(len(loc.lines) / len(tid_module), 3) if tid_module else 0,
|
|
}
|
|
return index
|
|
|
|
|
|
def _size(path: Path) -> int:
|
|
if path.is_file():
|
|
return path.stat().st_size
|
|
return sum(p.stat().st_size for p in path.rglob("*") if p.is_file())
|
|
|
|
|
|
def _is_engine_file(game_dir: Path, p: Path) -> bool:
|
|
parts = [x.lower() for x in p.relative_to(game_dir).parts]
|
|
return len(parts) > 1 and parts[0] in ("cache", "saves", "tl", "renpy")
|
|
|
|
|
|
def _resolve(path: str, aliases: dict[str, str]) -> str | None:
|
|
key = path.replace("\\", "/").lower().lstrip("/")
|
|
for prefix in ("", "images/", "audio/", "music/", "sound/", "sounds/"):
|
|
if (hit := aliases.get(prefix + key)) is not None:
|
|
return hit
|
|
return None
|
|
|
|
|
|
def _resolve_image(img: dict[str, Any], aliases: dict[str, str], missing: set[str]
|
|
) -> dict[str, Any]:
|
|
img = dict(img)
|
|
for key in ("file", "movie"):
|
|
if isinstance(img.get(key), str):
|
|
hit = _resolve(img[key], aliases)
|
|
if hit is None:
|
|
missing.add(img[key])
|
|
else:
|
|
img[key] = hit
|
|
if "switch" in img:
|
|
img["switch"] = [[c, _resolve_image(sub, aliases, missing)] for c, sub in img["switch"]]
|
|
if "composite" in img:
|
|
img["composite"] = [{**layer, "img": _resolve_image(layer["img"], aliases, missing)}
|
|
for layer in img["composite"]]
|
|
if "frames" in img:
|
|
img["frames"] = [{**f, "img": _resolve_image(f["img"], aliases, missing)}
|
|
for f in img["frames"]]
|
|
if "layered" in img:
|
|
layers = []
|
|
for layer in img["layered"]:
|
|
layer = dict(layer)
|
|
for key in ("always", "img"):
|
|
if key in layer:
|
|
layer[key] = _resolve_image(layer[key], aliases, missing)
|
|
if "cond" in layer:
|
|
layer["cond"] = [[c, _resolve_image(sub, aliases, missing)]
|
|
for c, sub in layer["cond"]]
|
|
layers.append(layer)
|
|
img["layered"] = layers
|
|
return img
|
|
|
|
|
|
def anchors(labels: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
|
|
"""Positions of `label` markers (nested labels, named menus) inside label bodies."""
|
|
found: dict[str, dict[str, Any]] = {}
|
|
|
|
def walk(nodes: list[dict[str, Any]], top: str, path: list[int]) -> None:
|
|
for i, n in enumerate(nodes):
|
|
here = path + [i]
|
|
if n["op"] == "label":
|
|
found[n["name"]] = {"label": top, "path": here, "params": n.get("params")}
|
|
if n["op"] == "if":
|
|
for b, branch in enumerate(n["branches"]):
|
|
walk(branch["body"], top, here + [b])
|
|
elif n["op"] == "menu":
|
|
for b, item in enumerate(n["items"]):
|
|
walk(item["body"], top, here + [b])
|
|
elif n["op"] == "while":
|
|
walk(n["body"], top, here + [0])
|
|
|
|
for label in labels:
|
|
walk(label["body"], label["name"], [])
|
|
return found
|
|
|
|
|
|
def _auto_images(data: ScriptData, aliases: dict[str, str]) -> None:
|
|
"""Ren'Py defines an image for every file under images/ (name = lowercase basename)."""
|
|
for rel, logical in aliases.items():
|
|
if not rel.startswith("images/") or assets.kind_of(Path(rel)) != "image":
|
|
continue
|
|
name = os.path.splitext(os.path.basename(rel))[0].lower().strip()
|
|
name = " ".join(name.split())
|
|
if name and name not in data.images:
|
|
data.images[name] = {"file": rel}
|
|
|
|
|
|
def _ui(gui: dict[str, Any], aliases: dict[str, str]) -> dict[str, Any]:
|
|
ui: dict[str, Any] = {}
|
|
for key, rel in (("textbox", "gui/textbox.png"), ("namebox", "gui/namebox.png"),
|
|
("mainMenu", "gui/main_menu.png"), ("nvl", "gui/nvl.png"),
|
|
("choice", "gui/button/choice_idle_background.png"),
|
|
("choiceHover", "gui/button/choice_hover_background.png")):
|
|
if hit := aliases.get(rel):
|
|
ui[key] = hit
|
|
for key in ("text_font", "name_text_font"):
|
|
if isinstance(gui.get(key), str) and (hit := _resolve(gui[key], aliases)):
|
|
ui[key] = hit
|
|
for key in ("text_size", "name_text_size", "text_color", "accent_color", "textbox_height",
|
|
"textbox_yalign", "dialogue_xpos", "dialogue_ypos", "dialogue_width",
|
|
"name_xpos", "name_ypos", "interface_text_color", "idle_color", "hover_color",
|
|
"choice_button_text_idle_color", "choice_button_text_hover_color"):
|
|
if isinstance(gui.get(key), (int, float, str)):
|
|
ui[key] = gui[key]
|
|
return ui
|