优化更新
This commit is contained in:
@@ -0,0 +1,244 @@
|
||||
"""硬件温度数据源组件的获取与启动。
|
||||
|
||||
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}）"
|
||||
Reference in New Issue
Block a user