pgames-manager/backend/app/converter/pipeline.py

"""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.compatibility import (
    CompatibilityError,
    Issue,
    audit_ir,
    blocking_names,
    detect,
    require_supported_scripts,
)
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

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 _profile_converter(detection):
    """The game profile that converts this game itself, or None for the built-in Ren'Py path."""
    from app.games.registry import select

    profile = select(detection.engine, detection.root)
    if profile is None:
        raise CompatibilityError(detection.engine, [Issue(
            "unsupported_engine", f"Для движка {detection.engine} нет профиля импорта")])
    return None if profile.converts_in_pipeline else profile


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 (profile := _profile_converter(detection)):
            return profile.convert(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 (profile := _profile_converter(detection)):
            return profile.convert(detection.root, work, progress)
        game_dir = find_game_dir(src_root)

    from app.games.renpy_builds import identify, placeholder_files, prepare_source, skipped_files, waive_known
    build = identify(game_dir)  # before the archives are unpacked and scripts decompiled
    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("Разбор скриптов")
    skipped = skipped_files(build)
    rpy_files = sorted((p for p in game_dir.rglob("*.rpy") if not _is_engine_file(game_dir, p)
                        and p.relative_to(game_dir).as_posix().lower() not in skipped),
                       key=lambda p: p.relative_to(game_dir).as_posix().lower())
    sources = {p: prepare_source(build, p.relative_to(game_dir).as_posix(), read_source(p)) for p in rpy_files}
    for p in rpy_files:
        data.prescan(sources[p])
    for p in rpy_files:
        rel = p.relative_to(game_dir).as_posix()
        try:
            RpyParser(data, rel).parse(sources[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)
    for message in waive_known(data, build):
        data.warn(f"Известный дефект оригинала: {message}", blocking=False)
    require_supported_scripts(data)

    files: dict[str, dict[str, Any]] = {}
    aliases: dict[str, str] = {}
    placeholders = placeholder_files(build, game_dir)
    for p in sorted(placeholders):
        data.warn("Файл-заглушка оригинала не конвертируется", p.relative_to(game_dir).as_posix(), blocking=False)
    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) and p not in placeholders]
    counts = {"image": 0, "audio": 0, "video": 0, "font": 0}
    failed: list[str] = []
    done = 0

    image_sizes: dict[str, list[int]] = {}

    def work_one(p: Path) -> tuple[str, str, str, int]:
        dimensions = _dimensions(p)
        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))
        if dimensions:
            image_sizes[logical] = dimensions  # the game asks for them (`renpy.image_size`) before anything is loaded
        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])

    # JSON the game reads itself (`renpy.file("data/x.json").read()`): kept as it is, loaded before the game starts
    data_files: list[str] = []
    for p in sorted(game_dir.rglob("*.json")):
        relative = p.relative_to(game_dir).as_posix()
        if _is_engine_file(game_dir, p) or relative.lower().startswith(("cache/", "saves/")) or p.stat().st_size > DATA_FILE_LIMIT:
            continue
        blob_id, size = put_file(p, ".json")
        files[f"data/{relative}"] = {"hash": blob_id, "size": size}
        data_files.append(relative)

    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,
        "atl": data.atl,
        "blocking": blocking_names(data),
        "screens": {name: data.screens[name] for name in sorted(_used_screens(data))},
        "styles": data.styles,
        "transitions": data.transitions,
        "audio": audio,
        "channels": data.channels,
        "build": data.build,
        "files": aliases,
        "dataFiles": data_files,
        "imageSizes": image_sizes,
        "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())


# Directories inside game/ that are not the story: engine caches, saves, translations, and
# add-on tooling players drop in (0x52-URM is an in-game console / save editor).
ENGINE_DIRS = ("cache", "saves", "tl", "renpy", "0x52-urm")


def _used_screens(data: ScriptData) -> set[str]:
    """Screens the game shows (and the ones those open); the rest stay out of game.json."""
    from app.converter.screens import referenced_screens

    used = {name for name in data.screen_used if name in data.screens}
    pending = list(used)
    while pending:
        for other in referenced_screens(data.screens[pending.pop()], data.screens):
            if other not in used:
                used.add(other)
                pending.append(other)
    return used


DATA_FILE_LIMIT = 4 * 1024 * 1024


def _dimensions(path: Path) -> list[int] | None:
    """Width and height of a picture, read from its header (None for other media)."""
    if assets.kind_of(path) != "image":
        return None
    try:
        from PIL import Image

        with Image.open(path) as img:
            return [int(img.size[0]), int(img.size[1])]
    except Exception:  # noqa: BLE001 - the conversion itself reports a broken picture
        return None


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 ENGINE_DIRS


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", "movieMask"):
        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 "alphaMask" in img:
        img["alphaMask"] = {key: _resolve_image(sub, aliases, missing) for key, sub in img["alphaMask"].items()}
    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 "timeline" in img:
        img["timeline"] = _resolve_timeline(img["timeline"], aliases, missing)
    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 _resolve_timeline(block: dict[str, Any], aliases: dict[str, str], missing: set[str]) -> dict[str, Any]:
    """Resolve the pictures an ATL timeline shows."""
    def step(s: dict[str, Any]) -> dict[str, Any]:
        s = dict(s)
        if s["op"] in ("image", "contains"):
            s["img"] = _resolve_image(s["img"], aliases, missing)
        elif s["op"] == "parallel":
            s["blocks"] = [_resolve_timeline(b, aliases, missing) for b in s["blocks"]]
        elif s["op"] == "choice":
            s["blocks"] = [{**b, "block": _resolve_timeline(b["block"], aliases, missing)} for b in s["blocks"]]
        elif s["op"] in ("block", "on"):
            s["block"] = _resolve_timeline(s["block"], aliases, missing)
        return s

    return {**block, "steps": [step(s) for s in block["steps"]]}


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