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_1_2,
}


def       a__      (					a							)		->	Any:
       A_							:       Optional[Any]	      =				text.strip()
       if not text:
              return []
       A_							:       List[Any]	      =				text.split()
       return tokens






class     __UpperCAmelCase(    A__				):






 """simple docstring"""




 def __init__(    self ,       __magic_name__=True ,       __magic_name__=None ,       __magic_name__=True ,       __magic_name__=None ,       __magic_name__=True ,       ):

        """simple docstring"""

        if never_split is None:
               A_							:       List[Any]	      =				[]
        A_							:       int	      =				do_lower_case
        A_							:       Union[str, Any]	      =				set(__magic_name__      )
        A_							:       Dict	      =				tokenize_chinese_chars
        A_							:       Optional[int]	      =				strip_accents
        A_							:       Union[str, Any]	      =				do_split_on_punc
 def  UpperCAmelCase   (    self ,       __magic_name__ ,       __magic_name__=None      ):

        """simple docstring"""

        A_							:       str	      =				self.never_split.union(set(__magic_name__      )      ) if never_split else self.never_split
        A_							:       Any	      =				self._clean_text(__magic_name__      )

        # 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(__magic_name__      )
        # prevents treating the same character with different unicode codepoints as different characters
        A_							:       Union[str, Any]	      =				unicodedata.normalize('''NFC''' ,       __magic_name__      )
        A_							:       Any	      =				whitespace_tokenize(__magic_name__      )
        A_							:       Optional[int]	      =				[]
        for token in orig_tokens:
               if token not in never_split:
                      if self.do_lower_case:
                             A_							:       Dict	      =				token.lower()
                             if self.strip_accents is not False:
                                    A_							:       Optional[Any]	      =				self._run_strip_accents(__magic_name__      )
                      elif self.strip_accents:
                             A_							:       List[Any]	      =				self._run_strip_accents(__magic_name__      )
               split_tokens.extend(self._run_split_on_punc(__magic_name__ ,       __magic_name__      )      )

        A_							:       int	      =				whitespace_tokenize(''' '''.join(__magic_name__      )      )
        return output_tokens
 def  UpperCAmelCase   (    self ,       __magic_name__      ):

        """simple docstring"""

        A_							:       Any	      =				unicodedata.normalize('''NFD''' ,       __magic_name__      )
        A_							:       Optional[Any]	      =				[]
        for char in text:
               A_							:       Optional[int]	      =				unicodedata.category(__magic_name__      )
               if cat == "Mn":
                      continue
               output.append(__magic_name__      )
        return "".join(__magic_name__      )
 def  UpperCAmelCase   (    self ,       __magic_name__ ,       __magic_name__=None      ):

        """simple docstring"""

        if not self.do_split_on_punc or (never_split is not None and text in never_split):
               return [text]
        A_							:       str	      =				list(__magic_name__      )
        A_							:       Dict	      =				0
        A_							:       Union[str, Any]	      =				True
        A_							:       int	      =				[]
        while i < len(__magic_name__      ):
               A_							:       List[str]	      =				chars[i]
               if _is_punctuation(__magic_name__      ):
                      output.append([char]      )
                      A_							:       Optional[Any]	      =				True
               else:
                      if start_new_word:
                             output.append([]      )
                      A_							:       Optional[int]	      =				False
                      output[-1].append(__magic_name__      )
               i += 1

        return ["".join(__magic_name__      ) for x in output]
 def  UpperCAmelCase   (    self ,       __magic_name__      ):

        """simple docstring"""

        A_							:       Tuple	      =				[]
        for char in text:
               A_							:       Tuple	      =				ord(__magic_name__      )
               if self._is_chinese_char(__magic_name__      ):
                      output.append(''' '''      )
                      output.append(__magic_name__      )
                      output.append(''' '''      )
               else:
                      output.append(__magic_name__      )
        return "".join(__magic_name__      )
 def  UpperCAmelCase   (    self ,       __magic_name__      ):

        """simple docstring"""

        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 ,       __magic_name__      ):

        """simple docstring"""

        A_							:       int	      =				[]
        for char in text:
               A_							:       Union[str, Any]	      =				ord(__magic_name__      )
               if cp == 0 or cp == 0xFF_FD or _is_control(__magic_name__      ):
                      continue
               if _is_whitespace(__magic_name__      ):
                      output.append(''' '''      )
               else:
                      output.append(__magic_name__      )
        return "".join(__magic_name__      )


def       a__      (					a							)		->	List[str]:
       A_							:       int	      =				set()
       A_							:       Any	      =				word[0]
       for char in word[1:]:
              pairs.add((prev_char, char)							)
              A_							:       Dict	      =				char
       return pairs


def       a__      (					a							)		->	int:
       A_							:       str	      =				text.replace('''—'''  ,   '''-'''							)
       A_							:       Union[str, Any]	      =				text.replace('''–'''  ,   '''-'''							)
       A_							:       int	      =				text.replace('''―'''  ,   '''-'''							)
       A_							:       Any	      =				text.replace('''…'''  ,   '''...'''							)
       A_							:       Union[str, Any]	      =				text.replace('''´'''  ,   '''\''''							)
       A_							:       List[Any]	      =				re.sub(R'''(-+|~+|!+|"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)'''  ,   R''' \1 '''  ,   a							)
       A_							:       int	      =				re.sub(R'''\s*\n\s*'''  ,   ''' \n '''  ,   a							)
       A_							:       Any	      =				re.sub(R'''[^\S\n]+'''  ,   ''' '''  ,   a							)
       return text.strip()






class     __UpperCAmelCase(    A__				):






 """simple docstring"""




 __magic_name__          =   VOCAB_FILES_NAMES
 __magic_name__          =   PRETRAINED_VOCAB_FILES_MAP
 __magic_name__          =   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 __magic_name__          =   ["""input_ids""", """attention_mask"""]
 def __init__(    self ,       __magic_name__ ,       __magic_name__ ,       __magic_name__="<unk>" ,       **__magic_name__      ):

        """simple docstring"""

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

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

               A_							:       Optional[int]	      =				English()
               A_							:       Union[str, Any]	      =				_nlp.tokenizer
               A_							:       List[str]	      =				ftfy.fix_text
        except ImportError:
               logger.warning('''ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.'''      )
               A_							:       Dict	      =				BasicTokenizer(do_lower_case=__magic_name__      )
               A_							:       str	      =				None

        with open(__magic_name__ ,       encoding='''utf-8'''      ) as vocab_handle:
               A_							:       Any	      =				json.load(__magic_name__      )
        A_							:       Optional[int]	      =				{v: k for k, v in self.encoder.items()}
        with open(__magic_name__ ,       encoding='''utf-8'''      ) as merges_handle:
               A_							:       List[Any]	      =				merges_handle.read().split('''\n'''      )[1:-1]
        A_							:       int	      =				[tuple(merge.split()      ) for merge in merges]
        A_							:       int	      =				dict(zip(__magic_name__ ,       range(len(__magic_name__      )      )      )      )
        A_							:       Optional[Any]	      =				{}
 @property
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        return True
 @property
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        return len(self.encoder      )
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        return dict(self.encoder ,       **self.added_tokens_encoder      )
 def  UpperCAmelCase   (    self ,       __magic_name__      ):

        """simple docstring"""

        A_							:       str	      =				tuple(token[:-1]      ) + (token[-1] + '''</w>''',)
        if token in self.cache:
               return self.cache[token]
        A_							:       Union[str, Any]	      =				get_pairs(__magic_name__      )

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

        while True:
               A_							:       Optional[Any]	      =				min(__magic_name__ ,       key=lambda __magic_name__      : self.bpe_ranks.get(__magic_name__ ,       float('''inf'''      )      )      )
               if bigram not in self.bpe_ranks:
                      break
               A_      ,							A_							:       Union[str, Any]	      =				bigram
               A_							:       int	      =				[]
               A_							:       List[Any]	      =				0
               while i < len(__magic_name__      ):
                      try:
                             A_							:       Union[str, Any]	      =				word.index(__magic_name__ ,       __magic_name__      )
                      except ValueError:
                             new_word.extend(word[i:]      )
                             break
                      else:
                             new_word.extend(word[i:j]      )
                             A_							:       Dict	      =				j

                      if word[i] == first and i < len(__magic_name__      ) - 1 and word[i + 1] == second:
                             new_word.append(first + second      )
                             i += 2
                      else:
                             new_word.append(word[i]      )
                             i += 1
               A_							:       Dict	      =				tuple(__magic_name__      )
               A_							:       List[Any]	      =				new_word
               if len(__magic_name__      ) == 1:
                      break
               else:
                      A_							:       List[Any]	      =				get_pairs(__magic_name__      )
        A_							:       Optional[Any]	      =				''' '''.join(__magic_name__      )
        if word == "\n  </w>":
               A_							:       List[str]	      =				'''\n</w>'''
        A_							:       Any	      =				word
        return word
 def  UpperCAmelCase   (    self ,       __magic_name__      ):

        """simple docstring"""

        A_							:       Tuple	      =				[]
        if self.fix_text is None:
               # Using BERT's BasicTokenizer
               A_							:       Optional[int]	      =				self.nlp.tokenize(__magic_name__      )
               for token in text:
                      split_tokens.extend(list(self.bpe(__magic_name__      ).split(''' '''      )      )      )
        else:
               # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
               A_							:       List[str]	      =				self.nlp(text_standardize(self.fix_text(__magic_name__      )      )      )
               for token in text:
                      split_tokens.extend(list(self.bpe(token.text.lower()      ).split(''' '''      )      )      )
        return split_tokens
 def  UpperCAmelCase   (    self ,       __magic_name__      ):

        """simple docstring"""

        return self.encoder.get(__magic_name__ ,       self.encoder.get(self.unk_token      )      )
 def  UpperCAmelCase   (    self ,       __magic_name__      ):

        """simple docstring"""

        return self.decoder.get(__magic_name__ ,       self.unk_token      )
 def  UpperCAmelCase   (    self ,       __magic_name__      ):

        """simple docstring"""

        A_							:       str	      =				''''''.join(__magic_name__      ).replace('''</w>''' ,       ''' '''      ).strip()
        return out_string
 def  UpperCAmelCase   (    self ,       __magic_name__ ,       __magic_name__ = None      ):

        """simple docstring"""

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

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

        A_							:       Any	      =				0
        with open(__magic_name__ ,       '''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 __magic_name__      : 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_							:       Dict	      =				token_index
                      writer.write(''' '''.join(__magic_name__      ) + '''\n'''      )
                      index += 1

        return vocab_file, merge_file



import os
from argparse import ArgumentParser
from pathlib import Path
from shutil import copyfile, rmtree
from typing import Generator

import datasets.config
from datasets.builder import DatasetBuilder
from datasets.commands import BaseDatasetsCLICommand
from datasets.download.download_manager import DownloadMode
from datasets.load import dataset_module_factory, import_main_class
from datasets.utils.filelock import logger as fl_logger
from datasets.utils.info_utils import VerificationMode
from datasets.utils.logging import ERROR, get_logger


_lowerCAmelCase        =       get_logger(__name__)


def       a__      (					a							)		->	Optional[int]:
       return TestCommand(
           args.dataset  ,   args.name  ,   args.cache_dir  ,   args.data_dir  ,   args.all_configs  ,   args.save_info or args.save_infos  ,   args.ignore_verifications  ,   args.force_redownload  ,   args.clear_cache  ,   )






class     __UpperCAmelCase(    A__				):






 """simple docstring"""




 __magic_name__          =   False  # to tell pytest it's not a test class
 @staticmethod
 def  UpperCAmelCase   (    __magic_name__      ):

        """simple docstring"""

        A_							:       Dict	      =				parser.add_parser('''test''' ,       help='''Test dataset implementation.'''      )
        test_parser.add_argument('''--name''' ,       type=__magic_name__ ,       default=__magic_name__ ,       help='''Dataset processing name'''      )
        test_parser.add_argument(
            '''--cache_dir''' ,       type=__magic_name__ ,       default=__magic_name__ ,       help='''Cache directory where the datasets are stored.''' ,       )
        test_parser.add_argument(
            '''--data_dir''' ,       type=__magic_name__ ,       default=__magic_name__ ,       help='''Can be used to specify a manual directory to get the files from.''' ,       )
        test_parser.add_argument('''--all_configs''' ,       action='''store_true''' ,       help='''Test all dataset configurations'''      )
        test_parser.add_argument(
            '''--save_info''' ,       action='''store_true''' ,       help='''Save the dataset infos in the dataset card (README.md)'''      )
        test_parser.add_argument(
            '''--ignore_verifications''' ,       action='''store_true''' ,       help='''Run the test without checksums and splits checks.''' ,       )
        test_parser.add_argument('''--force_redownload''' ,       action='''store_true''' ,       help='''Force dataset redownload'''      )
        test_parser.add_argument(
            '''--clear_cache''' ,       action='''store_true''' ,       help='''Remove downloaded files and cached datasets after each config test''' ,       )
        # aliases
        test_parser.add_argument('''--save_infos''' ,       action='''store_true''' ,       help='''alias to save_info'''      )
        test_parser.add_argument('''dataset''' ,       type=__magic_name__ ,       help='''Name of the dataset to download'''      )
        test_parser.set_defaults(func=_test_command_factory      )
 def __init__(    self ,       __magic_name__ ,       __magic_name__ ,       __magic_name__ ,       __magic_name__ ,       __magic_name__ ,       __magic_name__ ,       __magic_name__ ,       __magic_name__ ,       __magic_name__ ,       ):

        """simple docstring"""

        A_							:       Dict	      =				dataset
        A_							:       Any	      =				name
        A_							:       List[str]	      =				cache_dir
        A_							:       List[Any]	      =				data_dir
        A_							:       Any	      =				all_configs
        A_							:       int	      =				save_infos
        A_							:       Union[str, Any]	      =				ignore_verifications
        A_							:       List[Any]	      =				force_redownload
        A_							:       List[str]	      =				clear_cache
        if clear_cache and not cache_dir:
               print(
                   '''When --clear_cache is used, specifying a cache directory is mandatory.\n'''
                   '''The \'download\' folder of the cache directory and the dataset builder cache will be deleted after each configuration test.\n'''
                   '''Please provide a --cache_dir that will be used to test the dataset script.'''      )
               exit(1      )
        if save_infos:
               A_							:       Optional[Any]	      =				True
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        fl_logger().setLevel(__magic_name__      )
        if self._name is not None and self._all_configs:
               print('''Both parameters `config` and `all_configs` can\'t be used at once.'''      )
               exit(1      )
        A_      ,							A_							:       Tuple	      =				self._dataset, self._name
        A_							:       Dict	      =				dataset_module_factory(__magic_name__      )
        A_							:       int	      =				import_main_class(module.module_path      )
        A_							:       str	      =				len(builder_cls.BUILDER_CONFIGS      ) if self._all_configs and builder_cls.BUILDER_CONFIGS else 1

        def get_builders() -> Generator[DatasetBuilder, None, None]:
               if self._all_configs and builder_cls.BUILDER_CONFIGS:
                      for i, config in enumerate(builder_cls.BUILDER_CONFIGS      ):
                             if "config_name" in module.builder_kwargs:
                                    yield builder_cls(
                                        cache_dir=self._cache_dir ,       data_dir=self._data_dir ,       **module.builder_kwargs ,       )
                             else:
                                    yield builder_cls(
                                        config_name=config.name ,       cache_dir=self._cache_dir ,       data_dir=self._data_dir ,       **module.builder_kwargs ,       )
               else:
                      if "config_name" in module.builder_kwargs:
                             yield builder_cls(cache_dir=self._cache_dir ,       data_dir=self._data_dir ,       **module.builder_kwargs      )
                      else:
                             yield builder_cls(
                                 config_name=__magic_name__ ,       cache_dir=self._cache_dir ,       data_dir=self._data_dir ,       **module.builder_kwargs ,       )

        for j, builder in enumerate(get_builders()      ):
               print(f"""Testing builder '{builder.config.name}' ({j + 1}/{n_builders})"""      )
               A_							:       str	      =				os.path.exists(
                   os.path.join(builder.get_imported_module_dir() ,       datasets.config.DATASETDICT_INFOS_FILENAME      )      )  # record checksums only if we need to update a (deprecated) dataset_infos.json
               builder.download_and_prepare(
                   download_mode=DownloadMode.REUSE_CACHE_IF_EXISTS
                   if not self._force_redownload
                   else DownloadMode.FORCE_REDOWNLOAD ,       verification_mode=VerificationMode.NO_CHECKS
                   if self._ignore_verifications
                   else VerificationMode.ALL_CHECKS ,       try_from_hf_gcs=__magic_name__ ,       )
               builder.as_dataset()
               if self._save_infos:
                      builder._save_infos()

               # If save_infos=True, the dataset card (README.md) is created next to the loaded module file.
               # The dataset_infos are saved in the YAML part of the README.md

               # Let's move it to the original directory of the dataset script, to allow the user to
               # upload them on S3 at the same time afterwards.
               if self._save_infos:
                      A_							:       Union[str, Any]	      =				os.path.join(builder_cls.get_imported_module_dir() ,       '''README.md'''      )
                      A_							:       Union[str, Any]	      =				Path(__magic_name__      ).name + '''.py'''
                      A_							:       List[str]	      =				os.path.join(__magic_name__ ,       __magic_name__      )
                      if os.path.isfile(__magic_name__      ):
                             A_							:       str	      =				os.path.dirname(__magic_name__      )
                      elif os.path.isfile(__magic_name__      ):
                             A_							:       Any	      =				path
                      elif os.path.isdir(__magic_name__      ):  # for local directories containing only data files
                             A_							:       Optional[Any]	      =				path
                      else:  # in case of a remote dataset
                             A_							:       List[Any]	      =				None
                             print(f"""Dataset card saved at {dataset_readme_path}"""      )

                      # Move dataset_info back to the user
                      if dataset_dir is not None:
                             A_							:       Any	      =				os.path.join(__magic_name__ ,       '''README.md'''      )
                             copyfile(__magic_name__ ,       __magic_name__      )
                             print(f"""Dataset card saved at {user_dataset_readme_path}"""      )

            # If clear_cache=True, the download folder and the dataset builder cache directory are deleted
               if self._clear_cache:
                      if os.path.isdir(builder._cache_dir      ):
                             logger.warning(f"""Clearing cache at {builder._cache_dir}"""      )
                             rmtree(builder._cache_dir      )
                      A_							:       Optional[Any]	      =				os.path.join(self._cache_dir ,       datasets.config.DOWNLOADED_DATASETS_DIR      )
                      if os.path.isdir(__magic_name__      ):
                             logger.warning(f"""Clearing cache at {download_dir}"""      )
                             rmtree(__magic_name__      )

        print('''Test successful.'''      )


