diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/tests/test_bilibili_runtime.py b/tests/test_bilibili_runtime.py new file mode 100644 index 0000000..02d716f --- /dev/null +++ b/tests/test_bilibili_runtime.py @@ -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) diff --git a/uploader/bilibili_uploader/runtime.py b/uploader/bilibili_uploader/runtime.py new file mode 100644 index 0000000..4d782f8 --- /dev/null +++ b/uploader/bilibili_uploader/runtime.py @@ -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, + )