"""硬件温度数据源组件的获取与启动。 Windows 用户态没有任何 API 能直接读 CPU 核心温度（需要内核级 MSR 访问）， 因此本模块负责一键获取并启动开源组件 LibreHardwareMonitor（MPL-2.0）： 它由内核驱动读取传感器，再通过 WMI（root\\LibreHardwareMonitor）暴露出来， 温度窗口会自动读取并展示每个核心的温度。 下载策略： 1. 先查官方 GitHub API 拿到最新版本的精确地址与 SHA-256 摘要； 2. 优先直连官方地址；失败则改用公共加速镜像（仅在有官方摘要时使用）； 3. 无论走了哪条通道，都按官方 SHA-256 校验文件完整性后才解压。 组件安装到 %LOCALAPPDATA%\\niumasoftware\\tools\\LibreHardwareMonitor。 """ from __future__ import annotations import ctypes import hashlib import json import os import shutil import tempfile import urllib.request import zipfile from ctypes import wintypes from typing import Any, Callable API_URL = ( "https://api.github.com/repos/LibreHardwareMonitor/LibreHardwareMonitor/releases/latest" ) ASSET_NAME = "LibreHardwareMonitor.zip" LATEST_DOWNLOAD_URL = ( "https://github.com/LibreHardwareMonitor/LibreHardwareMonitor/releases/latest/" f"download/{ASSET_NAME}" ) RELEASE_PAGE = ( "https://github.com/LibreHardwareMonitor/LibreHardwareMonitor/releases/latest" ) # 公共 GitHub 加速前缀：仅用于「有官方摘要可校验」时的兜底通道 MIRROR_PREFIXES = ( "https://ghfast.top/", "https://gh-proxy.com/", "https://ghproxy.net/", ) EXE_NAME = "librehardwaremonitor.exe" PROCESS_NAME = "librehardwaremonitor" _release_cache: dict[str, Any] = {} def tool_dir() -> str: base = os.environ.get("LOCALAPPDATA") or tempfile.gettempdir() return os.path.join(base, "niumasoftware", "tools", "LibreHardwareMonitor") def find_exe() -> str: """返回已安装的 LibreHardwareMonitor.exe 路径，没有则返回空串。""" root = tool_dir() if not os.path.isdir(root): return "" for current, _dirs, files in os.walk(root): for name in files: if name.lower() == EXE_NAME: return os.path.join(current, name) return "" def is_running() -> bool: try: import psutil for proc in psutil.process_iter(["name"]): name = (proc.info.get("name") or "").lower() if name.startswith(PROCESS_NAME): return True except Exception: pass return False def _release_info() -> dict: """查询官方 release 信息：{url, digest, tag}。失败时返回只有 latest 地址的兜底值。""" if _release_cache: return _release_cache info: dict[str, Any] = {"url": LATEST_DOWNLOAD_URL, "digest": "", "tag": ""} try: request = urllib.request.Request( API_URL, headers={"User-Agent": "niumasoftware-temperature", "Accept": "application/vnd.github+json"}, ) with urllib.request.urlopen(request, timeout=12) as resp: data = json.loads(resp.read().decode("utf-8")) for asset in data.get("assets") or []: if str(asset.get("name")) != ASSET_NAME: continue info["url"] = str(asset.get("browser_download_url") or LATEST_DOWNLOAD_URL) digest = str(asset.get("digest") or "") if digest.startswith("sha256:"): info["digest"] = digest.split(":", 1)[1].strip().lower() info["tag"] = str(data.get("tag_name") or "") break except Exception: pass _release_cache.update(info) return _release_cache def _sha256(path: str) -> str: digest = hashlib.sha256() with open(path, "rb") as fh: while True: chunk = fh.read(1 << 20) if not chunk: break digest.update(chunk) return digest.hexdigest() def _download_one( url: str, dest_zip: str, progress: Callable[[int, int], None] | None, is_cancelled: Callable[[], bool] | None, ) -> None: request = urllib.request.Request( url, headers={"User-Agent": "niumasoftware-temperature"} ) with urllib.request.urlopen(request, timeout=30) as resp, open(dest_zip, "wb") as fh: total = int(resp.headers.get("Content-Length") or 0) if progress is not None: progress(0, total) got = 0 while True: if is_cancelled is not None and is_cancelled(): raise RuntimeError("已取消") chunk = resp.read(65536) if not chunk: break fh.write(chunk) got += len(chunk) if progress is not None: progress(got, total) def download( dest_zip: str, progress: Callable[[int, int], None] | None = None, is_cancelled: Callable[[], bool] | None = None, ) -> None: """下载组件压缩包，自动在官方源与镜像之间切换，并按官方摘要校验。""" info = _release_info() official = str(info.get("url") or LATEST_DOWNLOAD_URL) digest = str(info.get("digest") or "") candidates: list[tuple[str, str]] = [("官方源", official)] if digest: # 只有在能按官方 SHA-256 校验时才允许走第三方镜像 for prefix in MIRROR_PREFIXES: candidates.append(("镜像", f"{prefix}{official}")) errors: list[str] = [] for kind, url in candidates: try: _download_one(url, dest_zip, progress, is_cancelled) except Exception as exc: # noqa: BLE001 errors.append(f"{kind}：{exc}") continue if digest: actual = _sha256(dest_zip) if actual != digest: errors.append(f"{kind}：文件校验不一致") continue return raise RuntimeError("；".join(errors) if errors else "下载失败") def install( progress: Callable[[int, int], None] | None = None, is_cancelled: Callable[[], bool] | None = None, ) -> str: """下载并解压组件，返回可执行文件路径。失败抛异常。""" target = tool_dir() if os.path.isdir(target): shutil.rmtree(target, ignore_errors=True) os.makedirs(target, exist_ok=True) fd, temp_zip = tempfile.mkstemp(suffix=".zip", prefix="lhm_") os.close(fd) try: download(temp_zip, progress, is_cancelled) with zipfile.ZipFile(temp_zip) as zf: for member in zf.infolist(): name = member.filename.replace("\\", "/") if name.startswith("/") or ".." in name.split("/"): raise RuntimeError("下载包内容异常，已中止解压") zf.extractall(target) finally: try: os.remove(temp_zip) except OSError: pass exe = find_exe() if not exe: raise RuntimeError("下载包中未找到 LibreHardwareMonitor.exe") return exe def launch(exe_path: str, *, elevated: bool = True) -> tuple[bool, str]: """启动组件；elevated=True 会请求管理员权限（读取硬件传感器必需）。""" if not exe_path or not os.path.exists(exe_path): return False, "未找到可执行文件" try: shell32 = ctypes.WinDLL("shell32", use_last_error=True) shell32.ShellExecuteW.restype = ctypes.c_ssize_t shell32.ShellExecuteW.argtypes = [ wintypes.HWND, wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.LPCWSTR, ctypes.c_int, ] op = "runas" if elevated else "open" result = int( shell32.ShellExecuteW( None, op, exe_path, None, os.path.dirname(exe_path), 1 ) ) except Exception as exc: # noqa: BLE001 return False, str(exc) if result > 32: return True, "" if result == 5: # ERROR_ACCESS_DENIED：用户在 UAC 中点了“否” return False, "已取消管理员授权（读取硬件传感器需要管理员权限）" if result == 2: return False, "文件不存在" return False, f"启动失败（错误码 {result}）"