- FastAPI backend: admin bootstrap with forced credential change, admin-created users with one-time generated passwords, roles as permission sets - Games and versions with content-addressed blob storage; forks copy only the manifest, so mods share assets with the original - Worker: archive/RPA extraction, unrpyc for .rpyc, own .rpy parser, Python subset translated via ast, media converted to WebP/Opus/WebM - React panel (library search, game page, upload, users, roles) and a framework-free web player with saves, history, skip/auto, transitions - docker compose: postgres, backend, worker, nginx on a single HTTP port Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
"""Archive extraction and locating the game directory inside it."""
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
|
|
class ArchiveError(Exception):
|
|
pass
|
|
|
|
|
|
def _safe_target(root: Path, name: str) -> Path | None:
|
|
target = (root / name).resolve()
|
|
return target if target.is_relative_to(root.resolve()) else None
|
|
|
|
|
|
def extract(archive: Path, filename: str, dest: Path) -> None:
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
if filename.lower().endswith(".zip") and zipfile.is_zipfile(archive):
|
|
_extract_zip(archive, dest)
|
|
return
|
|
seven = shutil.which("7zz") or shutil.which("7z")
|
|
if not seven:
|
|
raise ArchiveError("Для этого формата нужен 7-Zip (7zz), он не установлен")
|
|
proc = subprocess.run([seven, "x", "-y", "-snl-", f"-o{dest}", str(archive)],
|
|
capture_output=True, text=True)
|
|
if proc.returncode not in (0, 1): # 1 = warnings
|
|
raise ArchiveError(f"7-Zip: {proc.stderr.strip() or proc.stdout.strip()[-500:]}")
|
|
_drop_escaping_links(dest)
|
|
# .tar.gz etc. extract to a single inner .tar first
|
|
inner = list(dest.iterdir())
|
|
if len(inner) == 1 and inner[0].suffix == ".tar":
|
|
tar = inner[0]
|
|
extract(tar, tar.name, dest / "_tar")
|
|
tar.unlink()
|
|
|
|
|
|
def _extract_zip(archive: Path, dest: Path) -> None:
|
|
with zipfile.ZipFile(archive) as zf:
|
|
for info in zf.infolist():
|
|
name = info.filename
|
|
if info.flag_bits & 0x800 == 0:
|
|
# legacy zips store names in cp437; many were really written as cp866/utf-8
|
|
try:
|
|
name = name.encode("cp437").decode("utf-8")
|
|
except (UnicodeEncodeError, UnicodeDecodeError):
|
|
pass
|
|
target = _safe_target(dest, name)
|
|
if target is None:
|
|
continue
|
|
if info.is_dir():
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
continue
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
with zf.open(info) as src, target.open("wb") as out:
|
|
shutil.copyfileobj(src, out, 1 << 20)
|
|
|
|
|
|
def _drop_escaping_links(root: Path) -> None:
|
|
for dirpath, dirnames, filenames in os.walk(root):
|
|
for name in dirnames + filenames:
|
|
p = Path(dirpath, name)
|
|
if p.is_symlink():
|
|
p.unlink()
|
|
|
|
|
|
def find_game_dir(root: Path) -> Path:
|
|
"""Find the Ren'Py `game` directory (the one holding scripts or archives)."""
|
|
best: tuple[int, Path] | None = None
|
|
for dirpath, _, filenames in os.walk(root):
|
|
exts = {Path(f).suffix.lower() for f in filenames}
|
|
if not exts & {".rpy", ".rpyc", ".rpa", ".rpymc"}:
|
|
continue
|
|
p = Path(dirpath)
|
|
depth = len(p.relative_to(root).parts)
|
|
score = (0 if p.name.lower() == "game" else 100) + depth
|
|
if p.name.lower() in ("renpy", "common") or "renpy" in [x.lower() for x in p.parts[-3:]]:
|
|
continue
|
|
if best is None or score < best[0]:
|
|
best = (score, p)
|
|
if best is None:
|
|
raise ArchiveError("В архиве не найдена игра Ren'Py (нет .rpy/.rpyc/.rpa)")
|
|
game = best[1]
|
|
# scripts in a subfolder of game/ — walk up to game/ itself
|
|
for parent in [game, *game.parents]:
|
|
if parent.name.lower() == "game":
|
|
return parent
|
|
if parent == root:
|
|
break
|
|
return game
|