245 lines
8.1 KiB
Python
245 lines
8.1 KiB
Python
"""硬件温度数据源组件的获取与启动。
|
||
|
||
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}）"
|