#!/usr/bin/env python

"""
CmpRuns - A simple tool for comparing two static analyzer runs to determine
which reports have been added, removed, or changed.

This is designed to support automated testing using the static analyzer, from
two perspectives:
  1. To monitor changes in the static analyzer's reports on real code bases,
     for regression testing.

  2. For use by end users who want to integrate regular static analyzer testing
     into a buildbot like environment.

Usage:

    # Load the results of both runs, to obtain lists of the corresponding
    # AnalysisDiagnostic objects.
    #
    resultsA = load_results_from_single_run(singleRunInfoA, delete_empty)
    resultsB = load_results_from_single_run(singleRunInfoB, delete_empty)

    # Generate a relation from diagnostics in run A to diagnostics in run B
    # to obtain a list of triples (a, b, confidence).
    diff = compare_results(resultsA, resultsB)

"""
import json
import os
import plistlib
import re
import sys

from math import log
from collections import defaultdict
from copy import copy
from enum import Enum
from typing import (
    Any,
    DefaultDict,
    Dict,
    List,
    NamedTuple,
    Optional,
    Sequence,
    Set,
    TextIO,
    TypeVar,
    Tuple,
    Union,
)


Number = Union[int, float]
Stats = Dict[str, Dict[str, Number]]
Plist = Dict[str, Any]
JSON = Dict[str, Any]
# Diff in a form: field -> (before, after)
JSONDiff = Dict[str, Tuple[str, str]]
# Type for generics
T = TypeVar("T")

STATS_REGEXP = re.compile(r"Statistics: (\{.+\})", re.MULTILINE | re.DOTALL)


class Colors:
    """
    Color for terminal highlight.
    """

    RED = "\x1b[2;30;41m"
    GREEN = "\x1b[6;30;42m"
    CLEAR = "\x1b[0m"


class HistogramType(str, Enum):
    RELATIVE = "relative"
    LOG_RELATIVE = "log-relative"
    ABSOLUTE = "absolute"


class ResultsDirectory(NamedTuple):
    path: str
    root: str = ""


class SingleRunInfo:
    """
    Information about analysis run:
    path - the analysis output directory
    root - the name of the root directory, which will be disregarded when
    determining the source file name
    """

    def __init__(self, results: ResultsDirectory, verbose_log: Optional[str] = None):
        self.path = results.path
        self.root = results.root.rstrip("/\\")
        self.verbose_log = verbose_log


class AnalysisDiagnostic:
    def __init__(
        self, data: Plist, report: "AnalysisReport", html_report: Optional[str]
    ):
        self._data = data
        self._loc = self._data["location"]
        self._report = report
        self._html_report = html_report
        self._report_size = len(self._data["path"])

    def get_file_name(self) -> str:
        root = self._report.run.root
        file_name = self._report.files[self._loc["file"]]

        if file_name.startswith(root) and len(root) > 0:
            return file_name[len(root) + 1 :]

        return file_name

    def get_root_file_name(self) -> str:
        path = self._data["path"]

        if not path:
            return self.get_file_name()

        p = path[0]
        if "location" in p:
            file_index = p["location"]["file"]
        else:  # control edge
            file_index = path[0]["edges"][0]["start"][0]["file"]

        out = self._report.files[file_index]
        root = self._report.run.root

        if out.startswith(root):
            return out[len(root) :]

        return out

    def get_line(self) -> int:
        return self._loc["line"]

    def get_column(self) -> int:
        return self._loc["col"]

    def get_path_length(self) -> int:
        return self._report_size

    def get_category(self) -> str:
        return self._data["category"]

    def get_description(self) -> str:
        return self._data["description"]

    def get_location(self) -> str:
        return f"{self.get_file_name()}:{self.get_line()}:{self.get_column()}"

    def get_issue_identifier(self) -> str:
        id = self.get_file_name() + "+"

        if "issue_context" in self._data:
            id += self._data["issue_context"] + "+"

        if "issue_hash_content_of_line_in_context" in self._data:
            id += str(self._data["issue_hash_content_of_line_in_context"])

        return id

    def get_html_report(self) -> str:
        if self._html_report is None:
            return " "

        return os.path.join(self._report.run.path, self._html_report)

    def get_readable_name(self) -> str:
        if "issue_context" in self._data:
            funcname_postfix = "#" + self._data["issue_context"]
        else:
            funcname_postfix = ""

        root_filename = self.get_root_file_name()
        file_name = self.get_file_name()

        if root_filename != file_name:
            file_prefix = f"[{root_filename}] {file_name}"
        else:
            file_prefix = root_filename

        line = self.get_line()
        col = self.get_column()
        return (
            f"{file_prefix}{funcname_postfix}:{line}:{col}"
            f", {self.get_category()}: {self.get_description()}"
        )

    KEY_FIELDS = ["check_name", "category", "description"]

    def is_similar_to(self, other: "AnalysisDiagnostic") -> bool:
        # We consider two diagnostics similar only if at least one
        # of the key fields is the same in both diagnostics.
        return len(self.get_diffs(other)) != len(self.KEY_FIELDS)

    def get_diffs(self, other: "AnalysisDiagnostic") -> JSONDiff:
        return {
            field: (self._data[field], other._data[field])
            for field in self.KEY_FIELDS
            if self._data[field] != other._data[field]
        }

    # Note, the data format is not an API and may change from one analyzer
    # version to another.
    def get_raw_data(self) -> Plist:
        return self._data

    def __eq__(self, other: object) -> bool:
        return hash(self) == hash(other)

    def __ne__(self, other: object) -> bool:
        return hash(self) != hash(other)

    def __hash__(self) -> int:
        return hash(self.get_issue_identifier())


class AnalysisRun:
    def __init__(self, info: SingleRunInfo):
        self.path = info.path
        self.root = info.root
        self.info = info
        self.reports: List[AnalysisReport] = []
        # Cumulative list of all diagnostics from all the reports.
        self.diagnostics: List[AnalysisDiagnostic] = []
        self.clang_version: Optional[str] = None
        self.raw_stats: List[JSON] = []

    def get_clang_version(self) -> Optional[str]:
        return self.clang_version

    def read_single_file(self, path: str, delete_empty: bool):
        with open(path, "rb") as plist_file:
            data = plistlib.load(plist_file)

        if "statistics" in data:
            self.raw_stats.append(json.loads(data["statistics"]))
            data.pop("statistics")

        # We want to retrieve the clang version even if there are no
        # reports. Assume that all reports were created using the same
        # clang version (this is always true and is more efficient).
        if "clang_version" in data:
            if self.clang_version is None:
                self.clang_version = data.pop("clang_version")
            else:
                data.pop("clang_version")

        # Ignore/delete empty reports.
        if not data["files"]:
            if delete_empty:
                os.remove(path)
            return

        # Extract the HTML reports, if they exists.
        htmlFiles = []
        for d in data["diagnostics"]:
            if "HTMLDiagnostics_files" in d:
                # FIXME: Why is this named files, when does it have multiple
                # files?
                assert len(d["HTMLDiagnostics_files"]) == 1
                htmlFiles.append(d.pop("HTMLDiagnostics_files")[0])
            else:
                htmlFiles.append(None)

        report = AnalysisReport(self, data.pop("files"))
        # Python 3.10 offers zip(..., strict=True). The following assertion
        # mimics it.
        assert len(data["diagnostics"]) == len(htmlFiles)
        diagnostics = [
            AnalysisDiagnostic(d, report, h)
            for d, h in zip(data.pop("diagnostics"), htmlFiles)
        ]

        assert not data

        report.diagnostics.extend(diagnostics)
        self.reports.append(report)
        self.diagnostics.extend(diagnostics)


class AnalysisReport:
    def __init__(self, run: AnalysisRun, files: List[str]):
        self.run = run
        self.files = files
        self.diagnostics: List[AnalysisDiagnostic] = []


def load_results(
    results: ResultsDirectory,
    delete_empty: bool = True,
    verbose_log: Optional[str] = None,
) -> AnalysisRun:
    """
    Backwards compatibility API.
    """
    return load_results_from_single_run(
        SingleRunInfo(results, verbose_log), delete_empty
    )


def load_results_from_single_run(
    info: SingleRunInfo, delete_empty: bool = True
) -> AnalysisRun:
    """
    # Load results of the analyzes from a given output folder.
    # - info is the SingleRunInfo object
    # - delete_empty specifies if the empty plist files should be deleted

    """
    path = info.path
    run = AnalysisRun(info)

    if os.path.isfile(path):
        run.read_single_file(path, delete_empty)
    else:
        for dirpath, dirnames, filenames in os.walk(path):
            for f in filenames:
                if not f.endswith("plist"):
                    continue

                p = os.path.join(dirpath, f)
                run.read_single_file(p, delete_empty)

    return run


def cmp_analysis_diagnostic(d):
    return d.get_issue_identifier()


AnalysisDiagnosticPair = Tuple[AnalysisDiagnostic, AnalysisDiagnostic]


class ComparisonResult:
    def __init__(self):
        self.present_in_both: List[AnalysisDiagnostic] = []
        self.present_only_in_old: List[AnalysisDiagnostic] = []
        self.present_only_in_new: List[AnalysisDiagnostic] = []
        self.changed_between_new_and_old: List[AnalysisDiagnosticPair] = []

    def add_common(self, issue: AnalysisDiagnostic):
        self.present_in_both.append(issue)

    def add_removed(self, issue: AnalysisDiagnostic):
        self.present_only_in_old.append(issue)

    def add_added(self, issue: AnalysisDiagnostic):
        self.present_only_in_new.append(issue)

    def add_changed(self, old_issue: AnalysisDiagnostic, new_issue: AnalysisDiagnostic):
        self.changed_between_new_and_old.append((old_issue, new_issue))


GroupedDiagnostics = DefaultDict[str, List[AnalysisDiagnostic]]


def get_grouped_diagnostics(
    diagnostics: List[AnalysisDiagnostic],
) -> GroupedDiagnostics:
    result: GroupedDiagnostics = defaultdict(list)
    for diagnostic in diagnostics:
        result[diagnostic.get_location()].append(diagnostic)
    return result


def compare_results(
    results_old: AnalysisRun,
    results_new: AnalysisRun,
    histogram: Optional[HistogramType] = None,
) -> ComparisonResult:
    """
    compare_results - Generate a relation from diagnostics in run A to
    diagnostics in run B.

    The result is the relation as a list of triples (a, b) where
    each element {a,b} is None or a matching element from the respective run
    """

    res = ComparisonResult()

    # Map size_before -> size_after
    path_difference_data: List[float] = []

    diags_old = get_grouped_diagnostics(results_old.diagnostics)
    diags_new = get_grouped_diagnostics(results_new.diagnostics)

    locations_old = set(diags_old.keys())
    locations_new = set(diags_new.keys())

    common_locations = locations_old & locations_new

    for location in common_locations:
        old = diags_old[location]
        new = diags_new[location]

        # Quadratic algorithms in this part are fine because 'old' and 'new'
        # are most commonly of size 1.
        common: Set[AnalysisDiagnostic] = set()
        for a in old:
            for b in new:
                if a.get_issue_identifier() == b.get_issue_identifier():
                    a_path_len = a.get_path_length()
                    b_path_len = b.get_path_length()

                    if a_path_len != b_path_len:

                        if histogram == HistogramType.RELATIVE:
                            path_difference_data.append(float(a_path_len) / b_path_len)

                        elif histogram == HistogramType.LOG_RELATIVE:
                            path_difference_data.append(
                                log(float(a_path_len) / b_path_len)
                            )

                        elif histogram == HistogramType.ABSOLUTE:
                            path_difference_data.append(a_path_len - b_path_len)

                    res.add_common(b)
                    common.add(a)

        old = filter_issues(old, common)
        new = filter_issues(new, common)
        common = set()

        for a in old:
            for b in new:
                if a.is_similar_to(b):
                    res.add_changed(a, b)
                    common.add(a)
                    common.add(b)

        old = filter_issues(old, common)
        new = filter_issues(new, common)

        # Whatever is left in 'old' doesn't have a corresponding diagnostic
        # in 'new', so we need to mark it as 'removed'.
        for a in old:
            res.add_removed(a)

        # Whatever is left in 'new' doesn't have a corresponding diagnostic
        # in 'old', so we need to mark it as 'added'.
        for b in new:
            res.add_added(b)

    only_old_locations = locations_old - common_locations
    for location in only_old_locations:
        for a in diags_old[location]:
            # These locations have been found only in the old build, so we
            # need to mark all of therm as 'removed'
            res.add_removed(a)

    only_new_locations = locations_new - common_locations
    for location in only_new_locations:
        for b in diags_new[location]:
            # These locations have been found only in the new build, so we
            # need to mark all of therm as 'added'
            res.add_added(b)

    # FIXME: Add fuzzy matching. One simple and possible effective idea would
    # be to bin the diagnostics, print them in a normalized form (based solely
    # on the structure of the diagnostic), compute the diff, then use that as
    # the basis for matching. This has the nice property that we don't depend
    # in any way on the diagnostic format.

    if histogram:
        from matplotlib import pyplot

        pyplot.hist(path_difference_data, bins=100)
        pyplot.show()

    return res


def filter_issues(
    origin: List[AnalysisDiagnostic], to_remove: Set[AnalysisDiagnostic]
) -> List[AnalysisDiagnostic]:
    return [diag for diag in origin if diag not in to_remove]


def compute_percentile(values: Sequence[T], percentile: float) -> T:
    """
    Return computed percentile.
    """
    return sorted(values)[int(round(percentile * len(values) + 0.5)) - 1]


def derive_stats(results: AnalysisRun) -> Stats:
    # Assume all keys are the same in each statistics bucket.
    combined_data = defaultdict(list)

    # Collect data on paths length.
    for report in results.reports:
        for diagnostic in report.diagnostics:
            combined_data["PathsLength"].append(diagnostic.get_path_length())

    for stat in results.raw_stats:
        for key, value in stat.items():
            combined_data[str(key)].append(value)

    combined_stats: Stats = {}

    for key, values in combined_data.items():
        combined_stats[key] = {
            "max": max(values),
            "min": min(values),
            "mean": sum(values) / len(values),
            "90th %tile": compute_percentile(values, 0.9),
            "95th %tile": compute_percentile(values, 0.95),
            "median": sorted(values)[len(values) // 2],
            "total": sum(values),
        }

    return combined_stats


# TODO: compare_results decouples comparison from the output, we should
#       do it here as well
def compare_stats(
    results_old: AnalysisRun, results_new: AnalysisRun, out: TextIO = sys.stdout
):
    stats_old = derive_stats(results_old)
    stats_new = derive_stats(results_new)

    old_keys = set(stats_old.keys())
    new_keys = set(stats_new.keys())
    keys = sorted(old_keys & new_keys)

    for key in keys:
        out.write(f"{key}\n")

        nested_keys = sorted(set(stats_old[key]) & set(stats_new[key]))

        for nested_key in nested_keys:
            val_old = float(stats_old[key][nested_key])
            val_new = float(stats_new[key][nested_key])

            report = f"{val_old:.3f} -> {val_new:.3f}"

            # Only apply highlighting when writing to TTY and it's not Windows
            if out.isatty() and os.name != "nt":
                if val_new != 0:
                    ratio = (val_new - val_old) / val_new
                    if ratio < -0.2:
                        report = Colors.GREEN + report + Colors.CLEAR
                    elif ratio > 0.2:
                        report = Colors.RED + report + Colors.CLEAR

            out.write(f"\t {nested_key} {report}\n")

    removed_keys = old_keys - new_keys
    if removed_keys:
        out.write(f"REMOVED statistics: {removed_keys}\n")

    added_keys = new_keys - old_keys
    if added_keys:
        out.write(f"ADDED statistics: {added_keys}\n")

    out.write("\n")


def dump_scan_build_results_diff(
    dir_old: ResultsDirectory,
    dir_new: ResultsDirectory,
    delete_empty: bool = True,
    out: TextIO = sys.stdout,
    show_stats: bool = False,
    stats_only: bool = False,
    histogram: Optional[HistogramType] = None,
    verbose_log: Optional[str] = None,
):
    """
    Compare directories with analysis results and dump results.

    :param delete_empty: delete empty plist files
    :param out: buffer to dump comparison results to.
    :param show_stats: compare execution stats as well.
    :param stats_only: compare ONLY execution stats.
    :param histogram: optional histogram type to plot path differences.
    :param verbose_log: optional path to an additional log file.
    """
    results_old = load_results(dir_old, delete_empty, verbose_log)
    results_new = load_results(dir_new, delete_empty, verbose_log)

    if show_stats or stats_only:
        compare_stats(results_old, results_new)
    if stats_only:
        return

    # Open the verbose log, if given.
    if verbose_log:
        aux_log: Optional[TextIO] = open(verbose_log, "w")
    else:
        aux_log = None

    diff = compare_results(results_old, results_new, histogram)
    found_diffs = 0
    total_added = 0
    total_removed = 0
    total_modified = 0

    for new in diff.present_only_in_new:
        out.write(f"ADDED: {new.get_readable_name()}\n\n")
        found_diffs += 1
        total_added += 1
        if aux_log:
            aux_log.write(
                f"('ADDED', {new.get_readable_name()}, " f"{new.get_html_report()})\n"
            )

    for old in diff.present_only_in_old:
        out.write(f"REMOVED: {old.get_readable_name()}\n\n")
        found_diffs += 1
        total_removed += 1
        if aux_log:
            aux_log.write(
                f"('REMOVED', {old.get_readable_name()}, " f"{old.get_html_report()})\n"
            )

    for old, new in diff.changed_between_new_and_old:
        out.write(f"MODIFIED: {old.get_readable_name()}\n")
        found_diffs += 1
        total_modified += 1
        diffs = old.get_diffs(new)
        str_diffs = [
            f"          '{key}' changed: " f"'{old_value}' -> '{new_value}'"
            for key, (old_value, new_value) in diffs.items()
        ]
        out.write(",\n".join(str_diffs) + "\n\n")
        if aux_log:
            aux_log.write(
                f"('MODIFIED', {old.get_readable_name()}, "
                f"{old.get_html_report()})\n"
            )

    total_reports = len(results_new.diagnostics)
    out.write(f"TOTAL REPORTS: {total_reports}\n")
    out.write(f"TOTAL ADDED: {total_added}\n")
    out.write(f"TOTAL REMOVED: {total_removed}\n")
    out.write(f"TOTAL MODIFIED: {total_modified}\n")

    if aux_log:
        aux_log.write(f"('TOTAL NEW REPORTS', {total_reports})\n")
        aux_log.write(f"('TOTAL DIFFERENCES', {found_diffs})\n")
        aux_log.close()

    # TODO: change to NamedTuple
    return found_diffs, len(results_old.diagnostics), len(results_new.diagnostics)


if __name__ == "__main__":
    print("CmpRuns.py should not be used on its own.")
    print("Please use 'SATest.py compare' instead")
    sys.exit(1)