20250904-初步功能已完成
This commit is contained in:
100
Utils/AiUtils.py
100
Utils/AiUtils.py
@@ -710,103 +710,3 @@ class AiUtils(object):
|
||||
except Exception as e:
|
||||
LogManager.error(f"[delete_anchors_by_ids] 写入失败: {e}")
|
||||
return deleted
|
||||
|
||||
@staticmethod
|
||||
def _get_data_path(filename: str) -> str:
|
||||
"""
|
||||
根据文件名生成 data 目录下的完整路径
|
||||
"""
|
||||
base_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__))) # 当前项目根目录
|
||||
data_dir = os.path.join(base_dir, "data")
|
||||
Path(data_dir).mkdir(parents=True, exist_ok=True) # 确保 data 目录存在
|
||||
return os.path.join(data_dir, filename)
|
||||
|
||||
@staticmethod
|
||||
def read_json(filename: str) -> dict:
|
||||
"""
|
||||
读取 JSON 文件，返回字典
|
||||
如果文件不存在，返回空字典
|
||||
"""
|
||||
file_path = AiUtils._get_data_path(filename)
|
||||
try:
|
||||
if not os.path.exists(file_path):
|
||||
return {}
|
||||
with open(file_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
return data if isinstance(data, dict) else {}
|
||||
except Exception as e:
|
||||
print(f"读取 JSON 文件失败: {e}")
|
||||
return {}
|
||||
|
||||
@staticmethod
|
||||
def write_json(filename: str, data: dict, overwrite: bool = True) -> bool:
|
||||
"""
|
||||
将字典写入 JSON 文件
|
||||
:param filename: 文件名（自动放在 data 目录下）
|
||||
:param data: 要写入的字典
|
||||
:param overwrite: True=覆盖写，False=合并更新
|
||||
"""
|
||||
file_path = AiUtils._get_data_path(filename)
|
||||
try:
|
||||
if not overwrite and os.path.exists(file_path):
|
||||
# 读取旧数据
|
||||
with open(file_path, "r", encoding="utf-8") as f:
|
||||
old_data = json.load(f)
|
||||
if not isinstance(old_data, dict):
|
||||
old_data = {}
|
||||
# 合并
|
||||
old_data.update(data)
|
||||
data = old_data
|
||||
|
||||
# 覆盖写（写回最终的 data）
|
||||
with open(file_path, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, ensure_ascii=False, indent=4)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"写入 JSON 文件失败: {e}")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def update_json(filename: str, new_data: dict) -> bool:
|
||||
"""
|
||||
修改 JSON 文件：
|
||||
- 如果 key 已存在，则修改其值
|
||||
- 如果 key 不存在，则新增
|
||||
"""
|
||||
try:
|
||||
# 读取旧数据（没有则为空字典）
|
||||
data = AiUtils.read_json(filename)
|
||||
if not isinstance(data, dict):
|
||||
data = {}
|
||||
|
||||
# 合并（有则修改，无则新增）
|
||||
data.update(new_data)
|
||||
|
||||
# 写回
|
||||
return AiUtils.write_json(filename, data)
|
||||
except Exception as e:
|
||||
print(f"更新 JSON 文件失败: {e}")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def delete_json_key(filename: str, key: str) -> bool:
|
||||
"""
|
||||
删除 JSON 文件中的某个 key
|
||||
- 如果 key 存在则删除
|
||||
- 如果 key 不存在则不处理
|
||||
"""
|
||||
try:
|
||||
# 读取旧数据
|
||||
data = AiUtils.read_json(filename)
|
||||
if not isinstance(data, dict):
|
||||
data = {}
|
||||
|
||||
# 删除指定 key
|
||||
if key in data:
|
||||
del data[key]
|
||||
|
||||
# 写回文件
|
||||
return AiUtils.write_json(filename, data)
|
||||
except Exception as e:
|
||||
print(f"删除 JSON key 失败: {e}")
|
||||
return False
|
||||
|
||||
100
Utils/JsonUtils.py
Normal file
100
Utils/JsonUtils.py
Normal file
@@ -0,0 +1,100 @@
|
||||
import os
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class JsonUtils:
|
||||
@staticmethod
|
||||
def _normalize_filename(filename: str) -> str:
|
||||
"""
|
||||
确保文件名以 .json 结尾
|
||||
"""
|
||||
if not filename.endswith(".json"):
|
||||
filename = f"{filename}.json"
|
||||
return filename
|
||||
|
||||
@staticmethod
|
||||
def _get_data_path(filename: str) -> str:
|
||||
"""
|
||||
根据文件名生成 data 目录下的完整路径
|
||||
"""
|
||||
filename = JsonUtils._normalize_filename(filename)
|
||||
base_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__))) # 当前项目根目录
|
||||
data_dir = os.path.join(base_dir, "data")
|
||||
Path(data_dir).mkdir(parents=True, exist_ok=True) # 确保 data 目录存在
|
||||
return os.path.join(data_dir, filename)
|
||||
|
||||
@staticmethod
|
||||
def read_json(filename: str) -> dict:
|
||||
"""
|
||||
读取 JSON 文件，返回字典
|
||||
如果文件不存在，返回空字典
|
||||
"""
|
||||
file_path = JsonUtils._get_data_path(filename)
|
||||
try:
|
||||
if not os.path.exists(file_path):
|
||||
return {}
|
||||
with open(file_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
return data if isinstance(data, dict) else {}
|
||||
except Exception as e:
|
||||
print(f"读取 JSON 文件失败: {e}")
|
||||
return {}
|
||||
|
||||
@staticmethod
|
||||
def write_json(filename: str, data: dict, overwrite: bool = True) -> bool:
|
||||
"""
|
||||
将字典写入 JSON 文件
|
||||
:param filename: 文件名（不用写后缀，自动补 .json）
|
||||
:param data: 要写入的字典
|
||||
:param overwrite: True=覆盖写，False=合并更新
|
||||
"""
|
||||
file_path = JsonUtils._get_data_path(filename)
|
||||
try:
|
||||
if not overwrite and os.path.exists(file_path):
|
||||
with open(file_path, "r", encoding="utf-8") as f:
|
||||
old_data = json.load(f)
|
||||
if not isinstance(old_data, dict):
|
||||
old_data = {}
|
||||
old_data.update(data)
|
||||
data = old_data
|
||||
|
||||
with open(file_path, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, ensure_ascii=False, indent=4)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"写入 JSON 文件失败: {e}")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def update_json(filename: str, new_data: dict) -> bool:
|
||||
"""
|
||||
修改 JSON 文件：
|
||||
- 如果 key 已存在，则修改其值
|
||||
- 如果 key 不存在，则新增
|
||||
"""
|
||||
try:
|
||||
data = JsonUtils.read_json(filename)
|
||||
if not isinstance(data, dict):
|
||||
data = {}
|
||||
data.update(new_data)
|
||||
return JsonUtils.write_json(filename, data)
|
||||
except Exception as e:
|
||||
print(f"更新 JSON 文件失败: {e}")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def delete_json_key(filename: str, key: str) -> bool:
|
||||
"""
|
||||
删除 JSON 文件中的某个 key
|
||||
"""
|
||||
try:
|
||||
data = JsonUtils.read_json(filename)
|
||||
if not isinstance(data, dict):
|
||||
data = {}
|
||||
if key in data:
|
||||
del data[key]
|
||||
return JsonUtils.write_json(filename, data)
|
||||
except Exception as e:
|
||||
print(f"删除 JSON key 失败: {e}")
|
||||
return False
|
||||
Reference in New Issue
Block a user