初始化奇妙小屏幕控制器项目
This commit is contained in:
@@ -0,0 +1,4 @@
|
||||
"""核桃派镜像配置编辑器。"""
|
||||
|
||||
__version__ = "2.0.0"
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
from .cli import run_command
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if len(sys.argv) > 1:
|
||||
return run_command(sys.argv[1:])
|
||||
from .gui import run_gui
|
||||
|
||||
return run_gui()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
from .config import deserialize_config
|
||||
from .operations import copy_write_and_verify, read_config
|
||||
|
||||
|
||||
USAGE = "用法：镜像配置编辑器.exe [--validate 镜像 | --apply 原镜像 配置.json 新镜像]"
|
||||
|
||||
|
||||
def run_command(arguments: list[str]) -> int:
|
||||
try:
|
||||
if len(arguments) == 2 and arguments[0] == "--validate":
|
||||
config = read_config(arguments[1])
|
||||
public = {
|
||||
"valid": True,
|
||||
"SoftwareVersion": config["software_version"],
|
||||
"Username": config["account"]["username"],
|
||||
"Ssid": config["wifi"]["ssid"],
|
||||
"Mode": config["ipv4"]["mode"],
|
||||
"Address": config["ipv4"]["address"],
|
||||
}
|
||||
# ASCII escapes match the 1.0.1 System.Text.Json output and remain stable in every Windows console code page.
|
||||
print(json.dumps(public, ensure_ascii=True, separators=(",", ":")))
|
||||
return 0
|
||||
if len(arguments) == 4 and arguments[0] == "--apply":
|
||||
try:
|
||||
text = Path(arguments[2]).read_text(encoding="utf-8-sig")
|
||||
except (OSError, UnicodeError) as exception:
|
||||
raise ValueError(f"无法读取配置 JSON：{exception}") from exception
|
||||
config = deserialize_config(text)
|
||||
destination, _verified = copy_write_and_verify(arguments[1], arguments[3], config)
|
||||
print(destination)
|
||||
return 0
|
||||
print(USAGE, file=sys.stderr)
|
||||
return 2
|
||||
except Exception as exception:
|
||||
print(str(exception), file=sys.stderr)
|
||||
return 1
|
||||
@@ -0,0 +1,200 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import ipaddress
|
||||
import json
|
||||
import re
|
||||
import struct
|
||||
from typing import Any
|
||||
|
||||
|
||||
PRODUCT_ID = "matrix-screen-controller-walnutpi"
|
||||
CONFIG_FILE_BYTES = 64 * 1024
|
||||
HEADER_BYTES = 4096
|
||||
SLOT_BYTES = (CONFIG_FILE_BYTES - HEADER_BYTES) // 2
|
||||
FILE_MAGIC = b"MSCCFG2\0"
|
||||
SLOT_MAGIC = b"MSCSLOT\0"
|
||||
FORMAT_VERSION = 1
|
||||
_HEADER = struct.Struct("<8sII")
|
||||
_SLOT_HEADER = struct.Struct("<8sQI32s")
|
||||
_USERNAME = re.compile(r"[a-z_][a-z0-9_-]{0,31}", re.ASCII)
|
||||
_VERSION = re.compile(r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)", re.ASCII)
|
||||
|
||||
|
||||
class ImageConfigError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def _exact_object(value: Any, fields: set[str], label: str) -> dict[str, Any]:
|
||||
if not isinstance(value, dict) or set(value) != fields:
|
||||
raise ImageConfigError(f"{label}字段无效")
|
||||
return value
|
||||
|
||||
|
||||
def _utf16_units(value: str) -> int:
|
||||
# .NET String.Length counts UTF-16 code units; retain 1.0.1 validation semantics.
|
||||
return len(value.encode("utf-16-le")) // 2
|
||||
|
||||
|
||||
def _validate_secret(value: Any, label: str, minimum: int, maximum: int) -> str:
|
||||
if not isinstance(value, str) or not minimum <= _utf16_units(value) <= maximum:
|
||||
raise ImageConfigError(f"{label}长度或字符无效")
|
||||
if any(character in value for character in ("\0", "\r", "\n")):
|
||||
raise ImageConfigError(f"{label}长度或字符无效")
|
||||
return value
|
||||
|
||||
|
||||
def _validate_text(value: Any, label: str, minimum: int, maximum_bytes: int) -> str:
|
||||
if not isinstance(value, str) or _utf16_units(value) < minimum or len(value.encode("utf-8")) > maximum_bytes:
|
||||
raise ImageConfigError(f"{label}长度或字符无效")
|
||||
if any(character in value for character in ("\0", "\r", "\n")):
|
||||
raise ImageConfigError(f"{label}长度或字符无效")
|
||||
return value
|
||||
|
||||
|
||||
def validate_config(value: Any) -> dict[str, Any]:
|
||||
source = _exact_object(
|
||||
value,
|
||||
{"schema_version", "product", "software_version", "account", "wifi", "ipv4"},
|
||||
"镜像配置",
|
||||
)
|
||||
schema = source["schema_version"]
|
||||
if type(schema) is not int or schema != FORMAT_VERSION or source["product"] != PRODUCT_ID:
|
||||
raise ImageConfigError("镜像产品或配置版本不受支持")
|
||||
software_version = source["software_version"]
|
||||
if not isinstance(software_version, str) or _VERSION.fullmatch(software_version) is None:
|
||||
raise ImageConfigError("软件版本格式无效")
|
||||
|
||||
account = _exact_object(source["account"], {"username", "password"}, "账户配置")
|
||||
username = account["username"]
|
||||
if not isinstance(username, str) or _USERNAME.fullmatch(username) is None:
|
||||
raise ImageConfigError("用户名只能使用小写字母、数字、下划线和连字符，且最长 32 个字符")
|
||||
account_password = _validate_secret(account["password"], "账户密码", 8, 128)
|
||||
|
||||
wifi = _exact_object(source["wifi"], {"ssid", "password"}, "Wi-Fi 配置")
|
||||
ssid = _validate_text(wifi["ssid"], "Wi-Fi 名称", 1, 32)
|
||||
wifi_password = _validate_secret(wifi["password"], "Wi-Fi 密码", 8, 63)
|
||||
|
||||
ipv4 = _exact_object(source["ipv4"], {"mode", "address", "prefix", "gateway", "dns"}, "IPv4 配置")
|
||||
mode = ipv4["mode"]
|
||||
if mode == "dhcp":
|
||||
normalized_ipv4: dict[str, Any] = {"mode": "dhcp", "address": "", "prefix": 0, "gateway": "", "dns": []}
|
||||
elif mode == "static":
|
||||
try:
|
||||
address = ipaddress.IPv4Address(ipv4["address"])
|
||||
gateway = ipaddress.IPv4Address(ipv4["gateway"])
|
||||
except (ipaddress.AddressValueError, TypeError) as exception:
|
||||
raise ImageConfigError("静态 IPv4 地址或网关无效") from exception
|
||||
prefix = ipv4["prefix"]
|
||||
if type(prefix) is not int or not 1 <= prefix <= 32:
|
||||
raise ImageConfigError("IPv4 前缀必须是 1 到 32")
|
||||
if gateway not in ipaddress.IPv4Network(f"{address}/{prefix}", strict=False):
|
||||
raise ImageConfigError("静态 IPv4 网关必须与地址处于同一子网")
|
||||
dns_source = ipv4["dns"]
|
||||
if not isinstance(dns_source, list) or not 1 <= len(dns_source) <= 4:
|
||||
raise ImageConfigError("DNS 必须包含 1 到 4 个 IPv4 地址")
|
||||
try:
|
||||
dns = [str(ipaddress.IPv4Address(item)) for item in dns_source]
|
||||
except (ipaddress.AddressValueError, TypeError) as exception:
|
||||
raise ImageConfigError("DNS 必须包含 1 到 4 个 IPv4 地址") from exception
|
||||
normalized_ipv4 = {
|
||||
"mode": "static",
|
||||
"address": str(ipv4["address"]),
|
||||
"prefix": prefix,
|
||||
"gateway": str(ipv4["gateway"]),
|
||||
"dns": dns,
|
||||
}
|
||||
else:
|
||||
raise ImageConfigError("IPv4 模式无效")
|
||||
|
||||
return {
|
||||
"schema_version": FORMAT_VERSION,
|
||||
"product": PRODUCT_ID,
|
||||
"software_version": software_version,
|
||||
"account": {"username": username, "password": account_password},
|
||||
"wifi": {"ssid": ssid, "password": wifi_password},
|
||||
"ipv4": normalized_ipv4,
|
||||
}
|
||||
|
||||
|
||||
def serialize_config(value: Any) -> str:
|
||||
return json.dumps(validate_config(value), ensure_ascii=False, separators=(",", ":"))
|
||||
|
||||
|
||||
def deserialize_config(text: str) -> dict[str, Any]:
|
||||
try:
|
||||
value = json.loads(text)
|
||||
except (UnicodeError, json.JSONDecodeError) as exception:
|
||||
raise ImageConfigError("配置 JSON 无效") from exception
|
||||
return validate_config(value)
|
||||
|
||||
|
||||
def _slot_offset(index: int) -> int:
|
||||
return HEADER_BYTES + index * SLOT_BYTES
|
||||
|
||||
|
||||
def _encode_slot(config: Any, generation: int) -> bytes:
|
||||
payload = (serialize_config(config) + "\n").encode("utf-8")
|
||||
if len(payload) > SLOT_BYTES - _SLOT_HEADER.size:
|
||||
raise ImageConfigError("配置内容过大")
|
||||
header = _SLOT_HEADER.pack(SLOT_MAGIC, generation, len(payload), hashlib.sha256(payload).digest())
|
||||
return header + payload + bytes(SLOT_BYTES - len(header) - len(payload))
|
||||
|
||||
|
||||
def create_config_file(config: Any) -> bytes:
|
||||
result = bytearray(CONFIG_FILE_BYTES)
|
||||
_HEADER.pack_into(result, 0, FILE_MAGIC, FORMAT_VERSION, CONFIG_FILE_BYTES)
|
||||
result[_slot_offset(0) : _slot_offset(0) + SLOT_BYTES] = _encode_slot(config, 1)
|
||||
return bytes(result)
|
||||
|
||||
|
||||
def read_config_file(data: bytes) -> tuple[dict[str, Any], int, int]:
|
||||
if len(data) != CONFIG_FILE_BYTES:
|
||||
raise ImageConfigError("镜像配置文件大小无效")
|
||||
try:
|
||||
magic, version, declared_size = _HEADER.unpack_from(data, 0)
|
||||
except struct.error as exception:
|
||||
raise ImageConfigError("镜像配置区头部无效") from exception
|
||||
if magic != FILE_MAGIC or version != FORMAT_VERSION or declared_size != CONFIG_FILE_BYTES:
|
||||
raise ImageConfigError("镜像配置区头部无效")
|
||||
|
||||
valid: list[tuple[int, int, dict[str, Any]]] = []
|
||||
for index in range(2):
|
||||
offset = _slot_offset(index)
|
||||
slot_magic, generation, payload_bytes, digest = _SLOT_HEADER.unpack_from(data, offset)
|
||||
if slot_magic != SLOT_MAGIC or payload_bytes <= 0 or payload_bytes > SLOT_BYTES - _SLOT_HEADER.size:
|
||||
continue
|
||||
payload_start = offset + _SLOT_HEADER.size
|
||||
payload = data[payload_start : payload_start + payload_bytes]
|
||||
if hashlib.sha256(payload).digest() != digest:
|
||||
continue
|
||||
try:
|
||||
document = deserialize_config(payload.decode("utf-8"))
|
||||
except (UnicodeError, ImageConfigError):
|
||||
continue
|
||||
valid.append((generation, index, document))
|
||||
if not valid:
|
||||
raise ImageConfigError("镜像配置区没有可恢复的有效副本")
|
||||
generation, index, document = max(valid, key=lambda item: (item[0], item[1]))
|
||||
return document, generation, index
|
||||
|
||||
|
||||
def encode_update(data: bytes, config: Any) -> tuple[bytes, int]:
|
||||
_current, generation, active = read_config_file(data)
|
||||
target = 1 - active
|
||||
return _encode_slot(config, generation + 1), target
|
||||
|
||||
|
||||
def update_config_file(data: bytes, config: Any) -> bytes:
|
||||
slot_data, target = encode_update(data, config)
|
||||
result = bytearray(data)
|
||||
offset = _slot_offset(target)
|
||||
result[offset : offset + SLOT_BYTES] = slot_data
|
||||
return bytes(result)
|
||||
|
||||
|
||||
def slot_offset(index: int) -> int:
|
||||
if index not in (0, 1):
|
||||
raise ValueError("配置槽索引无效")
|
||||
return _slot_offset(index)
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import os
|
||||
from pathlib import Path
|
||||
import struct
|
||||
from typing import BinaryIO, Iterator
|
||||
|
||||
|
||||
class Fat16Error(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FatEntry:
|
||||
first_cluster: int
|
||||
size: int
|
||||
|
||||
|
||||
class Fat16Disk:
|
||||
_SUPPORTED_PARTITION_TYPES = {0x04, 0x06, 0x0B, 0x0C, 0x0E}
|
||||
|
||||
def __init__(self, path: str | os.PathLike[str], *, writable: bool) -> None:
|
||||
self.path = Path(path)
|
||||
self._stream: BinaryIO = self.path.open("r+b" if writable else "rb", buffering=1024 * 1024)
|
||||
try:
|
||||
self._initialize()
|
||||
except Exception:
|
||||
self._stream.close()
|
||||
raise
|
||||
|
||||
def _initialize(self) -> None:
|
||||
self._stream.seek(0, os.SEEK_END)
|
||||
self._image_bytes = self._stream.tell()
|
||||
mbr = self._read_at(0, 512)
|
||||
if mbr[510:512] != b"\x55\xaa":
|
||||
raise Fat16Error("文件没有有效的 MBR 分区表")
|
||||
partition_type = mbr[450]
|
||||
partition_lba, partition_sectors = struct.unpack_from("<II", mbr, 454)
|
||||
if partition_type not in self._SUPPORTED_PARTITION_TYPES or partition_lba == 0 or partition_sectors == 0:
|
||||
raise Fat16Error("镜像第一分区不是受支持的 FAT 启动分区")
|
||||
self._partition_offset = partition_lba * 512
|
||||
self._partition_bytes = partition_sectors * 512
|
||||
self._partition_end = self._partition_offset + self._partition_bytes
|
||||
if self._partition_end > self._image_bytes:
|
||||
raise Fat16Error("镜像分区表超出文件边界")
|
||||
|
||||
bpb = self._read_at(self._partition_offset, 512)
|
||||
bytes_per_sector = struct.unpack_from("<H", bpb, 11)[0]
|
||||
sectors_per_cluster = bpb[13]
|
||||
reserved = struct.unpack_from("<H", bpb, 14)[0]
|
||||
fat_count = bpb[16]
|
||||
self._root_entries = struct.unpack_from("<H", bpb, 17)[0]
|
||||
fat_sectors = struct.unpack_from("<H", bpb, 22)[0]
|
||||
if (
|
||||
bytes_per_sector != 512
|
||||
or sectors_per_cluster == 0
|
||||
or reserved == 0
|
||||
or fat_count == 0
|
||||
or fat_sectors == 0
|
||||
or self._root_entries == 0
|
||||
):
|
||||
raise Fat16Error("镜像第一分区不是受支持的 FAT16")
|
||||
self._cluster_bytes = bytes_per_sector * sectors_per_cluster
|
||||
self._fat_bytes = fat_sectors * bytes_per_sector
|
||||
self._fat_offset = self._partition_offset + reserved * bytes_per_sector
|
||||
self._root_offset = self._fat_offset + fat_count * self._fat_bytes
|
||||
root_bytes = self._root_entries * 32
|
||||
root_sectors = (root_bytes + bytes_per_sector - 1) // bytes_per_sector
|
||||
self._data_offset = self._root_offset + root_sectors * bytes_per_sector
|
||||
if self._data_offset > self._partition_end or self._fat_offset + self._fat_bytes > self._partition_end:
|
||||
raise Fat16Error("FAT16 布局超出分区边界")
|
||||
|
||||
def __enter__(self) -> "Fat16Disk":
|
||||
return self
|
||||
|
||||
def __exit__(self, *_args: object) -> None:
|
||||
self.close()
|
||||
|
||||
def close(self) -> None:
|
||||
self._stream.close()
|
||||
|
||||
def _read_at(self, offset: int, count: int) -> bytes:
|
||||
if offset < 0 or count < 0 or offset + count > self._image_bytes:
|
||||
raise Fat16Error("镜像数据超出文件边界")
|
||||
self._stream.seek(offset)
|
||||
value = self._stream.read(count)
|
||||
if len(value) != count:
|
||||
raise Fat16Error("镜像数据被截断")
|
||||
return value
|
||||
|
||||
@staticmethod
|
||||
def _name83(name: str) -> bytes:
|
||||
parts = name.upper().split(".", 1)
|
||||
if not 1 <= len(parts[0]) <= 8 or (len(parts) == 2 and len(parts[1]) > 3):
|
||||
raise ValueError("文件名不是 8.3 格式")
|
||||
try:
|
||||
return (parts[0].ljust(8) + (parts[1] if len(parts) == 2 else "").ljust(3)).encode("ascii")
|
||||
except UnicodeEncodeError as exception:
|
||||
raise ValueError("文件名不是 ASCII 8.3 格式") from exception
|
||||
|
||||
def _find(self, name: str) -> FatEntry:
|
||||
expected = self._name83(name)
|
||||
root = self._read_at(self._root_offset, self._root_entries * 32)
|
||||
for index in range(self._root_entries):
|
||||
entry = root[index * 32 : (index + 1) * 32]
|
||||
if entry[0] == 0x00:
|
||||
break
|
||||
if entry[0] == 0xE5 or entry[11] == 0x0F:
|
||||
continue
|
||||
if entry[:11] == expected:
|
||||
first_cluster = struct.unpack_from("<H", entry, 26)[0]
|
||||
size = struct.unpack_from("<I", entry, 28)[0]
|
||||
return FatEntry(first_cluster, size)
|
||||
raise Fat16Error(f"镜像缺少 {name}")
|
||||
|
||||
def _chain(self, first_cluster: int, fat: bytes) -> Iterator[int]:
|
||||
seen: set[int] = set()
|
||||
cluster = first_cluster
|
||||
while 2 <= cluster < 0xFFF8:
|
||||
if cluster in seen:
|
||||
raise Fat16Error("FAT16 簇链循环")
|
||||
if cluster * 2 + 2 > len(fat):
|
||||
raise Fat16Error("FAT16 簇号超出 FAT 边界")
|
||||
offset = self._data_offset + (cluster - 2) * self._cluster_bytes
|
||||
if offset + self._cluster_bytes > self._partition_end:
|
||||
raise Fat16Error("FAT16 簇超出分区边界")
|
||||
seen.add(cluster)
|
||||
yield cluster
|
||||
cluster = struct.unpack_from("<H", fat, cluster * 2)[0]
|
||||
|
||||
def _cluster_offset(self, cluster: int) -> int:
|
||||
return self._data_offset + (cluster - 2) * self._cluster_bytes
|
||||
|
||||
def read_file(self, name: str) -> bytes:
|
||||
entry = self._find(name)
|
||||
if entry.size == 0:
|
||||
return b""
|
||||
fat = self._read_at(self._fat_offset, self._fat_bytes)
|
||||
result = bytearray(entry.size)
|
||||
written = 0
|
||||
for cluster in self._chain(entry.first_cluster, fat):
|
||||
count = min(self._cluster_bytes, entry.size - written)
|
||||
result[written : written + count] = self._read_at(self._cluster_offset(cluster), count)
|
||||
written += count
|
||||
if written == entry.size:
|
||||
break
|
||||
if written != entry.size:
|
||||
raise Fat16Error("镜像配置文件被截断")
|
||||
return bytes(result)
|
||||
|
||||
def write_file_range(self, name: str, file_offset: int, data: bytes) -> None:
|
||||
entry = self._find(name)
|
||||
if file_offset < 0 or file_offset + len(data) > entry.size:
|
||||
raise ValueError("写入范围超出配置文件")
|
||||
fat = self._read_at(self._fat_offset, self._fat_bytes)
|
||||
chain = list(self._chain(entry.first_cluster, fat))
|
||||
remaining = len(data)
|
||||
source_offset = 0
|
||||
position = file_offset
|
||||
while remaining:
|
||||
chain_index, within = divmod(position, self._cluster_bytes)
|
||||
if chain_index >= len(chain):
|
||||
raise Fat16Error("镜像配置文件簇链不足")
|
||||
count = min(remaining, self._cluster_bytes - within)
|
||||
absolute = self._cluster_offset(chain[chain_index]) + within
|
||||
if absolute + count > self._partition_end:
|
||||
raise Fat16Error("写入范围超出 FAT16 分区")
|
||||
self._stream.seek(absolute)
|
||||
self._stream.write(data[source_offset : source_offset + count])
|
||||
position += count
|
||||
source_offset += count
|
||||
remaining -= count
|
||||
self._stream.flush()
|
||||
os.fsync(self._stream.fileno())
|
||||
|
||||
@@ -0,0 +1,425 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from typing import Any, Callable
|
||||
|
||||
from PySide6.QtCore import QObject, QRunnable, QThreadPool, Qt, Signal
|
||||
from PySide6.QtGui import QCloseEvent, QDragEnterEvent, QDropEvent, QFont
|
||||
from PySide6.QtWidgets import (
|
||||
QApplication,
|
||||
QCheckBox,
|
||||
QComboBox,
|
||||
QDialog,
|
||||
QDialogButtonBox,
|
||||
QFileDialog,
|
||||
QFormLayout,
|
||||
QFrame,
|
||||
QGroupBox,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QLineEdit,
|
||||
QMainWindow,
|
||||
QPlainTextEdit,
|
||||
QProgressBar,
|
||||
QPushButton,
|
||||
QScrollArea,
|
||||
QSpinBox,
|
||||
QStyle,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from . import __version__
|
||||
from .config import validate_config
|
||||
from .operations import copy_write_and_verify, read_config, write_and_verify
|
||||
from .paths import validate_new_destination
|
||||
|
||||
|
||||
class AppDialog(QDialog):
|
||||
def __init__(self, parent: QWidget | None, message: str, title: str, icon: QStyle.StandardPixmap, confirm: bool) -> None:
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle(title)
|
||||
self.setModal(True)
|
||||
self.setMinimumSize(520, 260)
|
||||
self.resize(640, 320)
|
||||
|
||||
icon_label = QLabel()
|
||||
icon_label.setPixmap(self.style().standardIcon(icon).pixmap(48, 48))
|
||||
icon_label.setAlignment(Qt.AlignmentFlag.AlignTop | Qt.AlignmentFlag.AlignHCenter)
|
||||
body = QPlainTextEdit(message)
|
||||
body.setReadOnly(True)
|
||||
body.setLineWrapMode(QPlainTextEdit.LineWrapMode.WidgetWidth)
|
||||
|
||||
buttons = QDialogButtonBox(
|
||||
QDialogButtonBox.StandardButton.Yes | QDialogButtonBox.StandardButton.No
|
||||
if confirm
|
||||
else QDialogButtonBox.StandardButton.Ok
|
||||
)
|
||||
if confirm:
|
||||
buttons.button(QDialogButtonBox.StandardButton.Yes).setText("是")
|
||||
buttons.button(QDialogButtonBox.StandardButton.No).setText("否")
|
||||
buttons.accepted.connect(self.accept)
|
||||
buttons.rejected.connect(self.reject)
|
||||
else:
|
||||
buttons.button(QDialogButtonBox.StandardButton.Ok).setText("确定")
|
||||
buttons.accepted.connect(self.accept)
|
||||
|
||||
top = QHBoxLayout()
|
||||
top.addWidget(icon_label)
|
||||
top.addWidget(body, 1)
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(24, 24, 24, 24)
|
||||
layout.setSpacing(20)
|
||||
layout.addLayout(top, 1)
|
||||
layout.addWidget(buttons)
|
||||
|
||||
@classmethod
|
||||
def information(cls, parent: QWidget, message: str, title: str) -> None:
|
||||
cls(parent, message, title, QStyle.StandardPixmap.SP_MessageBoxInformation, False).exec()
|
||||
|
||||
@classmethod
|
||||
def error(cls, parent: QWidget | None, message: str, title: str) -> None:
|
||||
cls(parent, message, title, QStyle.StandardPixmap.SP_MessageBoxCritical, False).exec()
|
||||
|
||||
@classmethod
|
||||
def confirm(cls, parent: QWidget, message: str, title: str) -> bool:
|
||||
return cls(parent, message, title, QStyle.StandardPixmap.SP_MessageBoxWarning, True).exec() == QDialog.DialogCode.Accepted
|
||||
|
||||
|
||||
class WorkerSignals(QObject):
|
||||
progress = Signal(int)
|
||||
succeeded = Signal(object)
|
||||
failed = Signal(str)
|
||||
|
||||
|
||||
class Worker(QRunnable):
|
||||
def __init__(self, operation: Callable[[Callable[[int], None]], Any]) -> None:
|
||||
super().__init__()
|
||||
self.operation = operation
|
||||
self.signals = WorkerSignals()
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
result = self.operation(self.signals.progress.emit)
|
||||
except Exception as exception:
|
||||
self.signals.failed.emit(str(exception))
|
||||
else:
|
||||
self.signals.succeeded.emit(result)
|
||||
|
||||
|
||||
class MainWindow(QMainWindow):
|
||||
PREFERRED_SIZE = (960, 760)
|
||||
MINIMUM_SIZE = (760, 600)
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.setWindowTitle(f"核桃派镜像配置编辑器 {__version__}")
|
||||
self.setAcceptDrops(True)
|
||||
self._image_path: Path | None = None
|
||||
self._loaded: dict[str, Any] | None = None
|
||||
self._busy = False
|
||||
self._workers: set[Worker] = set()
|
||||
self._thread_pool = QThreadPool.globalInstance()
|
||||
self._screen_signal_connected = False
|
||||
|
||||
self.path = QLineEdit()
|
||||
self.path.setReadOnly(True)
|
||||
self.version = QLabel("尚未打开镜像")
|
||||
self.username = QLineEdit()
|
||||
self.account_password = QLineEdit()
|
||||
self.ssid = QLineEdit()
|
||||
self.wifi_password = QLineEdit()
|
||||
self.mode = QComboBox()
|
||||
self.mode.addItems(["自动获取（DHCP）", "静态 IPv4"])
|
||||
self.address = QLineEdit()
|
||||
self.prefix = QSpinBox()
|
||||
self.prefix.setRange(1, 32)
|
||||
self.prefix.setValue(24)
|
||||
self.gateway = QLineEdit()
|
||||
self.dns = QLineEdit()
|
||||
self.open_button = QPushButton("打开镜像…")
|
||||
self.save_button = QPushButton("修改原镜像")
|
||||
self.save_as_button = QPushButton("另存为…")
|
||||
self.progress = QProgressBar()
|
||||
self.progress.setRange(0, 100)
|
||||
self.progress.hide()
|
||||
self.save_button.setEnabled(False)
|
||||
self.save_as_button.setEnabled(False)
|
||||
|
||||
self.account_password.setEchoMode(QLineEdit.EchoMode.Password)
|
||||
self.wifi_password.setEchoMode(QLineEdit.EchoMode.Password)
|
||||
self.mode.currentIndexChanged.connect(self._update_network_fields)
|
||||
self.open_button.clicked.connect(self._open_dialog)
|
||||
self.save_button.clicked.connect(self._save_original)
|
||||
self.save_as_button.clicked.connect(self._save_as)
|
||||
self.setCentralWidget(self._build_ui())
|
||||
self._update_network_fields()
|
||||
|
||||
def _build_ui(self) -> QWidget:
|
||||
content = QWidget()
|
||||
outer = QVBoxLayout(content)
|
||||
outer.setContentsMargins(24, 24, 24, 24)
|
||||
outer.setSpacing(12)
|
||||
|
||||
header = QHBoxLayout()
|
||||
header.addWidget(self.path, 1)
|
||||
self.open_button.setMinimumSize(130, 38)
|
||||
header.addWidget(self.open_button)
|
||||
outer.addLayout(header)
|
||||
outer.addWidget(self.version)
|
||||
outer.addWidget(self._group("登录账户", [("用户名", self.username), ("密码", self._password_row(self.account_password))]))
|
||||
outer.addWidget(self._group("Wi-Fi", [("SSID", self.ssid), ("密码", self._password_row(self.wifi_password))]))
|
||||
outer.addWidget(
|
||||
self._group(
|
||||
"IPv4",
|
||||
[
|
||||
("方式", self.mode),
|
||||
("地址", self.address),
|
||||
("前缀", self.prefix),
|
||||
("网关", self.gateway),
|
||||
("DNS（逗号分隔）", self.dns),
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
footer = QHBoxLayout()
|
||||
footer.addWidget(self.progress, 1)
|
||||
footer.addStretch(1)
|
||||
self.save_button.setMinimumSize(150, 40)
|
||||
self.save_as_button.setMinimumSize(130, 40)
|
||||
footer.addWidget(self.save_button)
|
||||
footer.addWidget(self.save_as_button)
|
||||
outer.addLayout(footer)
|
||||
|
||||
scroll = QScrollArea()
|
||||
scroll.setFrameShape(QFrame.Shape.NoFrame)
|
||||
scroll.setWidgetResizable(True)
|
||||
scroll.setWidget(content)
|
||||
return scroll
|
||||
|
||||
@staticmethod
|
||||
def _group(title: str, rows: list[tuple[str, QWidget]]) -> QGroupBox:
|
||||
group = QGroupBox(title)
|
||||
form = QFormLayout(group)
|
||||
form.setContentsMargins(16, 20, 16, 16)
|
||||
form.setHorizontalSpacing(18)
|
||||
form.setVerticalSpacing(10)
|
||||
form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.AllNonFixedFieldsGrow)
|
||||
for label, widget in rows:
|
||||
widget.setMinimumWidth(320)
|
||||
form.addRow(label, widget)
|
||||
return group
|
||||
|
||||
@staticmethod
|
||||
def _password_row(field: QLineEdit) -> QWidget:
|
||||
wrapper = QWidget()
|
||||
layout = QHBoxLayout(wrapper)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setSpacing(12)
|
||||
show = QCheckBox("显示")
|
||||
show.toggled.connect(lambda checked: field.setEchoMode(QLineEdit.EchoMode.Normal if checked else QLineEdit.EchoMode.Password))
|
||||
layout.addWidget(field, 1)
|
||||
layout.addWidget(show)
|
||||
return wrapper
|
||||
|
||||
def showEvent(self, event: Any) -> None:
|
||||
super().showEvent(event)
|
||||
self._fit_to_screen()
|
||||
handle = self.windowHandle()
|
||||
if handle is not None and not self._screen_signal_connected:
|
||||
handle.screenChanged.connect(self._fit_to_screen)
|
||||
self._screen_signal_connected = True
|
||||
|
||||
def _fit_to_screen(self, *_args: object) -> None:
|
||||
screen = self.screen() or QApplication.primaryScreen()
|
||||
if screen is None:
|
||||
return
|
||||
area = screen.availableGeometry()
|
||||
maximum_width = max(320, int(area.width() * 0.9))
|
||||
maximum_height = max(240, int(area.height() * 0.9))
|
||||
minimum_width = min(self.MINIMUM_SIZE[0], maximum_width)
|
||||
minimum_height = min(self.MINIMUM_SIZE[1], maximum_height)
|
||||
self.setMinimumSize(minimum_width, minimum_height)
|
||||
width = max(minimum_width, min(self.PREFERRED_SIZE[0], maximum_width))
|
||||
height = max(minimum_height, min(self.PREFERRED_SIZE[1], maximum_height))
|
||||
self.resize(width, height)
|
||||
self.move(area.left() + max(0, (area.width() - width) // 2), area.top() + max(0, (area.height() - height) // 2))
|
||||
|
||||
def dragEnterEvent(self, event: QDragEnterEvent) -> None:
|
||||
urls = event.mimeData().urls() if event.mimeData().hasUrls() else []
|
||||
if not self._busy and len(urls) == 1 and urls[0].isLocalFile():
|
||||
event.acceptProposedAction()
|
||||
else:
|
||||
event.ignore()
|
||||
|
||||
def dropEvent(self, event: QDropEvent) -> None:
|
||||
urls = event.mimeData().urls() if event.mimeData().hasUrls() else []
|
||||
if not self._busy and len(urls) == 1 and urls[0].isLocalFile():
|
||||
event.acceptProposedAction()
|
||||
self._open_image(urls[0].toLocalFile())
|
||||
else:
|
||||
event.ignore()
|
||||
|
||||
def closeEvent(self, event: QCloseEvent) -> None:
|
||||
if self._busy:
|
||||
event.ignore()
|
||||
return
|
||||
super().closeEvent(event)
|
||||
|
||||
def _open_dialog(self) -> None:
|
||||
filename, _filter = QFileDialog.getOpenFileName(self, "打开核桃派镜像", "", "核桃派镜像 (*.img)")
|
||||
if filename:
|
||||
self._open_image(filename)
|
||||
|
||||
def _open_image(self, filename: str) -> None:
|
||||
try:
|
||||
config = read_config(filename)
|
||||
self._apply_loaded(Path(filename).resolve(), config)
|
||||
except Exception as exception:
|
||||
AppDialog.error(self, str(exception), "无法打开镜像")
|
||||
|
||||
def _apply_loaded(self, filename: Path, config: dict[str, Any]) -> None:
|
||||
self._image_path = filename
|
||||
self._loaded = config
|
||||
self.path.setText(str(filename))
|
||||
self.version.setText(f"软件版本：{config['software_version']} 配置校验：正常")
|
||||
self.username.setText(config["account"]["username"])
|
||||
self.account_password.setText(config["account"]["password"])
|
||||
self.ssid.setText(config["wifi"]["ssid"])
|
||||
self.wifi_password.setText(config["wifi"]["password"])
|
||||
ipv4 = config["ipv4"]
|
||||
self.mode.setCurrentIndex(1 if ipv4["mode"] == "static" else 0)
|
||||
self.address.setText(ipv4["address"])
|
||||
self.prefix.setValue(ipv4["prefix"] or 24)
|
||||
self.gateway.setText(ipv4["gateway"])
|
||||
self.dns.setText(",".join(ipv4["dns"]))
|
||||
self.save_button.setEnabled(True)
|
||||
self.save_as_button.setEnabled(True)
|
||||
self._update_network_fields()
|
||||
|
||||
def _collect(self) -> dict[str, Any]:
|
||||
if self._loaded is None:
|
||||
raise ValueError("尚未打开镜像")
|
||||
ipv4: dict[str, Any]
|
||||
if self.mode.currentIndex() == 0:
|
||||
ipv4 = {"mode": "dhcp", "address": "", "prefix": 0, "gateway": "", "dns": []}
|
||||
else:
|
||||
ipv4 = {
|
||||
"mode": "static",
|
||||
"address": self.address.text().strip(),
|
||||
"prefix": self.prefix.value(),
|
||||
"gateway": self.gateway.text().strip(),
|
||||
"dns": [item.strip() for item in self.dns.text().split(",") if item.strip()],
|
||||
}
|
||||
return validate_config(
|
||||
{
|
||||
"schema_version": self._loaded["schema_version"],
|
||||
"product": self._loaded["product"],
|
||||
"software_version": self._loaded["software_version"],
|
||||
"account": {"username": self.username.text().strip(), "password": self.account_password.text()},
|
||||
"wifi": {"ssid": self.ssid.text(), "password": self.wifi_password.text()},
|
||||
"ipv4": ipv4,
|
||||
}
|
||||
)
|
||||
|
||||
def _save_original(self) -> None:
|
||||
if self._image_path is None:
|
||||
return
|
||||
try:
|
||||
config = self._collect()
|
||||
except Exception as exception:
|
||||
AppDialog.error(self, str(exception), "保存失败")
|
||||
return
|
||||
if not AppDialog.confirm(
|
||||
self,
|
||||
"这会直接修改当前镜像。建议日常使用“另存为”保留原始包。\n\n确定继续吗？",
|
||||
"确认修改原镜像",
|
||||
):
|
||||
return
|
||||
target = self._image_path
|
||||
self._start_worker(
|
||||
lambda progress: write_and_verify(target, config, progress),
|
||||
lambda verified: self._save_succeeded(target, verified, "配置已保存，镜像与 SHA-256 摘要均已完成写后校验。", "保存成功"),
|
||||
)
|
||||
|
||||
def _save_as(self) -> None:
|
||||
if self._image_path is None:
|
||||
return
|
||||
try:
|
||||
config = self._collect()
|
||||
except Exception as exception:
|
||||
AppDialog.error(self, str(exception), "另存为失败")
|
||||
return
|
||||
dialog = QFileDialog(self, "请重命名，以便区分设备或使用地点", str(self._image_path.parent), "核桃派镜像 (*.img)")
|
||||
dialog.setAcceptMode(QFileDialog.AcceptMode.AcceptSave)
|
||||
dialog.setFileMode(QFileDialog.FileMode.AnyFile)
|
||||
dialog.setOption(QFileDialog.Option.DontConfirmOverwrite, True)
|
||||
dialog.selectFile(f"{self._image_path.stem}-自定义.img")
|
||||
if dialog.exec() != QDialog.DialogCode.Accepted or not dialog.selectedFiles():
|
||||
return
|
||||
selected = dialog.selectedFiles()[0]
|
||||
if Path(selected).suffix == "":
|
||||
selected += ".img"
|
||||
try:
|
||||
destination = validate_new_destination(self._image_path, selected)
|
||||
except Exception as exception:
|
||||
AppDialog.error(self, str(exception), "另存为失败")
|
||||
return
|
||||
source = self._image_path
|
||||
self._start_worker(
|
||||
lambda progress: copy_write_and_verify(source, destination, config, progress),
|
||||
lambda result: self._save_succeeded(
|
||||
result[0], result[1], "自定义镜像已保存并重新打开，镜像与 SHA-256 摘要均已完成写后校验。", "另存为成功"
|
||||
),
|
||||
)
|
||||
|
||||
def _start_worker(self, operation: Callable[[Callable[[int], None]], Any], on_success: Callable[[Any], None]) -> None:
|
||||
self._set_busy(True)
|
||||
worker = Worker(operation)
|
||||
self._workers.add(worker)
|
||||
worker.signals.progress.connect(self.progress.setValue)
|
||||
|
||||
def succeeded(result: Any) -> None:
|
||||
self._workers.discard(worker)
|
||||
self._set_busy(False)
|
||||
on_success(result)
|
||||
|
||||
def failed(message: str) -> None:
|
||||
self._workers.discard(worker)
|
||||
self._set_busy(False)
|
||||
AppDialog.error(self, message, "保存失败")
|
||||
|
||||
worker.signals.succeeded.connect(succeeded)
|
||||
worker.signals.failed.connect(failed)
|
||||
self._thread_pool.start(worker)
|
||||
|
||||
def _save_succeeded(self, target: Path, verified: dict[str, Any], message: str, title: str) -> None:
|
||||
self._apply_loaded(target, verified)
|
||||
AppDialog.information(self, message, title)
|
||||
|
||||
def _set_busy(self, busy: bool) -> None:
|
||||
self._busy = busy
|
||||
self.open_button.setEnabled(not busy)
|
||||
self.save_button.setEnabled(not busy and self._loaded is not None)
|
||||
self.save_as_button.setEnabled(not busy and self._loaded is not None)
|
||||
self.progress.setVisible(busy)
|
||||
self.progress.setValue(0)
|
||||
QApplication.setOverrideCursor(Qt.CursorShape.WaitCursor) if busy else QApplication.restoreOverrideCursor()
|
||||
|
||||
def _update_network_fields(self) -> None:
|
||||
enabled = self.mode.currentIndex() == 1
|
||||
for widget in (self.address, self.prefix, self.gateway, self.dns):
|
||||
widget.setEnabled(enabled)
|
||||
|
||||
|
||||
def run_gui() -> int:
|
||||
app = QApplication(sys.argv)
|
||||
app.setApplicationName("核桃派镜像配置编辑器")
|
||||
app.setApplicationVersion(__version__)
|
||||
if os.name == "nt":
|
||||
app.setFont(QFont("Microsoft YaHei UI", 10))
|
||||
window = MainWindow()
|
||||
window.show()
|
||||
return app.exec()
|
||||
@@ -0,0 +1,171 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
from typing import Any, Callable
|
||||
import uuid
|
||||
|
||||
from .config import encode_update, read_config_file, serialize_config, slot_offset, validate_config
|
||||
from .fat16 import Fat16Disk
|
||||
from .paths import (
|
||||
cleanup_new_destination,
|
||||
ensure_sidecar_writable,
|
||||
sidecar_path,
|
||||
validate_existing_image,
|
||||
validate_new_destination,
|
||||
)
|
||||
|
||||
|
||||
Progress = Callable[[int], None]
|
||||
_CHUNK_BYTES = 4 * 1024 * 1024
|
||||
|
||||
|
||||
class ImageOperationError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def _hash_file(path: Path, progress: Progress | None = None, *, start: int = 0, span: int = 100) -> str:
|
||||
digest = hashlib.sha256()
|
||||
size = path.stat().st_size
|
||||
completed = 0
|
||||
with path.open("rb", buffering=_CHUNK_BYTES) as stream:
|
||||
for chunk in iter(lambda: stream.read(_CHUNK_BYTES), b""):
|
||||
digest.update(chunk)
|
||||
completed += len(chunk)
|
||||
if progress is not None:
|
||||
fraction = completed / size if size else 1.0
|
||||
progress(min(100, start + int(fraction * span)))
|
||||
if progress is not None:
|
||||
progress(min(100, start + span))
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _read_sidecar_digest(path: Path, expected_filename: str) -> str:
|
||||
try:
|
||||
if path.stat().st_size > 128 * 1024:
|
||||
raise ImageOperationError("镜像 SHA-256 摘要文件过大")
|
||||
raw = path.read_bytes()
|
||||
except OSError as exception:
|
||||
raise ImageOperationError(f"无法读取镜像 SHA-256 摘要：{exception}") from exception
|
||||
if raw.startswith(b"\xef\xbb\xbf"):
|
||||
raise ImageOperationError("镜像 SHA-256 摘要必须使用无 BOM UTF-8 格式")
|
||||
try:
|
||||
content = raw.decode("utf-8", errors="strict")
|
||||
except UnicodeDecodeError as exception:
|
||||
raise ImageOperationError("镜像 SHA-256 摘要不是有效的 UTF-8 文本") from exception
|
||||
if len(content) < 68 or not content.endswith("\n") or "\r" in content or "\n" in content[:-1]:
|
||||
raise ImageOperationError("镜像 SHA-256 摘要文件必须是以 LF 结尾的单行规范格式")
|
||||
line = content[:-1]
|
||||
digest = line[:64]
|
||||
if len(line) < 67 or any(character not in "0123456789abcdef" for character in digest) or line[64:66] != " ":
|
||||
raise ImageOperationError("镜像 SHA-256 摘要文件格式无效")
|
||||
if line[66:] != expected_filename:
|
||||
raise ImageOperationError("镜像 SHA-256 摘要中的文件名与当前镜像不一致")
|
||||
return digest
|
||||
|
||||
|
||||
def verify_sidecar_if_present(image_path: Path, progress: Progress | None = None) -> None:
|
||||
sidecar = sidecar_path(image_path)
|
||||
if not sidecar.exists():
|
||||
return
|
||||
expected = _read_sidecar_digest(sidecar, image_path.name)
|
||||
actual = _hash_file(image_path, progress)
|
||||
if actual != expected:
|
||||
raise ImageOperationError("镜像内容与 SHA-256 摘要不符，禁止编辑")
|
||||
|
||||
|
||||
def read_config(image: str | os.PathLike[str], progress: Progress | None = None) -> dict[str, Any]:
|
||||
image_path = validate_existing_image(image)
|
||||
verify_sidecar_if_present(image_path, progress)
|
||||
with Fat16Disk(image_path, writable=False) as disk:
|
||||
config, _generation, _slot = read_config_file(disk.read_file("MSCCFG.BIN"))
|
||||
return config
|
||||
|
||||
|
||||
def write_config(image: str | os.PathLike[str], config: Any) -> None:
|
||||
image_path = validate_existing_image(image)
|
||||
ensure_sidecar_writable(image_path)
|
||||
verify_sidecar_if_present(image_path)
|
||||
normalized = validate_config(config)
|
||||
with Fat16Disk(image_path, writable=True) as disk:
|
||||
current = disk.read_file("MSCCFG.BIN")
|
||||
active, _generation, _slot = read_config_file(current)
|
||||
if active["software_version"] != normalized["software_version"]:
|
||||
raise ImageOperationError("禁止通过编辑器改变软件版本")
|
||||
slot_data, target = encode_update(current, normalized)
|
||||
disk.write_file_range("MSCCFG.BIN", slot_offset(target), slot_data)
|
||||
verified, _generation, _slot = read_config_file(disk.read_file("MSCCFG.BIN"))
|
||||
if serialize_config(verified) != serialize_config(normalized):
|
||||
raise ImageOperationError("配置写后校验失败")
|
||||
|
||||
|
||||
def write_sha256(image: str | os.PathLike[str], progress: Progress | None = None) -> None:
|
||||
image_path = validate_existing_image(image)
|
||||
ensure_sidecar_writable(image_path)
|
||||
digest = _hash_file(image_path, progress)
|
||||
sidecar = sidecar_path(image_path)
|
||||
temporary = Path(str(sidecar) + f".{uuid.uuid4().hex}.tmp")
|
||||
try:
|
||||
payload = f"{digest} {image_path.name}\n".encode("utf-8")
|
||||
with temporary.open("xb") as stream:
|
||||
stream.write(payload)
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(temporary, sidecar)
|
||||
stored = _read_sidecar_digest(sidecar, image_path.name)
|
||||
if stored != digest:
|
||||
raise ImageOperationError("镜像 SHA-256 摘要写后校验失败")
|
||||
finally:
|
||||
try:
|
||||
temporary.unlink(missing_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def write_and_verify(image: str | os.PathLike[str], config: Any, progress: Progress | None = None) -> dict[str, Any]:
|
||||
write_config(image, config)
|
||||
write_sha256(image, progress)
|
||||
return read_config(image)
|
||||
|
||||
|
||||
def _copy_file(source: Path, destination: Path, progress: Progress | None = None) -> None:
|
||||
total = source.stat().st_size
|
||||
completed = 0
|
||||
with source.open("rb", buffering=_CHUNK_BYTES) as incoming, destination.open("xb", buffering=_CHUNK_BYTES) as outgoing:
|
||||
for chunk in iter(lambda: incoming.read(_CHUNK_BYTES), b""):
|
||||
outgoing.write(chunk)
|
||||
completed += len(chunk)
|
||||
if progress is not None:
|
||||
progress(int((completed / total if total else 1.0) * 35))
|
||||
outgoing.flush()
|
||||
os.fsync(outgoing.fileno())
|
||||
try:
|
||||
shutil.copystat(source, destination)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def copy_write_and_verify(
|
||||
source: str | os.PathLike[str],
|
||||
destination: str | os.PathLike[str],
|
||||
config: Any,
|
||||
progress: Progress | None = None,
|
||||
) -> tuple[Path, dict[str, Any]]:
|
||||
source_path = validate_existing_image(source)
|
||||
target = validate_new_destination(source_path, destination)
|
||||
normalized = validate_config(config)
|
||||
read_config(source_path)
|
||||
try:
|
||||
_copy_file(source_path, target, progress)
|
||||
write_config(target, normalized)
|
||||
write_sha256(target, None if progress is None else lambda value: progress(35 + value * 65 // 100))
|
||||
verified = read_config(target)
|
||||
return target, verified
|
||||
except Exception as exception:
|
||||
cleanup_error = cleanup_new_destination(target)
|
||||
if cleanup_error is not None:
|
||||
raise ImageOperationError(f"{exception}\n清理失败：{cleanup_error}") from exception
|
||||
raise
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import uuid
|
||||
|
||||
|
||||
class ImagePathError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
_WINDOWS_RESERVED = re.compile(r"^(CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])(?:\..*)?$", re.IGNORECASE | re.ASCII)
|
||||
_WINDOWS_INVALID = set('<>:"/\\|?*')
|
||||
|
||||
|
||||
def _absolute(path: str | os.PathLike[str]) -> Path:
|
||||
if not str(path).strip():
|
||||
raise ImagePathError("镜像路径不能为空")
|
||||
try:
|
||||
return Path(path).expanduser().resolve(strict=False)
|
||||
except (OSError, RuntimeError, ValueError) as exception:
|
||||
raise ImagePathError("镜像路径格式无效") from exception
|
||||
|
||||
|
||||
def _validate_name_and_directory(path: str | os.PathLike[str]) -> Path:
|
||||
full = _absolute(path)
|
||||
name = full.name
|
||||
if not name.strip():
|
||||
raise ImagePathError("镜像文件名不能为空")
|
||||
if full.suffix.lower() != ".img":
|
||||
raise ImagePathError("镜像文件名必须以 .img 结尾")
|
||||
if os.name == "nt":
|
||||
if name != name.rstrip(" .") or any(character in _WINDOWS_INVALID or ord(character) < 32 for character in name):
|
||||
raise ImagePathError("镜像文件名包含 Windows 不允许的字符，或以空格、句点结尾")
|
||||
if _WINDOWS_RESERVED.fullmatch(full.stem):
|
||||
raise ImagePathError("镜像文件名使用了 Windows 保留名称，请更换名称")
|
||||
if not full.parent.is_dir():
|
||||
raise ImagePathError("镜像目标文件夹不存在")
|
||||
return full
|
||||
|
||||
|
||||
def sidecar_path(image_path: Path) -> Path:
|
||||
return Path(str(image_path) + ".sha256")
|
||||
|
||||
|
||||
def validate_existing_image(path: str | os.PathLike[str]) -> Path:
|
||||
full = _validate_name_and_directory(path)
|
||||
if not full.is_file():
|
||||
raise ImagePathError(f"镜像文件不存在：{full}")
|
||||
sidecar = sidecar_path(full)
|
||||
if sidecar.is_dir():
|
||||
raise ImagePathError("同名 SHA-256 摘要路径被文件夹占用")
|
||||
return full
|
||||
|
||||
|
||||
def validate_new_destination(source: str | os.PathLike[str], destination: str | os.PathLike[str]) -> Path:
|
||||
source_path = validate_existing_image(source)
|
||||
target = _validate_name_and_directory(destination)
|
||||
if os.path.normcase(str(target)) == os.path.normcase(str(source_path)):
|
||||
raise ImagePathError("另存为必须使用与原镜像不同的文件名")
|
||||
if target.exists():
|
||||
raise ImagePathError("目标镜像已经存在，请使用新的文件名")
|
||||
sidecar = sidecar_path(target)
|
||||
if sidecar.exists():
|
||||
raise ImagePathError("目标镜像的同名 SHA-256 摘要已经存在，请使用新的文件名")
|
||||
_probe_directory(target.parent)
|
||||
return target
|
||||
|
||||
|
||||
def ensure_sidecar_writable(image_path: Path) -> None:
|
||||
sidecar = sidecar_path(image_path)
|
||||
if sidecar.exists():
|
||||
if not sidecar.is_file():
|
||||
raise ImagePathError("同名 SHA-256 摘要不是普通文件")
|
||||
try:
|
||||
with sidecar.open("ab"):
|
||||
pass
|
||||
except OSError as exception:
|
||||
raise ImagePathError("同名 SHA-256 摘要不可写") from exception
|
||||
_probe_directory(image_path.parent)
|
||||
|
||||
|
||||
def _probe_directory(directory: Path) -> None:
|
||||
probe = directory / f".matrix-image-editor-{uuid.uuid4().hex}.tmp"
|
||||
try:
|
||||
with probe.open("xb") as stream:
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
except OSError as exception:
|
||||
raise ImagePathError("镜像目标文件夹不可写") from exception
|
||||
finally:
|
||||
try:
|
||||
probe.unlink(missing_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def cleanup_new_destination(destination: str | os.PathLike[str]) -> str | None:
|
||||
target = Path(destination)
|
||||
failures: list[str] = []
|
||||
for item in (sidecar_path(target), target):
|
||||
try:
|
||||
if item.is_file() or item.is_symlink():
|
||||
item.unlink()
|
||||
except OSError as exception:
|
||||
failures.append(f"{item.name}：{exception}")
|
||||
return "；".join(failures) if failures else None
|
||||
|
||||
Reference in New Issue
Block a user