
"""simple docstring"""



import json
import os
from typing import Optional, Tuple

import regex as re

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


a					:	Union[str, Any]							=		logging.get_logger(__name__)

a					:	Union[str, Any]							=		{
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

a					:	Optional[Any]							=		{
    '''vocab_file''': {'''ctrl''': '''https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json'''},
    '''merges_file''': {'''ctrl''': '''https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt'''},
}

a					:	Tuple							=		{
    '''ctrl''': 2_5_6,
}

a					:	List[Any]							=		{
    '''Pregnancy''': 1_6_8_6_2_9,
    '''Christianity''': 7_6_7_5,
    '''Explain''': 1_0_6_4_2_3,
    '''Fitness''': 6_3_4_4_0,
    '''Saving''': 6_3_1_6_3,
    '''Ask''': 2_7_1_7_1,
    '''Ass''': 9_5_9_8_5,
    '''Joke''': 1_6_3_5_0_9,
    '''Questions''': 4_5_6_2_2,
    '''Thoughts''': 4_9_6_0_5,
    '''Retail''': 5_2_3_4_2,
    '''Feminism''': 1_6_4_3_3_8,
    '''Writing''': 1_1_9_9_2,
    '''Atheism''': 1_9_2_2_6_3,
    '''Netflix''': 4_8_6_1_6,
    '''Computing''': 3_9_6_3_9,
    '''Opinion''': 4_3_2_1_3,
    '''Alone''': 4_4_9_6_7,
    '''Funny''': 5_8_9_1_7,
    '''Gaming''': 4_0_3_5_8,
    '''Human''': 4_0_8_8,
    '''India''': 1_3_3_1,
    '''Joker''': 7_7_1_3_8,
    '''Diet''': 3_6_2_0_6,
    '''Legal''': 1_1_8_5_9,
    '''Norman''': 4_9_3_9,
    '''Tip''': 7_2_6_8_9,
    '''Weight''': 5_2_3_4_3,
    '''Movies''': 4_6_2_7_3,
    '''Running''': 2_3_4_2_5,
    '''Science''': 2_0_9_0,
    '''Horror''': 3_7_7_9_3,
    '''Confession''': 6_0_5_7_2,
    '''Finance''': 1_2_2_5_0,
    '''Politics''': 1_6_3_6_0,
    '''Scary''': 1_9_1_9_8_5,
    '''Support''': 1_2_6_5_4,
    '''Technologies''': 3_2_5_1_6,
    '''Teenage''': 6_6_1_6_0,
    '''Event''': 3_2_7_6_9,
    '''Learned''': 6_7_4_6_0,
    '''Notion''': 1_8_2_7_7_0,
    '''Wikipedia''': 3_7_5_8_3,
    '''Books''': 6_6_6_5,
    '''Extract''': 7_6_0_5_0,
    '''Confessions''': 1_0_2_7_0_1,
    '''Conspiracy''': 7_5_9_3_2,
    '''Links''': 6_3_6_7_4,
    '''Narcissus''': 1_5_0_4_2_5,
    '''Relationship''': 5_4_7_6_6,
    '''Relationships''': 1_3_4_7_9_6,
    '''Reviews''': 4_1_6_7_1,
    '''News''': 4_2_5_6,
    '''Translation''': 2_6_8_2_0,
    '''multilingual''': 1_2_8_4_0_6,
}


def  _UpperCamelCase	(						_A      )    ->  Tuple:





    """simple docstring"""

    _UpperCAmelCase          =   set()
    _UpperCAmelCase          =   word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)      )
        _UpperCAmelCase          =   char

    _UpperCAmelCase          =   set(_A      )
    return pairs







class 		a_						( a__   ):
    a      :				List[Any]     =				VOCAB_FILES_NAMES
    a      :				Dict     =				PRETRAINED_VOCAB_FILES_MAP
    a      :				Tuple     =				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    a      :				Union[str, Any]     =				CONTROL_CODES







    def __init__(			self			:       Any     ,					__UpperCamelCase			:       Optional[int]     ,					__UpperCamelCase			:       Optional[int]     ,					__UpperCamelCase			:       Dict="<unk>"     ,					**__UpperCamelCase			:       Optional[Any]     )		->List[str]:






        '''simple docstring'''






        super().__init__(unk_token=lowerCamelCase_     ,					**lowerCamelCase_     )

        with open(lowerCamelCase_     ,					encoding="""utf-8"""     ) as vocab_handle:
            _UpperCAmelCase          =   json.load(lowerCamelCase_     )
        _UpperCAmelCase          =   {v: k for k, v in self.encoder.items()}
        with open(lowerCamelCase_     ,					encoding="""utf-8"""     ) as merges_handle:
            _UpperCAmelCase          =   merges_handle.read().split("""\n"""     )[1:-1]
        _UpperCAmelCase          =   [tuple(merge.split()     ) for merge in merges]
        _UpperCAmelCase          =   dict(zip(lowerCamelCase_     ,					range(len(lowerCamelCase_     )     )     )     )
        _UpperCAmelCase          =   {}







    @property
    def      _snake_case					(			self			:       str     )		->int:






        '''simple docstring'''






        return len(self.encoder     )







    def      _snake_case					(			self			:       List[str]     )		->List[Any]:






        '''simple docstring'''






        return dict(self.encoder     ,					**self.added_tokens_encoder     )







    def      _snake_case					(			self			:       str     ,					__UpperCamelCase			:       Any     )		->List[str]:






        '''simple docstring'''






        if token in self.cache:
            return self.cache[token]
        _UpperCAmelCase          =   tuple(lowerCamelCase_     )
        _UpperCAmelCase          =   tuple(list(word[:-1]     ) + [word[-1] + """</w>"""]     )
        _UpperCAmelCase          =   get_pairs(lowerCamelCase_     )

        if not pairs:
            return token

        while True:
            _UpperCAmelCase          =   min(lowerCamelCase_     ,					key=lambda __UpperCamelCase     : self.bpe_ranks.get(lowerCamelCase_     ,					float("""inf"""     )     )     )
            if bigram not in self.bpe_ranks:
                break
            _UpperCAmelCase    ,_UpperCAmelCase          =   bigram
            _UpperCAmelCase          =   []
            _UpperCAmelCase          =   0
            while i < len(lowerCamelCase_     ):
                try:
                    _UpperCAmelCase          =   word.index(lowerCamelCase_     ,					lowerCamelCase_     )
                except ValueError:
                    new_word.extend(word[i:]     )
                    break
                else:
                    new_word.extend(word[i:j]     )
                    _UpperCAmelCase          =   j

                if word[i] == first and i < len(lowerCamelCase_     ) - 1 and word[i + 1] == second:
                    new_word.append(first + second     )
                    i += 2
                else:
                    new_word.append(word[i]     )
                    i += 1
            _UpperCAmelCase          =   tuple(lowerCamelCase_     )
            _UpperCAmelCase          =   new_word
            if len(lowerCamelCase_     ) == 1:
                break
            else:
                _UpperCAmelCase          =   get_pairs(lowerCamelCase_     )
        _UpperCAmelCase          =   """@@ """.join(lowerCamelCase_     )
        _UpperCAmelCase          =   word[:-4]
        _UpperCAmelCase          =   word
        return word







    def      _snake_case					(			self			:       Union[str, Any]     ,					__UpperCamelCase			:       str     )		->List[Any]:






        '''simple docstring'''






        _UpperCAmelCase          =   []

        _UpperCAmelCase          =   re.findall(r"""\S+\n?"""     ,					lowerCamelCase_     )

        for token in words:
            split_tokens.extend(list(self.bpe(lowerCamelCase_     ).split(""" """     )     )     )
        return split_tokens







    def      _snake_case					(			self			:       Tuple     ,					__UpperCamelCase			:       Optional[Any]     )		->str:






        '''simple docstring'''






        return self.encoder.get(lowerCamelCase_     ,					self.encoder.get(self.unk_token     )     )







    def      _snake_case					(			self			:       List[str]     ,					__UpperCamelCase			:       Optional[Any]     )		->int:






        '''simple docstring'''






        return self.decoder.get(lowerCamelCase_     ,					self.unk_token     )







    def      _snake_case					(			self			:       Optional[int]     ,					__UpperCamelCase			:       Any     )		->Any:






        '''simple docstring'''






        _UpperCAmelCase          =   """ """.join(lowerCamelCase_     ).replace("""@@ """     ,					""""""     ).strip()
        return out_string







    def      _snake_case					(			self			:       Optional[int]     ,					__UpperCamelCase			:       Any     ,					__UpperCamelCase			:       List[str] = None     )		->Tuple[str]:






        '''simple docstring'''






        if not os.path.isdir(lowerCamelCase_     ):
            logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""     )
            return
        _UpperCAmelCase          =   os.path.join(
            lowerCamelCase_     ,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]     )
        _UpperCAmelCase          =   os.path.join(
            lowerCamelCase_     ,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]     )

        with open(lowerCamelCase_     ,					"""w"""     ,					encoding="""utf-8"""     ) as f:
            f.write(json.dumps(self.encoder     ,					indent=2     ,					sort_keys=lowerCamelCase_     ,					ensure_ascii=lowerCamelCase_     ) + """\n"""     )

        _UpperCAmelCase          =   0
        with open(lowerCamelCase_     ,					"""w"""     ,					encoding="""utf-8"""     ) as writer:
            writer.write("""#version: 0.2\n"""     )
            for bpe_tokens, token_index in sorted(self.bpe_ranks.items()     ,					key=lambda __UpperCamelCase     : kv[1]     ):
                if index != token_index:
                    logger.warning(
                        f"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
                        """ Please check that the tokenizer is not corrupted!"""     )
                    _UpperCAmelCase          =   token_index
                writer.write(""" """.join(lowerCamelCase_     ) + """\n"""     )
                index += 1

        return vocab_file, merge_file

    # def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
    #     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
    #     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
    #     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
    #     return ''.join(tokens_generated_so_far)

"""simple docstring"""



import enum
import warnings

from ..tokenization_utils import TruncationStrategy
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

 from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

if is_torch_available():
 from ..models.auto.modeling_auto import MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

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




class 		a_						( enum.Enum	):
    a      :				Optional[Any]     =				0
    a      :				Dict     =				1







@add_end_docstrings(_UpperCAmelCase	)
class 		a_						( _UpperCAmelCase	):
    a      :				List[Any]     =				'generated'







    def __init__(			self			:       int     ,					*__UpperCamelCase			:       Optional[int]     ,					**__UpperCamelCase			:       str     )		->Any:






        '''simple docstring'''






        super().__init__(*__UpperCamelCase     ,					**__UpperCamelCase     )

        self.check_model_type(
            TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
            if self.framework == """tf"""
            else MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING     )







    def      _snake_case					(			self			:       Optional[int]     ,					__UpperCamelCase			:       List[Any]=None     ,					__UpperCamelCase			:       Union[str, Any]=None     ,					__UpperCamelCase			:       Any=None     ,					__UpperCamelCase			:       int=None     ,					__UpperCamelCase			:       Tuple=None     ,					__UpperCamelCase			:       Dict=None     ,					**__UpperCamelCase			:       Any     ,					)		->Tuple:






        '''simple docstring'''






        _UpperCAmelCase          =   {}
        if truncation is not None:
            _UpperCAmelCase          =   truncation

        _UpperCAmelCase          =   generate_kwargs

        _UpperCAmelCase          =   {}
        if return_tensors is not None and return_type is None:
            _UpperCAmelCase          =   ReturnType.TENSORS if return_tensors else ReturnType.TEXT
        if return_type is not None:
            _UpperCAmelCase          =   return_type

        if clean_up_tokenization_spaces is not None:
            _UpperCAmelCase          =   clean_up_tokenization_spaces

        if stop_sequence is not None:
            _UpperCAmelCase          =   self.tokenizer.encode(__UpperCamelCase     ,					add_special_tokens=__UpperCamelCase     )
            if len(__UpperCamelCase     ) > 1:
                warnings.warn(
                    """Stopping on a multiple token sequence is not yet supported on transformers. The first token of"""
                    """ the stop sequence will be used as the stop sequence string in the interim."""     )
            _UpperCAmelCase          =   stop_sequence_ids[0]

        return preprocess_params, forward_params, postprocess_params







    def      _snake_case					(			self			:       int     ,					__UpperCamelCase			:       int     ,					__UpperCamelCase			:       int     ,					__UpperCamelCase			:       int     )		->Any:






        '''simple docstring'''






        return True







    def      _snake_case					(			self			:       Optional[Any]     ,					*__UpperCamelCase			:       Any     ,					__UpperCamelCase			:       Dict     )		->Dict:






        '''simple docstring'''






        _UpperCAmelCase          =   self.model.config.prefix if self.model.config.prefix is not None else """"""
        if isinstance(args[0]     ,					__UpperCamelCase     ):
            if self.tokenizer.pad_token_id is None:
                raise ValueError("""Please make sure that the tokenizer has a pad_token_id when using a batch input"""     )
            _UpperCAmelCase          =   ([prefix + arg for arg in args[0]],)
            _UpperCAmelCase          =   True

        elif isinstance(args[0]     ,					__UpperCamelCase     ):
            _UpperCAmelCase          =   (prefix + args[0],)
            _UpperCAmelCase          =   False
        else:
            raise ValueError(
                f""" `args[0]`: {args[0]} have the wrong format. The should be either of type `str` or type `list`"""     )
        _UpperCAmelCase          =   self.tokenizer(*__UpperCamelCase     ,					padding=__UpperCamelCase     ,					truncation=__UpperCamelCase     ,					return_tensors=self.framework     )
        # This is produced by tokenizers but is an invalid generate kwargs
        if "token_type_ids" in inputs:
            del inputs["token_type_ids"]
        return inputs







    def __call__(			self			:       Dict     ,					*__UpperCamelCase			:       str     ,					**__UpperCamelCase			:       Dict     )		->Optional[Any]:






        '''simple docstring'''






        _UpperCAmelCase          =   super().__call__(*__UpperCamelCase     ,					**__UpperCamelCase     )
        if (
            isinstance(args[0]     ,					__UpperCamelCase     )
            and all(isinstance(__UpperCamelCase     ,					__UpperCamelCase     ) for el in args[0]     )
            and all(len(__UpperCamelCase     ) == 1 for res in result     )
        ):
            return [res[0] for res in result]
        return result







    def      _snake_case					(			self			:       List[str]     ,					__UpperCamelCase			:       Any     ,					__UpperCamelCase			:       str=TruncationStrategy.DO_NOT_TRUNCATE     ,					**__UpperCamelCase			:       Optional[int]     )		->Optional[int]:






        '''simple docstring'''






        _UpperCAmelCase          =   self._parse_and_tokenize(__UpperCamelCase     ,					truncation=__UpperCamelCase     ,					**__UpperCamelCase     )
        return inputs







    def      _snake_case					(			self			:       str     ,					__UpperCamelCase			:       Dict     ,					**__UpperCamelCase			:       Dict     )		->Tuple:






        '''simple docstring'''






        if self.framework == "pt":
            _UpperCAmelCase    ,_UpperCAmelCase          =   model_inputs["""input_ids"""].shape
        elif self.framework == "tf":
            _UpperCAmelCase    ,_UpperCAmelCase          =   tf.shape(model_inputs["""input_ids"""]     ).numpy()

        _UpperCAmelCase          =   generate_kwargs.get("""min_length"""     ,					self.model.config.min_length     )
        _UpperCAmelCase          =   generate_kwargs.get("""max_length"""     ,					self.model.config.max_length     )
        self.check_inputs(__UpperCamelCase     ,					generate_kwargs["""min_length"""]     ,					generate_kwargs["""max_length"""]     )
        _UpperCAmelCase          =   self.model.generate(**__UpperCamelCase     ,					**__UpperCamelCase     )
        _UpperCAmelCase          =   output_ids.shape[0]
        if self.framework == "pt":
            _UpperCAmelCase          =   output_ids.reshape(__UpperCamelCase     ,					out_b // in_b     ,					*output_ids.shape[1:]     )
        elif self.framework == "tf":
            _UpperCAmelCase          =   tf.reshape(__UpperCamelCase     ,					(in_b, out_b // in_b, *output_ids.shape[1:])     )
        return {"output_ids": output_ids}







    def      _snake_case					(			self			:       Tuple     ,					__UpperCamelCase			:       str     ,					__UpperCamelCase			:       Union[str, Any]=ReturnType.TEXT     ,					__UpperCamelCase			:       int=False     )		->Any:






        '''simple docstring'''






        _UpperCAmelCase          =   []
        for output_ids in model_outputs["output_ids"][0]:
            if return_type == ReturnType.TENSORS:
                _UpperCAmelCase          =   {f"""{self.return_name}_token_ids""": output_ids}
            elif return_type == ReturnType.TEXT:
                _UpperCAmelCase          =   {
                    f"""{self.return_name}_text""": self.tokenizer.decode(
                        __UpperCamelCase     ,					skip_special_tokens=__UpperCamelCase     ,					clean_up_tokenization_spaces=__UpperCamelCase     ,					)
                }
            records.append(__UpperCamelCase     )
        return records







@add_end_docstrings(_UpperCAmelCase	)
class 		a_						( _UpperCAmelCase	):
    a      :				List[Any]     =				'summary'







    def __call__(			self			:       Optional[Any]     ,					*__UpperCamelCase			:       Optional[Any]     ,					**__UpperCamelCase			:       Optional[int]     )		->Any:






        '''simple docstring'''






        return super().__call__(*__UpperCamelCase     ,					**__UpperCamelCase     )







    def      _snake_case					(			self			:       str     ,					__UpperCamelCase			:       int     ,					__UpperCamelCase			:       int     ,					__UpperCamelCase			:       int     )		->bool:






        '''simple docstring'''






        if max_length < min_length:
            logger.warning(f"""Your min_length={min_length} must be inferior than your max_length={max_length}."""     )

        if input_length < max_length:
            logger.warning(
                f"""Your max_length is set to {max_length}, but your input_length is only {input_length}. Since this is """
                """a summarization task, where outputs shorter than the input are typically wanted, you might """
                f"""consider decreasing max_length manually, e.g. summarizer('...', max_length={input_length//2})"""     )







@add_end_docstrings(_UpperCAmelCase	)
class 		a_						( _UpperCAmelCase	):
    a      :				Optional[int]     =				'translation'







    def      _snake_case					(			self			:       Union[str, Any]     ,					__UpperCamelCase			:       int     ,					__UpperCamelCase			:       int     ,					__UpperCamelCase			:       int     )		->Any:






        '''simple docstring'''






        if input_length > 0.9 * max_length:
            logger.warning(
                f"""Your input_length: {input_length} is bigger than 0.9 * max_length: {max_length}. You might consider """
                """increasing your max_length manually, e.g. translator('...', max_length=400)"""     )
        return True







    def      _snake_case					(			self			:       Tuple     ,					*__UpperCamelCase			:       List[str]     ,					__UpperCamelCase			:       Tuple=TruncationStrategy.DO_NOT_TRUNCATE     ,					__UpperCamelCase			:       Tuple=None     ,					__UpperCamelCase			:       Union[str, Any]=None     )		->Tuple:






        '''simple docstring'''






        if getattr(self.tokenizer     ,					"""_build_translation_inputs"""     ,					__UpperCamelCase     ):
            return self.tokenizer._build_translation_inputs(
                *__UpperCamelCase     ,					return_tensors=self.framework     ,					truncation=__UpperCamelCase     ,					src_lang=__UpperCamelCase     ,					tgt_lang=__UpperCamelCase     )
        else:
            return super()._parse_and_tokenize(*__UpperCamelCase     ,					truncation=__UpperCamelCase     )







    def      _snake_case					(			self			:       List[str]     ,					__UpperCamelCase			:       int=None     ,					__UpperCamelCase			:       int=None     ,					**__UpperCamelCase			:       Any     )		->int:






        '''simple docstring'''






        _UpperCAmelCase    ,_UpperCAmelCase    ,_UpperCAmelCase          =   super()._sanitize_parameters(**__UpperCamelCase     )
        if src_lang is not None:
            _UpperCAmelCase          =   src_lang
        if tgt_lang is not None:
            _UpperCAmelCase          =   tgt_lang
        if src_lang is None and tgt_lang is None:
            # Backward compatibility, direct arguments use is preferred.
            _UpperCAmelCase          =   kwargs.get("""task"""     ,					self.task     )
            _UpperCAmelCase          =   task.split("""_"""     )
            if task and len(__UpperCamelCase     ) == 4:
                # translation, XX, to YY
                _UpperCAmelCase          =   items[1]
                _UpperCAmelCase          =   items[3]
        return preprocess_params, forward_params, postprocess_params







    def __call__(			self			:       List[str]     ,					*__UpperCamelCase			:       str     ,					**__UpperCamelCase			:       Optional[Any]     )		->int:






        '''simple docstring'''






        return super().__call__(*__UpperCamelCase     ,					**__UpperCamelCase     )