code: 
  
  
  
  
  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
  
  
  lowerCAmelCase__   	=				logging.get_logger(__name__)
  
  lowerCAmelCase__   	=				{
      'vocab_file': 'vocab.json',
      'merges_file': 'merges.txt',
  }
  
  lowerCAmelCase__   	=				{
      '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'},
  }
  
  lowerCAmelCase__   	=				{
      'openai-gpt': 5_12,
  }
  def 							_UpperCAmelCase       (UpperCamelCase__  :						List[str]		):
         _A   : Union[str, Any]       					=				text.strip()
         if not text:
                return []
         _A   : Tuple       					=				text.split()
         return tokens
  
  
  
  
  
  
  
  class   lowerCAmelCase__ (       a):
  
  
  
  
       '''simple docstring'''
  
  
       def __init__(		self				,		__lowerCamelCase=True				,		__lowerCamelCase=None				,		__lowerCamelCase=True				,		__lowerCamelCase=None				,		__lowerCamelCase=True				,		)    ->       int:
              if never_split is None:
                     _A   : List[Any]       					=				[]
              _A   : Union[str, Any]       					=				do_lower_case
              _A   : Tuple       					=				set(__lowerCamelCase)
              _A   : Optional[int]       					=				tokenize_chinese_chars
              _A   : int       					=				strip_accents
              _A   : Any       					=				do_split_on_punc
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase				,		__lowerCamelCase=None)    ->       Dict:
              _A   : str       					=				self.never_split.union(set(__lowerCamelCase)) if never_split else self.never_split
              _A   : int       					=				self._clean_text(__lowerCamelCase)
  
              # 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   : int       					=				self._tokenize_chinese_chars(__lowerCamelCase)
              # prevents treating the same character with different unicode codepoints as different characters
              _A   : Optional[int]       					=				unicodedata.normalize("NFC"				,		__lowerCamelCase)
              _A   : Optional[int]       					=				whitespace_tokenize(__lowerCamelCase)
              _A   : List[Any]       					=				[]
              for token in orig_tokens:
                     if token not in never_split:
                            if self.do_lower_case:
                                   _A   : str       					=				token.lower()
                                   if self.strip_accents is not False:
                                          _A   : List[str]       					=				self._run_strip_accents(__lowerCamelCase)
                            elif self.strip_accents:
                                   _A   : Any       					=				self._run_strip_accents(__lowerCamelCase)
                     split_tokens.extend(self._run_split_on_punc(__lowerCamelCase				,		__lowerCamelCase))
  
              _A   : List[Any]       					=				whitespace_tokenize(" ".join(__lowerCamelCase))
              return output_tokens
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       List[Any]:
              _A   : List[Any]       					=				unicodedata.normalize("NFD"				,		__lowerCamelCase)
              _A   : Dict       					=				[]
              for char in text:
                     _A   : Optional[int]       					=				unicodedata.category(__lowerCamelCase)
                     if cat == "Mn":
                            continue
                     output.append(__lowerCamelCase)
              return "".join(__lowerCamelCase)
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase				,		__lowerCamelCase=None)    ->       List[str]:
              if not self.do_split_on_punc or (never_split is not None and text in never_split):
                     return [text]
              _A   : List[Any]       					=				list(__lowerCamelCase)
              _A   : List[str]       					=				0
              _A   : List[Any]       					=				True
              _A   : Any       					=				[]
              while i < len(__lowerCamelCase):
                     _A   : Union[str, Any]       					=				chars[i]
                     if _is_punctuation(__lowerCamelCase):
                            output.append([char])
                            _A   : List[Any]       					=				True
                     else:
                            if start_new_word:
                                   output.append([])
                            _A   : Union[str, Any]       					=				False
                            output[-1].append(__lowerCamelCase)
                     i += 1
  
              return ["".join(__lowerCamelCase) for x in output]
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       List[Any]:
              _A   : Tuple       					=				[]
              for char in text:
                     _A   : List[str]       					=				ord(__lowerCamelCase)
                     if self._is_chinese_char(__lowerCamelCase):
                            output.append(" ")
                            output.append(__lowerCamelCase)
                            output.append(" ")
                     else:
                            output.append(__lowerCamelCase)
              return "".join(__lowerCamelCase)
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       List[str]:
              # 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 >= 0x4_e00 and cp <= 0x9_fff)
                  or (cp >= 0x3_400 and cp <= 0x4_dbf)  #
                  or (cp >= 0x20_000 and cp <= 0x2a_6df)  #
                  or (cp >= 0x2a_700 and cp <= 0x2b_73f)  #
                  or (cp >= 0x2b_740 and cp <= 0x2b_81f)  #
                  or (cp >= 0x2b_820 and cp <= 0x2c_eaf)  #
                  or (cp >= 0xf_900 and cp <= 0xf_aff)
                  or (cp >= 0x2f_800 and cp <= 0x2f_a1f)  #
              ):  #
                     return True
  
              return False
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       int:
              _A   : Optional[int]       					=				[]
              for char in text:
                     _A   : List[str]       					=				ord(__lowerCamelCase)
                     if cp == 0 or cp == 0xf_ffd or _is_control(__lowerCamelCase):
                            continue
                     if _is_whitespace(__lowerCamelCase):
                            output.append(" ")
                     else:
                            output.append(__lowerCamelCase)
              return "".join(__lowerCamelCase)
  
  
  
  
  def 							_UpperCAmelCase       (UpperCamelCase__  :						Any		):
         _A   : Tuple       					=				set()
         _A   : List[str]       					=				word[0]
         for char in word[1:]:
                pairs.add((prev_char, char)		)
                _A   : Dict       					=				char
         return pairs
  def 							_UpperCAmelCase       (UpperCamelCase__  :						int		):
         _A   : Optional[Any]       					=				text.replace("—"       ,							"-"		)
         _A   : Optional[int]       					=				text.replace("–"       ,							"-"		)
         _A   : Tuple       					=				text.replace("―"       ,							"-"		)
         _A   : List[str]       					=				text.replace("…"       ,							"..."		)
         _A   : Union[str, Any]       					=				text.replace("´"       ,							"'"		)
         _A   : Any       					=				re.sub(r"(-+|~+|!+|\"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)"       ,							r" \1 "       ,							UpperCamelCase__		)
         _A   : Optional[Any]       					=				re.sub(r"\s*\n\s*"       ,							" \n "       ,							UpperCamelCase__		)
         _A   : List[str]       					=				re.sub(r"[^\S\n]+"       ,							" "       ,							UpperCamelCase__		)
         return text.strip()
  
  
  
  
  
  
  
  class   lowerCAmelCase__ (       a):
  
  
  
  
       '''simple docstring'''
       __SCREAMING_SNAKE_CASE						    =				VOCAB_FILES_NAMES
       __SCREAMING_SNAKE_CASE						    =				PRETRAINED_VOCAB_FILES_MAP
       __SCREAMING_SNAKE_CASE						    =				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       __SCREAMING_SNAKE_CASE						    =				["input_ids", "attention_mask"]
  
  
       def __init__(		self				,		__lowerCamelCase				,		__lowerCamelCase				,		__lowerCamelCase="<unk>"				,		**__lowerCamelCase)    ->       Dict:
              super().__init__(unk_token=__lowerCamelCase				,		**__lowerCamelCase)
  
              try:
                     import ftfy
                     from spacy.lang.en import English
  
                     _A   : int       					=				English()
                     _A   : Dict       					=				_nlp.tokenizer
                     _A   : Dict       					=				ftfy.fix_text
              except ImportError:
                     logger.warning("ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.")
                     _A   : Optional[int]       					=				BasicTokenizer(do_lower_case=__lowerCamelCase)
                     _A   : str       					=				None
  
              with open(__lowerCamelCase				,		encoding="utf-8") as vocab_handle:
                     _A   : Tuple       					=				json.load(__lowerCamelCase)
              _A   : Any       					=				{v: k for k, v in self.encoder.items()}
              with open(__lowerCamelCase				,		encoding="utf-8") as merges_handle:
                     _A   : int       					=				merges_handle.read().split("\n")[1:-1]
              _A   : Tuple       					=				[tuple(merge.split()) for merge in merges]
              _A   : str       					=				dict(zip(__lowerCamelCase				,		range(len(__lowerCamelCase))))
              _A   : List[Any]       					=				{}
  
  
       @property
       def 				_lowerCamelCase				(		self)    ->       str:
              return True
  
  
       @property
       def 				_lowerCamelCase				(		self)    ->       str:
              return len(self.encoder)
  
  
       def 				_lowerCamelCase				(		self)    ->       Union[str, Any]:
              return dict(self.encoder				,		**self.added_tokens_encoder)
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       Tuple:
              _A   : int       					=				tuple(token[:-1]) + (token[-1] + "</w>",)
              if token in self.cache:
                     return self.cache[token]
              _A   : List[str]       					=				get_pairs(__lowerCamelCase)
  
              if not pairs:
                     return token + "</w>"
  
              while True:
                     _A   : List[str]       					=				min(__lowerCamelCase				,		key=lambda __lowerCamelCase: self.bpe_ranks.get(__lowerCamelCase				,		float("inf")))
                     if bigram not in self.bpe_ranks:
                            break
                     _A   : str       					=				bigram
                     _A   : Tuple       					=				[]
                     _A   : str       					=				0
                     while i < len(__lowerCamelCase):
                            try:
                                   _A   : Optional[Any]       					=				word.index(__lowerCamelCase				,		__lowerCamelCase)
                            except ValueError:
                                   new_word.extend(word[i:])
                                   break
                            else:
                                   new_word.extend(word[i:j])
                                   _A   : Optional[int]       					=				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
                     _A   : Optional[int]       					=				tuple(__lowerCamelCase)
                     _A   : List[Any]       					=				new_word
                     if len(__lowerCamelCase) == 1:
                            break
                     else:
                            _A   : Dict       					=				get_pairs(__lowerCamelCase)
              _A   : str       					=				" ".join(__lowerCamelCase)
              if word == "\n  </w>":
                     _A   : int       					=				"\n</w>"
              _A   : Dict       					=				word
              return word
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       Any:
              _A   : Tuple       					=				[]
              if self.fix_text is None:
                     # Using BERT's BasicTokenizer
                     _A   : Union[str, Any]       					=				self.nlp.tokenize(__lowerCamelCase)
                     for token in text:
                            split_tokens.extend(list(self.bpe(__lowerCamelCase).split(" ")))
              else:
                     # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
                     _A   : Union[str, Any]       					=				self.nlp(text_standardize(self.fix_text(__lowerCamelCase)))
                     for token in text:
                            split_tokens.extend(list(self.bpe(token.text.lower()).split(" ")))
              return split_tokens
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       str:
              return self.encoder.get(__lowerCamelCase				,		self.encoder.get(self.unk_token))
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       Tuple:
              return self.decoder.get(__lowerCamelCase				,		self.unk_token)
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       Union[str, Any]:
              _A   : Dict       					=				"".join(__lowerCamelCase).replace("</w>"				,		" ").strip()
              return out_string
  
  
       def 				_lowerCamelCase				(		self				,		__lowerCamelCase				,		__lowerCamelCase = None)    ->       Tuple[str]:
              if not os.path.isdir(__lowerCamelCase):
                     logger.error(F"Vocabulary path ({save_directory}) should be a directory")
                     return
              _A   : Any       					=				os.path.join(
                  __lowerCamelCase				,		(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"])
              _A   : str       					=				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")
  
              _A   : Optional[int]       					=				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 __lowerCamelCase: 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   : Optional[int]       					=				token_index
                            writer.write(" ".join(__lowerCamelCase) + "\n")
                            index += 1
  
              return vocab_file, merge_file
  
style_context: 
  
  
  import importlib
  import inspect
  from functools import wraps
  from typing import TYPE_CHECKING, Optional, Union
  
  from .download.streaming_download_manager import (
      xbasename,
      xdirname,
      xet_parse,
      xexists,
      xgetsize,
      xglob,
      xgzip_open,
      xisdir,
      xisfile,
      xjoin,
      xlistdir,
      xnumpy_load,
      xopen,
      xpandas_read_csv,
      xpandas_read_excel,
      xPath,
      xrelpath,
      xsio_loadmat,
      xsplit,
      xsplitext,
      xwalk,
      xxml_dom_minidom_parse,
  )
  from .utils.logging import get_logger
  from .utils.patching import patch_submodule
  from .utils.py_utils import get_imports
  
  
  lowerCAmelCase__   	=				get_logger(__name__)
  
  
  if TYPE_CHECKING:
      from .builder import DatasetBuilder
  def 							_UpperCAmelCase       (UpperCamelCase__  :						Optional[Any]       ,							UpperCamelCase__  :						Optional[Union[str, bool]] = None		):
         _A   : Dict       					=				importlib.import_module(UpperCamelCase__		)
  
         # TODO(QL): always update the module to add subsequent new authentication
         if hasattr(UpperCamelCase__       ,							"_patched_for_streaming"		) and module._patched_for_streaming:
                return
  
         def wrap_auth(UpperCamelCase__  :						Optional[int]		):
                @wraps(UpperCamelCase__		)
                def wrapper(*UpperCamelCase__  :						str       ,							**UpperCamelCase__  :						Optional[int]		):
                       return function(*UpperCamelCase__       ,							use_auth_token=UpperCamelCase__       ,							**UpperCamelCase__		)
  
                _A   : List[Any]       					=				"wrap_auth"
                return wrapper
  
         # open files in a streaming fashion
         patch_submodule(UpperCamelCase__       ,							"open"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"os.listdir"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"os.walk"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"glob.glob"       ,							wrap_auth(UpperCamelCase__		)		).start()
         # allow to navigate in remote zip files
         patch_submodule(UpperCamelCase__       ,							"os.path.join"       ,							UpperCamelCase__		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.dirname"       ,							UpperCamelCase__		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.basename"       ,							UpperCamelCase__		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.relpath"       ,							UpperCamelCase__		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.split"       ,							UpperCamelCase__		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.splitext"       ,							UpperCamelCase__		).start()
         # allow checks on paths
         patch_submodule(UpperCamelCase__       ,							"os.path.exists"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.isdir"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.isfile"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.getsize"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"pathlib.Path"       ,							UpperCamelCase__		).start()
         # file readers
         patch_submodule(UpperCamelCase__       ,							"gzip.open"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"numpy.load"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"pandas.read_csv"       ,							wrap_auth(UpperCamelCase__		)       ,							attrs=["__version__"]		).start()
         patch_submodule(UpperCamelCase__       ,							"pandas.read_excel"       ,							wrap_auth(UpperCamelCase__		)       ,							attrs=["__version__"]		).start()
         patch_submodule(UpperCamelCase__       ,							"scipy.io.loadmat"       ,							wrap_auth(UpperCamelCase__		)       ,							attrs=["__version__"]		).start()
         patch_submodule(UpperCamelCase__       ,							"xml.etree.ElementTree.parse"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"xml.dom.minidom.parse"       ,							wrap_auth(UpperCamelCase__		)		).start()
         _A   : Any       					=				True
  
  
  
  
  
  
  def 							_UpperCAmelCase       (UpperCamelCase__  :						"DatasetBuilder"		):
         extend_module_for_streaming(builder.__module__       ,							use_auth_token=builder.use_auth_token		)
         # if needed, we also have to extend additional internal imports (like wmt14 -> wmt_utils)
         if not builder.__module__.startswith("datasets."		):  # check that it's not a packaged builder like csv
                for imports in get_imports(inspect.getfile(builder.__class__		)		):
                       if imports[0] == "internal":
                              _A   : Optional[Any]       					=				imports[1]
                              _A   : int       					=				".".join(builder.__module__.split("."		)[:-1] + [internal_import_name]		)
                              extend_module_for_streaming(UpperCamelCase__       ,							use_auth_token=builder.use_auth_token		)
  
      # builders can inherit from other builders that might use streaming functionality
      # (for example, ImageFolder and AudioFolder inherit from FolderBuilder which implements examples generation)
      # but these parents builders are not patched automatically as they are not instantiated, so we patch them here
         from .builder import DatasetBuilder
  
         _A   : Tuple       					=				[
             cls.__module__
             for cls in type(UpperCamelCase__		).__mro__[1:]  # make sure it's not the same module we've already patched
             if issubclass(cls       ,							UpperCamelCase__		) and cls.__module__ != DatasetBuilder.__module__
         ]  # check it's not a standard builder from datasets.builder
         for module in parent_builder_modules:
                extend_module_for_streaming(UpperCamelCase__       ,							use_auth_token=builder.use_auth_token		)
  