使用Python和PyYAML检测YAML文件中特定键的重复值

-
    ip: 1.1.1.1
    status: Active
    type: 'typeA'
-
    ip: 1.1.1.1
    status: Disabled
    type: 'typeA'
-
    ip: 2.2.2.2
    status: Active
    type: 'typeC'
-
    ip: 3.3.3.3
    status: Active
    type: 'typeB'
-
    ip: 3.3.3.3
    status: Active
    type: 'typeC'
-
    ip: 2.2.2.2
    status: Active
    type: 'typeC'

IP 1.1.1.1, typeA duplicate
IP 2.2.2.2, typeC duplicate

pip install pyyaml

import yaml
from collections import defaultdict

# 假设YAML文件名为 'myyaml.yaml'
yaml_file_path = 'myyaml.yaml'

try:
    with open(yaml_file_path, 'r', encoding='utf-8') as file:
        data = yaml.safe_load(file)
except FileNotFoundError:
    print(f"错误：文件 '{yaml_file_path}' 未找到。")
    exit()
except yaml.YAMLError as e:
    print(f"错误：解析YAML文件时出错：{e}")
    exit()

if not isinstance(data, list):
    print("警告：YAML文件内容不是一个列表，可能无法按预期处理。")
    data = [] # 将data设置为空列表以避免后续错误

# 用于存储 (ip, type) 组合及其出现次数的字典
# 键是 (ip, type) 元组，值是出现次数
item_counts = defaultdict(int)

# 用于存储已识别的重复组合，避免重复打印
duplicates_found = set()

for entry in data:
    # 确保条目有效且包含 'ip' 和 'type' 键
    if isinstance(entry, dict) and 'ip' in entry and 'type' in entry:
        ip = entry['ip']
        entry_type = entry['type']

        # 将ip和type组合成一个元组作为字典的键
        item_key = (ip, entry_type)

        # 增加该组合的计数
        item_counts[item_key] += 1

        # 如果计数大于1，并且这个组合尚未被标记为已发现的重复项，则打印并添加到已发现集合
        if item_counts[item_key] > 1 and item_key not in duplicates_found:
            print(f"IP {ip}, {entry_type} duplicate")
            duplicates_found.add(item_key)
    else:
        # 打印警告信息，指出YAML数据中存在无效或不完整的条目
        print(f"警告：YAML数据中存在无效或不完整的条目：{entry}")

import yaml
from collections import defaultdict

def find_duplicate_yaml_entries(yaml_file_path):
    """
    查找YAML文件中特定键（ip和type）的重复条目。

    Args:
        yaml_file_path (str): YAML文件的路径。

    Returns:
        list: 包含重复条目信息的列表。
    """
    try:
        with open(yaml_file_path, 'r', encoding='utf-8') as file:
            data = yaml.safe_load(file)
    except FileNotFoundError:
        print(f"错误：文件 '{yaml_file_path}' 未找到。")
        return []
    except yaml.YAMLError as e:
        print(f"错误：解析YAML文件时出错：{e}")
        return []

    if not isinstance(data, list):
        print("警告：YAML文件内容不是一个列表，可能无法按预期处理。")
        return []

    item_counts = defaultdict(int)
    duplicates_reported = set()

    # 存储最终的重复项结果
    duplicate_results = []

    for entry in data:
        if isinstance(entry, dict) and 'ip' in entry and 'type' in entry:
            ip = entry['ip']
            entry_type = entry['type']
            item_key = (ip, entry_type)

            item_counts[item_key] += 1

            if item_counts[item_key] > 1 and item_key not in duplicates_reported:
                duplicate_results.append(f"IP {ip}, {entry_type} duplicate")
                duplicates_reported.add(item_key)
        else:
            print(f"警告：YAML数据中存在无效或不完整的条目，已跳过：{entry}")

    return duplicate_results

if __name__ == "__main__":
    # 创建一个示例YAML文件用于测试
    example_yaml_content = """
-
    ip: 1.1.1.1
    status: Active
    type: 'typeA'
-
    ip: 1.1.1.1
    status: Disabled
    type: 'typeA'
-
    ip: 2.2.2.2
    status: Active
    type: 'typeC'
-
    ip: 3.3.3.3
    status: Active
    type: 'typeB'
-
    ip: 3.3.3.3
    status: Active
    type: 'typeC'
-
    ip: 2.2.2.2
    status: Active
    type: 'typeC'
-
"""
    with open('myyaml.yaml', 'w', encoding='utf-8') as f:
        f.write(example_yaml_content)

    print("开始查找重复项...")
    found_duplicates = find_duplicate_yaml_entries('myyaml.yaml')

    if found_duplicates:
        for duplicate_info in found_duplicates:
            print(duplicate_info)
    else:
        print("未发现符合条件的重复条目。")