diff --git a/storage/base_storage.py b/storage/base_storage.py index 0d47842..a0e21d3 100644 --- a/storage/base_storage.py +++ b/storage/base_storage.py @@ -1,2 +1,11 @@ # 存储接口 +class BaseStorage: + def save(self, key: str, value): + raise NotImplementedError + + def load(self, key: str, default=None): + raise NotImplementedError + + def delete(self, key: str): + raise NotImplementedError diff --git a/storage/json_storage.py b/storage/json_storage.py index a8db291..b98cd73 100644 --- a/storage/json_storage.py +++ b/storage/json_storage.py @@ -1 +1,40 @@ -# json 文件实现 \ No newline at end of file +# json 文件实现 + +import json + +from storage.base_storage import BaseStorage + + +class JSONStorage(BaseStorage): + def __init__(self, filename="data.json"): + self.filename = filename + # 如果文件不存在,先创建空字典 + try: + with open(self.filename, "r") as f: + pass + except OSError: + with open(self.filename, "w") as f: + f.write("{}") + + def _read_all(self): + with open(self.filename, "r") as f: + return json.load(f) + + def _write_all(self, data): + with open(self.filename, "w") as f: + json.dump(data, f) + + def save(self, key, value): + data = self._read_all() + data[key] = value + self._write_all(data) + + def load(self, key, default=None): + data = self._read_all() + return data.get(key, default) + + def delete(self, key): + data = self._read_all() + if key in data: + del data[key] + self._write_all(data) diff --git a/utils/fs.py b/utils/fs.py new file mode 100644 index 0000000..c98c7dc --- /dev/null +++ b/utils/fs.py @@ -0,0 +1,57 @@ +# 兼容PC和MicroPython的文件操作 + +# compat_fs.py +try: + import uos as os # MicroPython + MICROPYTHON = True +except ImportError: + import os # CPython + MICROPYTHON = False + +def list_dir(path="."): + """列出目录内容""" + return os.listdir(path) + +def make_dir(path): + """创建目录""" + if MICROPYTHON: + os.mkdir(path) + else: + os.makedirs(path, exist_ok=True) + +def remove_file(path): + """删除文件""" + os.remove(path) + +def read_file(path, mode="r"): + """读取文件内容""" + with open(path, mode) as f: + return f.read() + +def write_file(path, data, mode="w"): + """写入文件内容""" + with open(path, mode) as f: + f.write(data) + +def is_file(path): + """判断是否是文件""" + try: + st = os.stat(path) + # MicroPython: stat()[0] >> 14 & 0xF == 8 表示普通文件 + if MICROPYTHON: + return (st[0] >> 14) & 0xF == 8 + else: + return os.path.isfile(path) + except OSError: + return False + +def is_dir(path): + """判断是否是目录""" + try: + st = os.stat(path) + if MICROPYTHON: + return (st[0] >> 14) & 0xF == 2 + else: + return os.path.isdir(path) + except OSError: + return False