MetaDiff.py

Go to the documentation of this file.

"""The function in this module you should look to be using is meta_diff"""
# -*- coding: utf-8 -*-
# Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
# This script reads metadata from a given file
 
import logging
import re
 
from PyUtils.MetaReader import read_metadata, trigger_keys
 
 
def summary(content):
    """Create a summary string for an object"""
    if isinstance(content, str):
        return content
 
    try:
        try:
            working_copy = content.items()
        except AttributeError:
            working_copy = content
        result = ''
        for key, value in working_copy:
            result += "{}: {}, ".format(key, summary(value))
        return result
    except (TypeError, ValueError,):
        pass
 
    try:
        if len(content) < 3:
            return str(content)
        return "[{}, {}, ..., {}]".format(
            summary(content[0]), summary(content[1]), summary(content[-1])
        )
    except TypeError:
        pass
 
    return str(content)
 
 
def truncateDict(value):
    """Create truncted string replaceing dicts with {...}"""
    return ', '.join(
        [
            '{}: {}'.format(
                k,
                '{...}' if isinstance(v, dict) else v
            )
            for k, v in sorted(value.items())
        ]
    )
 
 
def print_diff(parent_key, obj1, obj2, diff_format, filter_key, key_only):
    """build comparison string for two non-dictionary objects"""
 
    if filter_key is not None and filter_key(parent_key) is False:
        # skip this key
        return
 
    result = "\n"
 
    if diff_format == "simple":
        if not obj1:
            result += "{} has been inserted".format(parent_key)
        elif not obj2:
            result += "{} has been deleted".format(parent_key)
        else:
            if key_only:
                result += "{} has changed".format( parent_key )
            else:
                result += "{} has changed from '{}' to '{}'".format(
                    parent_key, obj1, obj2
                )
        result += "\n"
    else:
 
        if parent_key is not None:
 
            if key_only:
                result += "{}".format(parent_key)
            else:
                result += "{}:\n".format(parent_key)
                try:
                    overlap = set(obj1).intersection(set(obj2))
                    for item in overlap:
                        obj1.remove(item)
                        obj2.remove(item)
                except (AttributeError, TypeError,):
                    pass
                result += """\
                < {}
                ----------
                > {}
                """.format(
                    summary(obj1), summary(obj2)
                )
 
    return result
 
 
def print_diff_type(parent_key, obj1, obj2, diff_format, filter_key, key_only):
    """Build diff string for objet of different type"""
 
    if filter_key is not None and filter_key(parent_key) is False:
        # skip this key
        return
 
    result = "\n"
 
    if diff_format == "simple":
        if obj1 is None:
            result += "{} has been inserted".format(parent_key)
        elif obj2 is None:
            result += "{} has been deleted".format(parent_key)
        else:
            if key_only:
                result += (
                    "{} has changed changed type from {} to {}"
                ).format(parent_key, type(obj1), type(obj2))
            else:
                result += (
                    "{} has changed changed type from {} (value: '{}') to "
                    "{} (value: '{}')"
                ).format(parent_key,
                         type(obj1), obj1,
                         type(obj2), obj2)
        result += "\n"
    else:
        if parent_key is not None:
            if key_only:
                result += "{}".format(parent_key)
            else:
                result += "{}:\n".format(parent_key)
                result += """\
                < {} (type: {})
                ----------
                > {} (type: {})
                """.format(
                    summary(obj1), type(obj1), summary(obj2), type(obj2)
                )
 
    return result
 
 
def print_diff_dict_keys(parent_key, obj1, obj2, diff_format, filter_key, key_only):
    """build diff style string for dictionary objects"""
 
    if filter_key is not None and filter_key(parent_key) is False:
        # skip this key
        return
 
    result = '\n'
    if diff_format != 'simple':
        shared_keys = set(obj1.keys()).intersection(obj2.keys())
        for k in shared_keys:
            if obj1[k] == obj2[k]:
                try:
                    obj1.pop(k, None)
                    obj2.pop(k, None)
                except TypeError:
                    pass
 
    if diff_format == 'simple':
        if obj1 is None:
            result += "{} has been inserted".format(parent_key)
        elif obj2 is None:
            result += "{} has been deleted".format(parent_key)
        else:
 
            if key_only:
                result += "{} has changed".format(parent_key)
            else:
                value1 = truncateDict(obj1)
                value2 = truncateDict(obj2)
                result += "{} has changed from '{}' to '{}'".format(
                    parent_key, value1, value2
                )
    else:
        if parent_key is not None:
 
 
            if key_only:
                result += "{}".format(parent_key)
            else:
                result += "{}:\n".format(parent_key)
                result += """\
                < {}
                ----------
                > {}
                """.format( summary(obj1), summary(obj2) )
 
    result += "\n"
 
    return result
 
 
def compare(obj1, obj2, parent_key=None, ordered=False, diff_format="simple", filter_key=None, filter_metadata_items=None, key_only=False):
    """Caclulate difference between two objects
 
    Keyword arguments:
    obj1       -- first object in comparision
    obj2       -- second object in comparision
    parent_key -- the key of the objects in the parent, used in recursion
    ordered    -- whether to check order of list content
    """
    result = list()
 
    if not ordered and isinstance(obj1, list):
        obj1.sort()
 
    if not ordered and isinstance(obj2, list):
        obj2.sort()
 
    if obj1 == obj2:
        return result
 
    if isinstance(obj1, type(obj2)):
 
        if isinstance(obj1, dict):
            # filter out ignored metadata items
            if parent_key == "metadata_items":
                for key in set(list(obj1.keys()) + list(obj2.keys())):
                    if filter_metadata_items is not None and filter_metadata_items(key) is False:
                        obj1.pop(key, None)
                        obj2.pop(key, None)
 
            if sorted(obj1.keys()) != sorted(obj2.keys()):
                result += [
                    print_diff_dict_keys(parent_key, obj1, obj2, diff_format, filter_key, key_only)
                ]
            else:
                for key in sorted(set(obj1.keys()) | set(obj2.keys())):
                    if parent_key:
                        child_key = "{}/{}".format(parent_key, key)
                    else:
                        child_key = key
                    result += compare(
                        obj1[key], obj2[key], child_key, ordered, diff_format, filter_key, filter_metadata_items, key_only
                    )
 
        else:
            result += [print_diff(parent_key, obj1, obj2, diff_format, filter_key, key_only)]
 
    else:
        result += [print_diff_type(parent_key, obj1, obj2, diff_format, filter_key, key_only)]
 
    return result
 
 
def compare_dicts(test, reference, ordered=False, diff_format="simple", filter_key=None, filter_metadata_items=None, key_only=False):
    """Show the differences between two dictionaries
 
    Args:
        test          (dict): first object in comparision
        reference     (dict): second object in comparision
        ordered       (bool): whether to check order of list content
        diff_format (string): specify a format to display the difference in
    """
    result = list()
 
    keys = set(test.keys()).union(reference.keys())
    for key in keys:
 
        try:
            val1 = test[key]
        except KeyError:
            val1 = None
        try:
            val2 = reference[key]
        except KeyError:
            val2 = None
 
        result += compare(
            obj1=val1,
            obj2=val2,
            parent_key=key,
            ordered=ordered,
            diff_format=diff_format,
            filter_key=filter_key,
            filter_metadata_items=filter_metadata_items,
            key_only=key_only
        )
    return result
 
def meta_diff(
        files,
        verbose=False,
        ordered=False,
        drop=None,
        mode="lite",
        meta_key_filter=None,
        file_type=None,
        promote=False,
        diff_format="simple",
        regex=False,
        key_only=False,
        ignore_trigger=False,
):
    """
    Compare the in-file metadata in two given files. Uses PyUtils.MetaReader
    to obtain file content. Generates list of string that show difference.
    Returns empty list if no difference is found
 
    Keyword arguments:
    files   -- Names of two files to compare
    verbose -- toggle to get debug information
    ordered -- whether to check order of lists in the metadata
    drop    -- keys to drop from metadata retrieved by MetaReader
    mode    -- MetaReader argument setting amount of content (default 'lite').
               Allowed values are: tiny, lite, peeker, and full
    meta_key_filter -- MetaReader argument selecting keys to retrieve (default
                       get all)
    file_type    -- Type of files, POOL or BS (default: auto-configure)
    promote      -- MetaReader argument (default: False)
    diff_format  -- Return 'simple' or 'diff' style string (default: 'simple')
    regex        -- Use regex for the drop filter (default: False)
    key_only     -- Show only the keys instead of their value (default: False)
    """
    if len(files) != 2:
        raise ValueError("Wrong number of files passes, need two")
 
    reader_msg = logging.getLogger("MetaReader")
    reader_msg.setLevel(logging.INFO if verbose else logging.WARNING)
 
    msg = logging.getLogger("MetaDiff")
    msg.setLevel(logging.DEBUG if verbose else logging.INFO)
 
    msg.debug("Reading from %s and %s", files[0], files[1])
 
    metadata = read_metadata(
        files,
        file_type,
        mode=mode,
        meta_key_filter=meta_key_filter,
        promote=promote,
    )
 
    if drop is not None and regex:
        for i in range(len(drop)):
            drop[i] = re.compile( drop[i] )
 
    def filter_metadata_items(key):
        key_str = str(key)  # force conversion to plain Python string
 
        if drop is not None:
            for drop_key in drop:
                if not drop_key.startswith("metadata_items/"):
                    continue
 
                drop_key = drop_key[len("metadata_items/"):]
 
                if not regex:
                    if key_str.startswith(drop_key):
                        return False
                else:
                    if drop_key.match(key_str):
                        return False
 
        return True
 
    def filter_key(key):
        key_str = str(key)  # force conversion to plain Python string
 
        if drop is not None:
            for drop_key in drop:
                if not regex:
                    if key_str.startswith(drop_key):
                        return False
                else:
                    if drop_key.match(key_str):
                        return False
 
        if ignore_trigger:
            for trigger_key in trigger_keys:
                if key_str.startswith(trigger_key):
                    return False
 
        return True
 
    result = compare_dicts(
        metadata[files[0]],
        metadata[files[1]],
        ordered=ordered,
        diff_format=diff_format,
        filter_key=filter_key,
        filter_metadata_items=filter_metadata_items,
        key_only=key_only
    )
 
    if not result:
        msg.info("No differences found")
 
    return list(sorted([r for r in result if r is not None ]))