

import io
import os
import unicodedata
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


_UpperCAmelCase		:   List[str]										=       logging.get_logger(__name__)

_UpperCAmelCase		:   Union[str, Any]										=       """▁"""

_UpperCAmelCase		:   int										=       {"""vocab_file""": """vocab.txt""", """sentencepiece_model_ckpt""": """sentencepiece.bpe.model"""}

_UpperCAmelCase		:   List[Any]										=       {
    """sentencepiece_model_file""": """sentencepiece.bpe.model""",
    """vocab_file""": """vocab.txt""",
}

_UpperCAmelCase		:   str										=       {
    """vocab_file""": {
        """ernie-m-base""": """https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/vocab.txt""",
        """ernie-m-large""": """https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/vocab.txt""",
    },
    """sentencepiece_model_file""": {
        """ernie-m-base""": """https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/sentencepiece.bpe.model""",
        """ernie-m-large""": """https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/sentencepiece.bpe.model""",
    },
}

_UpperCAmelCase		:   Optional[Any]										=       {
    """ernie-m-base""": 514,
    """ernie-m-large""": 514,
}

_UpperCAmelCase		:   Optional[Any]										=       {
    """ernie-m-base""": {"""do_lower_case""": False},
    """ernie-m-large""": {"""do_lower_case""": False},
}







class  lowercase				(       lowercase_			):
				__SCREAMING_SNAKE_CASE					: List[str]       			=			["input_ids"]

				__SCREAMING_SNAKE_CASE					: int       			=			VOCAB_FILES_NAMES
				__SCREAMING_SNAKE_CASE					: Dict       			=			PRETRAINED_INIT_CONFIGURATION
				__SCREAMING_SNAKE_CASE					: List[Any]       			=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				__SCREAMING_SNAKE_CASE					: Any       			=			PRETRAINED_VOCAB_FILES_MAP
				__SCREAMING_SNAKE_CASE					: Union[str, Any]       			=			RESOURCE_FILES_NAMES




				def __init__(					self      ,   snake_case      ,   snake_case=None      ,   snake_case=False      ,   snake_case="utf8"      ,   snake_case="[UNK]"      ,   snake_case="[SEP]"      ,   snake_case="[PAD]"      ,   snake_case="[CLS]"      ,   snake_case="[MASK]"      ,   snake_case = None      ,   **snake_case      ,   ):
											# Mask token behave like a normal word, i.e. include the space before it and
											# is included in the raw text, there should be a match in a non-normalized sentence.

											snake_case_							=    {} if sp_model_kwargs is None else sp_model_kwargs
											super().__init__(
											    do_lower_case=snake_case      ,   unk_token=snake_case      ,   sep_token=snake_case      ,   pad_token=snake_case      ,   cls_token=snake_case      ,   mask_token=snake_case      ,   vocab_file=snake_case      ,   encoding=snake_case      ,   sp_model_kwargs=self.sp_model_kwargs      ,   **snake_case      ,   )
											snake_case_							=    do_lower_case
											snake_case_							=    sentencepiece_model_ckpt
											snake_case_							=    spm.SentencePieceProcessor(**self.sp_model_kwargs       )
											self.sp_model.Load(snake_case       )

											# to mimic paddlenlp.transformers.ernie_m.tokenizer.ErnieMTokenizer functioning
											if vocab_file is not None:
																		snake_case_							=    self.load_vocab(filepath=snake_case       )
											else:
																		snake_case_							=    {self.sp_model.id_to_piece(snake_case       ): id for id in range(self.sp_model.get_piece_size()       )}
											snake_case_							=    {v: k for k, v in self.vocab.items()}




				def     a							(					self      ,   snake_case       ):
											if text is None:
																		return None

											snake_case_							=    self.tokenize(snake_case       )
											snake_case_				,    snake_case_							=    '', []

											for i, ch in enumerate(snake_case       ):
																		if ch in self.SP_CHAR_MAPPING:
																									snake_case_							=    self.SP_CHAR_MAPPING.get(snake_case       )
																		else:
																									snake_case_							=    unicodedata.normalize('NFKC'      ,   snake_case       )
																		if self.is_whitespace(snake_case       ):
																									continue
																		normalized_text += ch
																		char_mapping.extend([i] * len(snake_case       )       )

											snake_case_				,    snake_case_				,    snake_case_							=    normalized_text, [], 0

											if self.do_lower_case:
																		snake_case_							=    text.lower()

											for token in split_tokens:
																		if token[:1] == "▁":
																									snake_case_							=    token[1:]
																		snake_case_							=    text[offset:].index(snake_case       ) + offset
																		snake_case_							=    start + len(snake_case       )

																		token_mapping.append((char_mapping[start], char_mapping[end - 1] + 1)       )
																		snake_case_							=    end
											return token_mapping




				@property
				def     a							(					self       ):
											return len(self.vocab       )




				def     a							(					self       ):
											return dict(self.vocab      ,   **self.added_tokens_encoder       )




				def __getstate__(					self       ):
											snake_case_							=    self.__dict__.copy()
											snake_case_							=    None
											return state




				def __setstate__(					self      ,   snake_case       ):
											snake_case_							=    d

											# for backward compatibility
											if not hasattr(self      ,   'sp_model_kwargs'       ):
																		snake_case_							=    {}

											snake_case_							=    spm.SentencePieceProcessor(**self.sp_model_kwargs       )
											self.sp_model.Load(self.sentencepiece_model_ckpt       )




				def     a							(					self      ,   snake_case       ):
											return "".join((self.SP_CHAR_MAPPING.get(snake_case      ,   snake_case       ) for c in text)       )




				def     a							(					self      ,   snake_case      ,   snake_case=False      ,   snake_case=64      ,   snake_case=0.1       ):

											if self.sp_model_kwargs.get('enable_sampling'       ) is True:
																		snake_case_							=    True
											if self.sp_model_kwargs.get('alpha'       ) is not None:
																		snake_case_							=    self.sp_model_kwargs.get('alpha'       )
											if self.sp_model_kwargs.get('nbest_size'       ) is not None:
																		snake_case_							=    self.sp_model_kwargs.get('nbest_size'       )

											if not enable_sampling:
																		snake_case_							=    self.sp_model.EncodeAsPieces(snake_case       )
											else:
																		snake_case_							=    self.sp_model.SampleEncodeAsPieces(snake_case      ,   snake_case      ,   snake_case       )
											snake_case_							=    []
											for pi, piece in enumerate(snake_case       ):
																		if piece == SPIECE_UNDERLINE:
																									if not pieces[pi + 1].startswith(snake_case       ) and pi != 0:
																																new_pieces.append(snake_case       )
																																continue
																									else:
																																continue
																		snake_case_							=    0
																		for i, chunk in enumerate(snake_case       ):
																									if chunk == SPIECE_UNDERLINE:
																																continue
																									if self.is_ch_char(snake_case       ) or self.is_punct(snake_case       ):
																																if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
																																							new_pieces.append(piece[lst_i:i]       )
																																new_pieces.append(snake_case       )
																																snake_case_							=    i + 1
																									elif chunk.isdigit() and i > 0 and not piece[i - 1].isdigit():
																																if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
																																							new_pieces.append(piece[lst_i:i]       )
																																snake_case_							=    i
																									elif not chunk.isdigit() and i > 0 and piece[i - 1].isdigit():
																																if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
																																							new_pieces.append(piece[lst_i:i]       )
																																snake_case_							=    i
																		if len(snake_case       ) > lst_i:
																									new_pieces.append(piece[lst_i:]       )
											return new_pieces




				def     a							(					self      ,   snake_case       ):
											snake_case_							=    ''.join(snake_case       ).replace(snake_case      ,   ' '       ).strip()
											return out_string




				def     a							(					self      ,   snake_case       ):
											snake_case_							=    self.convert_ids_to_tokens(snake_case       )
											snake_case_							=    ''.join(snake_case       ).replace(snake_case      ,   ' '       ).strip()
											return out_string




				def     a							(					self      ,   snake_case       ):
											return self.vocab.get(snake_case      ,   self.vocab.get(self.unk_token       )       )




				def     a							(					self      ,   snake_case       ):
											return self.reverse_vocab.get(snake_case      ,   self.unk_token       )




				def     a							(					self      ,   snake_case      ,   snake_case=None       ):
											if token_ids_a is None:
																		return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
											snake_case_							=    [self.cls_token_id]
											snake_case_							=    [self.sep_token_id]
											return _cls + token_ids_a + _sep + _sep + token_ids_a + _sep




				def     a							(					self      ,   snake_case      ,   snake_case=None       ):
											if offset_mapping_a is None:
																		return [(0, 0)] + offset_mapping_a + [(0, 0)]

											return [(0, 0)] + offset_mapping_a + [(0, 0), (0, 0)] + offset_mapping_a + [(0, 0)]




				def     a							(					self      ,   snake_case      ,   snake_case=None      ,   snake_case=False       ):

											if already_has_special_tokens:
																		if token_ids_a is not None:
																									raise ValueError(
																									    'You should not supply a second sequence if the provided sequence of '
																									    'ids is already formatted with special tokens for the model.'       )
																		return [1 if x in [self.sep_token_id, self.cls_token_id] else 0 for x in token_ids_a]

											if token_ids_a is not None:
																		return [1] + ([0] * len(snake_case       )) + [1, 1] + ([0] * len(snake_case       )) + [1]
											return [1] + ([0] * len(snake_case       )) + [1]




				def     a							(					self      ,   snake_case      ,   snake_case = None       ):
											# called when `add_special_tokens` is True, so align with `build_inputs_with_special_tokens` method
											if token_ids_a is None:
																		# [CLS] X [SEP]
																		return (len(snake_case       ) + 2) * [0]

											# [CLS] A [SEP] [SEP] B [SEP]
											return [0] * (len(snake_case       ) + 1) + [1] * (len(snake_case       ) + 3)




				def     a							(					self      ,   snake_case       ):
											if "\u4e00" <= char <= "\u9fff":
																		return True
											return False




				def     a							(					self      ,   snake_case       ):
											if ("a" <= char <= "z") or ("A" <= char <= "Z"):
																		return True
											return False




				def     a							(					self      ,   snake_case       ):
											if char in ",;:.?!~，；：。？！《》【】":
																		return True
											return False




				def     a							(					self      ,   snake_case       ):
											if char == " " or char == "\t" or char == "\n" or char == "\r":
																		return True
											if len(snake_case       ) == 1:
																		snake_case_							=    unicodedata.category(snake_case       )
																		if cat == "Zs":
																									return True
											return False




				def     a							(					self      ,   snake_case       ):
											snake_case_							=    {}
											with io.open(snake_case      ,   'r'      ,   encoding='utf-8'       ) as f:
																		for index, line in enumerate(snake_case       ):
																									snake_case_							=    line.rstrip('\n'       )
																									snake_case_							=    int(snake_case       )

											return token_to_idx




				def     a							(					self      ,   snake_case      ,   snake_case = None       ):
											snake_case_							=    0
											if os.path.isdir(snake_case       ):
																		snake_case_							=    os.path.join(
																		    snake_case      ,   (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']       )
											else:
																		snake_case_							=    (filename_prefix + '-' if filename_prefix else '') + save_directory
											with open(snake_case      ,   'w'      ,   encoding='utf-8'       ) as writer:
																		for token, token_index in sorted(self.vocab.items()      ,   key=lambda snake_case       : kv[1]       ):
																									if index != token_index:
																																logger.warning(
																																    F'''Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive.'''
																																    ' Please check that the vocabulary is not corrupted!'       )
																																snake_case_							=    token_index
																									writer.write(token + '\n'       )
																									index += 1

											snake_case_							=    os.path.join(snake_case      ,   'sentencepiece.bpe.model'       )
											with open(snake_case      ,   'wb'       ) as fi:
																		snake_case_							=    self.sp_model.serialized_model_proto()
																		fi.write(snake_case       )

											return (vocab_file,)







import uuid
from typing import Any, Dict, List, Optional, Union

from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
					import tensorflow as tf

if is_torch_available():
					import torch


_UpperCAmelCase		:   Dict										=       logging.get_logger(__name__)







class  lowercase				:




				def __init__(					self      ,   snake_case = None      ,   snake_case = None      ,   snake_case=None      ,   snake_case=None       ):
											if not conversation_id:
																		snake_case_							=    uuid.uuida()
											if past_user_inputs is None:
																		snake_case_							=    []
											if generated_responses is None:
																		snake_case_							=    []

											snake_case_							=    conversation_id
											snake_case_							=    past_user_inputs
											snake_case_							=    generated_responses
											snake_case_							=    text




				def __eq__(					self      ,   snake_case       ):
											if not isinstance(snake_case      ,   snake_case       ):
																		return False
											if self.uuid == other.uuid:
																		return True
											return (
											    self.new_user_input == other.new_user_input
											    and self.past_user_inputs == other.past_user_inputs
											    and self.generated_responses == other.generated_responses
											)




				def     a							(					self      ,   snake_case      ,   snake_case = False       ):
											if self.new_user_input:
																		if overwrite:
																									logger.warning(
																									    F'''User input added while unprocessed input was existing: "{self.new_user_input}" was overwritten '''
																									    F'''with: "{text}".'''       )
																									snake_case_							=    text
																		else:
																									logger.warning(
																									    F'''User input added while unprocessed input was existing: "{self.new_user_input}" new input '''
																									    F'''ignored: "{text}". Set `overwrite` to True to overwrite unprocessed user input'''       )
											else:
																		snake_case_							=    text




				def     a							(					self       ):
											if self.new_user_input:
																		self.past_user_inputs.append(self.new_user_input       )
											snake_case_							=    None




				def     a							(					self      ,   snake_case       ):
											self.generated_responses.append(snake_case       )




				def     a							(					self       ):
											for user_input, generated_response in zip(self.past_user_inputs      ,   self.generated_responses       ):
																		yield True, user_input
																		yield False, generated_response
											if self.new_user_input:
																		yield True, self.new_user_input




				def __repr__(					self       ):
											snake_case_							=    F'''Conversation id: {self.uuid} \n'''
											for is_user, text in self.iter_texts():
																		snake_case_							=    'user' if is_user else 'bot'
																		output += F'''{name} >> {text} \n'''
											return output







@add_end_docstrings(
    lowercase_	,  R'''
        min_length_for_response (`int`, *optional*, defaults to 32):
            The minimum length (in number of tokens) for a response.
        minimum_tokens (`int`, *optional*, defaults to 10):
            The minimum length of tokens to leave for a response.
    '''	,  )
class  lowercase				(       lowercase_			):




				def __init__(					self      ,   *snake_case      ,   **snake_case       ):
											super().__init__(*snake_case      ,   **snake_case       )
											if self.tokenizer.pad_token_id is None:
																		snake_case_							=    self.tokenizer.eos_token




				def     a							(					self      ,   snake_case=None      ,   snake_case=None      ,   snake_case=None      ,   **snake_case       ):
											snake_case_							=    {}
											snake_case_							=    {}
											snake_case_							=    {}

											if min_length_for_response is not None:
																		snake_case_							=    min_length_for_response
											if minimum_tokens is not None:
																		snake_case_							=    minimum_tokens

											if "max_length" in generate_kwargs:
																		snake_case_							=    generate_kwargs['max_length']
																		# self.max_length = generate_kwargs.get("max_length", self.model.config.max_length)
											if clean_up_tokenization_spaces is not None:
																		snake_case_							=    clean_up_tokenization_spaces

											if generate_kwargs:
																		forward_params.update(snake_case       )
											return preprocess_params, forward_params, postprocess_params




				def __call__(					self      ,   snake_case      ,   snake_case=0      ,   **snake_case       ):
											snake_case_							=    super().__call__(snake_case      ,   num_workers=snake_case      ,   **snake_case       )
											if isinstance(snake_case      ,   snake_case       ) and len(snake_case       ) == 1:
																		return outputs[0]
											return outputs




				def     a							(					self      ,   snake_case      ,   snake_case=32       ):
											if not isinstance(snake_case      ,   snake_case       ):
																		raise ValueError('ConversationalPipeline, expects Conversation as inputs'       )
											if conversation.new_user_input is None:
																		raise ValueError(
																		    F'''Conversation with UUID {type(conversation.uuid       )} does not contain new user input to process. '''
																		    'Add user inputs with the conversation\'s `add_user_input` method'       )
											if hasattr(self.tokenizer      ,   '_build_conversation_input_ids'       ):
																		snake_case_							=    self.tokenizer._build_conversation_input_ids(snake_case       )
											else:
																		# If the tokenizer cannot handle conversations, we default to only the old version
																		snake_case_							=    self._legacy_parse_and_tokenize(snake_case       )

											if self.framework == "pt":
																		snake_case_							=    torch.LongTensor([input_ids]       )
											elif self.framework == "tf":
																		snake_case_							=    tf.constant([input_ids]       )
											return {"input_ids": input_ids, "conversation": conversation}




				def     a							(					self      ,   snake_case      ,   snake_case=10      ,   **snake_case       ):
											snake_case_							=    generate_kwargs.get('max_length'      ,   self.model.config.max_length       )

											snake_case_							=    model_inputs['input_ids'].shape[1]
											if max_length - minimum_tokens < n:
																		logger.warning(F'''Conversation input is to long ({n}), trimming it to ({max_length} - {minimum_tokens})'''       )
																		snake_case_							=    max_length - minimum_tokens
																		snake_case_							=    model_inputs['input_ids'][:, -trim:]
																		if "attention_mask" in model_inputs:
																									snake_case_							=    model_inputs['attention_mask'][:, -trim:]
											snake_case_							=    model_inputs.pop('conversation'       )
											snake_case_							=    max_length
											snake_case_							=    self.model.generate(**snake_case      ,   **snake_case       )
											if self.model.config.is_encoder_decoder:
																		snake_case_							=    1
											else:
																		snake_case_							=    n
											return {"output_ids": output_ids[:, start_position:], "conversation": conversation}




				def     a							(					self      ,   snake_case      ,   snake_case=True       ):
											snake_case_							=    model_outputs['output_ids']
											snake_case_							=    self.tokenizer.decode(
											    output_ids[0]      ,   skip_special_tokens=snake_case      ,   clean_up_tokenization_spaces=snake_case      ,   )
											snake_case_							=    model_outputs['conversation']
											conversation.mark_processed()
											conversation.append_response(snake_case       )
											return conversation




				def     a							(					self      ,   snake_case       ):
											snake_case_							=    self.tokenizer.eos_token_id
											snake_case_							=    []
											for is_user, text in conversation.iter_texts():
																		if eos_token_id is not None:
																									input_ids.extend(self.tokenizer.encode(snake_case      ,   add_special_tokens=snake_case       ) + [eos_token_id]       )
																		else:
																									input_ids.extend(self.tokenizer.encode(snake_case      ,   add_special_tokens=snake_case       )       )

											if len(snake_case       ) > self.tokenizer.model_max_length:
																		snake_case_							=    input_ids[-self.tokenizer.model_max_length :]
											return input_ids




