Python通用JSON写入工具data_manager类write_to_json方法异常求助

需求说明 #

我们需要实现一个纯Python编写的通用JSON操作工具，核心能力由data_manager类提供，支持创建JSON文件、向指定层级追加或更新数据，初始化时生成的JSON文件根节点名称与文件名一致。
核心逻辑通过write_to_json方法承载，方法定义为：

def write_to_json(self, new_data, base = ""):

需要覆盖4类使用场景（match指待写入数据的key已存在于目标层级，base指待写入的目标节点，为空则默认写入根节点）：

• 场景1：无匹配key、未指定base：将数据追加到根节点列表
• 场景2：无匹配key、已指定base：将数据追加到指定base节点下
• 场景3：有匹配key、未指定base：若新旧值不同，替换根节点下对应key的值
• 场景4：有匹配key、已指定base：若新旧值不同，替换指定base节点下对应key的值

原代码已实现场景1、3的能力，但场景2、4缺少节点定位和写入逻辑，导致指定base的操作无法生效。

核心问题分析 #

原代码存在两个核心缺陷：

1. 没有实现base节点的定位逻辑，指定base参数后仅打印日志，没有实际操作数据
2. key匹配为全局搜索，未限定在指定base层级下，存在不同层级同名key匹配错误的问题

修复后的完整代码 #

import os
import json

class file_manager:
    def set_file_contents(self, file_name, contents):
        with open(file_name, "w", encoding="utf-8") as file:
            file.write(contents)

    def set_file_contents_append(self, file_name, contents):
        if not os.path.exists(file_name):
            open(file_name, 'a', encoding="utf-8").close()
        with open(file_name, "a", encoding="utf-8") as file:
            file.write(contents)

    def get_file_size(self,file_name):
        return os.path.getsize(file_name) if os.path.exists(file_name) else 0

    def get_file_content_json(self,file_name):
        if self.get_file_size(file_name) == 0:
            return None
        with open(file_name, "r", encoding="utf-8") as f:
            return json.load(f)
        
class data_manager(file_manager):
    def __init__(self,data_name):
        self.data_base = data_name
        self.data_file = data_name + '.json'
        self.create_data_file()

    def create_base(self,base):
        data = {base:[]}      
        self.set_file_contents(self.data_file,json.dumps(data,indent=4, ensure_ascii=False))
  
    def create_data_file(self):
        if not os.path.exists(self.data_file) or self.get_file_size(self.data_file) == 0:
            self.create_base(self.data_base)

    def check_file_size(self):
        print(self.get_file_size(self.data_file))

    # 新增：定位指定base的节点
    def locate_base_node(self, root_data, base_name):
        # 根节点直接返回
        if base_name == self.data_base:
            return root_data[self.data_base]
        # 递归搜索base节点
        for item in root_data[self.data_base]:
            if isinstance(item, dict):
                for k, v in item.items():
                    if k == base_name:
                        return v
                    res = self.locate_base_node({self.data_base: v}, base_name)
                    if res is not None:
                        return res
        return None

    # 新增：在指定节点下搜索key是否存在
    def check_key_in_node(self, node, key):
        if isinstance(node, list):
            for item in node:
                if isinstance(item, dict) and key in item:
                    return item
        elif isinstance(node, dict):
            if key in node:
                return node
        return None

    def check_if_exist(self, data_name):
        file_data = self.get_file_content_json(self.data_file) 
        for item in file_data[self.data_base]:
            if data_name in item.keys():
                print(f'{data_name}: {item[data_name]}')
                return True
        return False

    def get_data_value(self,data_name):
        d = self.get_file_content_json(self.data_file) 
        d = d[self.data_base]
        items = []
        for item in self.item_generator(d,data_name):
            items.append(item)
        return dict(items[0]) if items else None

    def item_generator(self,json_input, lookup_key):
        if isinstance(json_input, dict):
            for key, value in json_input.items():
                if key == lookup_key:      
                    yield {key:value} 
                else:
                    yield from self.item_generator(value, lookup_key)
        elif isinstance(json_input, list):
            for item in json_input:
                yield from self.item_generator(item, lookup_key)
                
    def replace_data_value_json(self, file_data, data_name, data_value):
        for item in file_data:
            if data_name in item.keys():
                item[data_name] = data_value
        return file_data

    def set_data_value(self, data_name, data_value):
        file_data = self.get_file_content_json(self.data_file) 
        for item in file_data[self.data_base]:
            if data_name in item.keys():
                item[data_name] = data_value
        self.set_file_contents(self.data_file,json.dumps(file_data,indent=4, ensure_ascii=False))

    def view_all_data(self):
        file_data = self.get_file_content_json(self.data_file) 
        print(json.dumps(file_data, indent=4, ensure_ascii=False))
     
    def remove_data_item(self, data_name):
        file_data = self.get_file_content_json(self.data_file) 
        for element in file_data[self.data_base]:
            if data_name in element:
                del element[data_name]
        self.set_file_contents(self.data_file,json.dumps(file_data,indent=4, ensure_ascii=False).replace('{}','').replace('{},','') )

    def prettyjson(self,data):
        return json.dumps(data,indent=4, ensure_ascii=False)

    def compare_equal(self, value1, value2):
        return str(value1) == str(value2)

    # 重写后的write_to_json方法
    def write_to_json(self, new_data, base = ""):
        file_data = self.get_file_content_json(self.data_file) 
        target_base = self.data_base if base == "" else base

        # 定位目标base节点
        target_node = self.locate_base_node(file_data, target_base)
        if target_node is None:
            raise ValueError(f"指定的base节点{target_base}不存在")

        for data_key, data_value in new_data.items():
            # 只在当前目标节点下搜索匹配的key
            exist_item = self.check_key_in_node(target_node, data_key)
            data_single_item = {data_key: data_value}

            if not exist_item:
                # 无匹配key，追加到目标节点
                if isinstance(target_node, list):
                    target_node.append(data_single_item)
                elif isinstance(target_node, dict):
                    target_node[data_key] = data_value
            else:
                # 有匹配key，值不同则替换
                if not self.compare_equal(exist_item[data_key], data_value):
                    exist_item[data_key] = data_value

        # 持久化写入
        self.set_file_contents(self.data_file, self.prettyjson(file_data))

    def add_data_single(self, data_name, data_value, base=""):
        new_data_item = {data_name: data_value}
        self.write_to_json(new_data_item, base)
    
    def add_data_multiple(self,data,base=""):
        self.write_to_json(data,base)


# 测试代码
if __name__ == "__main__":
    test = data_manager('people') 
    test.write_to_json({'John':[], 'Alex':[], 'Samantha':[]})
    # 测试场景2：指定base为John，追加数据
    test.write_to_json({"Favorite-Food":"tacos" , "Age":45}, "John")
    # 测试场景4：指定base为John，更新Age的值
    test.write_to_json({"Age":46}, "John")
    # 查看最终结果
    test.view_all_data()

运行效果 #

测试代码运行后，生成的people.json内容如下：

{
    "people": [
        {
            "John": [
                {
                    "Favorite-Food": "tacos"
                },
                {
                    "Age": 46
                }
            ]
        },
        {
            "Alex": []
        },
        {
            "Samantha": []
        }
    ]
}

所有4类场景均可正常运行。

内容的提问来源于stack exchange，提问作者John Anthony