ayaka.storage

  1'''读写本地文件'''
  2import json
  3from pathlib import Path
  4from typing import TYPE_CHECKING
  5from .json_ctrl import AbstractJsonCtrl
  6
  7if TYPE_CHECKING:
  8    from .ayaka import AyakaApp
  9
 10
 11class AyakaStorage:
 12    def __init__(self, app: "AyakaApp") -> None:
 13        self.app = app
 14
 15    def plugin_path(self, *names):
 16        '''获取路径 <create_app_file>/../*names'''
 17        names = [str(name) for name in names]
 18        path = Path(self.app.path.parent, *names)
 19        return AyakaPath(path)
 20
 21    def group_path(self, *names):
 22        '''获取路径 data/groups/<bod_id>/<group_id>/<app.name>/*names'''
 23        names = [
 24            "data", "groups",
 25            self.app.bot_id,
 26            self.app.group_id,
 27            self.app.name,
 28            *names
 29        ]
 30        names = [str(name) for name in names]
 31        path = Path(*names)
 32        return AyakaPath(path)
 33
 34
 35class AyakaPath:
 36    '''文件夹路径'''
 37
 38    def __init__(self, path=Path("test")) -> None:
 39        self.path = path
 40        if not path.exists():
 41            path.mkdir(parents=True)
 42
 43    def iterdir(self):
 44        return self.path.iterdir()
 45
 46    def file(self, name, default=None):
 47        path = self.path / str(name)
 48        file = AyakaFile(path)
 49        if not path.exists() and default is not None:
 50            file.save(default)
 51        return file
 52
 53    def json(self, name, default={}):
 54        path = self.path / str(name)
 55        path = path.with_suffix(".json")
 56        file = AyakaJsonFile(path)
 57        if not path.exists():
 58            file.save(default)
 59        return file
 60
 61
 62class AyakaFile:
 63    '''文件'''
 64
 65    def __init__(self, path: Path):
 66        self.path = path
 67
 68    def load(self):
 69        with self.path.open("r", encoding="utf8") as f:
 70            data = f.read()
 71        return data
 72
 73    def save(self, data):
 74        with self.path.open("w+", encoding="utf8") as f:
 75            f.write(str(data))
 76
 77
 78class AyakaJsonFile:
 79    '''JSON文件'''
 80
 81    def __init__(self, path: Path) -> None:
 82        self.path = path
 83
 84    def chain(self, *keys):
 85        return AyakaJsonFileCtrl(self.path, *keys)
 86
 87    def load(self):
 88        with self.path.open("r", encoding="utf8") as f:
 89            data = json.load(f)
 90        return data
 91
 92    def save(self, data):
 93        with self.path.open("w+", encoding="utf8") as f:
 94            json.dump(data, f, ensure_ascii=False)
 95
 96
 97class AyakaJsonFileCtrl(AbstractJsonCtrl):
 98    '''AyakaJsonFileCtrl实际上可兼容替代AyakaJsonFile，但是为了避免语义上的混乱，仍分作两个类'''
 99
100    def __init__(self, path: Path, *keys) -> None:
101        self._path = path
102        self._keys = [str(k) for k in keys]
103
104    def _load(self):
105        with self._path.open("r", encoding="utf8") as f:
106            data = json.load(f)
107        return data
108
109    def _save(self, data):
110        with self._path.open("w+", encoding="utf8") as f:
111            json.dump(data, f, ensure_ascii=False)
112
113    def chain(self, *keys):
114        return AyakaJsonFileCtrl(self._path, *self._keys, *keys)