#   Copyright (c) 2022 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# !/usr/bin/env python3
"""
Functions for explaining text classifiers.
"""
import itertools
import json
import math
import re
import time
from functools import partial

import LIME.explanation as explanation
import LIME.lime_base as lime_base
import numpy as np
import paddle
import scipy as sp
import sklearn
from sklearn.utils import check_random_state


class TextDomainMapper(explanation.DomainMapper):
    """Maps feature ids to words or word-positions"""

    def __init__(self, indexed_string, language):
        """Initializer.

        Args:
            indexed_string: lime_text.IndexedString, original string
        """
        self.indexed_string = indexed_string
        self.language = language

    def map_exp_ids(self, exp, positions=False):
        """Maps ids to words or word-position strings.

        Args:
            exp: list of tuples [(id, weight), (id,weight)]
            positions: if True, also return word positions

        Returns:
            list of tuples (word, weight), or (word_positions, weight) if
            examples: ('bad', 1) or ('bad_3-6-12', 1)
        """
        if positions:
            exp = [
                (
                    "%s_%s"
                    % (self.indexed_string.word(x[0]), "-".join(map(str, self.indexed_string.string_position(x[0])))),
                    x[1],
                )
                for x in exp
            ]
        else:
            exp = [(self.indexed_string.word(x[0]), x[1]) for x in exp]
        return exp

    def visualize_instance_html(self, exp, label, div_name, exp_object_name, text=True, opacity=True):
        """Adds text with highlighted words to visualization.

        Args:
             exp: list of tuples [(id, weight), (id,weight)]
             label: label id (integer)
             div_name: name of div object to be used for rendering(in js)
             exp_object_name: name of js explanation object
             text: if False, return empty
             opacity: if True, fade colors according to weight
        """
        if not text:
            return ""
        text = self.indexed_string.raw_string().encode("utf-8", "xmlcharrefreplace").decode("utf-8")
        text = re.sub(r"[<>&]", "|", text)
        exp = [(self.indexed_string.word(x[0]), self.indexed_string.string_position(x[0]), x[1]) for x in exp]
        all_occurrences = list(itertools.chain.from_iterable([itertools.product([x[0]], x[1], [x[2]]) for x in exp]))
        all_occurrences = [(x[0], int(x[1]), x[2]) for x in all_occurrences]
        ret = """
            %s.show_raw_text(%s, %d, %s, %s, %s);
            """ % (
            exp_object_name,
            json.dumps(all_occurrences),
            label,
            json.dumps(text),
            div_name,
            json.dumps(opacity),
        )
        return ret


class IndexedString(object):
    """String with various indexes."""

    def __init__(self, raw_string, split_expression=r"\W+", bow=True, mask_string=None, language="en"):
        """Initializer.

        Args:
            raw_string: string with raw text in it
            split_expression: Regex string or callable. If regex string, will be used with re.split.
                If callable, the function should return a list of tokens.
            bow: if True, a word is the same everywhere in the text - i.e. we
                 will index multiple occurrences of the same word. If False,
                 order matters, so that the same word will have different ids
                 according to position.
            mask_string: If not None, replace words with this if bow=False
                if None, default value is UNKWORDZ
        """
        self.raw = raw_string
        self.mask_string = "UNKWORDZ" if mask_string is None else mask_string
        self.language = language

        if callable(split_expression):
            tokens = split_expression(self.raw)
            self.as_list = self._segment_with_tokens(self.raw, tokens)
            tokens = set(tokens)

            def non_word(string):
                return string not in tokens

        else:
            # with the split_expression as a non-capturing group (?:), we don't need to filter out
            # the separator character from the split results.
            # splitter = re.compile(r'(%s)|$' % split_expression)
            # self.as_list = [s for s in splitter.split(self.raw) if s]
            if self.language == "ch":
                splitter = re.compile(r"([\u4e00-\u9fa5])")
                self.as_list = [w for w in splitter.split(self.raw) if len(w.strip()) > 0]
            else:
                splitter = re.compile(split_expression)
                self.as_list = [w for w in self.raw.strip().split() if len(w.strip()) > 0]
            valid_word = splitter.match

        self.as_np = np.array(self.as_list)
        self.string_start = np.hstack(([0], np.cumsum([len(x) for x in self.as_np[:-1]])))
        vocab = {}
        self.inverse_vocab = []
        self.positions = []
        self.bow = bow
        non_vocab = set()
        for i, word in enumerate(self.as_np):
            if word in non_vocab:
                continue
            if (valid_word(word) and self.language == "en") or (not valid_word(word) and self.language == "ch"):
                non_vocab.add(word)
                continue
            if bow:
                if word not in vocab:
                    vocab[word] = len(vocab)
                    self.inverse_vocab.append(word)
                    self.positions.append([])
                idx_word = vocab[word]
                self.positions[idx_word].append(i)
            else:
                self.inverse_vocab.append(word)
                self.positions.append(i)
        if not bow:
            self.positions = np.array(self.positions)

    def raw_string(self):
        """Returns the original raw string"""
        return self.raw

    def num_words(self):
        """Returns the number of tokens in the vocabulary for this document."""
        return len(self.inverse_vocab)

    def word(self, id_):
        """Returns the word that corresponds to id_ (int)"""
        return self.inverse_vocab[id_]

    def string_position(self, id_):
        """Returns a np array with indices to id_ (int) occurrences"""
        if self.bow:
            return self.string_start[self.positions[id_]]
        else:
            return self.string_start[[self.positions[id_]]]

    def inverse_removing(self, words_to_remove):
        """Returns a string after removing the appropriate words.

        If self.bow is false, replaces word with UNKWORDZ instead of removing it.

        Args:
            words_to_remove: list of ids (ints) to remove

        Returns:
            original raw string with appropriate words removed.
        """
        mask = np.ones(self.as_np.shape[0], dtype="bool")
        mask[self.__get_idxs(words_to_remove)] = False
        if self.language == "ch":
            if not self.bow:
                return "".join([self.as_list[i] if mask[i] else self.mask_string for i in range(mask.shape[0])])
            return "".join([self.as_list[v] for v in mask.nonzero()[0]])
        else:
            if not self.bow:
                return " ".join([self.as_list[i] if mask[i] else self.mask_string for i in range(mask.shape[0])])
            return " ".join([self.as_list[v] for v in mask.nonzero()[0]])

    @staticmethod
    def _segment_with_tokens(text, tokens):
        """Segment a string around the tokens created by a passed-in tokenizer"""
        list_form = []
        text_ptr = 0
        for token in tokens:
            inter_token_string = []
            while not text[text_ptr:].startswith(token):
                inter_token_string.append(text[text_ptr])
                text_ptr += 1
                if text_ptr >= len(text):
                    raise ValueError("Tokenization produced tokens that do not belong in string!")
            text_ptr += len(token)
            if inter_token_string:
                list_form.append("".join(inter_token_string))
            list_form.append(token)
        if text_ptr < len(text):
            list_form.append(text[text_ptr:])
        return list_form

    def __get_idxs(self, words):
        """Returns indexes to appropriate words."""
        if self.bow:
            return list(itertools.chain.from_iterable([self.positions[z] for z in words]))
        else:
            return self.positions[words]


class IndexedCharacters(object):
    """String with various indexes."""

    def __init__(self, raw_string, bow=True, mask_string=None):
        """Initializer.

        Args:
            raw_string: string with raw text in it
            bow: if True, a char is the same everywhere in the text - i.e. we
                 will index multiple occurrences of the same character. If False,
                 order matters, so that the same word will have different ids
                 according to position.
            mask_string: If not None, replace characters with this if bow=False
                if None, default value is chr(0)
        """
        self.raw = raw_string
        self.as_list = list(self.raw)
        self.as_np = np.array(self.as_list)
        self.mask_string = chr(0) if mask_string is None else mask_string
        self.string_start = np.arange(len(self.raw))
        vocab = {}
        self.inverse_vocab = []
        self.positions = []
        self.bow = bow
        non_vocab = set()
        for i, char in enumerate(self.as_np):
            if char in non_vocab:
                continue
            if bow:
                if char not in vocab:
                    vocab[char] = len(vocab)
                    self.inverse_vocab.append(char)
                    self.positions.append([])
                idx_char = vocab[char]
                self.positions[idx_char].append(i)
            else:
                self.inverse_vocab.append(char)
                self.positions.append(i)
        if not bow:
            self.positions = np.array(self.positions)

    def raw_string(self):
        """Returns the original raw string"""
        return self.raw

    def num_words(self):
        """Returns the number of tokens in the vocabulary for this document."""
        return len(self.inverse_vocab)

    def word(self, id_):
        """Returns the word that corresponds to id_ (int)"""
        return self.inverse_vocab[id_]

    def string_position(self, id_):
        """Returns a np array with indices to id_ (int) occurrences"""
        if self.bow:
            return self.string_start[self.positions[id_]]
        else:
            return self.string_start[[self.positions[id_]]]

    def inverse_removing(self, words_to_remove):
        """Returns a string after removing the appropriate words.

        If self.bow is false, replaces word with UNKWORDZ instead of removing
        it.

        Args:
            words_to_remove: list of ids (ints) to remove

        Returns:
            original raw string with appropriate words removed.
        """
        mask = np.ones(self.as_np.shape[0], dtype="bool")
        mask[self.__get_idxs(words_to_remove)] = False
        if not self.bow:
            return "".join([self.as_list[i] if mask[i] else self.mask_string for i in range(mask.shape[0])])
        return "".join([self.as_list[v] for v in mask.nonzero()[0]])

    def __get_idxs(self, words):
        """Returns indexes to appropriate words."""
        if self.bow:
            return list(itertools.chain.from_iterable([self.positions[z] for z in words]))
        else:
            return self.positions[words]


class LimeTextExplainer(object):
    """Explains text classifiers.
    Currently, we are using an exponential kernel on cosine distance, and
    restricting explanations to words that are present in documents."""

    def __init__(
        self,
        kernel_width=25,
        kernel=None,
        verbose=False,
        class_names=None,
        feature_selection="auto",
        split_expression=r"\W+",
        bow=True,
        mask_string=None,
        random_state=None,
        char_level=False,
        language="en",
    ):
        """Init function.

        Args:
            kernel_width: kernel width for the exponential kernel.
            kernel: similarity kernel that takes euclidean distances and kernel
                width as input and outputs weights in (0,1). If None, defaults to
                an exponential kernel.
            verbose: if true, print local prediction values from linear model
            class_names: list of class names, ordered according to whatever the
                classifier is using. If not present, class names will be '0',
                '1', ...
            feature_selection: feature selection method. can be
                'forward_selection', 'lasso_path', 'none' or 'auto'.
                See function 'explain_instance_with_data' in lime_base.py for
                details on what each of the options does.
            split_expression: Regex string or callable. If regex string, will be used with re.split.
                If callable, the function should return a list of tokens.
            bow: if True (bag of words), will perturb input data by removing
                all occurrences of individual words or characters.
                Explanations will be in terms of these words. Otherwise, will
                explain in terms of word-positions, so that a word may be
                important the first time it appears and unimportant the second.
                Only set to false if the classifier uses word order in some way
                (bigrams, etc), or if you set char_level=True.
            mask_string: String used to mask tokens or characters if bow=False
                if None, will be 'UNKWORDZ' if char_level=False, chr(0)
                otherwise.
            random_state: an integer or numpy.RandomState that will be used to
                generate random numbers. If None, the random state will be
                initialized using the internal numpy seed.
            char_level: an boolean identifying that we treat each character
                as an independent occurrence in the string
        """

        if kernel is None:

            def kernel(d, kernel_width):
                return np.sqrt(np.exp(-(d**2) / kernel_width**2))

        kernel_fn = partial(kernel, kernel_width=kernel_width)

        self.random_state = check_random_state(random_state)
        self.base = lime_base.LimeBase(kernel_fn, verbose, random_state=self.random_state)
        self.class_names = class_names
        self.vocabulary = None
        self.feature_selection = feature_selection
        self.bow = bow
        self.mask_string = mask_string
        self.split_expression = split_expression
        self.char_level = char_level
        self.language = language

    def explain_instance(
        self,
        text_instance: str,
        tokenizer,
        pred_label: int,
        classifier_fn,
        labels=(0, 1),
        top_labels=None,
        num_features=10,
        num_samples=5000,
        distance_metric="cosine",
        model_regressor=None,
        if_lstm=False,
    ):
        """Generates explanations for a prediction.

        First, we generate neighborhood data by randomly hiding features from
        the instance (see __data_labels_distance_mapping). We then learn
        locally weighted linear models on this neighborhood data to explain
        each of the classes in an interpretable way (see lime_base.py).

        Args:
            text_instance: raw text string to be explained.
            classifier_fn: classifier prediction probability function, which
                takes a list of d strings and outputs a (d, k) numpy array with
                prediction probabilities, where k is the number of classes.
                For ScikitClassifiers , this is classifier.predict_proba.
            labels: iterable with labels to be explained.
            top_labels: if not None, ignore labels and produce explanations for
                the K labels with highest prediction probabilities, where K is
                this parameter.
            num_features: maximum number of features present in explanation
            num_samples: size of the neighborhood to learn the linear model
            distance_metric: the distance metric to use for sample weighting,
                defaults to cosine similarity
            model_regressor: sklearn regressor to use in explanation. Defaults
            to Ridge regression in LimeBase. Must have model_regressor.coef_
            and 'sample_weight' as a parameter to model_regressor.fit()
        Returns:
            An Explanation object (see explanation.py) with the corresponding
            explanations.
        """
        indexed_string = (
            IndexedCharacters(text_instance, bow=self.bow, mask_string=self.mask_string)
            if self.char_level
            else IndexedString(
                text_instance,
                bow=self.bow,
                split_expression=self.split_expression,
                mask_string=self.mask_string,
                language=self.language,
            )
        )
        domain_mapper = TextDomainMapper(indexed_string, self.language)

        # 产生扰动数据集    第一条是原始数据
        # data: 解释器训练特征  list (num_samples, doc_size)
        # yss:  解释器训练标签  list (num_samples, class_num(2))
        # distances: 扰动样本到原始样本的距离 np.array(float) (num_samples, )
        data, yss, distances = self.__data_labels_distances(
            indexed_string, tokenizer, classifier_fn, num_samples, distance_metric=distance_metric, if_lstm=if_lstm
        )

        if self.class_names is None:
            self.class_names = [str(x) for x in range(yss[0].shape[0])]
        ret_exp = explanation.Explanation(
            domain_mapper=domain_mapper, class_names=self.class_names, random_state=self.random_state
        )
        ret_exp.predict_proba = yss[0]
        if top_labels:
            labels = np.argsort(yss[0])[-top_labels:]
            ret_exp.top_labels = list(labels)
            ret_exp.top_labels.reverse()

        num_features = indexed_string.num_words()  # 特征数量跟word_num相同

        (
            ret_exp.intercept[pred_label],
            ret_exp.local_exp[pred_label],
            ret_exp.score[pred_label],
            ret_exp.local_pred[pred_label],
            relative_err,
            err,
        ) = self.base.explain_instance_with_data(
            data,
            yss,
            distances,
            pred_label,
            num_features,
            model_regressor=model_regressor,
            feature_selection=self.feature_selection,
        )

        return ret_exp, indexed_string, relative_err, err

    def __data_labels_distances(
        self, indexed_string, tokenizer, classifier_fn, num_samples, distance_metric="cosine", if_lstm=False
    ):
        """Generates a neighborhood around a prediction.

        Generates neighborhood data by randomly removing words from
        the instance, and predicting with the classifier. Uses cosine distance
        to compute distances between original and perturbed instances.
        Args:
            indexed_string: document (IndexedString) to be explained,
            classifier_fn: classifier prediction probability function, which
                takes a string and outputs prediction probabilities. For
                ScikitClassifier, this is classifier.predict_proba.
            num_samples: size of the neighborhood to learn the linear model
            distance_metric: the distance metric to use for sample weighting,
                defaults to cosine similarity.

        Returns:
            A tuple (data, labels, distances), where:
                data: dense num_samples * K binary matrix, where K is the
                    number of tokens in indexed_string. The first row is the
                    original instance, and thus a row of ones.
                labels: num_samples * L matrix, where L is the number of target
                    labels
                distances: cosine distance between the original instance and
                    each perturbed instance (computed in the binary 'data'
                    matrix), times 100.
        """

        def distance_fn(x):
            return sklearn.metrics.pairwise.pairwise_distances(x, x[0], metric=distance_metric).ravel() * 100

        doc_size = indexed_string.num_words()

        if doc_size > 1:
            sample = self.random_state.randint(
                1, doc_size, num_samples - 1
            )  # sample: [int(1 ~ doc_size-1) * num_samples-1]
        else:
            sample = [0 for i in range(num_samples - 1)]
        data = np.ones((num_samples, doc_size))
        data[0] = np.ones(doc_size)
        features_range = range(doc_size)
        perturb_text = [indexed_string.raw_string()]  # [文本 * num_samples]

        for i, size in enumerate(sample, start=1):
            # inactive: 从range（0， doc_size）中随机取出的size个数组成的list, 要去掉的字的id
            inactive = self.random_state.choice(
                features_range, size, replace=False  # [0, doc_size)  # int: 该扰动样本中remove token的数量
            )

            text = indexed_string.inverse_removing(inactive)  # 原文本去掉了inactive中的字后的文本

            data[i, inactive] = 0
            perturb_text.append(text)

        prev_time = time.time()
        # inverse_data: 扰动数据集 [扰动样本 str] * num_samples
        labels = []
        token_ids_list, s_ids_list, seq_len_list = [], [], []
        token_ids_max_len = 0

        valid_idxs = []

        for idx, text in enumerate(perturb_text):
            if self.language == "en":
                if if_lstm:
                    pad_id = [tokenizer.vocab.token_to_idx.get("[PAD]", 0)]

                    token_ids = tokenizer.encode(text)
                    token_ids_max_len = max(token_ids_max_len, len(token_ids))
                    seq_len = len(token_ids)
                    if seq_len == 0:
                        continue
                    else:
                        valid_idxs.append(idx)
                    seq_len_list.append(seq_len)
                    pad_id = [tokenizer.vocab.token_to_idx.get("[PAD]", 0)]

                else:
                    pad_id = tokenizer.convert_tokens_to_ids(["[PAD]"])

                    tokens = tokenizer.tokenize(text)
                    token_ids = tokenizer.convert_tokens_to_ids(tokens)
                    token_ids = (
                        tokenizer.convert_tokens_to_ids(["[CLS]"])
                        + token_ids
                        + tokenizer.convert_tokens_to_ids(["[SEP]"])
                    )
                    token_ids_max_len = max(token_ids_max_len, len(token_ids))

                token_ids_list.append(token_ids)
            else:
                if len(text) == 0:  # TODO
                    text = perturb_text[0]
                tokens = tokenizer.tokenize(text)
                token_ids = tokenizer.convert_tokens_to_ids(tokens)

                if if_lstm:
                    seq_len = len(token_ids)
                    if seq_len == 0:
                        continue
                    else:
                        valid_idxs.append(idx)
                    seq_len_list.append(seq_len)
                else:
                    token_ids = (
                        tokenizer.convert_tokens_to_ids(["[CLS]"])
                        + token_ids
                        + tokenizer.convert_tokens_to_ids(["[SEP]"])
                    )

                # padding
                token_ids = token_ids + tokenizer.convert_tokens_to_ids(["[PAD]"]) * (
                    len(perturb_text[0]) + 2 - len(token_ids)
                )
                token_ids_list.append(token_ids)
                s_ids = [0 for _ in range(len(token_ids))]
                s_ids_list.append(s_ids)

        if self.language == "en":
            for token_ids in token_ids_list:
                while len(token_ids) < token_ids_max_len:
                    token_ids += pad_id

                s_ids = [0 for _ in range(len(token_ids))]
                s_ids_list.append(s_ids)

        token_ids_np = np.array(token_ids_list)
        s_ids_np = np.array(s_ids_list)
        seq_len_np = np.array(seq_len_list)

        prev_time = time.time()

        batch = 0
        if self.language == "ch":
            length = len(perturb_text[0])

            if if_lstm:
                batch = 128
            else:
                batch = 64 if length < 130 else 50
        else:
            batch = 32

        epoch_num = math.ceil(len(token_ids_np) / batch)
        for idx in range(epoch_num):
            token_ids_tensor = paddle.Tensor(
                value=token_ids_np[idx * batch : (idx + 1) * batch], place=paddle.CUDAPlace(0), stop_gradient=True
            )
            if if_lstm:
                seq_len_tensor = paddle.Tensor(
                    value=seq_len_np[idx * batch : (idx + 1) * batch],
                    place=token_ids_tensor.place,
                    stop_gradient=token_ids_tensor.stop_gradient,
                )
                label = classifier_fn(token_ids_tensor, seq_len_tensor)[0]  # label: Tensor[num_samples, 2]
            else:
                s_ids_tensor = paddle.Tensor(
                    value=s_ids_np[idx * batch : (idx + 1) * batch],
                    place=token_ids_tensor.place,
                    stop_gradient=token_ids_tensor.stop_gradient,
                )
                label = classifier_fn(token_ids_tensor, s_ids_tensor)[0]  # label: Tensor[num_samples, 2]

            labels.extend(label.numpy().tolist())

        labels = np.array(labels)  # labels: nsp.array(num_samples, 2)

        print("mode forward time: %.5f" % (time.time() - prev_time))

        distances = distance_fn(sp.sparse.csr_matrix(data))

        return data, labels, distances
