限制记录的项目数/ json 长度

Question

我正在开发一个返回 JSON 的 API。我正在记录我的回复，有时 JSON 太长了，基本上会阻塞我的日志文件。有没有一种简洁的方法来减少 JSON 的长度，纯粹是为了直观地记录数据？ （在生产中无效）

基本方法是将长度超过 5 的 数组 减少到 [first 2, "...", last 2] 和超过 4 的 dictionaries {first 4, "..." : "..."}

的项目

下面的代码很难看。我知道它应该是一个递归解决方案，以相同的方式减少任意深度的 JSON 的项目 - 它目前只对深度 2 这样做。

def log_reducer(response_log):
original_response_log = response_log
try:
    if type(response_log) == dict:
        if len(response_log) >= 4:  # {123456}
            response_log = dict(list(response_log.items())[:4])
            response_log.update({"...": "..."})  # {1234...}
        for key, value in response_log.items():
            if type(value) == list:
                if len(value) >= 5:  # {key:[123456]}
                    new_item = value[:2] + ['...'] + value[-2:]  # {[12...56]}
                    response_log.update({key: new_item})
            if type(value) == dict:
                if len(value) >= 4:  # {key:{123456}}
                    reduced_dict = dict(list(value.items())[:4])
                    reduced_dict.update({"...": "..."})
                    response_log.update({key: reduced_dict})  # {{1234...}}

    elif type(response_log) == list:
        if len(response_log) >= 5:  # [123456]
            response_log = response_log[:2] + ['...'] + response_log[-2:]  # [12...56]
        for inner_item in response_log:
            if type(inner_item) == list:
                if len(inner_item) >= 5:  # [[123456]]
                    reduced_list = inner_item[:2] + ['...'] + inner_item[-2:]  # [[12...56]]
                    response_log.remove(inner_item)
                    response_log.append(reduced_list)
            if type(inner_item) == dict:
                if len(inner_item) >= 4:  # [{123456}]
                    reduced_dict = dict(list(inner_item.items())[:4])
                    reduced_dict.update({"...": "..."})  # [{1234...}]
                    response_log.remove(inner_item)
                    response_log.append(reduced_dict)
except Exception as e:
    return original_response_log
return response_log

然后使用 logger.info(str(response_log)) 记录返回的 response_log

如您所见，每个级别都可以有数组或字典这一事实使这项任务更加复杂，我正在努力寻找可以简化此任务的任何类型的库或代码片段。如果有人想试一试，我将不胜感激。

您可以使用这样的测试 JSON 来查看它的效果：

test_json = {"works": [1, 2, 3, 4, 5, 6],
             "not_affected": [{"1": "1", "2": "2", "3": "3", "4": "4", "5": "5"}],
             "1": "1", "2": "2", "3": "3",
             "removed": "removed"
             }
print("original", test_json)
reduced_log = log_reducer(test_json)
print("reduced", reduced_log)

print("original", test_json)
reduced_log = log_reducer([test_json])  # <- increases nesting depth
print("reduced", reduced_log)

Answer 1

这个答案使用@calceamenta的想法，但实现了实际的削减逻辑：

def recursive_reduce(obj):
    if isinstance(obj, (float, str, int, bool, type(None))):
        return obj

    if isinstance(obj, dict):
        keys = list(sorted(obj))
        obj['...'] = '...'

        if len(keys) > 5:
            new_keys = keys[:2] + ["..."] + keys[-2:]
        else:
            new_keys = keys

        new_dict = {x:obj[x] for x in new_keys}
        for k, v in new_dict.items():
            new_dict[k] = recursive_reduce(v)

        return new_dict

    if isinstance(obj, list):
        if len(obj) > 5:
            new_list = obj[:2] + ["..."] + obj[-2:]
        else:
            new_list = obj

        for i, v in enumerate(new_list):
            new_list[i] = recursive_reduce(v)

        return new_list

    return str(obj)

test_json = {"works": [1, 2, 3, 4, 5, 6],
             "not_affected": [{"1": "1", "2": "2", "3": "3", "4": "4", "5": "5"}],
             "1": "1", "2": "2", "3": "3",
             "removed": "removed"
             }

print("original", test_json)
reduced_log = recursive_reduce(test_json)
print("reduced", reduced_log)

输出：

original {'works': [1, 2, 3, 4, 5, 6], 'not_affected': [{'1': '1', '2': '2', '3': '3', '4': '4', '5': '5'}], '1': '1', '2': '2', '3': '3', 'removed': 'removed'}
reduced {'1': '1', '2': '2', '...': '...', 'removed': 'removed', 'works': [1, 2, '...', 5, 6]}

希望这会有所帮助:)

Answer 2

您可以使用 def __str__(): 方法覆盖 Python 中字典和列表的 字符串表示。使用它只是递归调用所有元素的打印函数。它可以有一个像这样的简单样板：

def custom_print(obj):
    log_str = ''
    if type(obj) == list:
        for item in obj:
            log_str += custom_print(item)
    elif type(obj) == dict:
        for k, item in obj.items():
            custom_print(item)

使用此自定义日志功能根据您的日志文件格式打印到您的日志文件中。