"""Validated PGVN editing and deterministic three-way merges (no source execution).""" import copy import difflib import json from uuid import NAMESPACE_URL, uuid4, uuid5 from app.blobs import BLOB_ID_RE from app.converter.locales import walk from app.converter.pipeline import anchors from app.games.registry import for_runtime from app.pgvn import expression, statements from app.translation import file_json, put_json OPS = set("label camera say scene show hide with play stop queue voice pause menu if while jump call return exec window noop nvl".split()) OPS |= {"nani_add_choice", "nani_menu", "nani_clear_choice", "nani_hide_characters", "nani_exit"} OPS.add("nani_command") OPS |= {"call_screen", "show_screen", "hide_screen"} MISSING = object() def document(manifest): game = copy.deepcopy(file_json(manifest, "game.json")) return {"game": game, "scripts": {p: copy.deepcopy(file_json(manifest, p)) for p in game["scripts"]}} def check_path(path): if not isinstance(path, str) or not path or "\\" in path or path.startswith("/") or any(p in ("", ".", "..") for p in path.split("/")): raise ValueError("Некорректный путь файла") return path def validate(doc, manifest): """Validate the whole playable graph before writing any script blobs.""" if not isinstance(doc, dict) or not isinstance(doc.get("game"), dict) or not isinstance(doc.get("scripts"), dict): raise ValueError("Нужны game и scripts") game, scripts = doc["game"], doc["scripts"] paths = game.get("scripts") if game.get("format") != "pgvn/1" or not isinstance(paths, list) or any(not isinstance(p, str) for p in paths) or len(set(paths)) != len(paths) or set(paths) != set(scripts): raise ValueError("Список модулей не совпадает с game.scripts") resolution = game.get("resolution") if not isinstance(resolution, list) or len(resolution) != 2 or any(type(n) is not int or not 1 <= n <= 32768 for n in resolution): raise ValueError("Некорректное разрешение игры") for key in ("characters", "images", "transforms", "transitions", "audio", "files", "ui"): if not isinstance(game.get(key), dict): raise ValueError(f"Нужен объект game.{key}") if not isinstance(game.get("defaults"), list) or not isinstance(game.get("init"), list): raise ValueError("Некорректные defaults/init") statements(game["init"]) for item in game["defaults"]: if not isinstance(item, list) or len(item) != 2 or not isinstance(item[0], str): raise ValueError("Некорректное значение default") expression(item[1]) names, ids, tids = set(), set(), set() targets = [] budget = 0 def identity(obj): ident = obj.get("id") if not isinstance(ident, str) or not ident or len(ident) > 128 or ident in ids: raise ValueError("Отсутствующий или повторяющийся ID") ids.add(ident) def name(value): if not isinstance(value, str) or not value or value in names: raise ValueError("Пустая или повторяющаяся метка") names.add(value) def body(nodes, depth=0): nonlocal budget if depth > 64 or not isinstance(nodes, list): raise ValueError("Некорректная вложенность скрипта") for n in nodes: budget += 1 if budget > 200000 or not isinstance(n, dict) or n.get("op") not in OPS: raise ValueError("Некорректный оператор") identity(n) op = n["op"] for key in ("expr", "c", "d", "whatExpr", "whoExpr", "tagExpr"): if isinstance(n.get(key), dict): expression(n[key]) if op in ("call_screen", "show_screen"): if not isinstance(n.get("name"), str) or not isinstance(n.get("args"), list): raise ValueError("Некорректный вызов screen") for item in [*n["args"], *(n.get("kw") or {}).values()]: expression(item) if op == "hide_screen" and not isinstance(n.get("name"), str): raise ValueError("Некорректный вызов screen") if op in {"say", "nani_add_choice"} or (op == "nani_command" and "what" in n): if not isinstance(n.get("what"), str) or not isinstance(n.get("tid"), str) or n["tid"] in tids: raise ValueError("Некорректная реплика или повторяющийся tid") tids.add(n["tid"]) if op == "label": name(n.get("name")) if op in ("jump", "call", "nani_add_choice"): if isinstance(n.get("target"), str): targets.append(current_scope + n["target"] if n["target"].startswith(".") else n["target"]) elif not isinstance(n.get("expr"), dict): raise ValueError("Переход должен содержать target или expr") if op in ("if", "menu"): children = n.get("branches" if op == "if" else "items") if not isinstance(children, list) or not children: raise ValueError("Пустое условие или меню") for child in children: if not isinstance(child, dict): raise ValueError("Некорректная ветка") identity(child) cond = child.get("if" if op == "menu" else "c") if cond is not None and not isinstance(cond, dict): raise ValueError("Условие ветки должно быть PGVN-выражением") if cond is not None: expression(cond) if op == "menu" and not isinstance(child.get("text"), str): raise ValueError("Нужен текст варианта выбора") body(child.get("body"), depth + 1) if op == "menu" and n.get("say"): body([n["say"]], depth + 1) if op == "while": if not isinstance(n.get("c"), dict): raise ValueError("Нужно условие цикла") body(n.get("body"), depth + 1) if op == "exec": statements(n.get("code")) if op == "nani_command" and not for_runtime(game.get("runtime")): raise ValueError("Исходные команды Naninovel требуют отдельный плеер игры") for path, module in scripts.items(): check_path(path) if not path.startswith("script/") or not path.endswith(".json") or not isinstance(module, dict) or not isinstance(module.get("labels"), list): raise ValueError("Некорректный модуль") scope = "" for label in module["labels"]: identity(label) label_name = label.get("name") if not isinstance(label_name, str): raise ValueError("Нужно имя метки") if not label_name.startswith("."): scope = label_name resolved_name = scope + label_name if label_name.startswith(".") else label_name current_scope = resolved_name.split(".")[0] name(resolved_name) body(label.get("body")) if game.get("entry") not in names: raise ValueError("Точка входа не найдена") if any(t not in names for t in targets): raise ValueError("Переход на отсутствующую метку") if profile := for_runtime(game.get("runtime")): profile.validate_mod(game, scripts, manifest) files = manifest["files"] for path, entry in files.items(): check_path(path) if not BLOB_ID_RE.fullmatch(entry.get("hash", "")): raise ValueError("Некорректный хэш") def refs(value): if isinstance(value, dict): for key, v in value.items(): if key in ("file", "movie", "font", "strings", "meta", "text_font", "name_text_font", "textbox") and isinstance(v, str) and v and v not in files and v not in scripts: raise ValueError(f"Отсутствует файл: {v}") refs(v) elif isinstance(value, list): for v in value: refs(v) refs(game) for module in scripts.values(): for label in module["labels"]: for node in walk(label["body"]): refs({k: v for k, v in node.items() if k in ("img", "file", "movie")}) def prepare(doc): """One-time compatibility IDs for labels/branches; existing IDs never change.""" for path, module in doc["scripts"].items(): for label in module["labels"]: label.setdefault("id", str(uuid5(NAMESPACE_URL, path + ":" + label["name"]))) for n in walk(label["body"]): n.setdefault("id", str(uuid4())) for i, c in enumerate(n.get("branches", []) + n.get("items", [])): c.setdefault("id", str(uuid5(NAMESPACE_URL, n["id"] + ":branch:" + str(i)))) if n.get("op") == "say": n.setdefault("tid", "mod_" + n["id"]) return doc def store(doc, manifest): validate(doc, manifest) files = dict(manifest["files"]) old_scripts = set(file_json(manifest, "game.json")["scripts"]) for path in old_scripts - set(doc["scripts"]): files.pop(path, None) for path, module in doc["scripts"].items(): module = {**module, "anchors": anchors(module["labels"])} put_json(files, path, module) put_json(files, "game.json", doc["game"]) from app.translation import stats game = copy.deepcopy(doc["game"]) for code, info in game.get("locales", {}).items(): st = stats({"files": files}, code) info.update(lines=st["translated"], progress=st["progress"], strings_count=st["stringsTranslated"]) if game != doc["game"]: put_json(files, "game.json", game) return {**manifest, "files": files} def merge_order(base, left, right): if left == base: return right if right == base or left == right: return left changes = [] for side in (left, right): for tag, a, b, c, d in difflib.SequenceMatcher(a=base, b=side, autojunk=False).get_opcodes(): if tag != "equal": item = (a, b, side[c:d]) if item not in changes: changes.append(item) changes.sort(key=lambda x: (x[0], x[1])) for x, y in zip(changes, changes[1:]): if y[0] < x[1] or y[0] == x[0]: return None out, pos = [], 0 for a, b, value in changes: out.extend(base[pos:a]) out.extend(value) pos = b return out + base[pos:] def merge_value(base, left, right, path, conflicts, choices): if left == right or right == base: return left if left == base: return right if all(isinstance(v, dict) and "hash" in v for v in (base, left, right)): return conflict(path, base, left, right, conflicts, choices) if base is MISSING and isinstance(left, dict) and isinstance(right, dict): base = {} if all(isinstance(v, dict) for v in (base, left, right)): out = {} for key in sorted(set(base) | set(left) | set(right)): part = path + "/" + key.replace("~", "~0").replace("/", "~1") v = merge_value(base.get(key, MISSING), left.get(key, MISSING), right.get(key, MISSING), part, conflicts, choices) if v is not MISSING: out[key] = v return out if all(isinstance(v, list) for v in (base, left, right)): all_items = base + left + right if all_items and all(isinstance(v, dict) and isinstance(v.get("id"), str) for v in all_items): maps = [{v["id"]: v for v in seq} for seq in (base, left, right)] items = {} for key in sorted(set(maps[0]) | set(maps[1]) | set(maps[2])): item = merge_value(*(m.get(key, MISSING) for m in maps), path + "/@" + key, conflicts, choices) if item is not MISSING: items[key] = item seqs = [[v["id"] for v in seq if v["id"] in items] for seq in (base, left, right)] order = merge_order(*seqs) if order is None: order = conflict(path + "/$order", *seqs, conflicts, choices) order = list(dict.fromkeys(order + [k for k in items if k not in order])) return [items[k] for k in order if k in items] return conflict(path, base, left, right, conflicts, choices) def conflict(path, base, left, right, conflicts, choices): if choices.get(path) in ("left", "right"): return left if choices[path] == "left" else right conflicts.append({"path": path, "base": None if base is MISSING else base, "left": None if left is MISSING else left, "right": None if right is MISSING else right, "leftDeleted": left is MISSING, "rightDeleted": right is MISSING}) return left def comparable(doc): doc = copy.deepcopy(doc) for module in doc["scripts"].values(): module.pop("anchors", None) return doc def differences(base, current, path=""): """Bounded structural diff, using IDs rather than array positions for commands.""" if base == current: return [] if isinstance(base, dict) and isinstance(current, dict): out = [] for key in sorted(set(base) | set(current)): out.extend(differences(base.get(key), current.get(key), path + "/" + key)) return out if isinstance(base, list) and isinstance(current, list) and base + current and all(isinstance(v, dict) and "id" in v for v in base + current): a, b = ({v["id"]: v for v in seq} for seq in (base, current)) out = differences(a, b, path) if [v["id"] for v in base] != [v["id"] for v in current]: out.append({"path": path + "/$order", "before": list(a), "after": list(b)}) return out return [{"path": path, "before": base, "after": current}] def fingerprint(node): return json.dumps({k: v for k, v in node.items() if k not in ("id", "tid", "body", "items", "branches", "say")}, sort_keys=True, ensure_ascii=False) def align(base, target, manual=None): """Only unique semantic matches within a named label. Ordinal IDs are NEVER evidence.""" target = copy.deepcopy(target) manual = manual or {} pairs, unresolved, used = {}, [], set() old_nodes, new_nodes = {}, {} for path, module in base["scripts"].items(): other = target["scripts"].get(path, {}) by_name = {l["name"]: l for l in other.get("labels", [])} for label in module["labels"]: new_label = by_name.get(label["name"]) old = list(walk(label["body"])) old_nodes.update((n["id"], n) for n in old) new = list(walk(new_label["body"])) if new_label else [] new_nodes.update((n["id"], n) for n in new) if new_label: new_label["id"] = label["id"] for n in old: same_old = [o for o in old if fingerprint(o) == fingerprint(n)] candidates = [o for o in new if fingerprint(o) == fingerprint(n)] chosen = manual.get(n["id"]) if chosen is not None: candidates = [o for o in new if o["id"] == chosen] if len(candidates) != 1 or candidates[0]["op"] != n["op"]: raise ValueError("Некорректное ручное сопоставление") if len(candidates) == 1 and (chosen is not None or len(same_old) == 1): nid = candidates[0]["id"] if nid in used: raise ValueError("Узел сопоставлен дважды") used.add(nid) pairs[nid] = n else: unresolved.append({"id": n["id"], "module": path, "label": label["name"], "op": n["op"], "text": n.get("what", n.get("op")), "candidates": [{"id": o["id"], "text": o.get("what", o["op"])} for o in new if o["op"] == n["op"]]}) unknown = set(manual) - set(old_nodes) if unknown: raise ValueError("Неизвестный ID базы") for module in target["scripts"].values(): for label in module["labels"]: for node in walk(label["body"]): original = pairs.get(node["id"]) node["id"] = original["id"] if original else str(uuid5(NAMESPACE_URL, "rebase:" + node["id"] + ":" + fingerprint(node))) if original and "tid" in original: node["tid"] = original["tid"] for key in ("branches", "items"): previous = original.get(key, []) if original else [] for i, child in enumerate(node.get(key, [])): matches = [p for p in previous if {k: v for k, v in p.items() if k not in ("id", "body")} == {k: v for k, v in child.items() if k not in ("id", "body")}] child["id"] = matches[0]["id"] if len(matches) == 1 else str(uuid5(NAMESPACE_URL, node["id"] + ":" + key + ":" + str(i))) return target, unresolved