mirror of
https://github.com/mofeng-git/One-KVM.git
synced 2025-12-12 01:00:29 +08:00
refactoring
This commit is contained in:
parent
317a127f58
commit
941b3bbd75
@ -25,14 +25,24 @@ import dataclasses
|
|||||||
|
|
||||||
|
|
||||||
# =====
|
# =====
|
||||||
|
class PartitionType:
|
||||||
|
MSD = "otgmsd"
|
||||||
|
PST = "pst"
|
||||||
|
ALL = (
|
||||||
|
MSD,
|
||||||
|
PST,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@dataclasses.dataclass(frozen=True)
|
@dataclasses.dataclass(frozen=True)
|
||||||
class FstabStorage:
|
class Partition:
|
||||||
mount_path: str
|
mount_path: str
|
||||||
root_path: str
|
root_path: str
|
||||||
user: str
|
user: str
|
||||||
|
|
||||||
|
|
||||||
def find_storage(target: str) -> FstabStorage:
|
def find_partition(part_type: str) -> Partition:
|
||||||
|
assert part_type in PartitionType.ALL
|
||||||
fstab_path = "/etc/fstab"
|
fstab_path = "/etc/fstab"
|
||||||
with open(fstab_path) as file:
|
with open(fstab_path) as file:
|
||||||
for line in file.read().split("\n"):
|
for line in file.read().split("\n"):
|
||||||
@ -40,11 +50,11 @@ def find_storage(target: str) -> FstabStorage:
|
|||||||
if line and not line.startswith("#"):
|
if line and not line.startswith("#"):
|
||||||
parts = line.split()
|
parts = line.split()
|
||||||
if len(parts) == 6:
|
if len(parts) == 6:
|
||||||
options = dict(re.findall(r"X-kvmd\.%s-(root|user)(?:=([^,]+))?" % (target), parts[3]))
|
options = dict(re.findall(r"X-kvmd\.%s-(root|user)(?:=([^,]+))?" % (part_type), parts[3]))
|
||||||
if options:
|
if options:
|
||||||
return FstabStorage(
|
return Partition(
|
||||||
mount_path=parts[1],
|
mount_path=parts[1],
|
||||||
root_path=(options.get("root", "") or parts[1]),
|
root_path=(options.get("root", "") or parts[1]),
|
||||||
user=options.get("user", ""),
|
user=options.get("user", ""),
|
||||||
)
|
)
|
||||||
raise RuntimeError(f"Can't find {target!r} mountpoint in {fstab_path}")
|
raise RuntimeError(f"Can't find {part_type!r} mountpoint in {fstab_path}")
|
||||||
|
|||||||
@ -64,27 +64,26 @@ def main() -> None:
|
|||||||
if len(sys.argv) != 2 or sys.argv[1] not in ["ro", "rw"]:
|
if len(sys.argv) != 2 or sys.argv[1] not in ["ro", "rw"]:
|
||||||
raise SystemExit(f"Usage: {sys.argv[0]} [ro|rw]")
|
raise SystemExit(f"Usage: {sys.argv[0]} [ro|rw]")
|
||||||
|
|
||||||
target = ""
|
part_type = ""
|
||||||
dirs: list[str] = []
|
dirs: list[str] = []
|
||||||
app = os.path.basename(sys.argv[0])
|
app = os.path.basename(sys.argv[0])
|
||||||
if app == "kvmd-helper-otgmsd-remount":
|
if app == "kvmd-helper-otgmsd-remount":
|
||||||
target = "otgmsd"
|
part_type = fstab.PartitionType.MSD
|
||||||
dirs = ["images", "meta"]
|
dirs = ["images", "meta"]
|
||||||
elif app == "kvmd-helper-pst-remount":
|
elif app == "kvmd-helper-pst-remount":
|
||||||
target = "pst"
|
part_type = fstab.PartitionType.PST
|
||||||
dirs = ["data"]
|
dirs = ["data"]
|
||||||
else:
|
else:
|
||||||
raise SystemExit("Unknown application target")
|
raise SystemExit("Unknown application target")
|
||||||
|
|
||||||
rw = (sys.argv[1] == "rw")
|
rw = (sys.argv[1] == "rw")
|
||||||
|
|
||||||
assert target
|
part = fstab.find_partition(part_type)
|
||||||
storage = fstab.find_storage(target)
|
_remount(part.mount_path, rw)
|
||||||
_remount(storage.mount_path, rw)
|
if rw and part.root_path:
|
||||||
if rw and storage.root_path:
|
|
||||||
for name in dirs:
|
for name in dirs:
|
||||||
path = os.path.join(storage.root_path, name)
|
path = os.path.join(part.root_path, name)
|
||||||
_mkdir(path)
|
_mkdir(path)
|
||||||
if storage.user:
|
if part.user:
|
||||||
_chown(path, storage.user)
|
_chown(path, part.user)
|
||||||
_log(f"Storage in the {'RW' if rw else 'RO'}-mode now")
|
_log(f"Storage in the {'RW' if rw else 'RO'}-mode now")
|
||||||
|
|||||||
@ -160,7 +160,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
|
|||||||
self.__initial_cdrom: bool = initial["cdrom"]
|
self.__initial_cdrom: bool = initial["cdrom"]
|
||||||
|
|
||||||
self.__drive = Drive(gadget, instance=0, lun=0)
|
self.__drive = Drive(gadget, instance=0, lun=0)
|
||||||
self.__storage = Storage(fstab.find_storage("otgmsd").root_path)
|
self.__storage = Storage(fstab.find_partition(fstab.PartitionType.MSD).root_path)
|
||||||
|
|
||||||
self.__reader: (MsdFileReader | None) = None
|
self.__reader: (MsdFileReader | None) = None
|
||||||
self.__writer: (MsdFileWriter | None) = None
|
self.__writer: (MsdFileWriter | None) = None
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user