如何检测和缩进较长的非 json 文本中的 json 子字符串?
How to detect and indent json substrings inside longer non-json text?
我有一个现有的 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 脚本自行处理,该脚本环绕我的代码并在出现故障时通过电子邮件发送日志。
我想要的是获取每个日志条目(或 stream/list 个条目)并应用此缩进的方法。
所以我想要一个接受字符串的函数,并检测该字符串的哪些子集是 json,然后插入 \n
和 </code> 漂亮地打印 json.</p>
<p><strong>示例输入:</strong></p>
<blockquote>
<p>Hello, {"a": {"b": "c"}} is some json data, but also {"c": [1,2,3]} is too</p>
</blockquote>
<p><strong>示例输出</strong></p>
<pre><code>Hello,
{
"a": {
"b": "c"
}
}
is some json data, but also
{
"c": [
1,
2,
3
]
}
is too
我考虑过将每个条目分成第一个 {
之前和之后的所有内容。留下左半边,右半边传给json.dumps(json.loads(x), indent=4)
.
但是,如果日志文件中 json 对象之后有内容怎么办?
好的,我们可以 select 第一个 {
之后和最后一个 }
之前的所有内容。
然后将中间位传递给JSON库。
但是如果此日志条目中有两个 JSON 对象怎么办? (就像上面的例子一样。)我们必须使用一个堆栈来确定在所有先前的 {
已经用相应的 }
.[=31 关闭之后是否出现任何 {
=]
但是如果有类似 {"a": "\}"}
的东西呢?嗯,好的,我们需要处理转义。
现在我发现自己不得不从头开始编写整个 json 解析器。
有什么简单的方法可以做到这一点吗?
我想我可以使用正则表达式将整个存储库中的每个 json.dumps(x)
实例替换为 json.dumps(x, indent=4)
。但是 json.dumps
有时会在日志语句之外使用,它只会让我所有的日志行变得更长一点。有没有一个简洁优雅的解决方案?
(如果它可以解析和缩进 str(x)
在 python 中生成的类似 json 的输出,则加分。这基本上是 json 用单引号而不是双引号.)
要从字符串中提取 JSON 个对象,请参阅 this answer。该答案中的 extract_json_objects()
函数将处理 JSON 对象和嵌套的 JSON 对象,但仅此而已。如果您的日志中有一个 JSON 对象 之外的列表,则不会被拾取。
在你的情况下,将函数修改为 also return strings/text around all the JSON objects,这样你就可以把他们一起进入日志(或替换日志行):
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
使用该函数处理您的日志行,将它们添加到 字符串列表 ,然后您将这些字符串连接在一起为您的输出、记录器等生成单个字符串:
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)
,或 jsonify_logline(x)
。这样,如果您需要更改 所有 日志的 JSON 格式,您只需更改那个函数;不需要质量"search & replace",它有自己的问题和边缘情况。
- 您甚至可以向它添加一个可选参数
pretty=True
来决定是否要缩进。
- 您可以批量搜索并替换所有现有日志行来完成
logger.blah(jsonify_logline(<previous log f-string or text>))
- 如果您正在 JSON-转储自定义 objects/class 实例,则将它们的
__str__
method to always output pretty-printed JSON. And the __repr__
用作 non-pretty/compact。
- 那么你根本不需要修改日志行。执行
logger.info(f'here is my object {x}')
会直接调用 obj.__str__
.
我有一个现有的 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 脚本自行处理,该脚本环绕我的代码并在出现故障时通过电子邮件发送日志。
我想要的是获取每个日志条目(或 stream/list 个条目)并应用此缩进的方法。
所以我想要一个接受字符串的函数,并检测该字符串的哪些子集是 json,然后插入 \n
和 </code> 漂亮地打印 json.</p>
<p><strong>示例输入:</strong></p>
<blockquote>
<p>Hello, {"a": {"b": "c"}} is some json data, but also {"c": [1,2,3]} is too</p>
</blockquote>
<p><strong>示例输出</strong></p>
<pre><code>Hello,
{
"a": {
"b": "c"
}
}
is some json data, but also
{
"c": [
1,
2,
3
]
}
is too
我考虑过将每个条目分成第一个 {
之前和之后的所有内容。留下左半边,右半边传给json.dumps(json.loads(x), indent=4)
.
但是,如果日志文件中 json 对象之后有内容怎么办?
好的,我们可以 select 第一个 {
之后和最后一个 }
之前的所有内容。
然后将中间位传递给JSON库。
但是如果此日志条目中有两个 JSON 对象怎么办? (就像上面的例子一样。)我们必须使用一个堆栈来确定在所有先前的 {
已经用相应的 }
.[=31 关闭之后是否出现任何 {
=]
但是如果有类似 {"a": "\}"}
的东西呢?嗯,好的,我们需要处理转义。
现在我发现自己不得不从头开始编写整个 json 解析器。
有什么简单的方法可以做到这一点吗?
我想我可以使用正则表达式将整个存储库中的每个 json.dumps(x)
实例替换为 json.dumps(x, indent=4)
。但是 json.dumps
有时会在日志语句之外使用,它只会让我所有的日志行变得更长一点。有没有一个简洁优雅的解决方案?
(如果它可以解析和缩进 str(x)
在 python 中生成的类似 json 的输出,则加分。这基本上是 json 用单引号而不是双引号.)
要从字符串中提取 JSON 个对象,请参阅 this answer。该答案中的 extract_json_objects()
函数将处理 JSON 对象和嵌套的 JSON 对象,但仅此而已。如果您的日志中有一个 JSON 对象 之外的列表,则不会被拾取。
在你的情况下,将函数修改为 also return strings/text around all the JSON objects,这样你就可以把他们一起进入日志(或替换日志行):
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
使用该函数处理您的日志行,将它们添加到 字符串列表 ,然后您将这些字符串连接在一起为您的输出、记录器等生成单个字符串:
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)
,或jsonify_logline(x)
。这样,如果您需要更改 所有 日志的 JSON 格式,您只需更改那个函数;不需要质量"search & replace",它有自己的问题和边缘情况。- 您甚至可以向它添加一个可选参数
pretty=True
来决定是否要缩进。
- 您甚至可以向它添加一个可选参数
- 您可以批量搜索并替换所有现有日志行来完成
logger.blah(jsonify_logline(<previous log f-string or text>))
- 如果您正在 JSON-转储自定义 objects/class 实例,则将它们的
__str__
method to always output pretty-printed JSON. And the__repr__
用作 non-pretty/compact。- 那么你根本不需要修改日志行。执行
logger.info(f'here is my object {x}')
会直接调用obj.__str__
.
- 那么你根本不需要修改日志行。执行