如何在较长的非json文本中检测和缩进json子字符串？

Question

我有一个现有的Python应用程序，其日志记录如下：

import logging
import json
logger = logging.getLogger()

some_var = 'abc'
data = {
   1: 2,
   'blah': {
      ['hello']
   }
}

logger.info(f"The value of some_var is {some_var} and data is {json.dumps(data)}")

因此logger.info函数被赋予：

The value of some_var is abc and data is {1: 2,"blah": {["hello"]}}

当前，我的日志进入AWS CloudWatch，它做了一些魔术，并使用缩进方式呈现该文本：

The value of some_var is abc and data is {
   1: 2,
   "blah": {
      ["hello"]
   }
}

这使日志超清晰可读。

现在，我想对日志记录进行一些更改，使用另一个环绕我的代码的python脚本自己处理它，并在发生故障时通过电子邮件发送日志。

我想要的是获取每个日志条目（或条目的流/列表）并应用此缩进的某种方式。

所以我想要一个函数，它接收一个字符串，并检测该字符串的哪些子集是json，然后插入\n和以漂亮地打印该json。

示例输入：

你好，{“ a”：{“ b”：“ c”}}是一些json数据，但是{“ c”：[1,2,3]}也是]

示例输出

Hello, 
{
  "a": {
    "b": "c"
  }
} 
is some json data, but also 
{
  "c": [
    1,
    2,
    3
  ]
}
is too

我已经考虑过将每个条目分成第一个{之前和之后的所有内容。保留左半部分不变，然后将右半部分传递给json.dumps(json.loads(x), indent=4)。

但是如果日志文件中的json对象后面有东西怎么办？好的，我们只需选择第一个{之后和最后一个}之前的所有内容。然后将中间位传递给JSON库。

但是如果此日志条目中有两个JSON对象怎么办？ （就像上面的示例一样。）我们必须使用堆栈来确定在用相应的{关闭所有先前的{之后是否出现任何}。

但是如果出现类似{"a": "\}"}的情况该怎么办。嗯，好的，我们需要处理转义。现在，我发现自己不得不从头开始编写整个json解析器。

有没有简单的方法可以做到这一点？

我想我可以使用正则表达式将整个仓库中的每个json.dumps(x)实例替换为json.dumps(x, indent=4)。但是json.dumps有时在日志记录语句之外使用，这只会使我的所有日​​志记录行加长一点。有没有一个简洁而优雅的解决方案？

（（如果可以解析并缩进str(x)在python中生成的类似json的输出，奖金会指出。这基本上是带有单引号而不是双引号的json。）

Answer 1

为了从字符串中提取JSON对象，请参见this answer。该答案的extract_json_objects()函数将处理JSON对象和嵌套的JSON对象，但仅此而已。如果您的日志中有一个列表[[JSON对象之外，则不会被选中。

在您的情况下，将函数修改为

also

返回所有JSON对象周围的字符串/文本，以便您可以将它们全部放入日志中（或替换日志行）：

from json import JSONDecoder

def extract_json_objects(text, decoder=JSONDecoder()):
    pos = 0
    while True:
        match = text.find('{', pos)
        if match == -1:
            yield text[pos:]  # return the remaining text
            break
        yield text[pos:match]  # modification for the non-JSON parts
        try:
            result, index = decoder.raw_decode(text[match:])
            yield result
            pos = match + index
        except ValueError:
            pos = match + 1

使用该函数来处理日志行，将它们添加到
strings

列表中，然后将它们合并在一起以为输出，记录器等生成单个字符串：

def jsonify_logline(line):
    line_parts = []
    for result in extract_json_objects(line):
        if isinstance(result, dict):  # got a JSON obj
            line_parts.append(json.dumps(result, indent=4))
        else:                         # got text/non-JSON-obj
            line_parts.append(result)
    # (don't make that a list comprehension, quite un-readable)

    return ''.join(line_parts)

示例：
>>> demo_text = """Hello, {"a": {"b": "c"}} is some json data, but also {"c": [1,2,3]} is too"""
>>> print(jsonify_logline(demo_text))
Hello, {
    "a": {
        "b": "c"
    }
} is some json data, but also {
    "c": [
        1,
        2,
        3
    ]
} is too
>>>

其他没有直接关系的东西会有所帮助：

代替在所有日志行中使用json.dumps(x)，而是在DRY principle之后创建一个像logdump(x)一样的函数，该函数可以执行您想做的任何事情，例如json.dumps(x)，json.dumps(x, indent=4)或[C0 ]。这样，如果您需要更改
• all

日志的JSON格式，则只需更改该功能即可；无需进行大量的“搜索与替换”，而这本身就带有问题和极端情况。

您甚至可以向其添加一个可选参数jsonify_logline(x)，以决定是否希望其缩进。

您可以批量搜索并替换所有现有的日志行以进行pretty=True

如果您是JSON转储的自定义对象/类实例，请使用其logger.blah(jsonify_logline(<previous log f-string or text>))方法始终输出漂亮打印的JSON。并且__str__不漂亮/紧凑。

然后您根本不需要修改日志行。执行__str__将直接调用__repr__。