| 1 | from __future__ import annotations |
| 2 | |
| 3 | import platform |
| 4 | import shutil |
| 5 | import stat |
| 6 | import subprocess |
| 7 | import tarfile |
| 8 | import tempfile |
| 9 | import zipfile |
| 10 | from pathlib import Path |
| 11 | |
| 12 | import requests |
| 13 | |
| 14 | |
| 15 | GITHUB_RELEASE_API = "https://api.github.com/repos/biliup/biliup/releases/latest" |
| 16 | |
| 17 | |
| 18 | def get_biliup_runtime_root() -> Path: |
| 19 | return Path.home() / ".social-auto-upload" / "tools" / "biliup" |
| 20 | |
| 21 | |
| 22 | def _normalize_system(system_name: str | None = None) -> str: |
| 23 | system_value = (system_name or platform.system()).strip().lower() |
| 24 | if system_value == "darwin": |
| 25 | return "macos" |
| 26 | return system_value |
| 27 | |
| 28 | |
| 29 | def _normalize_machine(machine_name: str | None = None) -> str: |
| 30 | machine_value = (machine_name or platform.machine()).strip().lower() |
| 31 | aliases = { |
| 32 | "amd64": "x86_64", |
| 33 | "x64": "x86_64", |
| 34 | "arm64": "aarch64", |
| 35 | } |
| 36 | return aliases.get(machine_value, machine_value) |
| 37 | |
| 38 | |
| 39 | def _build_platform_key(system_name: str | None = None, machine_name: str | None = None) -> str: |
| 40 | return f"{_normalize_system(system_name)}-{_normalize_machine(machine_name)}" |
| 41 | |
| 42 | |
| 43 | def build_biliup_runtime_path(system_name: str | None = None) -> Path: |
| 44 | executable_name = "biliup.exe" if _normalize_system(system_name) == "windows" else "biliup" |
| 45 | return get_biliup_runtime_root() / _build_platform_key(system_name) / executable_name |
| 46 | |
| 47 | |
| 48 | def _build_biliup_version_path(system_name: str | None = None) -> Path: |
| 49 | return build_biliup_runtime_path(system_name).with_name("version.txt") |
| 50 | |
| 51 | |
| 52 | def _select_release_asset(assets: list[dict]) -> dict: |
| 53 | platform_key = _build_platform_key() |
| 54 | preferred_patterns = { |
| 55 | "windows-x86_64": ("x86_64-windows.zip",), |
| 56 | "linux-x86_64": ("x86_64-linux.tar.xz",), |
| 57 | "linux-aarch64": ("aarch64-linux.tar.xz",), |
| 58 | "linux-arm": ("arm-linux.tar.xz",), |
| 59 | "macos-x86_64": ("x86_64-macos.tar.xz",), |
| 60 | "macos-aarch64": ("aarch64-macos.tar.xz",), |
| 61 | } |
| 62 | patterns = preferred_patterns.get(platform_key) |
| 63 | if not patterns: |
| 64 | raise RuntimeError(f"Unsupported biliup platform: {platform_key}") |
| 65 | |
| 66 | for asset in assets: |
| 67 | asset_name = asset.get("name", "") |
| 68 | if any(pattern in asset_name for pattern in patterns): |
| 69 | return { |
| 70 | "asset_name": asset_name, |
| 71 | "asset_url": asset.get("browser_download_url", ""), |
| 72 | } |
| 73 | |
| 74 | raise RuntimeError(f"No matching biliup release asset found for platform: {platform_key}") |
| 75 | |
| 76 | |
| 77 | def fetch_latest_release() -> dict: |
| 78 | response = requests.get( |
| 79 | GITHUB_RELEASE_API, |
| 80 | headers={ |
| 81 | "Accept": "application/vnd.github+json", |
| 82 | "User-Agent": "social-auto-upload", |
| 83 | }, |
| 84 | timeout=30, |
| 85 | ) |
| 86 | response.raise_for_status() |
| 87 | payload = response.json() |
| 88 | selected_asset = _select_release_asset(payload.get("assets", [])) |
| 89 | return { |
| 90 | "tag_name": payload.get("tag_name", ""), |
| 91 | "asset_name": selected_asset["asset_name"], |
| 92 | "asset_url": selected_asset["asset_url"], |
| 93 | } |
| 94 | |
| 95 | |
| 96 | def read_local_biliup_version() -> str | None: |
| 97 | version_path = _build_biliup_version_path() |
| 98 | if not version_path.exists(): |
| 99 | return None |
| 100 | return version_path.read_text(encoding="utf-8").strip() or None |
| 101 | |
| 102 | |
| 103 | def write_local_biliup_version(version: str) -> None: |
| 104 | version_path = _build_biliup_version_path() |
| 105 | version_path.parent.mkdir(parents=True, exist_ok=True) |
| 106 | version_path.write_text(version, encoding="utf-8") |
| 107 | |
| 108 | |
| 109 | def _pick_executable(extract_root: Path) -> Path: |
| 110 | candidates = [] |
| 111 | for path in extract_root.rglob("*"): |
| 112 | if not path.is_file(): |
| 113 | continue |
| 114 | lower_name = path.name.lower() |
| 115 | if lower_name in {"biliup", "biliup.exe", "biliupr", "biliupr.exe"}: |
| 116 | candidates.append(path) |
| 117 | if not candidates: |
| 118 | raise RuntimeError("Downloaded biliup archive does not contain a runnable executable") |
| 119 | candidates.sort(key=lambda item: len(str(item))) |
| 120 | return candidates[0] |
| 121 | |
| 122 | |
| 123 | def download_biliup_asset(release: dict, destination: Path) -> Path: |
| 124 | destination.parent.mkdir(parents=True, exist_ok=True) |
| 125 | with tempfile.TemporaryDirectory(prefix="biliup-download-") as temp_dir: |
| 126 | temp_root = Path(temp_dir) |
| 127 | archive_path = temp_root / release["asset_name"] |
| 128 | with requests.get(release["asset_url"], stream=True, timeout=120) as response: |
| 129 | response.raise_for_status() |
| 130 | with archive_path.open("wb") as file_obj: |
| 131 | for chunk in response.iter_content(chunk_size=1024 * 1024): |
| 132 | if chunk: |
| 133 | file_obj.write(chunk) |
| 134 | |
| 135 | extract_root = temp_root / "extract" |
| 136 | extract_root.mkdir(parents=True, exist_ok=True) |
| 137 | if archive_path.suffix.lower() == ".zip": |
| 138 | with zipfile.ZipFile(archive_path) as zip_file: |
| 139 | zip_file.extractall(extract_root) |
| 140 | else: |
| 141 | with tarfile.open(archive_path, "r:xz") as tar_file: |
| 142 | tar_file.extractall(extract_root) |
| 143 | |
| 144 | extracted_binary = _pick_executable(extract_root) |
| 145 | temp_binary = destination.with_suffix(f"{destination.suffix}.tmp") |
| 146 | shutil.copy2(extracted_binary, temp_binary) |
| 147 | temp_binary.replace(destination) |
| 148 | if _normalize_system() != "windows": |
| 149 | destination.chmod(destination.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) |
| 150 | return destination |
| 151 | |
| 152 | |
| 153 | def ensure_biliup_binary(force_check: bool = True) -> Path: |
| 154 | binary_path = build_biliup_runtime_path() |
| 155 | local_version = read_local_biliup_version() |
| 156 | |
| 157 | # 默认优先复用本地已存在的 biliup，避免每次执行都去请求 GitHub latest release。 |
| 158 | if binary_path.exists() and not force_check: |
| 159 | return binary_path |
| 160 | |
| 161 | try: |
| 162 | release = fetch_latest_release() |
| 163 | except Exception: |
| 164 | # 如果本地已经有可执行的 biliup，就在 GitHub 限流/网络失败时直接复用本地版本。 |
| 165 | if binary_path.exists(): |
| 166 | return binary_path |
| 167 | raise |
| 168 | |
| 169 | latest_version = release["tag_name"] |
| 170 | if binary_path.exists() and local_version == latest_version: |
| 171 | return binary_path |
| 172 | |
| 173 | download_biliup_asset(release, binary_path) |
| 174 | write_local_biliup_version(latest_version) |
| 175 | return binary_path |
| 176 | |
| 177 | |
| 178 | def run_biliup_command(arguments: list[str], interactive: bool = False) -> subprocess.CompletedProcess[str]: |
| 179 | binary_path = ensure_biliup_binary(force_check=False) |
| 180 | command = [str(binary_path), *arguments] |
| 181 | if interactive: |
| 182 | return subprocess.run(command, check=False) |
| 183 | return subprocess.run( |
| 184 | command, |
| 185 | check=False, |
| 186 | capture_output=True, |
| 187 | text=True, |
| 188 | encoding="utf-8", |
| 189 | errors="replace", |
| 190 | ) |
| 191 |