Add .gitattributes: LF in the repository, CRLF for batch files
Normalize the files that had been committed with CRLF (line endings only, no content changes). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
mrleo1nidandClaude Opus 5.5 committed
1 parent
e1d7b29f77
commit
51a7e77e56
14 files changed
+1461
-1447
No files matched your search
@@ -1,86 +1,86 @@
|
||||
"""Media conversion into web-friendly formats."""
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image, ImageSequence
|
||||
|
||||
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif", ".tga", ".avif"}
|
||||
AUDIO_EXTS = {".ogg", ".mp3", ".wav", ".opus", ".flac", ".m4a", ".oga"}
|
||||
VIDEO_EXTS = {".webm", ".ogv", ".mp4", ".mkv", ".avi", ".mpg", ".mpeg", ".m4v"}
|
||||
FONT_EXTS = {".ttf", ".otf", ".woff", ".woff2", ".ttc"}
|
||||
|
||||
WEBP_MAX = 16383
|
||||
|
||||
|
||||
def kind_of(path: Path) -> str | None:
|
||||
ext = path.suffix.lower()
|
||||
if ext in IMAGE_EXTS:
|
||||
return "image"
|
||||
if ext in AUDIO_EXTS:
|
||||
return "audio"
|
||||
if ext in VIDEO_EXTS:
|
||||
return "video"
|
||||
if ext in FONT_EXTS:
|
||||
return "font"
|
||||
return None
|
||||
|
||||
|
||||
def target_ext(path: Path) -> str:
|
||||
kind = kind_of(path)
|
||||
return {"image": ".webp", "audio": ".ogg", "video": ".webm"}.get(kind, path.suffix.lower())
|
||||
|
||||
|
||||
def convert(src: Path, out_dir: Path) -> tuple[Path, str]:
|
||||
"""Convert one file into out_dir. Returns (output file, extension used for the blob)."""
|
||||
kind = kind_of(src)
|
||||
tmp = str(out_dir / uuid.uuid4().hex)
|
||||
if kind == "image":
|
||||
return _image(src, Path(tmp))
|
||||
if kind == "audio":
|
||||
return _ffmpeg(src, Path(tmp + ".ogg"),
|
||||
["-vn", "-c:a", "libopus", "-b:a", "112k", "-vbr", "on"]), ".ogg"
|
||||
if kind == "video":
|
||||
if src.suffix.lower() == ".webm":
|
||||
out = Path(tmp + ".webm")
|
||||
shutil.copyfile(src, out)
|
||||
return out, ".webm"
|
||||
return _ffmpeg(src, Path(tmp + ".webm"),
|
||||
["-c:v", "libvpx-vp9", "-crf", "34", "-b:v", "0", "-deadline", "realtime",
|
||||
"-cpu-used", "8", "-row-mt", "1", "-c:a", "libopus", "-b:a", "112k"]
|
||||
), ".webm"
|
||||
out = Path(tmp + src.suffix.lower())
|
||||
shutil.copyfile(src, out)
|
||||
return out, src.suffix.lower()
|
||||
|
||||
|
||||
def _image(src: Path, tmp: Path) -> tuple[Path, str]:
|
||||
with Image.open(src) as img:
|
||||
animated = getattr(img, "is_animated", False)
|
||||
if max(img.size) > WEBP_MAX:
|
||||
out = tmp.with_suffix(".png")
|
||||
img.save(out, "PNG", optimize=False)
|
||||
return out, ".png"
|
||||
out = tmp.with_suffix(".webp")
|
||||
if animated:
|
||||
frames = [f.convert("RGBA") for f in ImageSequence.Iterator(img)]
|
||||
frames[0].save(out, "WEBP", save_all=True, append_images=frames[1:], quality=85,
|
||||
duration=img.info.get("duration", 100), loop=0)
|
||||
return out, ".webp"
|
||||
has_alpha = img.mode in ("RGBA", "LA", "PA") or (img.mode == "P" and "transparency" in img.info)
|
||||
img = img.convert("RGBA" if has_alpha else "RGB")
|
||||
img.save(out, "WEBP", quality=88, method=4, alpha_quality=95)
|
||||
return out, ".webp"
|
||||
|
||||
|
||||
def _ffmpeg(src: Path, out: Path, args: list[str]) -> Path:
|
||||
proc = subprocess.run(
|
||||
["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y", "-i", str(src),
|
||||
*args, str(out)],
|
||||
capture_output=True, text=True, timeout=3600,
|
||||
)
|
||||
if proc.returncode != 0:
|
||||
raise RuntimeError(f"ffmpeg: {proc.stderr.strip()[-300:]}")
|
||||
return out
|
||||
"""Media conversion into web-friendly formats."""
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image, ImageSequence
|
||||
|
||||
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif", ".tga", ".avif"}
|
||||
AUDIO_EXTS = {".ogg", ".mp3", ".wav", ".opus", ".flac", ".m4a", ".oga"}
|
||||
VIDEO_EXTS = {".webm", ".ogv", ".mp4", ".mkv", ".avi", ".mpg", ".mpeg", ".m4v"}
|
||||
FONT_EXTS = {".ttf", ".otf", ".woff", ".woff2", ".ttc"}
|
||||
|
||||
WEBP_MAX = 16383
|
||||
|
||||
|
||||
def kind_of(path: Path) -> str | None:
|
||||
ext = path.suffix.lower()
|
||||
if ext in IMAGE_EXTS:
|
||||
return "image"
|
||||
if ext in AUDIO_EXTS:
|
||||
return "audio"
|
||||
if ext in VIDEO_EXTS:
|
||||
return "video"
|
||||
if ext in FONT_EXTS:
|
||||
return "font"
|
||||
return None
|
||||
|
||||
|
||||
def target_ext(path: Path) -> str:
|
||||
kind = kind_of(path)
|
||||
return {"image": ".webp", "audio": ".ogg", "video": ".webm"}.get(kind, path.suffix.lower())
|
||||
|
||||
|
||||
def convert(src: Path, out_dir: Path) -> tuple[Path, str]:
|
||||
"""Convert one file into out_dir. Returns (output file, extension used for the blob)."""
|
||||
kind = kind_of(src)
|
||||
tmp = str(out_dir / uuid.uuid4().hex)
|
||||
if kind == "image":
|
||||
return _image(src, Path(tmp))
|
||||
if kind == "audio":
|
||||
return _ffmpeg(src, Path(tmp + ".ogg"),
|
||||
["-vn", "-c:a", "libopus", "-b:a", "112k", "-vbr", "on"]), ".ogg"
|
||||
if kind == "video":
|
||||
if src.suffix.lower() == ".webm":
|
||||
out = Path(tmp + ".webm")
|
||||
shutil.copyfile(src, out)
|
||||
return out, ".webm"
|
||||
return _ffmpeg(src, Path(tmp + ".webm"),
|
||||
["-c:v", "libvpx-vp9", "-crf", "34", "-b:v", "0", "-deadline", "realtime",
|
||||
"-cpu-used", "8", "-row-mt", "1", "-c:a", "libopus", "-b:a", "112k"]
|
||||
), ".webm"
|
||||
out = Path(tmp + src.suffix.lower())
|
||||
shutil.copyfile(src, out)
|
||||
return out, src.suffix.lower()
|
||||
|
||||
|
||||
def _image(src: Path, tmp: Path) -> tuple[Path, str]:
|
||||
with Image.open(src) as img:
|
||||
animated = getattr(img, "is_animated", False)
|
||||
if max(img.size) > WEBP_MAX:
|
||||
out = tmp.with_suffix(".png")
|
||||
img.save(out, "PNG", optimize=False)
|
||||
return out, ".png"
|
||||
out = tmp.with_suffix(".webp")
|
||||
if animated:
|
||||
frames = [f.convert("RGBA") for f in ImageSequence.Iterator(img)]
|
||||
frames[0].save(out, "WEBP", save_all=True, append_images=frames[1:], quality=85,
|
||||
duration=img.info.get("duration", 100), loop=0)
|
||||
return out, ".webp"
|
||||
has_alpha = img.mode in ("RGBA", "LA", "PA") or (img.mode == "P" and "transparency" in img.info)
|
||||
img = img.convert("RGBA" if has_alpha else "RGB")
|
||||
img.save(out, "WEBP", quality=88, method=4, alpha_quality=95)
|
||||
return out, ".webp"
|
||||
|
||||
|
||||
def _ffmpeg(src: Path, out: Path, args: list[str]) -> Path:
|
||||
proc = subprocess.run(
|
||||
["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y", "-i", str(src),
|
||||
*args, str(out)],
|
||||
capture_output=True, text=True, timeout=3600,
|
||||
)
|
||||
if proc.returncode != 0:
|
||||
raise RuntimeError(f"ffmpeg: {proc.stderr.strip()[-300:]}")
|
||||
return out
|
||||
+189
-189
@@ -1,189 +1,189 @@
|
||||
"""Translate the subset of Python used by visual novels into PGVN IR.
|
||||
|
||||
Expressions become JSON trees ``{"k": kind, ...}``; statements become
|
||||
``{"op": ..., ...}`` nodes executed by the player's JS interpreter.
|
||||
Anything outside the subset raises :class:`Unsupported` so the caller can
|
||||
report it and degrade gracefully.
|
||||
"""
|
||||
|
||||
import ast
|
||||
from typing import Any
|
||||
|
||||
Expr = dict[str, Any]
|
||||
Stmt = dict[str, Any]
|
||||
|
||||
BIN_OPS = {
|
||||
ast.Add: "+", ast.Sub: "-", ast.Mult: "*", ast.Div: "/", ast.FloorDiv: "//",
|
||||
ast.Mod: "%", ast.Pow: "**", ast.BitAnd: "&", ast.BitOr: "|", ast.BitXor: "^",
|
||||
ast.LShift: "<<", ast.RShift: ">>",
|
||||
}
|
||||
CMP_OPS = {
|
||||
ast.Eq: "==", ast.NotEq: "!=", ast.Lt: "<", ast.LtE: "<=", ast.Gt: ">", ast.GtE: ">=",
|
||||
ast.In: "in", ast.NotIn: "not in", ast.Is: "is", ast.IsNot: "is not",
|
||||
}
|
||||
UNARY_OPS = {ast.Not: "not", ast.USub: "-", ast.UAdd: "+", ast.Invert: "~"}
|
||||
|
||||
# Modules whose import is harmless: the player provides the functions it supports.
|
||||
KNOWN_MODULES = {"random", "math", "renpy", "store", "time", "datetime"}
|
||||
|
||||
|
||||
class Unsupported(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def const(v: Any) -> Expr:
|
||||
return {"k": "const", "v": v}
|
||||
|
||||
|
||||
def parse_expr(source: str) -> Expr:
|
||||
try:
|
||||
tree = ast.parse(source.strip(), mode="eval")
|
||||
except SyntaxError as e:
|
||||
raise Unsupported(f"синтаксическая ошибка в выражении: {source!r}") from e
|
||||
return expr(tree.body)
|
||||
|
||||
|
||||
def parse_stmts(source: str) -> list[Stmt]:
|
||||
try:
|
||||
tree = ast.parse(_dedent(source))
|
||||
except SyntaxError as e:
|
||||
raise Unsupported(f"синтаксическая ошибка в python: {e.msg} ({source[:60]!r})") from e
|
||||
return stmts(tree.body)
|
||||
|
||||
|
||||
def _dedent(source: str) -> str:
|
||||
import textwrap
|
||||
return textwrap.dedent(source)
|
||||
|
||||
|
||||
def literal(source: str) -> Any:
|
||||
"""Evaluate a constant expression at conversion time, or raise Unsupported."""
|
||||
try:
|
||||
return ast.literal_eval(source.strip())
|
||||
except (ValueError, SyntaxError, TypeError, MemoryError, RecursionError) as e:
|
||||
raise Unsupported(f"не константа: {source!r}") from e
|
||||
|
||||
|
||||
def expr(node: ast.AST) -> Expr:
|
||||
match node:
|
||||
case ast.Constant(value=v):
|
||||
if isinstance(v, bytes):
|
||||
v = v.decode("utf-8", "replace")
|
||||
if isinstance(v, complex) or v is Ellipsis:
|
||||
raise Unsupported("неподдерживаемая константа")
|
||||
return const(v)
|
||||
case ast.Name(id=name):
|
||||
return {"k": "name", "id": name}
|
||||
case ast.Attribute(value=ast.Name(id="store"), attr=attr):
|
||||
return {"k": "name", "id": attr}
|
||||
case ast.Attribute(value=value, attr=attr):
|
||||
return {"k": "attr", "o": expr(value), "a": attr}
|
||||
case ast.Subscript(value=value, slice=ast.Slice(lower=lo, upper=hi, step=st)):
|
||||
return {"k": "slice", "o": expr(value), "lo": _opt(lo), "hi": _opt(hi), "st": _opt(st)}
|
||||
case ast.Subscript(value=value, slice=index):
|
||||
return {"k": "sub", "o": expr(value), "i": expr(index)}
|
||||
case ast.BinOp(left=l, op=op, right=r):
|
||||
if type(op) not in BIN_OPS:
|
||||
raise Unsupported(f"оператор {type(op).__name__}")
|
||||
return {"k": "bin", "op": BIN_OPS[type(op)], "l": expr(l), "r": expr(r)}
|
||||
case ast.BoolOp(op=op, values=values):
|
||||
return {"k": "bool", "op": "and" if isinstance(op, ast.And) else "or",
|
||||
"v": [expr(v) for v in values]}
|
||||
case ast.UnaryOp(op=op, operand=operand):
|
||||
return {"k": "unary", "op": UNARY_OPS[type(op)], "v": expr(operand)}
|
||||
case ast.Compare(left=l, ops=ops, comparators=rs):
|
||||
return {"k": "cmp", "l": expr(l), "ops": [CMP_OPS[type(o)] for o in ops],
|
||||
"r": [expr(r) for r in rs]}
|
||||
case ast.IfExp(test=t, body=b, orelse=e):
|
||||
return {"k": "if", "c": expr(t), "t": expr(b), "e": expr(e)}
|
||||
case ast.List(elts=elts) | ast.Set(elts=elts):
|
||||
return {"k": "list", "v": [expr(e) for e in elts]}
|
||||
case ast.Tuple(elts=elts):
|
||||
return {"k": "tuple", "v": [expr(e) for e in elts]}
|
||||
case ast.Dict(keys=keys, values=values):
|
||||
if any(k is None for k in keys):
|
||||
raise Unsupported("распаковка ** в словаре")
|
||||
return {"k": "dict", "v": [[expr(k), expr(v)] for k, v in zip(keys, values)]}
|
||||
case ast.Call(func=func, args=args, keywords=kws):
|
||||
if any(isinstance(a, ast.Starred) for a in args) or any(k.arg is None for k in kws):
|
||||
raise Unsupported("распаковка аргументов *args/**kwargs")
|
||||
return {"k": "call", "f": expr(func), "a": [expr(a) for a in args],
|
||||
"kw": {k.arg: expr(k.value) for k in kws}}
|
||||
case ast.JoinedStr(values=values):
|
||||
parts: list[Any] = []
|
||||
for v in values:
|
||||
if isinstance(v, ast.Constant):
|
||||
parts.append(v.value)
|
||||
elif isinstance(v, ast.FormattedValue):
|
||||
parts.append(expr(v.value))
|
||||
return {"k": "fmt", "v": parts}
|
||||
case (ast.ListComp(elt=elt, generators=[ast.comprehension(target=t, iter=it, ifs=ifs,
|
||||
is_async=0)])
|
||||
| ast.GeneratorExp(elt=elt, generators=[ast.comprehension(target=t, iter=it,
|
||||
ifs=ifs, is_async=0)])
|
||||
| ast.SetComp(elt=elt, generators=[ast.comprehension(target=t, iter=it, ifs=ifs,
|
||||
is_async=0)])):
|
||||
return {"k": "comp", "e": expr(elt), "t": target(t), "it": expr(it),
|
||||
"if": [expr(i) for i in ifs]}
|
||||
case ast.Lambda():
|
||||
raise Unsupported("lambda")
|
||||
raise Unsupported(f"выражение {type(node).__name__}")
|
||||
|
||||
|
||||
def _opt(node: ast.AST | None) -> Expr | None:
|
||||
return None if node is None else expr(node)
|
||||
|
||||
|
||||
def target(node: ast.AST) -> Expr:
|
||||
match node:
|
||||
case ast.Name() | ast.Attribute() | ast.Subscript():
|
||||
return expr(node)
|
||||
case ast.Tuple(elts=elts) | ast.List(elts=elts):
|
||||
return {"k": "tuple", "v": [target(e) for e in elts]}
|
||||
raise Unsupported(f"присваивание в {type(node).__name__}")
|
||||
|
||||
|
||||
def stmts(nodes: list[ast.stmt]) -> list[Stmt]:
|
||||
out: list[Stmt] = []
|
||||
for n in nodes:
|
||||
out.extend(stmt(n))
|
||||
return out
|
||||
|
||||
|
||||
def stmt(node: ast.stmt) -> list[Stmt]:
|
||||
match node:
|
||||
case ast.Assign(targets=targets, value=value):
|
||||
v = expr(value)
|
||||
return [{"op": "set", "t": target(t), "v": v} for t in targets]
|
||||
case ast.AnnAssign(target=t, value=value) if value is not None:
|
||||
return [{"op": "set", "t": target(t), "v": expr(value)}]
|
||||
case ast.AugAssign(target=t, op=op, value=value):
|
||||
if type(op) not in BIN_OPS:
|
||||
raise Unsupported(f"оператор {type(op).__name__}")
|
||||
return [{"op": "aug", "t": target(t), "o": BIN_OPS[type(op)], "v": expr(value)}]
|
||||
case ast.Expr(value=ast.Constant()):
|
||||
return [] # docstring / bare literal
|
||||
case ast.Expr(value=value):
|
||||
return [{"op": "eval", "e": expr(value)}]
|
||||
case ast.If(test=test, body=body, orelse=orelse):
|
||||
return [{"op": "if", "c": expr(test), "then": stmts(body), "else": stmts(orelse)}]
|
||||
case ast.While(test=test, body=body, orelse=[]):
|
||||
return [{"op": "while", "c": expr(test), "body": stmts(body)}]
|
||||
case ast.For(target=t, iter=it, body=body, orelse=[]):
|
||||
return [{"op": "for", "t": target(t), "it": expr(it), "body": stmts(body)}]
|
||||
case ast.Delete(targets=targets):
|
||||
return [{"op": "del", "t": target(t)} for t in targets]
|
||||
case ast.Pass() | ast.Global() | ast.Nonlocal():
|
||||
return []
|
||||
case ast.Break():
|
||||
return [{"op": "break"}]
|
||||
case ast.Continue():
|
||||
return [{"op": "continue"}]
|
||||
case ast.Import(names=names) | ast.ImportFrom(names=names):
|
||||
mod = getattr(node, "module", None) or names[0].name
|
||||
if mod.split(".")[0] in KNOWN_MODULES:
|
||||
return []
|
||||
raise Unsupported(f"импорт модуля {mod}")
|
||||
case ast.FunctionDef(name=name) | ast.ClassDef(name=name):
|
||||
raise Unsupported(f"определение {name}()")
|
||||
raise Unsupported(f"оператор {type(node).__name__}")
|
||||
"""Translate the subset of Python used by visual novels into PGVN IR.
|
||||
|
||||
Expressions become JSON trees ``{"k": kind, ...}``; statements become
|
||||
``{"op": ..., ...}`` nodes executed by the player's JS interpreter.
|
||||
Anything outside the subset raises :class:`Unsupported` so the caller can
|
||||
report it and degrade gracefully.
|
||||
"""
|
||||
|
||||
import ast
|
||||
from typing import Any
|
||||
|
||||
Expr = dict[str, Any]
|
||||
Stmt = dict[str, Any]
|
||||
|
||||
BIN_OPS = {
|
||||
ast.Add: "+", ast.Sub: "-", ast.Mult: "*", ast.Div: "/", ast.FloorDiv: "//",
|
||||
ast.Mod: "%", ast.Pow: "**", ast.BitAnd: "&", ast.BitOr: "|", ast.BitXor: "^",
|
||||
ast.LShift: "<<", ast.RShift: ">>",
|
||||
}
|
||||
CMP_OPS = {
|
||||
ast.Eq: "==", ast.NotEq: "!=", ast.Lt: "<", ast.LtE: "<=", ast.Gt: ">", ast.GtE: ">=",
|
||||
ast.In: "in", ast.NotIn: "not in", ast.Is: "is", ast.IsNot: "is not",
|
||||
}
|
||||
UNARY_OPS = {ast.Not: "not", ast.USub: "-", ast.UAdd: "+", ast.Invert: "~"}
|
||||
|
||||
# Modules whose import is harmless: the player provides the functions it supports.
|
||||
KNOWN_MODULES = {"random", "math", "renpy", "store", "time", "datetime"}
|
||||
|
||||
|
||||
class Unsupported(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def const(v: Any) -> Expr:
|
||||
return {"k": "const", "v": v}
|
||||
|
||||
|
||||
def parse_expr(source: str) -> Expr:
|
||||
try:
|
||||
tree = ast.parse(source.strip(), mode="eval")
|
||||
except SyntaxError as e:
|
||||
raise Unsupported(f"синтаксическая ошибка в выражении: {source!r}") from e
|
||||
return expr(tree.body)
|
||||
|
||||
|
||||
def parse_stmts(source: str) -> list[Stmt]:
|
||||
try:
|
||||
tree = ast.parse(_dedent(source))
|
||||
except SyntaxError as e:
|
||||
raise Unsupported(f"синтаксическая ошибка в python: {e.msg} ({source[:60]!r})") from e
|
||||
return stmts(tree.body)
|
||||
|
||||
|
||||
def _dedent(source: str) -> str:
|
||||
import textwrap
|
||||
return textwrap.dedent(source)
|
||||
|
||||
|
||||
def literal(source: str) -> Any:
|
||||
"""Evaluate a constant expression at conversion time, or raise Unsupported."""
|
||||
try:
|
||||
return ast.literal_eval(source.strip())
|
||||
except (ValueError, SyntaxError, TypeError, MemoryError, RecursionError) as e:
|
||||
raise Unsupported(f"не константа: {source!r}") from e
|
||||
|
||||
|
||||
def expr(node: ast.AST) -> Expr:
|
||||
match node:
|
||||
case ast.Constant(value=v):
|
||||
if isinstance(v, bytes):
|
||||
v = v.decode("utf-8", "replace")
|
||||
if isinstance(v, complex) or v is Ellipsis:
|
||||
raise Unsupported("неподдерживаемая константа")
|
||||
return const(v)
|
||||
case ast.Name(id=name):
|
||||
return {"k": "name", "id": name}
|
||||
case ast.Attribute(value=ast.Name(id="store"), attr=attr):
|
||||
return {"k": "name", "id": attr}
|
||||
case ast.Attribute(value=value, attr=attr):
|
||||
return {"k": "attr", "o": expr(value), "a": attr}
|
||||
case ast.Subscript(value=value, slice=ast.Slice(lower=lo, upper=hi, step=st)):
|
||||
return {"k": "slice", "o": expr(value), "lo": _opt(lo), "hi": _opt(hi), "st": _opt(st)}
|
||||
case ast.Subscript(value=value, slice=index):
|
||||
return {"k": "sub", "o": expr(value), "i": expr(index)}
|
||||
case ast.BinOp(left=l, op=op, right=r):
|
||||
if type(op) not in BIN_OPS:
|
||||
raise Unsupported(f"оператор {type(op).__name__}")
|
||||
return {"k": "bin", "op": BIN_OPS[type(op)], "l": expr(l), "r": expr(r)}
|
||||
case ast.BoolOp(op=op, values=values):
|
||||
return {"k": "bool", "op": "and" if isinstance(op, ast.And) else "or",
|
||||
"v": [expr(v) for v in values]}
|
||||
case ast.UnaryOp(op=op, operand=operand):
|
||||
return {"k": "unary", "op": UNARY_OPS[type(op)], "v": expr(operand)}
|
||||
case ast.Compare(left=l, ops=ops, comparators=rs):
|
||||
return {"k": "cmp", "l": expr(l), "ops": [CMP_OPS[type(o)] for o in ops],
|
||||
"r": [expr(r) for r in rs]}
|
||||
case ast.IfExp(test=t, body=b, orelse=e):
|
||||
return {"k": "if", "c": expr(t), "t": expr(b), "e": expr(e)}
|
||||
case ast.List(elts=elts) | ast.Set(elts=elts):
|
||||
return {"k": "list", "v": [expr(e) for e in elts]}
|
||||
case ast.Tuple(elts=elts):
|
||||
return {"k": "tuple", "v": [expr(e) for e in elts]}
|
||||
case ast.Dict(keys=keys, values=values):
|
||||
if any(k is None for k in keys):
|
||||
raise Unsupported("распаковка ** в словаре")
|
||||
return {"k": "dict", "v": [[expr(k), expr(v)] for k, v in zip(keys, values)]}
|
||||
case ast.Call(func=func, args=args, keywords=kws):
|
||||
if any(isinstance(a, ast.Starred) for a in args) or any(k.arg is None for k in kws):
|
||||
raise Unsupported("распаковка аргументов *args/**kwargs")
|
||||
return {"k": "call", "f": expr(func), "a": [expr(a) for a in args],
|
||||
"kw": {k.arg: expr(k.value) for k in kws}}
|
||||
case ast.JoinedStr(values=values):
|
||||
parts: list[Any] = []
|
||||
for v in values:
|
||||
if isinstance(v, ast.Constant):
|
||||
parts.append(v.value)
|
||||
elif isinstance(v, ast.FormattedValue):
|
||||
parts.append(expr(v.value))
|
||||
return {"k": "fmt", "v": parts}
|
||||
case (ast.ListComp(elt=elt, generators=[ast.comprehension(target=t, iter=it, ifs=ifs,
|
||||
is_async=0)])
|
||||
| ast.GeneratorExp(elt=elt, generators=[ast.comprehension(target=t, iter=it,
|
||||
ifs=ifs, is_async=0)])
|
||||
| ast.SetComp(elt=elt, generators=[ast.comprehension(target=t, iter=it, ifs=ifs,
|
||||
is_async=0)])):
|
||||
return {"k": "comp", "e": expr(elt), "t": target(t), "it": expr(it),
|
||||
"if": [expr(i) for i in ifs]}
|
||||
case ast.Lambda():
|
||||
raise Unsupported("lambda")
|
||||
raise Unsupported(f"выражение {type(node).__name__}")
|
||||
|
||||
|
||||
def _opt(node: ast.AST | None) -> Expr | None:
|
||||
return None if node is None else expr(node)
|
||||
|
||||
|
||||
def target(node: ast.AST) -> Expr:
|
||||
match node:
|
||||
case ast.Name() | ast.Attribute() | ast.Subscript():
|
||||
return expr(node)
|
||||
case ast.Tuple(elts=elts) | ast.List(elts=elts):
|
||||
return {"k": "tuple", "v": [target(e) for e in elts]}
|
||||
raise Unsupported(f"присваивание в {type(node).__name__}")
|
||||
|
||||
|
||||
def stmts(nodes: list[ast.stmt]) -> list[Stmt]:
|
||||
out: list[Stmt] = []
|
||||
for n in nodes:
|
||||
out.extend(stmt(n))
|
||||
return out
|
||||
|
||||
|
||||
def stmt(node: ast.stmt) -> list[Stmt]:
|
||||
match node:
|
||||
case ast.Assign(targets=targets, value=value):
|
||||
v = expr(value)
|
||||
return [{"op": "set", "t": target(t), "v": v} for t in targets]
|
||||
case ast.AnnAssign(target=t, value=value) if value is not None:
|
||||
return [{"op": "set", "t": target(t), "v": expr(value)}]
|
||||
case ast.AugAssign(target=t, op=op, value=value):
|
||||
if type(op) not in BIN_OPS:
|
||||
raise Unsupported(f"оператор {type(op).__name__}")
|
||||
return [{"op": "aug", "t": target(t), "o": BIN_OPS[type(op)], "v": expr(value)}]
|
||||
case ast.Expr(value=ast.Constant()):
|
||||
return [] # docstring / bare literal
|
||||
case ast.Expr(value=value):
|
||||
return [{"op": "eval", "e": expr(value)}]
|
||||
case ast.If(test=test, body=body, orelse=orelse):
|
||||
return [{"op": "if", "c": expr(test), "then": stmts(body), "else": stmts(orelse)}]
|
||||
case ast.While(test=test, body=body, orelse=[]):
|
||||
return [{"op": "while", "c": expr(test), "body": stmts(body)}]
|
||||
case ast.For(target=t, iter=it, body=body, orelse=[]):
|
||||
return [{"op": "for", "t": target(t), "it": expr(it), "body": stmts(body)}]
|
||||
case ast.Delete(targets=targets):
|
||||
return [{"op": "del", "t": target(t)} for t in targets]
|
||||
case ast.Pass() | ast.Global() | ast.Nonlocal():
|
||||
return []
|
||||
case ast.Break():
|
||||
return [{"op": "break"}]
|
||||
case ast.Continue():
|
||||
return [{"op": "continue"}]
|
||||
case ast.Import(names=names) | ast.ImportFrom(names=names):
|
||||
mod = getattr(node, "module", None) or names[0].name
|
||||
if mod.split(".")[0] in KNOWN_MODULES:
|
||||
return []
|
||||
raise Unsupported(f"импорт модуля {mod}")
|
||||
case ast.FunctionDef(name=name) | ast.ClassDef(name=name):
|
||||
raise Unsupported(f"определение {name}()")
|
||||
raise Unsupported(f"оператор {type(node).__name__}")
|
||||
+109
-109
@@ -1,109 +1,109 @@
|
||||
"""Reader for Ren'Py .rpa archives (RPA-2.0, RPA-3.x and common variants).
|
||||
|
||||
The index is a zlib-compressed pickle, so it is loaded with a restricted
|
||||
unpickler that only allows plain containers — archives are untrusted input.
|
||||
"""
|
||||
|
||||
import io
|
||||
import pickle
|
||||
import zlib
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class RpaError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class _SafeUnpickler(pickle.Unpickler):
|
||||
_ALLOWED = {("_codecs", "encode"), ("builtins", "set"), ("builtins", "frozenset"),
|
||||
("__builtin__", "set"), ("builtins", "bytearray"), ("__builtin__", "bytearray"),
|
||||
("builtins", "bytes"), ("__builtin__", "bytes")}
|
||||
|
||||
def find_class(self, module, name):
|
||||
if (module, name) in self._ALLOWED:
|
||||
import builtins
|
||||
import codecs
|
||||
return codecs.encode if module == "_codecs" else getattr(builtins, name)
|
||||
raise pickle.UnpicklingError(f"запрещённый тип в индексе RPA: {module}.{name}")
|
||||
|
||||
|
||||
def _text(v) -> str:
|
||||
return v.decode("utf-8", "replace") if isinstance(v, bytes) else str(v)
|
||||
|
||||
|
||||
def read_index(path: Path) -> dict[str, list[tuple[int, int, bytes]]]:
|
||||
with path.open("rb") as f:
|
||||
header = f.readline(200).decode("latin-1").strip()
|
||||
parts = header.split()
|
||||
if not parts or not parts[0].startswith("RPA-"):
|
||||
raise RpaError(f"{path.name}: неизвестный формат архива ({header[:20]!r})")
|
||||
version = parts[0]
|
||||
try:
|
||||
offset = int(parts[1], 16)
|
||||
key = 0
|
||||
if version != "RPA-2.0":
|
||||
for token in parts[2:]:
|
||||
key ^= int(token, 16)
|
||||
except (IndexError, ValueError) as e:
|
||||
raise RpaError(f"{path.name}: повреждён заголовок") from e
|
||||
f.seek(offset)
|
||||
try:
|
||||
raw = _SafeUnpickler(io.BytesIO(zlib.decompress(f.read())), encoding="bytes").load()
|
||||
except (zlib.error, pickle.UnpicklingError, EOFError) as e:
|
||||
raise RpaError(f"{path.name}: не удалось прочитать индекс: {e}") from e
|
||||
|
||||
index: dict[str, list[tuple[int, int, bytes]]] = {}
|
||||
for name, entries in raw.items():
|
||||
fixed = []
|
||||
for entry in entries:
|
||||
off, length = entry[0] ^ key, entry[1] ^ key
|
||||
prefix = entry[2] if len(entry) > 2 else b""
|
||||
if isinstance(prefix, str):
|
||||
prefix = prefix.encode("latin-1")
|
||||
fixed.append((off, length, prefix))
|
||||
index[_text(name)] = fixed
|
||||
return index
|
||||
|
||||
|
||||
def extract_rpa(path: Path, dest: Path, only: Callable[[str], bool] | None = None) -> int:
|
||||
"""Extract files of an archive into dest (all, or those `only` accepts)."""
|
||||
index = read_index(path)
|
||||
root = dest.resolve()
|
||||
count = 0
|
||||
with path.open("rb") as f:
|
||||
for name, entries in index.items():
|
||||
if only is not None and not only(name):
|
||||
continue
|
||||
target = (dest / name).resolve()
|
||||
if not target.is_relative_to(root):
|
||||
continue
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
with target.open("wb") as out:
|
||||
for off, length, prefix in entries:
|
||||
out.write(prefix)
|
||||
f.seek(off)
|
||||
remaining = length - len(prefix)
|
||||
while remaining > 0:
|
||||
chunk = f.read(min(remaining, 1 << 20))
|
||||
if not chunk:
|
||||
break
|
||||
out.write(chunk)
|
||||
remaining -= len(chunk)
|
||||
count += 1
|
||||
return count
|
||||
|
||||
|
||||
def write_rpa(path: Path, files: dict[str, bytes], key: int = 0x42424242) -> None:
|
||||
"""Write an RPA-3.0 archive (used by tests)."""
|
||||
with path.open("wb") as f:
|
||||
f.write(b"RPA-3.0 " + b"0" * 16 + b" " + b"0" * 8 + b"\n")
|
||||
index = {}
|
||||
for name, data in files.items():
|
||||
off = f.tell()
|
||||
f.write(data)
|
||||
index[name] = [(off ^ key, len(data) ^ key, b"")]
|
||||
idx_off = f.tell()
|
||||
f.write(zlib.compress(pickle.dumps(index, protocol=2)))
|
||||
f.seek(0)
|
||||
f.write(f"RPA-3.0 {idx_off:016x} {key:08x}\n".encode())
|
||||
"""Reader for Ren'Py .rpa archives (RPA-2.0, RPA-3.x and common variants).
|
||||
|
||||
The index is a zlib-compressed pickle, so it is loaded with a restricted
|
||||
unpickler that only allows plain containers — archives are untrusted input.
|
||||
"""
|
||||
|
||||
import io
|
||||
import pickle
|
||||
import zlib
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class RpaError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class _SafeUnpickler(pickle.Unpickler):
|
||||
_ALLOWED = {("_codecs", "encode"), ("builtins", "set"), ("builtins", "frozenset"),
|
||||
("__builtin__", "set"), ("builtins", "bytearray"), ("__builtin__", "bytearray"),
|
||||
("builtins", "bytes"), ("__builtin__", "bytes")}
|
||||
|
||||
def find_class(self, module, name):
|
||||
if (module, name) in self._ALLOWED:
|
||||
import builtins
|
||||
import codecs
|
||||
return codecs.encode if module == "_codecs" else getattr(builtins, name)
|
||||
raise pickle.UnpicklingError(f"запрещённый тип в индексе RPA: {module}.{name}")
|
||||
|
||||
|
||||
def _text(v) -> str:
|
||||
return v.decode("utf-8", "replace") if isinstance(v, bytes) else str(v)
|
||||
|
||||
|
||||
def read_index(path: Path) -> dict[str, list[tuple[int, int, bytes]]]:
|
||||
with path.open("rb") as f:
|
||||
header = f.readline(200).decode("latin-1").strip()
|
||||
parts = header.split()
|
||||
if not parts or not parts[0].startswith("RPA-"):
|
||||
raise RpaError(f"{path.name}: неизвестный формат архива ({header[:20]!r})")
|
||||
version = parts[0]
|
||||
try:
|
||||
offset = int(parts[1], 16)
|
||||
key = 0
|
||||
if version != "RPA-2.0":
|
||||
for token in parts[2:]:
|
||||
key ^= int(token, 16)
|
||||
except (IndexError, ValueError) as e:
|
||||
raise RpaError(f"{path.name}: повреждён заголовок") from e
|
||||
f.seek(offset)
|
||||
try:
|
||||
raw = _SafeUnpickler(io.BytesIO(zlib.decompress(f.read())), encoding="bytes").load()
|
||||
except (zlib.error, pickle.UnpicklingError, EOFError) as e:
|
||||
raise RpaError(f"{path.name}: не удалось прочитать индекс: {e}") from e
|
||||
|
||||
index: dict[str, list[tuple[int, int, bytes]]] = {}
|
||||
for name, entries in raw.items():
|
||||
fixed = []
|
||||
for entry in entries:
|
||||
off, length = entry[0] ^ key, entry[1] ^ key
|
||||
prefix = entry[2] if len(entry) > 2 else b""
|
||||
if isinstance(prefix, str):
|
||||
prefix = prefix.encode("latin-1")
|
||||
fixed.append((off, length, prefix))
|
||||
index[_text(name)] = fixed
|
||||
return index
|
||||
|
||||
|
||||
def extract_rpa(path: Path, dest: Path, only: Callable[[str], bool] | None = None) -> int:
|
||||
"""Extract files of an archive into dest (all, or those `only` accepts)."""
|
||||
index = read_index(path)
|
||||
root = dest.resolve()
|
||||
count = 0
|
||||
with path.open("rb") as f:
|
||||
for name, entries in index.items():
|
||||
if only is not None and not only(name):
|
||||
continue
|
||||
target = (dest / name).resolve()
|
||||
if not target.is_relative_to(root):
|
||||
continue
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
with target.open("wb") as out:
|
||||
for off, length, prefix in entries:
|
||||
out.write(prefix)
|
||||
f.seek(off)
|
||||
remaining = length - len(prefix)
|
||||
while remaining > 0:
|
||||
chunk = f.read(min(remaining, 1 << 20))
|
||||
if not chunk:
|
||||
break
|
||||
out.write(chunk)
|
||||
remaining -= len(chunk)
|
||||
count += 1
|
||||
return count
|
||||
|
||||
|
||||
def write_rpa(path: Path, files: dict[str, bytes], key: int = 0x42424242) -> None:
|
||||
"""Write an RPA-3.0 archive (used by tests)."""
|
||||
with path.open("wb") as f:
|
||||
f.write(b"RPA-3.0 " + b"0" * 16 + b" " + b"0" * 8 + b"\n")
|
||||
index = {}
|
||||
for name, data in files.items():
|
||||
off = f.tell()
|
||||
f.write(data)
|
||||
index[name] = [(off ^ key, len(data) ^ key, b"")]
|
||||
idx_off = f.tell()
|
||||
f.write(zlib.compress(pickle.dumps(index, protocol=2)))
|
||||
f.seek(0)
|
||||
f.write(f"RPA-3.0 {idx_off:016x} {key:08x}\n".encode())
|
||||
Reference in new issue
Block a user