diffoscope 93ubuntu1 / usr / lib / python3 / dist-packages / diffoscope / difference.py

# -*- coding: utf-8 -*-
#
# diffoscope: in-depth comparison of files, archives, and directories
#
# Copyright © 2014-2015 Jérémy Bobbio <lunar@debian.org>
#
# diffoscope is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# diffoscope is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with diffoscope.  If not, see <https://www.gnu.org/licenses/>.

import heapq
import logging

from . import feeders
from .exc import RequiredToolNotFound
from .diff import diff, reverse_unified_diff, diff_split_lines
from .excludes import command_excluded

logger = logging.getLogger(__name__)


class Difference(object):
    def __init__(self, unified_diff, path1, path2, source=None, comment=None,
                 has_internal_linenos=False, details=None, visuals=None):
        self._unified_diff = unified_diff

        self._comments = []
        if comment:
            if type(comment) is list:
                self._comments.extend(comment)
            else:
                self._comments.append(comment)

        # Allow to override declared file paths, useful when comparing
        # tempfiles
        if source:
            if type(source) is list:
                self._source1, self._source2 = source
            else:
                self._source1 = source
                self._source2 = source
        else:
            self._source1 = path1
            self._source2 = path2

        # Ensure renderable types
        if not isinstance(self._source1, str):
            raise TypeError("path1/source[0] is not a string")
        if not isinstance(self._source2, str):
            raise TypeError("path2/source[1] is not a string")

        # Whether the unified_diff already contains line numbers inside itself
        self._has_internal_linenos = has_internal_linenos
        self._details = details or []
        self._visuals = visuals or []
        self._size_cache = None

    def __repr__(self):
        return "<Difference %s -- %s %s>" % (
            self._source1,
            self._source2,
            self._details,
        )

    def map_lines(self, f_diff, f_comment):
        unified_diff = self.unified_diff
        return self.__class__(
            "".join(map(f_diff, diff_split_lines(unified_diff))) if unified_diff is not None else None,
            self.source1,
            self.source2,
            comment=["".join(map(f_comment, diff_split_lines(comment))) for comment in self._comments],
            has_internal_linenos=self.has_internal_linenos,
            details=self._details[:],
            visuals=self._visuals[:],
        )

    def fmap(self, f):
        return f(self.__class__(
            self.unified_diff,
            self.source1,
            self.source2,
            comment=self._comments[:],
            has_internal_linenos=self.has_internal_linenos,
            details=[d.fmap(f) for d in self._details],
            visuals=self._visuals[:],
        ))

    def _reverse_self(self):
        # assumes we're being called from get_reverse()
        if self._visuals:
            raise NotImplementedError(
                "_reverse_self on VisualDifference is not yet implemented",
            )
        return self.__class__(
            reverse_unified_diff(self.unified_diff) if self.unified_diff is not None else None,
            self.source2,
            self.source1,
            comment=self._comments,  # already copied by fmap in get_reverse
            has_internal_linenos=self.has_internal_linenos,
            details=self._details,  # already reversed by fmap in get_reverse, no need to copy
        )

    def get_reverse(self):
        logger.debug("Reverse orig %s %s", self.source1, self.source2)
        return self.fmap(Difference._reverse_self)

    def equals(self, other):
        return self == other or (
            self.unified_diff == other.unified_diff and
            self.source1 == other.source1 and
            self.source2 == other.source2 and
            self._comments == other._comments and
            self.has_internal_linenos == other.has_internal_linenos and
            all(x.equals(y) for x, y in zip(self._details, other._details)) and
            all(x.equals(y) for x, y in zip(self._visuals, other._visuals)))

    def size(self):
        if self._size_cache is None:
            self._size_cache = sum(d.size_self() for d in self.traverse_depth())
        return self._size_cache

    def size_self(self):
        """Size, excluding children."""
        return ((len(self.unified_diff) if self.unified_diff else 0) +
                (len(self.source1) if self.source1 else 0) +
                (len(self.source2) if self.source2 else 0) +
                sum(map(len, self.comments)) +
                sum(v.size() for v in self._visuals))

    def has_visible_children(self):
        """
        Whether there are visible children.

        Useful for e.g. choosing whether to display [+]/[-] controls.
        """
        return (self._unified_diff is not None or
                self._comments or self._details or self._visuals)

    def traverse_depth(self, depth=-1):
        yield self
        if depth != 0:
            for d in self._details:
                yield from d.traverse_depth(depth-1)

    def traverse_breadth(self, queue=None):
        queue = queue if queue is not None else [self]
        if queue:
            top = queue.pop(0)
            yield top
            queue.extend(top._details)
            yield from self.traverse_breadth(queue)

    def traverse_heapq(self, scorer, yield_score=False, queue=None):
        """Traverse the difference tree using a priority queue, where each node
        is scored according to a user-supplied function, and nodes with smaller
        scores are traversed first (after they have been added to the queue).

        The function `scorer` takes two arguments, a node to score and the
        score of its parent node (or None if there is no parent). It should
        return the score of the input node.
        """
        queue = queue if queue is not None else [(scorer(self, None), self)]
        while queue:
            val, top = heapq.heappop(queue)
            prune_descendants = yield ((top, val) if yield_score else top)
            if not prune_descendants:
                for d in top._details:
                    heapq.heappush(queue, (scorer(d, val), d))

    @staticmethod
    def from_feeder(feeder1, feeder2, path1, path2, source=None, comment=None, **kwargs):
        try:
            unified_diff = diff(feeder1, feeder2)
            if not unified_diff:
                return None
            return Difference(
                unified_diff,
                path1,
                path2,
                source,
                comment,
                **kwargs
            )
        except RequiredToolNotFound:
            difference = Difference(None, path1, path2, source)
            difference.add_comment("diff is not available")
            if comment:
                difference.add_comment(comment)
            return difference

    @staticmethod
    def from_text(content1, content2, *args, **kwargs):
        return Difference.from_feeder(
            feeders.from_text(content1),
            feeders.from_text(content2),
            *args,
            **kwargs
        )

    @staticmethod
    def from_raw_readers(file1, file2, *args, **kwargs):
        return Difference.from_feeder(
            feeders.from_raw_reader(file1),
            feeders.from_raw_reader(file2),
            *args,
            **kwargs
        )

    @staticmethod
    def from_text_readers(file1, file2, *args, **kwargs):
        return Difference.from_feeder(
            feeders.from_text_reader(file1),
            feeders.from_text_reader(file2),
            *args,
            **kwargs
        )

    @staticmethod
    def from_command(klass, path1, path2, *args, **kwargs):
        return Difference.from_command_exc(klass, path1, path2, *args, **kwargs)[0]

    @staticmethod
    def from_command_exc(klass, path1, path2, *args, **kwargs):
        command_args = []
        if 'command_args' in kwargs:
            command_args = kwargs['command_args']
            del kwargs['command_args']

        def command_and_feeder(path):
            command = None
            if path == '/dev/null':
                feeder = feeders.empty()
            else:
                command = klass(path, *command_args)
                feeder = feeders.from_command(command)
                if command_excluded(command.shell_cmdline()):
                    return None, None, True
                command.start()
            return feeder, command, False

        feeder1, command1, excluded1 = command_and_feeder(path1)
        feeder2, command2, excluded2 = command_and_feeder(path2)
        if not feeder1 or not feeder2:
            assert excluded1 or excluded2
            return None, True

        if 'source' not in kwargs:
            source_cmd = command1 or command2
            kwargs['source'] = source_cmd.shell_cmdline()

        difference = Difference.from_feeder(
            feeder1,
            feeder2,
            path1,
            path2,
            *args,
            **kwargs
        )
        if not difference:
            return None, False

        if command1 and command1.stderr:
            difference.add_comment("stderr from `{}`:".format(
                ' '.join(command1.cmdline()),
            ))
            difference.add_comment(command1.stderr)
        if command2 and command2.stderr:
            difference.add_comment("stderr from `{}`:".format(
                ' '.join(command2.cmdline()),
            ))
            difference.add_comment(command2.stderr)

        return difference, False

    @property
    def comment(self):
        return '\n'.join(self._comments)

    @property
    def comments(self):
        return self._comments

    def add_comment(self, comment):
        for line in comment.splitlines():
            self._comments.append(line)
        self._size_cache = None

    @property
    def source1(self):
        return self._source1

    @property
    def source2(self):
        return self._source2

    @property
    def unified_diff(self):
        return self._unified_diff

    @property
    def has_internal_linenos(self):
        return self._has_internal_linenos

    @property
    def details(self):
        return self._details

    @property
    def visuals(self):
        return self._visuals

    def add_details(self, differences):
        if len([d for d in differences if type(d) is not Difference]) > 0:
            raise TypeError("'differences' must contains Difference objects'")
        self._details.extend(differences)
        self._size_cache = None

    def add_visuals(self, visuals):
        if any([type(v) is not VisualDifference for v in visuals]):
            raise TypeError("'visuals' must contain VisualDifference objects'")
        self._visuals.extend(visuals)
        self._size_cache = None


class VisualDifference(object):
    def __init__(self, data_type, content, source):
        self._data_type = data_type
        self._content = content
        self._source = source

    @property
    def data_type(self):
        return self._data_type

    @property
    def content(self):
        return self._content

    @property
    def source(self):
        return self._source

    def size(self):
        return len(self.data_type) + len(self.content) + len(self.source)

    def equals(self, other):
        return self == other or (
            self._data_type == other._data_type and
            self._content == other._content and
            self._source == other._source)