dreagonmon/backup_box
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

vnode

Browse Source
This commit is contained in:
Dreagonmon
2025-11-05 17:36:53 +08:00
parent 5c564dfc66
commit 70fc4f0be5
5 changed files with 198 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
backup_box/app.py
View File

@@ -32,31 +32,18 @@ def parse_args():
config.apply_user_config(args.config) config.apply_user_config(args.config)
print("Args:", args) print("Args:", args)
def main(): async def a_main():
parse_args() parse_args()
print("Config:", config.get_config()) print("Config:", config.get_config())
print("Hello World") print("Hello World")
from .storage.vnode import VFileSystem, new_storage_file, new_storage_dir, dump_storage_item_lines from .storage.local_storage import LocalStorage
from pathlib import PurePosixPath from pathlib import PurePosixPath
fs = VFileSystem() fs = LocalStorage(r"D:\AuberyZhao\BackupBox\backup_box")
d1 = new_storage_dir("dirt1", [ async for rt in fs.sync_storage():
new_storage_file("tgn1"), print(rt)
new_storage_file("tgn2"), print(fs.dump_fs_tree())
new_storage_file("tgn3"),
new_storage_file("tgn4"), def main():
]) import asyncio
d2 = new_storage_dir("dirt2", [ asyncio.run(a_main())
d1,
new_storage_file("tgn1"),
new_storage_file("tgn2"),
new_storage_file("tgn3"),
new_storage_file("tgn4"),
])
d3 = new_storage_dir("dirt3", [])
fs._root["c"].append(d2)
fs._root["c"].append(new_storage_file(n="fsafsa"))
fs._root["c"].append(d3)
print(fs)
fs.remove(PurePosixPath("dirt2/dirt1/tgn1"))
print(fs)

51
backup_box/storage/local_storage.py
View File

@@ -1,15 +1,13 @@
from collections.abc import Awaitable from backup_box.storage.storage import AsyncFileReader, AsyncFileWriter
from .storage import Storage
from pathlib import PurePath, PurePosixPath from pathlib import PurePath, PurePosixPath
from ..config import StorageItem from ..config import StorageItem
from .vnode import VFileSystem from .vnode import VFileSystem, VFile, VDir
from typing import BinaryIO, Literal from typing import Literal, overload
class LocalStorage(Storage): class LocalStorage(VFileSystem):
def __init__(self, base_path: str) -> None: def __init__(self, base_path: str) -> None:
super().__init__() super().__init__()
self.base = PurePath(base_path) self._base = PurePath(base_path)
self.vfs = VFileSystem()
@classmethod @classmethod
def from_config(cls, cfg: StorageItem) -> "LocalStorage": def from_config(cls, cfg: StorageItem) -> "LocalStorage":
@@ -17,5 +15,44 @@ class LocalStorage(Storage):
raise ValueError("config type mismatch!") raise ValueError("config type mismatch!")
return LocalStorage(cfg["path"]) return LocalStorage(cfg["path"])
def _vnode_path_to_syspath(self, *path: VFile | VDir):
filtered = []
# filter abslute path
for n in path:
name = n["n"].strip()
if name.startswith("/") or name.startswith("\\"):
continue
if name in [".", "..", ""]:
continue
filtered.append(name)
return self._base.joinpath(*filtered)
async def sync_storage(self):
curr = 0
total = 1
parents: list[VDir] = []
self._root["c"].clear()
node = self._root
yield curr, total
while True:
if node["ty"] == "dir":
pass
if len(parents) <= 0:
break
async def _on_remove_node(self, node_path: list[VFile | VDir]) -> None:
raise NotImplementedError
async def _on_create_node(self, node_path: list[VFile | VDir]) -> None:
raise NotImplementedError
@overload
async def _on_open_node(self, node_path: list[VFile | VDir], mode: Literal['r'] = "r") -> AsyncFileReader: ...
@overload
async def _on_open_node(self, node_path: list[VFile | VDir], mode: Literal['w']) -> AsyncFileWriter: ...
async def _on_open_node(self, node_path: list[VFile | VDir], mode: Literal['r'] | Literal['w'] = "r") -> AsyncFileWriter | AsyncFileReader:
raise NotImplementedError

45
backup_box/storage/storage.py
View File

@@ -1,8 +1,25 @@
from ..config import StorageItem from ..config import StorageItem
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from pathlib import PurePosixPath from pathlib import PurePosixPath
from collections.abc import Awaitable, Iterable, AsyncIterable from collections.abc import Awaitable, AsyncIterable
from typing import BinaryIO, Literal from typing import overload, Literal, Protocol, SupportsBytes
class AsyncFileReader(Protocol):
async def read(self, n: int = -1) -> bytes:
...
async def close(self) -> None:
...
class AsyncFileWriter(Protocol):
async def write(self, s: SupportsBytes) -> int:
...
async def close(self) -> None:
...
class AsyncFileLike(AsyncFileReader, AsyncFileWriter):
...
class Storage(ABC): class Storage(ABC):
@classmethod @classmethod
@@ -15,46 +32,52 @@ class Storage(ABC):
... ...
@abstractmethod @abstractmethod
def sync_storage() -> Iterable[tuple[int,int]] | AsyncIterable[tuple[int,int]]: def sync_storage(self) -> AsyncIterable[tuple[int,int]]:
""" sync storage, yield (current, total) progress """ """ sync storage, yield (current, total) progress """
... ...
@abstractmethod @abstractmethod
def listdir(self, path=PurePosixPath("")) -> list[PurePosixPath] | Awaitable[list[PurePosixPath]]: def listdir(self, path=PurePosixPath("")) -> AsyncIterable[PurePosixPath]:
""" list storage file in 'path' """ """ list storage file in 'path' """
... ...
@abstractmethod @abstractmethod
def exists(self, path: PurePosixPath) -> bool | Awaitable[bool]: def exists(self, path: PurePosixPath) -> Awaitable[bool]:
""" if the file or dir is exists """ """ if the file or dir is exists """
... ...
@abstractmethod @abstractmethod
def is_dir(self, path: PurePosixPath) -> bool | Awaitable[bool]: def is_dir(self, path: PurePosixPath) -> Awaitable[bool]:
""" if the path is dir """ """ if the path is dir """
... ...
@abstractmethod @abstractmethod
def is_file(self, path: PurePosixPath) -> bool | Awaitable[bool]: def is_file(self, path: PurePosixPath) -> Awaitable[bool]:
""" if the path is file """ """ if the path is file """
... ...
@overload
@abstractmethod @abstractmethod
def open(self, path: PurePosixPath, mode: Literal["r", "w"] = "r") -> BinaryIO | Awaitable[BinaryIO]: def open(self, path: PurePosixPath, mode: Literal["r"] = "r") -> Awaitable[AsyncFileReader]: ...
@overload
@abstractmethod
def open(self, path: PurePosixPath, mode: Literal["w"]) -> Awaitable[AsyncFileWriter]: ...
@abstractmethod
def open(self, path: PurePosixPath, mode: Literal["r", "w"] = "w") -> Awaitable[AsyncFileWriter] | Awaitable[AsyncFileReader]:
""" open file-like object for file operation """ """ open file-like object for file operation """
... ...
@abstractmethod @abstractmethod
def remove(self, path: PurePosixPath) -> None | Awaitable[None]: def remove(self, path: PurePosixPath) -> Awaitable[None]:
""" delete file or dir from storage, dir should be empty """ """ delete file or dir from storage, dir should be empty """
... ...
@abstractmethod @abstractmethod
def rmtree(self, path: PurePosixPath) -> None | Awaitable[None]: def rmtree(self, path: PurePosixPath) -> Awaitable[None]:
""" remove dir tree, remove all contents """ """ remove dir tree, remove all contents """
... ...
@abstractmethod @abstractmethod
def makedirs(self, path: PurePosixPath, exists_ok=False) -> None | Awaitable[None]: def makedirs(self, path: PurePosixPath, exists_ok=False) -> Awaitable[None]:
""" create dir """ """ create dir """
... ...

158
backup_box/storage/vnode.py
View File

@@ -1,46 +1,30 @@
from typing import TypedDict, Literal, BinaryIO from typing import TypedDict, Literal, overload
from collections.abc import Awaitable
from pathlib import PurePosixPath from pathlib import PurePosixPath
from functools import cmp_to_key from functools import cmp_to_key
from io import BytesIO from .storage import Storage, AsyncFileWriter, AsyncFileReader
from abc import abstractmethod
class StorageFile(TypedDict): class VFile(TypedDict):
ty: Literal["file"] # type ty: Literal["file"] # type
n: str # virtual file name n: str # virtual file name
mt: int # modify time mt: int # modify time
sz: int # size sz: int # size
class StorageDir(TypedDict): class VDir(TypedDict):
ty: Literal["dir"] # type ty: Literal["dir"] # type
n: str # virtual file name n: str # virtual file name
c: list["StorageItem"] # dir content c: list["VFSItem"] # dir content
StorageItem = StorageFile | StorageDir VFSItem = VFile | VDir
def new_storage_dir(n="", c=[]) -> StorageDir: def new_storage_dir(n="", c=[]) -> VDir:
return StorageDir(ty="dir", n=n, c=c) c = list(c)
return VDir(ty="dir", n=n, c=c)
def new_storage_file(n="", mt=0, sz=0) -> StorageFile: def new_storage_file(n="", mt=0, sz=0) -> VFile:
return StorageFile(ty="file", n=n, mt=mt, sz=sz) return VFile(ty="file", n=n, mt=mt, sz=sz)
class VFSError(RuntimeError): def find_vnode(root: VDir, path: PurePosixPath) -> list[VFSItem]:
def __init__(self, *args: object) -> None:
super().__init__(*args)
class FileNotFoundError(VFSError):
def __init__(self, p) -> None:
super().__init__(f"FileNotFound: {p}")
class NotADirectoryError(VFSError):
def __init__(self, p) -> None:
super().__init__(f"NotADirectory: {p}")
class DirectoryNotEmptyError(VFSError):
def __init__(self, p) -> None:
super().__init__(f"DirectoryNotEmpty: {p}")
def find_vnode(root: StorageDir, path: PurePosixPath) -> list[StorageItem]:
if len(path.parts) > 0: if len(path.parts) > 0:
p = path.parts[0] p = path.parts[0]
for n in root["c"]: for n in root["c"]:
@@ -55,7 +39,7 @@ def find_vnode(root: StorageDir, path: PurePosixPath) -> list[StorageItem]:
raise FileNotFoundError(path) raise FileNotFoundError(path)
return [root] return [root]
def _cmp_storage_item1(item1: StorageItem, item2: StorageItem): def _cmp_storage_item1(item1: VFSItem, item2: VFSItem):
""" cmp, sort dir first, then name """ """ cmp, sort dir first, then name """
if item1["ty"] != item2["ty"]: if item1["ty"] != item2["ty"]:
if item1["ty"] == "dir": if item1["ty"] == "dir":
@@ -69,7 +53,7 @@ def _cmp_storage_item1(item1: StorageItem, item2: StorageItem):
return -1 return -1
return 0 return 0
def _cmp_storage_item2(item1: StorageItem, item2: StorageItem): def _cmp_storage_item2(item1: VFSItem, item2: VFSItem):
""" cmp, sort file first, then name """ """ cmp, sort file first, then name """
if item1["ty"] != item2["ty"]: if item1["ty"] != item2["ty"]:
if item1["ty"] == "dir": if item1["ty"] == "dir":
@@ -83,10 +67,10 @@ def _cmp_storage_item2(item1: StorageItem, item2: StorageItem):
return -1 return -1
return 0 return 0
def sort_storage_items(items: list[StorageItem], dir_first = True): def sort_storage_items(items: list[VFSItem], dir_first = True):
items.sort(key=cmp_to_key(_cmp_storage_item1 if dir_first else _cmp_storage_item2)) items.sort(key=cmp_to_key(_cmp_storage_item1 if dir_first else _cmp_storage_item2))
def dump_storage_item_lines(item: StorageItem, indent=0): def dump_storage_item_lines(item: VFSItem, indent=0):
indent_text = " " * indent indent_text = " " * indent
if item["ty"] == "file": if item["ty"] == "file":
return [f"{indent_text}📄{item["n"]}: size {item["sz"]}"] return [f"{indent_text}📄{item["n"]}: size {item["sz"]}"]
@@ -98,46 +82,85 @@ def dump_storage_item_lines(item: StorageItem, indent=0):
return lines return lines
return [] return []
class VFileSystem(): class VFileSystem(Storage):
def __init__(self) -> None: def __init__(self) -> None:
self._root = new_storage_dir() self._root = new_storage_dir()
def __repr__(self): def dump_fs_tree(self):
return "\n".join(dump_storage_item_lines(self._root)) return "\n".join(dump_storage_item_lines(self._root))
def _on_remove_node(self, node_path: list[StorageItem]) -> None | Awaitable[None]: @abstractmethod
parents_path = PurePosixPath( *(n["n"] for n in node_path) ) async def _on_remove_node(self, node_path: list[VFSItem]) -> None:
print("remove:", parents_path) """ remove a file or dir, dir is guaranteed to be empty. """
pass path = PurePosixPath( *(n["n"] for n in node_path) )
print("remove:", path)
...
def _op_open_node(self, node_path: list[StorageItem]) -> BinaryIO | Awaitable[BinaryIO]: @abstractmethod
return BytesIO() async def _on_create_node(self, node_path: list[VFSItem]) -> None:
""" create a file or dir, parent dir is always created first. """
path = PurePosixPath( *(n["n"] for n in node_path) )
print("create:", path)
...
def listdir(self, path=PurePosixPath("")) -> list[PurePosixPath]: @overload
@abstractmethod
async def _on_open_node(self, node_path: list[VFSItem], mode: Literal["r"] = "r") -> AsyncFileReader: ...
@overload
@abstractmethod
async def _on_open_node(self, node_path: list[VFSItem], mode: Literal["w"]) -> AsyncFileWriter: ...
@abstractmethod
async def _on_open_node(self, node_path: list[VFSItem], mode: Literal["r", "w"] = "r") -> AsyncFileWriter | AsyncFileReader:
print("open", mode, ":", node_path)
...
async def listdir(self, path=PurePosixPath("")):
nodes = find_vnode(self._root, path) nodes = find_vnode(self._root, path)
base_path = PurePosixPath( *(n["n"] for n in nodes) ) base_path = PurePosixPath( *(n["n"] for n in nodes) )
item = nodes[-1] item = nodes[-1]
if item["ty"] != "dir": if item["ty"] != "dir":
raise NotADirectoryError(path) raise NotADirectoryError(path)
sort_storage_items(item["c"], True) sort_storage_items(item["c"], True)
return [ base_path.joinpath(n["n"]) for n in item["c"]] for n in item["c"]:
yield base_path.joinpath(n["n"])
def exists(self, path: PurePosixPath) -> bool: async def exists(self, path: PurePosixPath) -> bool:
try: try:
_ = find_vnode(self._root, path) _ = find_vnode(self._root, path)
return True return True
except FileNotFoundError: except FileNotFoundError:
return False return False
def is_dir(self, path: PurePosixPath) -> bool: async def is_dir(self, path: PurePosixPath) -> bool:
node = find_vnode(self._root, path)[-1] node = find_vnode(self._root, path)[-1]
return node["ty"] == "dir" return node["ty"] == "dir"
def is_file(self, path: PurePosixPath) -> bool: async def is_file(self, path: PurePosixPath) -> bool:
node = find_vnode(self._root, path)[-1] node = find_vnode(self._root, path)[-1]
return node["ty"] == "file" return node["ty"] == "file"
def remove(self, path: PurePosixPath) -> None: @overload
async def open(self, path: PurePosixPath, mode: Literal["r"] = "r") -> AsyncFileReader: ...
@overload
async def open(self, path: PurePosixPath, mode: Literal["w"]) -> AsyncFileWriter: ...
async def open(self, path: PurePosixPath, mode: Literal["r", "w"] = "r") -> AsyncFileReader | AsyncFileWriter:
try:
nodes = find_vnode(self._root, path)
except FileNotFoundError:
if mode == "r":
raise
# try to create file
nodes = find_vnode(self._root, path.parent)
filenode = new_storage_file(path.name)
if nodes[-1]["ty"] == "dir":
nodes[-1]["c"].append(filenode)
nodes.append(filenode)
await self._on_create_node(nodes)
else:
raise NotADirectoryError(path.parent)
return await self._on_open_node(nodes, mode)
async def remove(self, path: PurePosixPath) -> None:
nodes = find_vnode(self._root, path) nodes = find_vnode(self._root, path)
if len(nodes) == 1: if len(nodes) == 1:
# in the root # in the root
@@ -149,13 +172,13 @@ class VFileSystem():
last = nodes[-1] last = nodes[-1]
if last["ty"] == "dir": if last["ty"] == "dir":
if len(last["c"]) > 0: if len(last["c"]) > 0:
raise DirectoryNotEmptyError(path) raise OSError(f"{path} is not empty.")
# remove callback # remove callback
self._on_remove_node(nodes) await self._on_remove_node(nodes)
if parent["ty"] == "dir": if parent["ty"] == "dir":
parent["c"].remove(last) parent["c"].remove(last)
def rmtree(self, path: PurePosixPath) -> None: async def rmtree(self, path: PurePosixPath) -> None:
nodes = find_vnode(self._root, path) nodes = find_vnode(self._root, path)
if len(nodes) == 1: if len(nodes) == 1:
# in the root # in the root
@@ -180,7 +203,7 @@ class VFileSystem():
# remove callback # remove callback
node_path = list(prefix) node_path = list(prefix)
node_path.extend(vstack) node_path.extend(vstack)
self._on_remove_node(node_path) await self._on_remove_node(node_path)
# remove self from parent # remove self from parent
if parent["ty"] == "dir": if parent["ty"] == "dir":
parent["c"].remove(last) parent["c"].remove(last)
@@ -197,3 +220,36 @@ class VFileSystem():
if len(vstack) <= 0: if len(vstack) <= 0:
break break
async def makedirs(self, path: PurePosixPath, exists_ok=False) -> None:
build: list[VFSItem] = []
# ensure parents dir
parent = self._root
for p in path.parent.parts:
for n in parent["c"]:
if n["n"] == p:
if n["ty"] != "dir":
raise NotADirectoryError(n["n"])
parent = n
build.append(n)
break
else:
# not found, create dir
node = new_storage_dir(p)
parent["c"].append(node)
parent = node
build.append(node)
await self._on_create_node(build)
# create dir
for n in parent["c"]:
if n["n"] == path.name:
if not exists_ok:
raise FileExistsError(path)
if n["ty"] != "dir":
raise NotADirectoryError(path)
break
else:
node = new_storage_dir(path.name)
parent["c"].append(node)
build.append(node)
await self._on_create_node(build)

5
pyproject.toml
View File

@@ -4,8 +4,9 @@ version = "0.0.1"
license = "MIT" license = "MIT"
authors = [{ name="Dreagonmon" }] authors = [{ name="Dreagonmon" }]
dependencies = [ dependencies = [
"textual>=6.5.0", "textual~=6.5.0",
"tomli-w >= 1.2.0", "tomli-w~=1.2.0",
"aiofiles~=25.1.0"
] ]
requires-python = ">=3.11" requires-python = ">=3.11"