使用 Python 在 YAML 中获取重复键
Getting duplicate keys in YAML using Python
我们需要解析包含重复键的 YAML 文件，所有这些都需要解析。跳过重复项是不够的。我知道这违反了 YAML 规范，我不想这样做，但我们使用的第三方工具支持这种用法，我们需要处理它。
文件示例：
build:
step: 'step1'
build:
step: 'step2'
解析后我们应该有一个类似的数据结构：
yaml.load('file.yml')
# [('build', [('step', 'step1')]), ('build', [('step', 'step2')])]
dict不能再用来表示解析后的内容
我正在 Python 中寻找解决方案，但我没有找到支持这个的库，我错过了什么吗？
或者，我很乐意编写自己的东西，但希望它尽可能简单。 ruamel.yaml 看起来是 Python 中最先进的 YAML 解析器，并且它看起来具有适度的可扩展性，是否可以扩展以支持重复字段？
如果您可以非常轻微地修改输入数据，您应该可以通过将单个类似 yaml 的文件转换为多个 yaml 文档来做到这一点。如果 yaml 文档单独在一行中由 --- 分隔，则它们可以位于同一文件中，并且您很容易看到条目由两个相邻的换行符分隔：
with open('file.yml', 'r') as f:
data = f.read()
data = data.replace('\n\n', '\n---\n')
for document in yaml.load_all(data):
print(document)
输出：
{'build': {'step': 'step1'}}
{'build': {'step': 'step2'}}
PyYAML 只会默默地覆盖第一个条目，ruamel.yaml¹ 如果与旧版 API 一起使用，将给出 DuplicateKeyFutureWarning，并使用新的 DuplicateKeyError API。
如果您不想为所有类型创建完整的 Constructor，覆盖 SafeConstructor 中的映射构造函数应该可以完成工作：
import sys
from ruamel.yaml import YAML
from ruamel.yaml.constructor import SafeConstructor
yaml_str = """\
build:
step: 'step1'
build:
step: 'step2'
"""
def construct_yaml_map(self, node):
# test if there are duplicate node keys
data = []
yield data
for key_node, value_node in node.value:
key = self.construct_object(key_node, deep=True)
val = self.construct_object(value_node, deep=True)
data.append((key, val))
SafeConstructor.add_constructor(u'tag:yaml.org,2002:map', construct_yaml_map)
yaml = YAML(typ='safe')
data = yaml.load(yaml_str)
print(data)
给出：
[('build', [('step', 'step1')]), ('build', [('step', 'step2')])]
不过好像没必要把step: 'step1'做成列表。以下将仅在存在重复项时创建列表（如有必要，可以通过缓存 self.construct_object(key_node, deep=True) 的结果进行优化）：
def construct_yaml_map(self, node):
# test if there are duplicate node keys
keys = set()
for key_node, value_node in node.value:
key = self.construct_object(key_node, deep=True)
if key in keys:
break
keys.add(key)
else:
data = {} # type: Dict[Any, Any]
yield data
value = self.construct_mapping(node)
data.update(value)
return
data = []
yield data
for key_node, value_node in node.value:
key = self.construct_object(key_node, deep=True)
val = self.construct_object(value_node, deep=True)
data.append((key, val))
给出：
[('build', {'step': 'step1'}), ('build', {'step': 'step2'})]
几点：
- 可能不用说，这不适用于 YAML merge keys (
<<: *xyz)
- 如果您需要 ruamel.yaml 的往返能力 (
yaml = YAML())，则需要更复杂的 construct_yaml_map.
如果你想转储输出，你应该为此实例化一个新的 YAML() 实例，而不是重新使用用于加载的 "patched" 实例（它可能工作，这只是为了确定）：
yaml_out = YAML(typ='safe')
yaml_out.dump(data, sys.stdout)
给出（第一个 construct_yaml_map）：
- - build
- - [step, step1]
- - build
- - [step, step2]
在 PyYAML 和 ruamel.yaml 中都不起作用的是 yaml.load('file.yml')。如果您不想自己 open() 文件，您可以这样做：
from pathlib import Path # or: from ruamel.std.pathlib import Path
yaml = YAML(typ='safe')
yaml.load(Path('file.yml')
¹ 免责声明：我是该软件包的作者。
您可以覆盖 pyyaml 加载密钥的方式。例如，您可以将 defaultdict 与每个键的值列表一起使用：
from collections import defaultdict
import yaml
def parse_preserving_duplicates(src):
# We deliberately define a fresh class inside the function,
# because add_constructor is a class method and we don't want to
# mutate pyyaml classes.
class PreserveDuplicatesLoader(yaml.loader.Loader):
pass
def map_constructor(loader, node, deep=False):
"""Walk the mapping, recording any duplicate keys.
"""
mapping = defaultdict(list)
for key_node, value_node in node.value:
key = loader.construct_object(key_node, deep=deep)
value = loader.construct_object(value_node, deep=deep)
mapping[key].append(value)
return mapping
PreserveDuplicatesLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, map_constructor)
return yaml.load(src, PreserveDuplicatesLoader)
这是基于 Anthon 的回答和 ruamel.yaml 的替代实现。它相当通用，使用列表来表示重复项，而其他条目保持不变。
from collections import Counter
from ruamel.yaml import YAML
from ruamel.yaml.constructor import SafeConstructor
yaml_str = '''
a: 1
b: 2
b: 2
'''
def construct_yaml_map(self, node):
data = {}
yield data
keys = [self.construct_object(node, deep=True) for node, _ in node.value]
vals = [self.construct_object(node, deep=True) for _, node in node.value]
key_count = Counter(keys)
for key, val in zip(keys, vals):
if key_count[key] > 1:
if key not in data:
data[key] = []
data[key].append(val)
else:
data[key] = val
SafeConstructor.add_constructor(u'tag:yaml.org,2002:map', construct_yaml_map)
yaml = YAML(typ='safe')
data = yaml.load(yaml_str)
print(data)
输出：
{'a': 1, 'b': [2, 2]}
pyyaml 包（灵感来自 Wilfred Hughes 的回答）也是可能的：
from collections import Counter
import yaml
yaml_str = '''
a: 1
b: 2
b: 2
'''
def parse_preserving_duplicates(src):
class PreserveDuplicatesLoader(yaml.loader.Loader):
pass
def map_constructor(loader, node, deep=False):
keys = [loader.construct_object(node, deep=deep) for node, _ in node.value]
vals = [loader.construct_object(node, deep=deep) for _, node in node.value]
key_count = Counter(keys)
data = {}
for key, val in zip(keys, vals):
if key_count[key] > 1:
if key not in data:
data[key] = []
data[key].append(val)
else:
data[key] = val
return data
PreserveDuplicatesLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, map_constructor)
return yaml.load(src, PreserveDuplicatesLoader)
print(parse_preserving_duplicates(yaml_str))
输出：
{'a': 1, 'b': [2, 2]}
我们需要解析包含重复键的 YAML 文件，所有这些都需要解析。跳过重复项是不够的。我知道这违反了 YAML 规范，我不想这样做，但我们使用的第三方工具支持这种用法，我们需要处理它。
文件示例：
build:
step: 'step1'
build:
step: 'step2'
解析后我们应该有一个类似的数据结构：
yaml.load('file.yml')
# [('build', [('step', 'step1')]), ('build', [('step', 'step2')])]
dict不能再用来表示解析后的内容
我正在 Python 中寻找解决方案，但我没有找到支持这个的库，我错过了什么吗？
或者，我很乐意编写自己的东西，但希望它尽可能简单。 ruamel.yaml 看起来是 Python 中最先进的 YAML 解析器，并且它看起来具有适度的可扩展性，是否可以扩展以支持重复字段？
如果您可以非常轻微地修改输入数据，您应该可以通过将单个类似 yaml 的文件转换为多个 yaml 文档来做到这一点。如果 yaml 文档单独在一行中由 --- 分隔，则它们可以位于同一文件中，并且您很容易看到条目由两个相邻的换行符分隔：
with open('file.yml', 'r') as f:
data = f.read()
data = data.replace('\n\n', '\n---\n')
for document in yaml.load_all(data):
print(document)
输出：
{'build': {'step': 'step1'}}
{'build': {'step': 'step2'}}
PyYAML 只会默默地覆盖第一个条目，ruamel.yaml¹ 如果与旧版 API 一起使用，将给出 DuplicateKeyFutureWarning，并使用新的 DuplicateKeyError API。
如果您不想为所有类型创建完整的 Constructor，覆盖 SafeConstructor 中的映射构造函数应该可以完成工作：
import sys
from ruamel.yaml import YAML
from ruamel.yaml.constructor import SafeConstructor
yaml_str = """\
build:
step: 'step1'
build:
step: 'step2'
"""
def construct_yaml_map(self, node):
# test if there are duplicate node keys
data = []
yield data
for key_node, value_node in node.value:
key = self.construct_object(key_node, deep=True)
val = self.construct_object(value_node, deep=True)
data.append((key, val))
SafeConstructor.add_constructor(u'tag:yaml.org,2002:map', construct_yaml_map)
yaml = YAML(typ='safe')
data = yaml.load(yaml_str)
print(data)
给出：
[('build', [('step', 'step1')]), ('build', [('step', 'step2')])]
不过好像没必要把step: 'step1'做成列表。以下将仅在存在重复项时创建列表（如有必要，可以通过缓存 self.construct_object(key_node, deep=True) 的结果进行优化）：
def construct_yaml_map(self, node):
# test if there are duplicate node keys
keys = set()
for key_node, value_node in node.value:
key = self.construct_object(key_node, deep=True)
if key in keys:
break
keys.add(key)
else:
data = {} # type: Dict[Any, Any]
yield data
value = self.construct_mapping(node)
data.update(value)
return
data = []
yield data
for key_node, value_node in node.value:
key = self.construct_object(key_node, deep=True)
val = self.construct_object(value_node, deep=True)
data.append((key, val))
给出：
[('build', {'step': 'step1'}), ('build', {'step': 'step2'})]
几点：
- 可能不用说，这不适用于 YAML merge keys (
<<: *xyz) - 如果您需要 ruamel.yaml 的往返能力 (
yaml = YAML())，则需要更复杂的construct_yaml_map. 如果你想转储输出，你应该为此实例化一个新的
YAML()实例，而不是重新使用用于加载的 "patched" 实例（它可能工作，这只是为了确定）：yaml_out = YAML(typ='safe') yaml_out.dump(data, sys.stdout)给出（第一个
construct_yaml_map）：- - build - - [step, step1] - - build - - [step, step2]在 PyYAML 和 ruamel.yaml 中都不起作用的是
yaml.load('file.yml')。如果您不想自己open()文件，您可以这样做：from pathlib import Path # or: from ruamel.std.pathlib import Path yaml = YAML(typ='safe') yaml.load(Path('file.yml')
¹ 免责声明：我是该软件包的作者。
您可以覆盖 pyyaml 加载密钥的方式。例如，您可以将 defaultdict 与每个键的值列表一起使用：
from collections import defaultdict
import yaml
def parse_preserving_duplicates(src):
# We deliberately define a fresh class inside the function,
# because add_constructor is a class method and we don't want to
# mutate pyyaml classes.
class PreserveDuplicatesLoader(yaml.loader.Loader):
pass
def map_constructor(loader, node, deep=False):
"""Walk the mapping, recording any duplicate keys.
"""
mapping = defaultdict(list)
for key_node, value_node in node.value:
key = loader.construct_object(key_node, deep=deep)
value = loader.construct_object(value_node, deep=deep)
mapping[key].append(value)
return mapping
PreserveDuplicatesLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, map_constructor)
return yaml.load(src, PreserveDuplicatesLoader)
这是基于 Anthon 的回答和 ruamel.yaml 的替代实现。它相当通用，使用列表来表示重复项，而其他条目保持不变。
from collections import Counter
from ruamel.yaml import YAML
from ruamel.yaml.constructor import SafeConstructor
yaml_str = '''
a: 1
b: 2
b: 2
'''
def construct_yaml_map(self, node):
data = {}
yield data
keys = [self.construct_object(node, deep=True) for node, _ in node.value]
vals = [self.construct_object(node, deep=True) for _, node in node.value]
key_count = Counter(keys)
for key, val in zip(keys, vals):
if key_count[key] > 1:
if key not in data:
data[key] = []
data[key].append(val)
else:
data[key] = val
SafeConstructor.add_constructor(u'tag:yaml.org,2002:map', construct_yaml_map)
yaml = YAML(typ='safe')
data = yaml.load(yaml_str)
print(data)
输出：
{'a': 1, 'b': [2, 2]}
pyyaml 包（灵感来自 Wilfred Hughes 的回答）也是可能的：
from collections import Counter
import yaml
yaml_str = '''
a: 1
b: 2
b: 2
'''
def parse_preserving_duplicates(src):
class PreserveDuplicatesLoader(yaml.loader.Loader):
pass
def map_constructor(loader, node, deep=False):
keys = [loader.construct_object(node, deep=deep) for node, _ in node.value]
vals = [loader.construct_object(node, deep=deep) for _, node in node.value]
key_count = Counter(keys)
data = {}
for key, val in zip(keys, vals):
if key_count[key] > 1:
if key not in data:
data[key] = []
data[key].append(val)
else:
data[key] = val
return data
PreserveDuplicatesLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, map_constructor)
return yaml.load(src, PreserveDuplicatesLoader)
print(parse_preserving_duplicates(yaml_str))
输出：
{'a': 1, 'b': [2, 2]}