'''simple docstring'''






import os
import warnings
from pathlib import Path
from typing import Optional, Union

from .dynamic_module_utils import custom_object_save
from .tokenization_utils_base import PreTrainedTokenizerBase
from .utils import PushToHubMixin, copy_func, direct_transformers_import, logging


_UpperCAmelCase						:    Any     					=					logging.get_logger(__name__)

# Dynamically import the Transformers module to grab the attribute classes of the processor form their names.
_UpperCAmelCase						:    Any     					=					direct_transformers_import(Path(__file__).parent)


_UpperCAmelCase						:    int     					=					{
    '''AutoTokenizer''': '''PreTrainedTokenizerBase''',
    '''AutoFeatureExtractor''': '''FeatureExtractionMixin''',
    '''AutoImageProcessor''': '''ImageProcessingMixin''',
}







class lowercase_					(		_UpperCamelCase ):







   """simple docstring"""

   __lowerCAmelCase    						=      ["feature_extractor", "tokenizer"]
   # Names need to be attr_class for attr in attributes
   __lowerCAmelCase    						=      None
   __lowerCAmelCase    						=      None
   __lowerCAmelCase    						=      None
   def __init__(			self			: Union[str, Any],   *UpperCamelCase__			: List[Any],   **UpperCamelCase__			: Optional[int]    )     -> List[Any]:
        # Sanitize args and kwargs
        for key in kwargs:
             if key not in self.attributes:
                  raise TypeError(f'Unexpected keyword argument {key}.'    )
        for arg, attribute_name in zip(UpperCamelCase__,   self.attributes    ):
             if attribute_name in kwargs:
                  raise TypeError(f'Got multiple values for argument {attribute_name}.'    )
             else:
                  _A								=     arg

        if len(UpperCamelCase__    ) != len(self.attributes    ):
             raise ValueError(
                 f'This processor requires {len(self.attributes    )} arguments: {", ".join(self.attributes    )}. Got '
                 f'{len(UpperCamelCase__    )} arguments instead.'    )

        # Check each arg is of the proper class (this will also catch a user initializing in the wrong order)
        for attribute_name, arg in kwargs.items():
             _A								=     getattr(self,   f'{attribute_name}_class'    )
             # Nothing is ever going to be an instance of "AutoXxx", in that case we check the base class.
             _A								=     AUTO_TO_BASE_CLASS_MAPPING.get(UpperCamelCase__,   UpperCamelCase__    )
             if isinstance(UpperCamelCase__,   UpperCamelCase__    ):
                  _A								=     tuple(getattr(UpperCamelCase__,   UpperCamelCase__    ) for n in class_name if n is not None    )
             else:
                  _A								=     getattr(UpperCamelCase__,   UpperCamelCase__    )

             if not isinstance(UpperCamelCase__,   UpperCamelCase__    ):
                  raise ValueError(
                      f'Received a {type(UpperCamelCase__    ).__name__} for argument {attribute_name}, but a {class_name} was expected.'    )

             setattr(self,   UpperCamelCase__,   UpperCamelCase__    )
   def __repr__(			self			: Optional[Any]    )     -> Dict:
        _A								=     [f'- {name}: {repr(getattr(self,   UpperCamelCase__    )    )}' for name in self.attributes]
        _A								=     '\n'.join(UpperCamelCase__    )
        return f'{self.__class__.__name__}:\n{attributes_repr}'
   def   __UpperCAmelCase      (			self			: Tuple,   UpperCamelCase__			: Any,   UpperCamelCase__			: bool = False,   **UpperCamelCase__			: Tuple    )     -> Tuple:
        os.makedirs(UpperCamelCase__,   exist_ok=UpperCamelCase__    )

        if push_to_hub:
             _A								=     kwargs.pop('commit_message',   UpperCamelCase__    )
             _A								=     kwargs.pop('repo_id',   save_directory.split(os.path.sep    )[-1]    )
             _A								=     self._create_repo(UpperCamelCase__,   **UpperCamelCase__    )
             _A								=     self._get_files_timestamps(UpperCamelCase__    )
        # If we have a custom config, we copy the file defining it in the folder and set the attributes so it can be
        # loaded from the Hub.
        if self._auto_class is not None:
             _A								=     [getattr(self,   UpperCamelCase__    ) for attribute_name in self.attributes]
             _A								=     [(a.init_kwargs if isinstance(UpperCamelCase__,   UpperCamelCase__    ) else a) for a in attrs]
             custom_object_save(self,   UpperCamelCase__,   config=UpperCamelCase__    )

        for attribute_name in self.attributes:
             _A								=     getattr(self,   UpperCamelCase__    )
             # Include the processor class in the attribute config so this processor can then be reloaded with the
             # `AutoProcessor` API.
             if hasattr(UpperCamelCase__,   '_set_processor_class'    ):
                  attribute._set_processor_class(self.__class__.__name__    )
             attribute.save_pretrained(UpperCamelCase__    )

        if self._auto_class is not None:
             # We added an attribute to the init_kwargs of the tokenizers, which needs to be cleaned up.
             for attribute_name in self.attributes:
                  _A								=     getattr(self,   UpperCamelCase__    )
                  if isinstance(UpperCamelCase__,   UpperCamelCase__    ):
                       del attribute.init_kwargs["auto_map"]

        if push_to_hub:
             self._upload_modified_files(
                 UpperCamelCase__,   UpperCamelCase__,   UpperCamelCase__,   commit_message=UpperCamelCase__,   token=kwargs.get('use_auth_token'    ),   )
   @classmethod
   def   __UpperCAmelCase      (			cls			: Optional[int],   UpperCamelCase__			: Union[str, os.PathLike],   UpperCamelCase__			: Optional[Union[str, os.PathLike]] = None,   UpperCamelCase__			: bool = False,   UpperCamelCase__			: bool = False,   UpperCamelCase__			: Optional[Union[str, bool]] = None,   UpperCamelCase__			: str = "main",   **UpperCamelCase__			: Optional[int],   )     -> Union[str, Any]:
        _A								=     cache_dir
        _A								=     force_download
        _A								=     local_files_only
        _A								=     revision

        _A								=     kwargs.pop('use_auth_token',   UpperCamelCase__    )
        if use_auth_token is not None:
             warnings.warn(
                 'The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers.',   UpperCamelCase__    )
             if token is not None:
                  raise ValueError(
                      '`token` and `use_auth_token` are both specified. Please set only the argument `token`.'    )
             _A								=     use_auth_token

        if token is not None:
             # change to `token` in a follow-up PR
             _A								=     token

        _A								=     cls._get_arguments_from_pretrained(UpperCamelCase__,   **UpperCamelCase__    )
        return cls(*UpperCamelCase__    )
   @classmethod
   def   __UpperCAmelCase      (			cls			: int,   UpperCamelCase__			: int="AutoProcessor"    )     -> Tuple:
        if not isinstance(UpperCamelCase__,   UpperCamelCase__    ):
             _A								=     auto_class.__name__

        import transformers.models.auto as auto_module

        if not hasattr(UpperCamelCase__,   UpperCamelCase__    ):
             raise ValueError(f'{auto_class} is not a valid auto class.'    )

        _A								=     auto_class
   @classmethod
   def   __UpperCAmelCase      (			cls			: int,   UpperCamelCase__			: Union[str, Any],   **UpperCamelCase__			: Tuple    )     -> Any:
        _A								=     []
        for attribute_name in cls.attributes:
             _A								=     getattr(cls,   f'{attribute_name}_class'    )
             if isinstance(UpperCamelCase__,   UpperCamelCase__    ):
                  _A								=     tuple(getattr(UpperCamelCase__,   UpperCamelCase__    ) if n is not None else None for n in class_name    )
                  _A								=     kwargs.get('use_fast',   UpperCamelCase__    )
                  if use_fast and classes[1] is not None:
                       _A								=     classes[1]
                  else:
                       _A								=     classes[0]
             else:
                  _A								=     getattr(UpperCamelCase__,   UpperCamelCase__    )

             args.append(attribute_class.from_pretrained(UpperCamelCase__,   **UpperCamelCase__    )    )
        return args

   @property
   def   __UpperCAmelCase      (			self			: List[Any]    )     -> int:
        _A								=     getattr(self,   self.attributes[0]    )
        return getattr(UpperCamelCase__,   'model_input_names',   UpperCamelCase__    )






_UpperCAmelCase						:    List[str]     					=					copy_func(ProcessorMixin.push_to_hub)
if ProcessorMixin.push_to_hub.__doc__ is not None:
   _UpperCAmelCase						:    str     					=					ProcessorMixin.push_to_hub.__doc__.format(
       object='''processor''', object_class='''AutoProcessor''', object_files='''processor files'''
   )




'''simple docstring'''






import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


_UpperCAmelCase						:    Optional[int]     					=					logging.get_logger(__name__)

_UpperCAmelCase						:    Optional[Any]     					=					{
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

_UpperCAmelCase						:    Optional[Any]     					=					{
    '''vocab_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/vocab.json'''},
    '''merges_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/merges.txt'''},
}

_UpperCAmelCase						:    int     					=					{
    '''openai-gpt''': 5_12,
}





def    _SCREAMING_SNAKE_CASE		(			__snake_case       : Any							):
     _A								=     text.strip()
     if not text:
          return []
     _A								=     text.split()
     return tokens







class lowercase_					(		_UpperCamelCase ):







   """simple docstring"""

   def __init__(			self			: Optional[int],   UpperCamelCase__			: Optional[int]=True,   UpperCamelCase__			: Optional[Any]=None,   UpperCamelCase__			: Optional[int]=True,   UpperCamelCase__			: Dict=None,   UpperCamelCase__			: Any=True,   )     -> str:
        if never_split is None:
             _A								=     []
        _A								=     do_lower_case
        _A								=     set(UpperCamelCase__    )
        _A								=     tokenize_chinese_chars
        _A								=     strip_accents
        _A								=     do_split_on_punc
   def   __UpperCAmelCase      (			self			: Any,   UpperCamelCase__			: List[str],   UpperCamelCase__			: int=None    )     -> Optional[Any]:
        _A								=     self.never_split.union(set(UpperCamelCase__    )    ) if never_split else self.never_split
        _A								=     self._clean_text(UpperCamelCase__    )

        # This was added on November 1st, 2018 for the multilingual and Chinese
        # models. This is also applied to the English models now, but it doesn't
        # matter since the English models were not trained on any Chinese data
        # and generally don't have any Chinese data in them (there are Chinese
        # characters in the vocabulary because Wikipedia does have some Chinese
        # words in the English Wikipedia.).
        if self.tokenize_chinese_chars:
             _A								=     self._tokenize_chinese_chars(UpperCamelCase__    )
        # prevents treating the same character with different unicode codepoints as different characters
        _A								=     unicodedata.normalize('NFC',   UpperCamelCase__    )
        _A								=     whitespace_tokenize(UpperCamelCase__    )
        _A								=     []
        for token in orig_tokens:
             if token not in never_split:
                  if self.do_lower_case:
                       _A								=     token.lower()
                       if self.strip_accents is not False:
                            _A								=     self._run_strip_accents(UpperCamelCase__    )
                  elif self.strip_accents:
                       _A								=     self._run_strip_accents(UpperCamelCase__    )
             split_tokens.extend(self._run_split_on_punc(UpperCamelCase__,   UpperCamelCase__    )    )

        _A								=     whitespace_tokenize(' '.join(UpperCamelCase__    )    )
        return output_tokens
   def   __UpperCAmelCase      (			self			: List[Any],   UpperCamelCase__			: Dict    )     -> Dict:
        _A								=     unicodedata.normalize('NFD',   UpperCamelCase__    )
        _A								=     []
        for char in text:
             _A								=     unicodedata.category(UpperCamelCase__    )
             if cat == "Mn":
                  continue
             output.append(UpperCamelCase__    )
        return "".join(UpperCamelCase__    )
   def   __UpperCAmelCase      (			self			: Union[str, Any],   UpperCamelCase__			: Tuple,   UpperCamelCase__			: Any=None    )     -> Optional[Any]:
        if not self.do_split_on_punc or (never_split is not None and text in never_split):
             return [text]
        _A								=     list(UpperCamelCase__    )
        _A								=     0
        _A								=     True
        _A								=     []
        while i < len(UpperCamelCase__    ):
             _A								=     chars[i]
             if _is_punctuation(UpperCamelCase__    ):
                  output.append([char]    )
                  _A								=     True
             else:
                  if start_new_word:
                       output.append([]    )
                  _A								=     False
                  output[-1].append(UpperCamelCase__    )
             i += 1

        return ["".join(UpperCamelCase__    ) for x in output]
   def   __UpperCAmelCase      (			self			: int,   UpperCamelCase__			: str    )     -> int:
        _A								=     []
        for char in text:
             _A								=     ord(UpperCamelCase__    )
             if self._is_chinese_char(UpperCamelCase__    ):
                  output.append(' '    )
                  output.append(UpperCamelCase__    )
                  output.append(' '    )
             else:
                  output.append(UpperCamelCase__    )
        return "".join(UpperCamelCase__    )
   def   __UpperCAmelCase      (			self			: int,   UpperCamelCase__			: List[Any]    )     -> Union[str, Any]:
        # This defines a "chinese character" as anything in the CJK Unicode block:
        #   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
        #
        # Note that the CJK Unicode block is NOT all Japanese and Korean characters,
        # despite its name. The modern Korean Hangul alphabet is a different block,
        # as is Japanese Hiragana and Katakana. Those alphabets are used to write
        # space-separated words, so they are not treated specially and handled
        # like the all of the other languages.
        if (
            (cp >= 0x4e_00 and cp <= 0x9f_ff)
            or (cp >= 0x34_00 and cp <= 0x4d_bf)  #
            or (cp >= 0x2_00_00 and cp <= 0x2_a6_df)  #
            or (cp >= 0x2_a7_00 and cp <= 0x2_b7_3f)  #
            or (cp >= 0x2_b7_40 and cp <= 0x2_b8_1f)  #
            or (cp >= 0x2_b8_20 and cp <= 0x2_ce_af)  #
            or (cp >= 0xf9_00 and cp <= 0xfa_ff)
            or (cp >= 0x2_f8_00 and cp <= 0x2_fa_1f)  #
        ):  #
             return True

        return False

   def   __UpperCAmelCase      (			self			: Optional[int],   UpperCamelCase__			: Union[str, Any]    )     -> str:
        _A								=     []
        for char in text:
             _A								=     ord(UpperCamelCase__    )
             if cp == 0 or cp == 0xff_fd or _is_control(UpperCamelCase__    ):
                  continue
             if _is_whitespace(UpperCamelCase__    ):
                  output.append(' '    )
             else:
                  output.append(UpperCamelCase__    )
        return "".join(UpperCamelCase__    )






def    _SCREAMING_SNAKE_CASE		(			__snake_case       : Optional[int]							):
     _A								=     set()
     _A								=     word[0]
     for char in word[1:]:
          pairs.add((prev_char, char)							)
          _A								=     char
     return pairs





def    _SCREAMING_SNAKE_CASE		(			__snake_case       : Any							):
     _A								=     text.replace('—'      ,		'-'							)
     _A								=     text.replace('–'      ,		'-'							)
     _A								=     text.replace('―'      ,		'-'							)
     _A								=     text.replace('…'      ,		'...'							)
     _A								=     text.replace('´'      ,		'\''							)
     _A								=     re.sub(R'(-+|~+|!+|"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)'      ,		R' \1 '      ,		__snake_case							)
     _A								=     re.sub(R'\s*\n\s*'      ,		' \n '      ,		__snake_case							)
     _A								=     re.sub(R'[^\S\n]+'      ,		' '      ,		__snake_case							)
     return text.strip()







class lowercase_					(		_UpperCamelCase ):







   """simple docstring"""

   __lowerCAmelCase    						=      VOCAB_FILES_NAMES
   __lowerCAmelCase    						=      PRETRAINED_VOCAB_FILES_MAP
   __lowerCAmelCase    						=      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   __lowerCAmelCase    						=      ["input_ids", "attention_mask"]
   def __init__(			self			: Tuple,   UpperCamelCase__			: Tuple,   UpperCamelCase__			: List[str],   UpperCamelCase__			: Any="<unk>",   **UpperCamelCase__			: List[str]    )     -> Dict:
        super().__init__(unk_token=UpperCamelCase__,   **UpperCamelCase__    )

        try:
             import ftfy
             from spacy.lang.en import English

             _A								=     English()
             _A								=     _nlp.tokenizer
             _A								=     ftfy.fix_text
        except ImportError:
             logger.warning('ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.'    )
             _A								=     BasicTokenizer(do_lower_case=UpperCamelCase__    )
             _A								=     None

        with open(UpperCamelCase__,   encoding='utf-8'    ) as vocab_handle:
             _A								=     json.load(UpperCamelCase__    )
        _A								=     {v: k for k, v in self.encoder.items()}
        with open(UpperCamelCase__,   encoding='utf-8'    ) as merges_handle:
             _A								=     merges_handle.read().split('\n'    )[1:-1]
        _A								=     [tuple(merge.split()    ) for merge in merges]
        _A								=     dict(zip(UpperCamelCase__,   range(len(UpperCamelCase__    )    )    )    )
        _A								=     {}
   @property
   def   __UpperCAmelCase      (			self			: Optional[int]    )     -> Tuple:
        return True
   @property
   def   __UpperCAmelCase      (			self			: str    )     -> List[str]:
        return len(self.encoder    )
   def   __UpperCAmelCase      (			self			: Tuple    )     -> Dict:
        return dict(self.encoder,   **self.added_tokens_encoder    )
   def   __UpperCAmelCase      (			self			: List[str],   UpperCamelCase__			: Dict    )     -> List[Any]:
        _A								=     tuple(token[:-1]    ) + (token[-1] + '</w>',)
        if token in self.cache:
             return self.cache[token]
        _A								=     get_pairs(UpperCamelCase__    )

        if not pairs:
             return token + "</w>"

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

                  if word[i] == first and i < len(UpperCamelCase__    ) - 1 and word[i + 1] == second:
                       new_word.append(first + second    )
                       i += 2
                  else:
                       new_word.append(word[i]    )
                       i += 1
             _A								=     tuple(UpperCamelCase__    )
             _A								=     new_word
             if len(UpperCamelCase__    ) == 1:
                  break
             else:
                  _A								=     get_pairs(UpperCamelCase__    )
        _A								=     ' '.join(UpperCamelCase__    )
        if word == "\n  </w>":
             _A								=     '\n</w>'
        _A								=     word
        return word
   def   __UpperCAmelCase      (			self			: Dict,   UpperCamelCase__			: Any    )     -> Union[str, Any]:
        _A								=     []
        if self.fix_text is None:
             # Using BERT's BasicTokenizer
             _A								=     self.nlp.tokenize(UpperCamelCase__    )
             for token in text:
                  split_tokens.extend(list(self.bpe(UpperCamelCase__    ).split(' '    )    )    )
        else:
             # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
             _A								=     self.nlp(text_standardize(self.fix_text(UpperCamelCase__    )    )    )
             for token in text:
                  split_tokens.extend(list(self.bpe(token.text.lower()    ).split(' '    )    )    )
        return split_tokens
   def   __UpperCAmelCase      (			self			: Any,   UpperCamelCase__			: List[Any]    )     -> List[Any]:
        return self.encoder.get(UpperCamelCase__,   self.encoder.get(self.unk_token    )    )
   def   __UpperCAmelCase      (			self			: int,   UpperCamelCase__			: List[str]    )     -> Tuple:
        return self.decoder.get(UpperCamelCase__,   self.unk_token    )
   def   __UpperCAmelCase      (			self			: Any,   UpperCamelCase__			: List[str]    )     -> Any:
        _A								=     ''.join(UpperCamelCase__    ).replace('</w>',   ' '    ).strip()
        return out_string

   def   __UpperCAmelCase      (			self			: Dict,   UpperCamelCase__			: str,   UpperCamelCase__			: Optional[str] = None    )     -> Tuple[str]:
        if not os.path.isdir(UpperCamelCase__    ):
             logger.error(f'Vocabulary path ({save_directory}) should be a directory'    )
             return
        _A								=     os.path.join(
            UpperCamelCase__,   (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']    )
        _A								=     os.path.join(
            UpperCamelCase__,   (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']    )

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

        _A								=     0
        with open(UpperCamelCase__,   '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!'    )
                       _A								=     token_index
                  writer.write(' '.join(UpperCamelCase__    ) + '\n'    )
                  index += 1

        return vocab_file, merge_file



