forked from anixart-mod/patcher
46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
from plumbum import local, ProcessExecutionError
|
|
from rich.progress import Progress
|
|
from rich.console import Console
|
|
from pathlib import Path
|
|
from typing import List
|
|
import httpx
|
|
import typer
|
|
|
|
|
|
TOOLS = Path("tools")
|
|
ORIGINAL = Path("original")
|
|
MODIFIED = Path("modified")
|
|
DECOMPILED = Path("decompiled")
|
|
PATCHES = Path("patches")
|
|
CONFIGS = Path("configs")
|
|
|
|
|
|
def ensure_dirs():
|
|
for d in [TOOLS, ORIGINAL, MODIFIED, DECOMPILED, PATCHES, CONFIGS]:
|
|
d.mkdir(exist_ok=True)
|
|
|
|
def run(console: Console, cmd: List[str], hide_output=True):
|
|
prog = local[cmd[0]][cmd[1:]]
|
|
try:
|
|
prog() if hide_output else prog & FG # type: ignore [reportUndefinedVariable]
|
|
except ProcessExecutionError as e:
|
|
console.print(f"[red]Ошибка при выполнении команды: {' '.join(cmd)}")
|
|
console.print(e.stderr)
|
|
raise typer.Exit(1)
|
|
|
|
def download(console: Console, url: str, dest: Path):
|
|
console.print(f"[cyan]Скачивание {url} → {dest.name}")
|
|
|
|
with httpx.Client(follow_redirects=True, timeout=60.0) as client:
|
|
with client.stream("GET", url) as response:
|
|
response.raise_for_status()
|
|
total = int(response.headers.get("Content-Length", 0))
|
|
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with open(dest, "wb") as f, Progress(console=console) as progress:
|
|
task = progress.add_task("Загрузка", total=total if total else None)
|
|
for chunk in response.iter_bytes(chunk_size=8192):
|
|
f.write(chunk)
|
|
progress.update(task, advance=len(chunk))
|