Spaces:
Paused
Paused
File size: 3,543 Bytes
99a7ebb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from services.storage.base import StorageBackend
class JSONStorageBackend(StorageBackend):
"""本地 JSON 文件存储后端"""
def __init__(self, file_path: Path, auth_keys_path: Path | None = None):
self.file_path = file_path
self.auth_keys_path = auth_keys_path or file_path.with_name("auth_keys.json")
self.file_path.parent.mkdir(parents=True, exist_ok=True)
self.auth_keys_path.parent.mkdir(parents=True, exist_ok=True)
@staticmethod
def _load_json_list(file_path: Path) -> list[dict[str, Any]]:
if not file_path.exists():
return []
try:
data = json.loads(file_path.read_text(encoding="utf-8"))
return data if isinstance(data, list) else []
except (json.JSONDecodeError, Exception):
return []
@staticmethod
def _save_json_list(file_path: Path, items: list[dict[str, Any]]) -> None:
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(
json.dumps(items, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def load_accounts(self) -> list[dict[str, Any]]:
"""从 JSON 文件加载账号数据"""
return self._load_json_list(self.file_path)
def save_accounts(self, accounts: list[dict[str, Any]]) -> None:
"""保存账号数据到 JSON 文件"""
self._save_json_list(self.file_path, accounts)
def load_auth_keys(self) -> list[dict[str, Any]]:
"""从 JSON 文件加载鉴权密钥数据"""
if not self.auth_keys_path.exists():
return []
try:
data = json.loads(self.auth_keys_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, Exception):
return []
if isinstance(data, dict):
data = data.get("items")
return data if isinstance(data, list) else []
def save_auth_keys(self, auth_keys: list[dict[str, Any]]) -> None:
"""保存鉴权密钥数据到 JSON 文件"""
self.auth_keys_path.parent.mkdir(parents=True, exist_ok=True)
self.auth_keys_path.write_text(
json.dumps({"items": auth_keys}, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def health_check(self) -> dict[str, Any]:
"""健康检查"""
try:
# 检查文件是否可读写
if self.file_path.exists():
self.file_path.read_text(encoding="utf-8")
return {
"status": "healthy",
"backend": "json",
"file_exists": self.file_path.exists(),
"file_path": str(self.file_path),
"auth_keys_file_exists": self.auth_keys_path.exists(),
"auth_keys_file_path": str(self.auth_keys_path),
}
except Exception as e:
return {
"status": "unhealthy",
"backend": "json",
"error": str(e),
}
def get_backend_info(self) -> dict[str, Any]:
"""获取存储后端信息"""
return {
"type": "json",
"description": "本地 JSON 文件存储",
"file_path": str(self.file_path),
"file_exists": self.file_path.exists(),
"auth_keys_file_path": str(self.auth_keys_path),
"auth_keys_file_exists": self.auth_keys_path.exists(),
}
|