139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
"""Download a release archive and extract it. A complete local file is extracted without downloading again."""
|
|
import tarfile
|
|
import urllib.error
|
|
import urllib.request
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
UA = "LLLL-tools"
|
|
|
|
|
|
def _archive_name(path: Path) -> str:
|
|
name = path.name.lower()
|
|
if name.endswith(".part"):
|
|
name = name[: -len(".part")]
|
|
return name
|
|
|
|
|
|
def archive_complete(path: Path) -> bool:
|
|
if not path.is_file() or path.stat().st_size == 0:
|
|
return False
|
|
name = _archive_name(path)
|
|
try:
|
|
if name.endswith(".zip") or name.endswith(".ipa"):
|
|
with zipfile.ZipFile(path) as archive:
|
|
return bool(archive.infolist())
|
|
if ".tar." in name or name.endswith(".tgz"):
|
|
mode = "r:xz" if name.endswith(".tar.xz") else "r:gz"
|
|
with tarfile.open(path, mode) as archive:
|
|
for _ in archive:
|
|
pass
|
|
return True
|
|
except (OSError, tarfile.TarError, zipfile.BadZipFile):
|
|
return False
|
|
return False
|
|
|
|
|
|
def _total_size(resp, have: int, code: int) -> int | None:
|
|
if code == 206:
|
|
content_range = resp.headers.get("Content-Range")
|
|
if content_range and "/" in content_range:
|
|
total = content_range.rsplit("/", 1)[-1]
|
|
if total.isdigit():
|
|
return int(total)
|
|
length = resp.headers.get("Content-Length")
|
|
if length and length.isdigit():
|
|
return have + int(length)
|
|
return None
|
|
length = resp.headers.get("Content-Length")
|
|
if length and length.isdigit():
|
|
return int(length)
|
|
return None
|
|
|
|
|
|
def ensure_archive(url: str, dest: Path) -> None:
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
part = dest.with_name(dest.name + ".part")
|
|
if archive_complete(dest):
|
|
return
|
|
if archive_complete(part):
|
|
part.replace(dest)
|
|
return
|
|
if dest.exists() and not part.exists():
|
|
dest.replace(part)
|
|
elif dest.exists():
|
|
dest.unlink()
|
|
|
|
have = part.stat().st_size if part.exists() else 0
|
|
headers = {"User-Agent": UA}
|
|
if have:
|
|
headers["Range"] = f"bytes={have}-"
|
|
request = urllib.request.Request(url, headers=headers)
|
|
try:
|
|
response = urllib.request.urlopen(request, timeout=120)
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code == 416 and archive_complete(part):
|
|
part.replace(dest)
|
|
return
|
|
print(f"下载失败：{exc.code}")
|
|
raise SystemExit(2)
|
|
except (urllib.error.URLError, TimeoutError, OSError):
|
|
print("下载未完成")
|
|
raise SystemExit(2)
|
|
|
|
try:
|
|
with response:
|
|
code = getattr(response, "status", 200)
|
|
if code == 200:
|
|
have = 0
|
|
elif code != 206:
|
|
print(f"下载失败：{code}")
|
|
raise SystemExit(2)
|
|
total = _total_size(response, have, code)
|
|
shown = -1
|
|
with part.open("wb" if code == 200 else "ab") as output:
|
|
done = have if code == 206 else 0
|
|
while True:
|
|
chunk = response.read(1024 * 1024)
|
|
if not chunk:
|
|
break
|
|
output.write(chunk)
|
|
done += len(chunk)
|
|
if not total:
|
|
continue
|
|
pct = min(100, done * 100 // total)
|
|
if pct // 10 != shown:
|
|
shown = pct // 10
|
|
print(f"\r{pct}%", end="", flush=True)
|
|
if shown >= 0:
|
|
print()
|
|
except (urllib.error.URLError, TimeoutError, OSError):
|
|
print("下载未完成")
|
|
raise SystemExit(2)
|
|
|
|
if archive_complete(part):
|
|
part.replace(dest)
|
|
return
|
|
if total and part.exists() and part.stat().st_size == total:
|
|
part.unlink(missing_ok=True)
|
|
print("下载内容无法解压")
|
|
raise SystemExit(2)
|
|
print("下载未完成")
|
|
raise SystemExit(2)
|
|
|
|
|
|
def extract_archive(archive: Path, dest: Path) -> None:
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
root = dest.resolve()
|
|
name = archive.name.lower()
|
|
if name.endswith(".zip"):
|
|
with zipfile.ZipFile(archive) as zipped:
|
|
for info in zipped.infolist():
|
|
target = (dest / info.filename).resolve()
|
|
if not target.is_relative_to(root):
|
|
raise SystemExit("压缩包路径不安全")
|
|
zipped.extractall(dest)
|
|
return
|
|
with tarfile.open(archive) as packed:
|
|
packed.extractall(dest, filter="data")
|