mirror of
https://github.com/dreammis/social-auto-upload.git
synced 2026-08-28 17:43:28 +08:00
feat: add biliup runtime bootstrap
This commit is contained in:
@@ -0,0 +1 @@
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from uploader.bilibili_uploader.runtime import (
|
||||
build_biliup_runtime_path,
|
||||
ensure_biliup_binary,
|
||||
run_biliup_command,
|
||||
)
|
||||
|
||||
|
||||
class BiliupRuntimeTests(unittest.TestCase):
|
||||
def test_build_biliup_runtime_path_returns_platform_path(self):
|
||||
path = build_biliup_runtime_path("Windows")
|
||||
self.assertTrue(str(path).endswith("biliup.exe"))
|
||||
|
||||
@patch("uploader.bilibili_uploader.runtime.fetch_latest_release")
|
||||
def test_ensure_biliup_binary_downloads_when_missing(self, mock_release):
|
||||
mock_release.return_value = {
|
||||
"tag_name": "v1.0.0",
|
||||
"asset_url": "https://example.invalid/biliup.exe",
|
||||
"asset_name": "biliup.exe",
|
||||
}
|
||||
with patch("uploader.bilibili_uploader.runtime.download_biliup_asset") as mock_download:
|
||||
ensure_biliup_binary(force_check=True)
|
||||
mock_download.assert_called_once()
|
||||
|
||||
@patch("uploader.bilibili_uploader.runtime.fetch_latest_release")
|
||||
def test_ensure_biliup_binary_reuses_local_when_up_to_date(self, mock_release):
|
||||
mock_release.return_value = {
|
||||
"tag_name": "v1.0.0",
|
||||
"asset_url": "https://example.invalid/biliup.exe",
|
||||
"asset_name": "biliup.exe",
|
||||
}
|
||||
with patch("uploader.bilibili_uploader.runtime.read_local_biliup_version", return_value="v1.0.0"):
|
||||
with patch("pathlib.Path.exists", return_value=True):
|
||||
with patch("uploader.bilibili_uploader.runtime.download_biliup_asset") as mock_download:
|
||||
ensure_biliup_binary(force_check=True)
|
||||
mock_download.assert_not_called()
|
||||
|
||||
@patch("uploader.bilibili_uploader.runtime.subprocess.run")
|
||||
@patch("uploader.bilibili_uploader.runtime.ensure_biliup_binary")
|
||||
def test_run_biliup_command_returns_completed_process(self, mock_ensure_binary, mock_run):
|
||||
mock_ensure_binary.return_value = build_biliup_runtime_path("Windows")
|
||||
mock_run.return_value = Mock(returncode=0, stdout="ok", stderr="")
|
||||
result = run_biliup_command(["login"])
|
||||
self.assertEqual(result.returncode, 0)
|
||||
@@ -0,0 +1,174 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import platform
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import tarfile
|
||||
import tempfile
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
GITHUB_RELEASE_API = "https://api.github.com/repos/biliup/biliup/releases/latest"
|
||||
|
||||
|
||||
def get_biliup_runtime_root() -> Path:
|
||||
return Path.home() / ".social-auto-upload" / "tools" / "biliup"
|
||||
|
||||
|
||||
def _normalize_system(system_name: str | None = None) -> str:
|
||||
system_value = (system_name or platform.system()).strip().lower()
|
||||
if system_value == "darwin":
|
||||
return "macos"
|
||||
return system_value
|
||||
|
||||
|
||||
def _normalize_machine(machine_name: str | None = None) -> str:
|
||||
machine_value = (machine_name or platform.machine()).strip().lower()
|
||||
aliases = {
|
||||
"amd64": "x86_64",
|
||||
"x64": "x86_64",
|
||||
"arm64": "aarch64",
|
||||
}
|
||||
return aliases.get(machine_value, machine_value)
|
||||
|
||||
|
||||
def _build_platform_key(system_name: str | None = None, machine_name: str | None = None) -> str:
|
||||
return f"{_normalize_system(system_name)}-{_normalize_machine(machine_name)}"
|
||||
|
||||
|
||||
def build_biliup_runtime_path(system_name: str | None = None) -> Path:
|
||||
executable_name = "biliup.exe" if _normalize_system(system_name) == "windows" else "biliup"
|
||||
return get_biliup_runtime_root() / _build_platform_key(system_name) / executable_name
|
||||
|
||||
|
||||
def _build_biliup_version_path(system_name: str | None = None) -> Path:
|
||||
return build_biliup_runtime_path(system_name).with_name("version.txt")
|
||||
|
||||
|
||||
def _select_release_asset(assets: list[dict]) -> dict:
|
||||
platform_key = _build_platform_key()
|
||||
preferred_patterns = {
|
||||
"windows-x86_64": ("x86_64-windows.zip",),
|
||||
"linux-x86_64": ("x86_64-linux.tar.xz",),
|
||||
"linux-aarch64": ("aarch64-linux.tar.xz",),
|
||||
"linux-arm": ("arm-linux.tar.xz",),
|
||||
"macos-x86_64": ("x86_64-macos.tar.xz",),
|
||||
"macos-aarch64": ("aarch64-macos.tar.xz",),
|
||||
}
|
||||
patterns = preferred_patterns.get(platform_key)
|
||||
if not patterns:
|
||||
raise RuntimeError(f"Unsupported biliup platform: {platform_key}")
|
||||
|
||||
for asset in assets:
|
||||
asset_name = asset.get("name", "")
|
||||
if any(pattern in asset_name for pattern in patterns):
|
||||
return {
|
||||
"asset_name": asset_name,
|
||||
"asset_url": asset.get("browser_download_url", ""),
|
||||
}
|
||||
|
||||
raise RuntimeError(f"No matching biliup release asset found for platform: {platform_key}")
|
||||
|
||||
|
||||
def fetch_latest_release() -> dict:
|
||||
response = requests.get(
|
||||
GITHUB_RELEASE_API,
|
||||
headers={
|
||||
"Accept": "application/vnd.github+json",
|
||||
"User-Agent": "social-auto-upload",
|
||||
},
|
||||
timeout=30,
|
||||
)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
selected_asset = _select_release_asset(payload.get("assets", []))
|
||||
return {
|
||||
"tag_name": payload.get("tag_name", ""),
|
||||
"asset_name": selected_asset["asset_name"],
|
||||
"asset_url": selected_asset["asset_url"],
|
||||
}
|
||||
|
||||
|
||||
def read_local_biliup_version() -> str | None:
|
||||
version_path = _build_biliup_version_path()
|
||||
if not version_path.exists():
|
||||
return None
|
||||
return version_path.read_text(encoding="utf-8").strip() or None
|
||||
|
||||
|
||||
def write_local_biliup_version(version: str) -> None:
|
||||
version_path = _build_biliup_version_path()
|
||||
version_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
version_path.write_text(version, encoding="utf-8")
|
||||
|
||||
|
||||
def _pick_executable(extract_root: Path) -> Path:
|
||||
candidates = []
|
||||
for path in extract_root.rglob("*"):
|
||||
if not path.is_file():
|
||||
continue
|
||||
lower_name = path.name.lower()
|
||||
if lower_name in {"biliup", "biliup.exe", "biliupr", "biliupr.exe"}:
|
||||
candidates.append(path)
|
||||
if not candidates:
|
||||
raise RuntimeError("Downloaded biliup archive does not contain a runnable executable")
|
||||
candidates.sort(key=lambda item: len(str(item)))
|
||||
return candidates[0]
|
||||
|
||||
|
||||
def download_biliup_asset(release: dict, destination: Path) -> Path:
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
with tempfile.TemporaryDirectory(prefix="biliup-download-") as temp_dir:
|
||||
temp_root = Path(temp_dir)
|
||||
archive_path = temp_root / release["asset_name"]
|
||||
with requests.get(release["asset_url"], stream=True, timeout=120) as response:
|
||||
response.raise_for_status()
|
||||
with archive_path.open("wb") as file_obj:
|
||||
for chunk in response.iter_content(chunk_size=1024 * 1024):
|
||||
if chunk:
|
||||
file_obj.write(chunk)
|
||||
|
||||
extract_root = temp_root / "extract"
|
||||
extract_root.mkdir(parents=True, exist_ok=True)
|
||||
if archive_path.suffix.lower() == ".zip":
|
||||
with zipfile.ZipFile(archive_path) as zip_file:
|
||||
zip_file.extractall(extract_root)
|
||||
else:
|
||||
with tarfile.open(archive_path, "r:xz") as tar_file:
|
||||
tar_file.extractall(extract_root)
|
||||
|
||||
extracted_binary = _pick_executable(extract_root)
|
||||
shutil.copy2(extracted_binary, destination)
|
||||
if _normalize_system() != "windows":
|
||||
destination.chmod(destination.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
|
||||
return destination
|
||||
|
||||
|
||||
def ensure_biliup_binary(force_check: bool = True) -> Path:
|
||||
binary_path = build_biliup_runtime_path()
|
||||
local_version = read_local_biliup_version()
|
||||
if binary_path.exists() and local_version and not force_check:
|
||||
return binary_path
|
||||
|
||||
release = fetch_latest_release()
|
||||
latest_version = release["tag_name"]
|
||||
if binary_path.exists() and local_version == latest_version:
|
||||
return binary_path
|
||||
|
||||
download_biliup_asset(release, binary_path)
|
||||
write_local_biliup_version(latest_version)
|
||||
return binary_path
|
||||
|
||||
|
||||
def run_biliup_command(arguments: list[str]) -> subprocess.CompletedProcess[str]:
|
||||
binary_path = ensure_biliup_binary(force_check=True)
|
||||
return subprocess.run(
|
||||
[str(binary_path), *arguments],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
Reference in New Issue
Block a user