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

"""Dry run of the script part of the converter on unpacked games (no media work).

    python -m app.converter.analyze /games/Some-Game [/games/Other ...]

Prints, per game, how much of the script was understood and the top warnings.
"""

import json
import re
import shutil
import sys
import tempfile
import time
from collections import Counter
from pathlib import Path

from app.converter import assets, locales
from app.converter.compatibility import Issue, audit_ir
from dataclasses import asdict
from app.converter.archive import ArchiveError, find_game_dir
from app.converter.pipeline import _auto_images, _is_engine_file, _resolve_image, anchors
from app.converter.rpa import RpaError, extract_rpa, read_index
from app.converter.rpy_lexer import ParseError
from app.converter.rpy_parser import RpyParser, ScriptData
from app.converter.rpyc import decompile_missing

SCRIPT_EXTS = (".rpy", ".rpyc", ".rpym", ".rpymc")


def renpy_version(root: Path) -> str:
    for vc in root.rglob("renpy/vc_version.py"):
        m = re.search(r"version\s*=\s*u?['\"]([^'\"]+)", vc.read_text(errors="replace"))
        if m:
            return m.group(1)
    for init in root.rglob("renpy/__init__.py"):
        m = re.search(r"version_tuple\s*=\s*\(([^)]*)\)", init.read_text(errors="replace"))
        if m:
            return ".".join(x.strip() for x in m.group(1).split(",")[:3])
    return "?"


def analyze(src: Path, dump: Path | None = None) -> dict:
    started = time.monotonic()
    game_dir = find_game_dir(src)
    data = ScriptData()
    work = Path(tempfile.mkdtemp(prefix="analyze-"))
    try:
        files: dict[str, str] = {}  # lowercase path -> path, for asset resolution
        for p in game_dir.rglob("*"):
            if p.is_file():
                rel = p.relative_to(game_dir).as_posix()
                if p.suffix.lower() in SCRIPT_EXTS:
                    dest = work / rel
                    dest.parent.mkdir(parents=True, exist_ok=True)
                    shutil.copyfile(p, dest)
                elif p.suffix.lower() != ".rpa":
                    files[rel.lower()] = rel
        rpa_errors = []
        for rpa in game_dir.rglob("*.rpa"):
            try:
                for name in read_index(rpa):
                    if not name.lower().endswith(SCRIPT_EXTS):
                        files[name.lower()] = name
                extract_rpa(rpa, work, only=lambda n: n.lower().endswith(SCRIPT_EXTS))
            except RpaError as e:
                rpa_errors.append(str(e))
        decompiled = decompile_missing(work, data.warn)
        if dump is not None:
            shutil.copytree(work, dump / src.name, dirs_exist_ok=True,
                            ignore=shutil.ignore_patterns("*.rpyc", "*.rpymc"))
        rpy = sorted((p for p in work.rglob("*.rpy") if not _is_engine_file(work, p)),
                     key=lambda p: p.relative_to(work).as_posix().lower())
        lines = 0
        for p in rpy:
            rel = p.relative_to(work).as_posix()
            text = p.read_text(encoding="utf-8", errors="replace")
            lines += text.count("\n")
            try:
                RpyParser(data, rel).parse(text)
            except ParseError as e:
                data.warn(f"Ошибка разбора файла: {e.msg}", f"{e.filename}:{e.lineno}")
            except RecursionError:
                data.warn("Слишком глубокая вложенность", rel)

        by_line = locales.assign_ids(data.modules)
        audit_ir(data)
        translations = locales.import_tl(work, data.modules, by_line, data.warn)
        says = len(by_line) or 1
        language = data.config.get("language")
        source_lang = (locales.lang_code(language) if isinstance(language, str)
                       else locales.detect_language(data.modules))

        # media aliases as the real pipeline would produce them
        aliases = {k: str(Path(v).with_suffix(assets.target_ext(Path(v))).as_posix())
                   for k, v in files.items() if assets.kind_of(Path(v))}
        _auto_images(data, aliases)
        missing: set[str] = set()
        for img in data.images.values():
            _resolve_image(img, aliases, missing)
        labels = {lbl["name"] for mod in data.modules.values() for lbl in mod}
        labels |= {name for mod in data.modules.values() for name in anchors(mod)}
        targets = Counter()
        _collect_targets(data.modules, targets)
        media = Counter(assets.kind_of(Path(v)) for v in files.values() if assets.kind_of(Path(v)))
        return {
            "issues": [asdict(Issue("unsupported_script", msg,
                          data.warning_examples.get(msg, ""), count))
                       for msg, count in data.blockers.items()] +
                      [asdict(Issue("archive_error", msg)) for msg in rpa_errors],
            "game": src.name,
            "renpy": renpy_version(src),
            "seconds": round(time.monotonic() - started, 1),
            "rpyFiles": len(rpy),
            "decompiled": decompiled,
            "lines": lines,
            "labels": len(labels),
            "hasStart": "start" in labels,
            "stats": dict(data.stats),
            "media": dict(media),
            "images": len(data.images),
            "sourceLang": source_lang,
            "locales": {code: {"lines": len(t.lines), "strings": len(t.strings),
                               "progress": round(len(t.lines) / says, 3), "unmatched": t.unmatched}
                        for code, t in translations.items()},
            "characters": len(data.characters),
            "missingJumpTargets": sorted(t for t in targets if t not in labels)[:15],
            "missingFiles": sorted(missing)[:15],
            "rpaErrors": rpa_errors,
            "warningsTotal": sum(data.warnings.values()),
            "warnings": [{"n": n, "msg": m, "ex": data.warning_examples.get(m)}
                         for m, n in data.warnings.most_common(25)],
        }
    finally:
        shutil.rmtree(work, ignore_errors=True)


def _collect_targets(modules, out: Counter) -> None:
    def walk(nodes):
        for n in nodes:
            if n["op"] in ("jump", "call") and n.get("target"):
                out[n["target"]] += 1
            for b in n.get("branches", []):
                walk(b["body"])
            for it in n.get("items", []):
                walk(it["body"])
            walk(n.get("body", []))
    for mod in modules.values():
        for lbl in mod:
            walk(lbl["body"])


if __name__ == "__main__":
    args = sys.argv[1:]
    dump = None
    if args[:1] == ["--dump"]:
        dump, args = Path(args[1]), args[2:]
    for arg in args:
        try:
            result = analyze(Path(arg), dump)
        except (ArchiveError, OSError) as e:
            result = {"game": Path(arg).name, "error": str(e)}
        print(json.dumps(result, ensure_ascii=False), flush=True)