DazedAnon
ef7a3f8235
- Add auto-detected RPG Maker and generic loose-PNG profiles - Make scanning read-only and patch images through validated transactions - Preserve RPG Maker compatibility with backups and legacy migration - Document the workflow and expand image manager coverage - Enable core RPG Maker dialogue and choice codes by default
673 lines
22 KiB
Python
673 lines
22 KiB
Python
"""Engine-neutral image workspace, profile registry, and loose-image support."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from typing import Callable, Iterable
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import uuid
|
|
|
|
|
|
PROFILE_AUTO = "auto"
|
|
PROFILE_RPGMAKER_MVMZ = "rpgmaker_mvmz"
|
|
PROFILE_GENERIC = "generic"
|
|
SUPPORTED_PLAIN_EXTENSIONS = (".png",)
|
|
_EXCLUDED_SCAN_DIRS = {".dazedtl", ".git", "__pycache__"}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ImageAsset:
|
|
"""One logical image and its editable/runtime representations."""
|
|
|
|
asset_id: str
|
|
relative_png: Path
|
|
plain_path: Path
|
|
encrypted_path: Path | None = None
|
|
runtime_plain_path: Path | None = None
|
|
engine_id: str = PROFILE_RPGMAKER_MVMZ
|
|
|
|
@property
|
|
def has_plain(self) -> bool:
|
|
return self.plain_path.is_file()
|
|
|
|
@property
|
|
def has_encrypted(self) -> bool:
|
|
return self.encrypted_path is not None and self.encrypted_path.is_file()
|
|
|
|
@property
|
|
def has_runtime_plain(self) -> bool:
|
|
return self.runtime_plain_path is not None and self.runtime_plain_path.is_file()
|
|
|
|
@property
|
|
def runtime_path(self) -> Path | None:
|
|
if self.has_encrypted:
|
|
return self.encrypted_path
|
|
return self.runtime_plain_path
|
|
|
|
|
|
@dataclass
|
|
class ImageActionResult:
|
|
completed: int = 0
|
|
skipped: int = 0
|
|
errors: list[str] | None = None
|
|
patch_files: list[Path] | None = None
|
|
gitignore_files: list[Path] | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.errors is None:
|
|
self.errors = []
|
|
if self.patch_files is None:
|
|
self.patch_files = []
|
|
if self.gitignore_files is None:
|
|
self.gitignore_files = []
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ImageEngineDetection:
|
|
engine_id: str
|
|
confidence: str
|
|
reason: str
|
|
suggested_image_root: Path | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ImageEngineProfile:
|
|
"""Discoverable capabilities for an image-engine backend."""
|
|
|
|
engine_id: str
|
|
label: str
|
|
configurable_image_root: bool
|
|
supports_encryption: bool
|
|
supported_extensions: tuple[str, ...] = SUPPORTED_PLAIN_EXTENSIONS
|
|
|
|
|
|
IMAGE_ENGINE_PROFILES = (
|
|
ImageEngineProfile(
|
|
PROFILE_RPGMAKER_MVMZ,
|
|
"RPG Maker MV/MZ",
|
|
configurable_image_root=False,
|
|
supports_encryption=True,
|
|
),
|
|
ImageEngineProfile(
|
|
PROFILE_GENERIC,
|
|
"Generic / Loose Images",
|
|
configurable_image_root=True,
|
|
supports_encryption=False,
|
|
),
|
|
)
|
|
|
|
|
|
def registered_image_profiles() -> tuple[ImageEngineProfile, ...]:
|
|
return IMAGE_ENGINE_PROFILES
|
|
|
|
|
|
def get_image_profile(engine_id: str) -> ImageEngineProfile:
|
|
for profile in IMAGE_ENGINE_PROFILES:
|
|
if profile.engine_id == engine_id:
|
|
return profile
|
|
raise ValueError(f"Unsupported image engine profile: {engine_id}")
|
|
|
|
|
|
def _resolved_root(game_root: str | Path) -> Path:
|
|
root = Path(game_root).expanduser().resolve()
|
|
if not root.is_dir():
|
|
raise FileNotFoundError(f"Game folder not found: {root}")
|
|
return root
|
|
|
|
|
|
def _inside(root: Path, path: Path, *, label: str = "Image path") -> Path:
|
|
try:
|
|
return path.resolve().relative_to(root.resolve())
|
|
except ValueError as exc:
|
|
raise ValueError(f"{label} is outside the selected game folder: {path}") from exc
|
|
|
|
|
|
def _atomic_write(path: Path, data: bytes) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
handle = tempfile.NamedTemporaryFile(
|
|
mode="wb", prefix=f".{path.name}.", suffix=".tmp", dir=path.parent, delete=False
|
|
)
|
|
temp_path = Path(handle.name)
|
|
try:
|
|
with handle:
|
|
handle.write(data)
|
|
os.replace(temp_path, path)
|
|
except Exception:
|
|
temp_path.unlink(missing_ok=True)
|
|
raise
|
|
|
|
|
|
def editable_workspace_root(game_root: str | Path) -> Path:
|
|
return Path(game_root).expanduser().resolve() / ".dazedtl" / "images"
|
|
|
|
|
|
def image_manager_state_root(game_root: str | Path) -> Path:
|
|
return Path(game_root).expanduser().resolve() / ".dazedtl" / "image_manager"
|
|
|
|
|
|
def _ensure_workspace_ignore(root: Path) -> None:
|
|
ignore_path = root / ".gitignore"
|
|
existing = (
|
|
ignore_path.read_text(encoding="utf-8", errors="surrogateescape")
|
|
if ignore_path.exists()
|
|
else ""
|
|
)
|
|
if "/.dazedtl/" in existing.splitlines():
|
|
return
|
|
text = existing
|
|
if text and not text.endswith("\n"):
|
|
text += "\n"
|
|
if "# DazedTL image manager working files" not in text:
|
|
text += "\n# DazedTL image manager working files\n"
|
|
text += "/.dazedtl/\n"
|
|
_atomic_write(ignore_path, text.encode("utf-8", errors="surrogateescape"))
|
|
|
|
|
|
def ensure_editable_workspace(game_root: str | Path) -> Path:
|
|
"""Create the workspace without migrating or touching runtime assets."""
|
|
|
|
root = _resolved_root(game_root)
|
|
workspace = editable_workspace_root(root)
|
|
workspace.mkdir(parents=True, exist_ok=True)
|
|
_ensure_workspace_ignore(root)
|
|
return workspace
|
|
|
|
|
|
def normalize_generic_image_root(
|
|
game_root: str | Path, image_root: str | Path | None
|
|
) -> Path:
|
|
root = _resolved_root(game_root)
|
|
if image_root is None or not str(image_root).strip():
|
|
raise ValueError("Choose an image folder for the Generic / Loose Images profile.")
|
|
candidate = Path(image_root).expanduser()
|
|
if not candidate.is_absolute():
|
|
candidate = root / candidate
|
|
candidate = candidate.resolve()
|
|
_inside(root, candidate, label="Generic image folder")
|
|
if not candidate.is_dir():
|
|
raise FileNotFoundError(f"Image folder not found: {candidate}")
|
|
return candidate
|
|
|
|
|
|
def _iter_plain_images(root: Path) -> Iterable[Path]:
|
|
for current, dirnames, filenames in os.walk(root, followlinks=False):
|
|
current_path = Path(current)
|
|
kept: list[str] = []
|
|
for name in dirnames:
|
|
child = current_path / name
|
|
if name in _EXCLUDED_SCAN_DIRS or child.is_symlink():
|
|
continue
|
|
kept.append(name)
|
|
dirnames[:] = kept
|
|
for name in filenames:
|
|
path = current_path / name
|
|
if path.is_symlink() or path.suffix.casefold() not in SUPPORTED_PLAIN_EXTENSIONS:
|
|
continue
|
|
yield path
|
|
|
|
|
|
def scan_generic_image_assets(
|
|
game_root: str | Path, image_root: str | Path | None
|
|
) -> list[ImageAsset]:
|
|
root = _resolved_root(game_root)
|
|
source_root = normalize_generic_image_root(root, image_root)
|
|
source_relative = _inside(root, source_root, label="Generic image folder")
|
|
workspace = editable_workspace_root(root)
|
|
workspace_source = workspace / source_relative
|
|
by_id: dict[str, dict[str, Path]] = {}
|
|
|
|
for runtime in _iter_plain_images(source_root):
|
|
relative = _inside(root, runtime)
|
|
asset_id = relative.as_posix()
|
|
by_id[asset_id] = {
|
|
"relative": relative,
|
|
"runtime": runtime,
|
|
"plain": workspace / relative,
|
|
}
|
|
|
|
if workspace_source.is_dir():
|
|
for editable in _iter_plain_images(workspace_source):
|
|
relative_under_source = editable.relative_to(workspace_source)
|
|
relative = source_relative / relative_under_source
|
|
asset_id = relative.as_posix()
|
|
by_id.setdefault(
|
|
asset_id,
|
|
{
|
|
"relative": relative,
|
|
"runtime": root / relative,
|
|
"plain": editable,
|
|
},
|
|
)["plain"] = editable
|
|
|
|
return sorted(
|
|
(
|
|
ImageAsset(
|
|
asset_id=asset_id,
|
|
relative_png=entry["relative"],
|
|
plain_path=entry["plain"],
|
|
runtime_plain_path=entry["runtime"],
|
|
engine_id=PROFILE_GENERIC,
|
|
)
|
|
for asset_id, entry in by_id.items()
|
|
),
|
|
key=lambda asset: asset.asset_id.casefold(),
|
|
)
|
|
|
|
|
|
def _sha256_path(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _validate_png_bytes(data: bytes, asset_id: str) -> None:
|
|
from PIL import Image
|
|
|
|
try:
|
|
with Image.open(BytesIO(data)) as image:
|
|
if image.format != "PNG":
|
|
raise ValueError(f"expected PNG data, found {image.format or 'unknown'}")
|
|
image.verify()
|
|
except Exception as exc:
|
|
raise ValueError(f"editable image is not a valid PNG: {asset_id}: {exc}") from exc
|
|
|
|
|
|
def _validate_asset_collisions(assets: list[ImageAsset]) -> list[ImageAsset]:
|
|
seen: dict[str, str] = {}
|
|
for asset in assets:
|
|
folded = asset.asset_id.casefold()
|
|
previous = seen.get(folded)
|
|
if previous is not None and previous != asset.asset_id:
|
|
raise ValueError(
|
|
"Image paths differ only by letter case and cannot be patched safely on "
|
|
f"Windows: {previous} and {asset.asset_id}"
|
|
)
|
|
seen[folded] = asset.asset_id
|
|
return assets
|
|
|
|
|
|
def _manifest_path(game_root: str | Path) -> Path:
|
|
return image_manager_state_root(game_root) / "manifest.json"
|
|
|
|
|
|
def load_image_manifest(game_root: str | Path) -> dict:
|
|
path = _manifest_path(game_root)
|
|
if not path.is_file():
|
|
return {"version": 1, "assets": {}}
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return {"version": 1, "assets": {}}
|
|
if not isinstance(data, dict) or not isinstance(data.get("assets"), dict):
|
|
return {"version": 1, "assets": {}}
|
|
return data
|
|
|
|
|
|
def save_image_manifest(game_root: str | Path, manifest: dict) -> None:
|
|
payload = json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
|
_atomic_write(_manifest_path(game_root), payload.encode("utf-8"))
|
|
|
|
|
|
def _manifest_key(asset: ImageAsset) -> str:
|
|
return f"{asset.engine_id}:{asset.asset_id}"
|
|
|
|
|
|
def record_asset_baseline(game_root: str | Path, asset: ImageAsset) -> None:
|
|
record_asset_baselines(game_root, [asset])
|
|
|
|
|
|
def record_asset_baselines(
|
|
game_root: str | Path, assets: Iterable[ImageAsset]
|
|
) -> None:
|
|
root = Path(game_root).expanduser().resolve()
|
|
manifest = load_image_manifest(root)
|
|
changed = False
|
|
for asset in assets:
|
|
runtime = asset.runtime_path
|
|
if runtime is None or not runtime.is_file():
|
|
continue
|
|
manifest["assets"][_manifest_key(asset)] = {
|
|
"engine": asset.engine_id,
|
|
"asset_id": asset.asset_id,
|
|
"runtime_path": runtime.resolve().relative_to(root).as_posix(),
|
|
"baseline_sha256": _sha256_path(runtime),
|
|
}
|
|
changed = True
|
|
if changed:
|
|
save_image_manifest(root, manifest)
|
|
|
|
|
|
def asset_source_conflict(game_root: str | Path, asset: ImageAsset) -> bool:
|
|
runtime = asset.runtime_path
|
|
if runtime is None or not runtime.is_file():
|
|
return False
|
|
entry = load_image_manifest(game_root)["assets"].get(_manifest_key(asset), {})
|
|
baseline = entry.get("baseline_sha256")
|
|
return bool(baseline and baseline != _sha256_path(runtime))
|
|
|
|
|
|
def copy_plain_assets_to_workspace(
|
|
game_root: str | Path,
|
|
assets: Iterable[ImageAsset],
|
|
*,
|
|
progress: Callable[[str], None] | None = None,
|
|
) -> ImageActionResult:
|
|
root = _resolved_root(game_root)
|
|
workspace = ensure_editable_workspace(root)
|
|
result = ImageActionResult()
|
|
for asset in assets:
|
|
try:
|
|
if asset.has_plain:
|
|
result.skipped += 1
|
|
continue
|
|
runtime = asset.runtime_plain_path
|
|
if runtime is None or not runtime.is_file():
|
|
raise FileNotFoundError("runtime PNG no longer exists")
|
|
_inside(root, runtime)
|
|
_inside(workspace, asset.plain_path, label="Editable image target")
|
|
if progress:
|
|
progress(f"Making {asset.asset_id} editable")
|
|
_atomic_write(asset.plain_path, runtime.read_bytes())
|
|
record_asset_baseline(root, asset)
|
|
result.completed += 1
|
|
except Exception as exc:
|
|
result.errors.append(f"{asset.asset_id}: {exc}")
|
|
return result
|
|
|
|
|
|
def prepare_plain_assets_for_patch(
|
|
game_root: str | Path,
|
|
assets: Iterable[ImageAsset],
|
|
*,
|
|
progress: Callable[[str], None] | None = None,
|
|
) -> ImageActionResult:
|
|
"""Publish loose PNG edits through the transactional patch path."""
|
|
|
|
return _prepare_assets_transaction(
|
|
PROFILE_GENERIC, game_root, assets, None, progress=progress
|
|
)
|
|
|
|
|
|
def _prepare_assets_transaction(
|
|
engine_id: str,
|
|
game_root: str | Path,
|
|
assets: Iterable[ImageAsset],
|
|
key: bytes | None,
|
|
*,
|
|
progress: Callable[[str], None] | None = None,
|
|
) -> ImageActionResult:
|
|
"""Preflight, stage, and atomically publish a set of runtime image files."""
|
|
|
|
from util.rpgmaker_images import (
|
|
add_patch_exceptions,
|
|
decrypt_image_bytes,
|
|
encrypt_image_bytes,
|
|
)
|
|
|
|
root = _resolved_root(game_root)
|
|
result = ImageActionResult()
|
|
candidates: list[tuple[ImageAsset, Path, Path, bytes]] = []
|
|
transaction_root = (
|
|
root / ".dazedtl" / "image_patch_stage" / uuid.uuid4().hex
|
|
)
|
|
|
|
for asset in assets:
|
|
try:
|
|
if not asset.has_plain:
|
|
raise FileNotFoundError("make this image editable before patching it")
|
|
runtime = asset.runtime_path
|
|
if runtime is None or not runtime.is_file():
|
|
raise FileNotFoundError("runtime image no longer exists")
|
|
relative = _inside(root, runtime)
|
|
if asset_source_conflict(root, asset):
|
|
raise RuntimeError(
|
|
"runtime image changed after the editable copy was created; "
|
|
"remove/recreate the editable copy or resolve the conflict manually"
|
|
)
|
|
editable = asset.plain_path.read_bytes()
|
|
_validate_png_bytes(editable, asset.asset_id)
|
|
current = runtime.read_bytes()
|
|
if asset.has_encrypted:
|
|
if engine_id != PROFILE_RPGMAKER_MVMZ:
|
|
raise ValueError("the active engine profile cannot encode this image")
|
|
if key is None:
|
|
raise ValueError("the RPG Maker encryption key is required")
|
|
if decrypt_image_bytes(current, key) == editable:
|
|
result.skipped += 1
|
|
continue
|
|
output = encrypt_image_bytes(editable, key)
|
|
else:
|
|
if current == editable:
|
|
result.skipped += 1
|
|
continue
|
|
output = editable
|
|
staged_new = transaction_root / "new" / relative
|
|
staged_old = transaction_root / "old" / relative
|
|
_atomic_write(staged_new, output)
|
|
_atomic_write(staged_old, current)
|
|
candidates.append((asset, runtime, relative, output))
|
|
except Exception as exc:
|
|
result.errors.append(f"{asset.asset_id}: {exc}")
|
|
|
|
# Treat a selected batch as one transaction. A conflict or invalid image
|
|
# must not leave the other selected files partially published.
|
|
if result.errors:
|
|
if transaction_root.exists():
|
|
shutil.rmtree(transaction_root)
|
|
return result
|
|
if not candidates:
|
|
if transaction_root.exists():
|
|
shutil.rmtree(transaction_root)
|
|
return result
|
|
|
|
published: list[tuple[Path, Path]] = []
|
|
targets = [runtime for _asset, runtime, _relative, _output in candidates]
|
|
try:
|
|
for asset, runtime, relative, output in candidates:
|
|
backup = root / ".dazedtl" / "image_backups" / relative
|
|
if not backup.exists():
|
|
backup.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(transaction_root / "old" / relative, backup)
|
|
if progress:
|
|
action = "Encrypting" if asset.has_encrypted else "Publishing"
|
|
progress(f"{action} {asset.asset_id}")
|
|
_atomic_write(runtime, output)
|
|
published.append((runtime, transaction_root / "old" / relative))
|
|
|
|
if progress:
|
|
progress("Updating .gitignore image allow-rules")
|
|
result.gitignore_files = add_patch_exceptions(root, targets)
|
|
record_asset_baselines(
|
|
root, (asset for asset, _runtime, _relative, _output in candidates)
|
|
)
|
|
except Exception as exc:
|
|
rollback_errors: list[str] = []
|
|
for runtime, staged_old in reversed(published):
|
|
try:
|
|
_atomic_write(runtime, staged_old.read_bytes())
|
|
except Exception as rollback_exc:
|
|
rollback_errors.append(f"{runtime}: {rollback_exc}")
|
|
result.errors.append(f"Patch transaction rolled back: {exc}")
|
|
result.errors.extend(f"Rollback failed: {error}" for error in rollback_errors)
|
|
return result
|
|
finally:
|
|
if transaction_root.exists():
|
|
shutil.rmtree(transaction_root)
|
|
|
|
result.completed = len(candidates)
|
|
result.patch_files = targets
|
|
return result
|
|
|
|
|
|
def detect_image_engine(game_root: str | Path) -> ImageEngineDetection:
|
|
root = _resolved_root(game_root)
|
|
markers = (
|
|
root / "Game.rpgproject",
|
|
root / "game.rmmzproject",
|
|
root / "www" / "data" / "System.json",
|
|
root / "data" / "System.json",
|
|
)
|
|
for content in (root / "www", root):
|
|
image_root = content / "img"
|
|
if not image_root.is_dir():
|
|
continue
|
|
encrypted = next(
|
|
(
|
|
path
|
|
for pattern in ("*.rpgmvp", "*.png_")
|
|
for path in image_root.rglob(pattern)
|
|
),
|
|
None,
|
|
)
|
|
if encrypted is not None or any(marker.exists() for marker in markers):
|
|
evidence = encrypted.name if encrypted is not None else "System/project markers"
|
|
return ImageEngineDetection(
|
|
PROFILE_RPGMAKER_MVMZ,
|
|
"high",
|
|
f"RPG Maker MV/MZ detected from {evidence} and {image_root}",
|
|
image_root,
|
|
)
|
|
|
|
candidates = (
|
|
root / "images",
|
|
root / "image",
|
|
root / "assets" / "images",
|
|
root / "assets",
|
|
root / "game" / "images",
|
|
root / "img",
|
|
)
|
|
suggested = next(
|
|
(
|
|
candidate
|
|
for candidate in candidates
|
|
if candidate.is_dir() and next(_iter_plain_images(candidate), None) is not None
|
|
),
|
|
None,
|
|
)
|
|
reason = (
|
|
f"No RPG Maker encryption markers found; loose PNGs found under {suggested}"
|
|
if suggested
|
|
else "No supported engine markers found; choose the folder containing loose PNGs"
|
|
)
|
|
return ImageEngineDetection(PROFILE_GENERIC, "fallback", reason, suggested)
|
|
|
|
|
|
def profile_label(engine_id: str) -> str:
|
|
try:
|
|
return get_image_profile(engine_id).label
|
|
except ValueError:
|
|
return engine_id
|
|
|
|
|
|
def scan_profile_assets(
|
|
engine_id: str,
|
|
game_root: str | Path,
|
|
image_root: str | Path | None = None,
|
|
) -> list[ImageAsset]:
|
|
if engine_id == PROFILE_RPGMAKER_MVMZ:
|
|
from util.rpgmaker_images import scan_image_assets
|
|
|
|
return _validate_asset_collisions(scan_image_assets(game_root))
|
|
if engine_id == PROFILE_GENERIC:
|
|
return _validate_asset_collisions(
|
|
scan_generic_image_assets(game_root, image_root)
|
|
)
|
|
raise ValueError(f"Unsupported image engine profile: {engine_id}")
|
|
|
|
|
|
def make_profile_assets_editable(
|
|
engine_id: str,
|
|
game_root: str | Path,
|
|
assets: Iterable[ImageAsset],
|
|
key: bytes | None,
|
|
*,
|
|
progress: Callable[[str], None] | None = None,
|
|
) -> ImageActionResult:
|
|
assets = list(assets)
|
|
if engine_id == PROFILE_RPGMAKER_MVMZ:
|
|
from util.rpgmaker_images import decrypt_assets
|
|
|
|
encrypted = [asset for asset in assets if asset.has_encrypted]
|
|
plain = [asset for asset in assets if not asset.has_encrypted]
|
|
result = ImageActionResult()
|
|
if encrypted:
|
|
if key is None:
|
|
result.errors.append("RPG Maker encryption key is required.")
|
|
else:
|
|
before = {asset.asset_id: asset.has_plain for asset in encrypted}
|
|
decrypted = decrypt_assets(
|
|
encrypted, key, game_root=game_root, overwrite=False, progress=progress
|
|
)
|
|
result.completed += decrypted.completed
|
|
result.skipped += decrypted.skipped
|
|
result.errors.extend(decrypted.errors)
|
|
for asset in encrypted:
|
|
if not before[asset.asset_id] and asset.plain_path.is_file():
|
|
record_asset_baseline(game_root, asset)
|
|
copied = copy_plain_assets_to_workspace(game_root, plain, progress=progress)
|
|
result.completed += copied.completed
|
|
result.skipped += copied.skipped
|
|
result.errors.extend(copied.errors)
|
|
return result
|
|
if engine_id == PROFILE_GENERIC:
|
|
return copy_plain_assets_to_workspace(game_root, assets, progress=progress)
|
|
raise ValueError(f"Unsupported image engine profile: {engine_id}")
|
|
|
|
|
|
def prepare_profile_assets_for_patch(
|
|
engine_id: str,
|
|
game_root: str | Path,
|
|
assets: Iterable[ImageAsset],
|
|
key: bytes | None,
|
|
*,
|
|
progress: Callable[[str], None] | None = None,
|
|
) -> ImageActionResult:
|
|
assets = list(assets)
|
|
if engine_id == PROFILE_GENERIC:
|
|
return prepare_plain_assets_for_patch(game_root, assets, progress=progress)
|
|
if engine_id == PROFILE_RPGMAKER_MVMZ:
|
|
return _prepare_assets_transaction(
|
|
PROFILE_RPGMAKER_MVMZ,
|
|
game_root,
|
|
assets,
|
|
key,
|
|
progress=progress,
|
|
)
|
|
raise ValueError(f"Unsupported image engine profile: {engine_id}")
|
|
|
|
|
|
def preview_profile_png_bytes(
|
|
engine_id: str, asset: ImageAsset, key: bytes | None
|
|
) -> bytes:
|
|
if engine_id == PROFILE_RPGMAKER_MVMZ:
|
|
from util.rpgmaker_images import preview_png_bytes
|
|
|
|
return preview_png_bytes(asset, key)
|
|
if asset.has_plain:
|
|
return asset.plain_path.read_bytes()
|
|
if asset.has_runtime_plain:
|
|
return asset.runtime_plain_path.read_bytes()
|
|
raise FileNotFoundError(asset.runtime_plain_path or asset.plain_path)
|
|
|
|
|
|
def thumbnail_profile_png_bytes(
|
|
engine_id: str, asset: ImageAsset, key: bytes | None, size: int = 128
|
|
) -> bytes:
|
|
from PIL import Image
|
|
|
|
raw = preview_profile_png_bytes(engine_id, asset, key)
|
|
with Image.open(BytesIO(raw)) as image:
|
|
image.thumbnail((size, size))
|
|
if image.mode not in {"RGB", "RGBA"}:
|
|
image = image.convert("RGBA")
|
|
output = BytesIO()
|
|
image.save(output, format="PNG")
|
|
return output.getvalue()
|