


'''simple docstring'''







import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
       import jieba

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


lowerCAmelCase     :  List[Any]											=    logging.get_logger(__name__)

lowerCAmelCase     :  List[str]											=    {"""vocab_file""": """vocab.txt"""}

lowerCAmelCase     :  Dict											=    {
    """vocab_file""": {
        """openbmb/cpm-ant-10b""": """https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt""",
    },
}

lowerCAmelCase     :  str											=    {
    """openbmb/cpm-ant-10b""": 10_24,
}






def       lowercase					(_A       ):







       """simple docstring"""







       _lowerCAmelCase     :				List[str]       =							collections.OrderedDict()
       with open(__UpperCamelCase ,    'r' ,    encoding='utf-8'       ) as reader:
              _lowerCAmelCase     :				Optional[Any]       =							reader.readlines()
       for index, token in enumerate(__UpperCamelCase       ):
              _lowerCAmelCase     :				List[str]       =							token.rstrip('\n'       )
              _lowerCAmelCase     :				Union[str, Any]       =							index
       return vocab
class 						UpperCamelCase__       (			SCREAMING_SNAKE_CASE_						):



     """simple docstring"""






     def __init__(			self   ,						snake_case__   ,						snake_case__="<unk>"   ,						snake_case__=200 ):

            '''simple docstring'''

            _lowerCAmelCase     :				str       =							vocab
            _lowerCAmelCase     :				Union[str, Any]       =							unk_token
            _lowerCAmelCase     :				Optional[Any]       =							max_input_chars_per_word







     def a						(			self   ,						snake_case__ ):

            '''simple docstring'''

            _lowerCAmelCase     :				Union[str, Any]       =							list(__UpperCAmelCase )
            if len(__UpperCAmelCase ) > self.max_input_chars_per_word:
                   return [self.unk_token]

            _lowerCAmelCase     :				Optional[Any]       =							0
            _lowerCAmelCase     :				Tuple       =							[]
            while start < len(__UpperCAmelCase ):
                   _lowerCAmelCase     :				Dict       =							len(__UpperCAmelCase )
                   _lowerCAmelCase     :				Union[str, Any]       =							None
                   while start < end:
                          _lowerCAmelCase     :				List[Any]       =							"".join(chars[start:end] )
                          if substr in self.vocab:
                                 _lowerCAmelCase     :				List[str]       =							substr
                                 break
                          end -= 1
                   if cur_substr is None:
                          sub_tokens.append(self.unk_token )
                          start += 1
                   else:
                          sub_tokens.append(__UpperCAmelCase )
                          _lowerCAmelCase     :				List[Any]       =							end

            return sub_tokens
class 						UpperCamelCase__       (			SCREAMING_SNAKE_CASE_						):



     """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"]
     __magic_name__           =     False







     def __init__(			self   ,						snake_case__   ,						snake_case__="<d>"   ,						snake_case__="</d>"   ,						snake_case__="<s>"   ,						snake_case__="</s>"   ,						snake_case__="<pad>"   ,						snake_case__="<unk>"   ,						snake_case__="</n>"   ,						snake_case__="</_>"   ,						snake_case__="left"   ,						**snake_case__   ,						):

            '''simple docstring'''

            requires_backends(self   ,						['jieba'] )
            super().__init__(
                bod_token=__UpperCAmelCase   ,						eod_token=__UpperCAmelCase   ,						bos_token=__UpperCAmelCase   ,						eos_token=__UpperCAmelCase   ,						pad_token=__UpperCAmelCase   ,						unk_token=__UpperCAmelCase   ,						line_token=__UpperCAmelCase   ,						space_token=__UpperCAmelCase   ,						padding_side=__UpperCAmelCase   ,						**__UpperCAmelCase   ,						)
            _lowerCAmelCase     :				Any       =							bod_token
            _lowerCAmelCase     :				List[str]       =							eod_token
            _lowerCAmelCase     :				Optional[int]       =							load_vocab(__UpperCAmelCase )
            _lowerCAmelCase     :				Dict       =							self.encoder[space_token]
            _lowerCAmelCase     :				Dict       =							self.encoder[line_token]

            del self.encoder[space_token]
            del self.encoder[line_token]

            _lowerCAmelCase     :				int       =							collections.OrderedDict(sorted(self.encoder.items()   ,						key=lambda snake_case__ : x[1] ) )
            _lowerCAmelCase     :				Optional[int]       =							{v: k for k, v in self.encoder.items()}

            _lowerCAmelCase     :				Optional[int]       =							WordpieceTokenizer(vocab=self.encoder   ,						unk_token=self.unk_token )







     @property
     def a						(			self ):

            '''simple docstring'''

            return self.encoder[self.bod_token]







     @property
     def a						(			self ):

            '''simple docstring'''

            return self.encoder[self.eod_token]







     @property
     def a						(			self ):

            '''simple docstring'''

            return self.encoder["\n"]







     @property
     def a						(			self ):

            '''simple docstring'''

            return len(self.encoder )







     def a						(			self ):

            '''simple docstring'''

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







     def a						(			self   ,						snake_case__ ):

            '''simple docstring'''

            _lowerCAmelCase     :				Optional[int]       =							[]
            for x in jieba.cut(__UpperCAmelCase   ,						cut_all=__UpperCAmelCase ):
                   output_tokens.extend(self.wordpiece_tokenizer.tokenize(__UpperCAmelCase ) )
            return output_tokens







     def a						(			self   ,						snake_case__   ,						**snake_case__ ):

            '''simple docstring'''

            _lowerCAmelCase     :				int       =							[i for i in token_ids if i >= 0]
            _lowerCAmelCase     :				Optional[int]       =							[
                x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
            ]
            return super()._decode(__UpperCAmelCase   ,						**__UpperCAmelCase )







     def a						(			self   ,						snake_case__ ):

            '''simple docstring'''

            return token in self.encoder







     def a						(			self   ,						snake_case__ ):

            '''simple docstring'''

            return "".join(__UpperCAmelCase )







     def a						(			self   ,						snake_case__ ):

            '''simple docstring'''

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







     def a						(			self   ,						snake_case__ ):

            '''simple docstring'''

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







     def a						(			self   ,						snake_case__   ,						snake_case__ = None ):

            '''simple docstring'''

            if os.path.isdir(__UpperCAmelCase ):
                   _lowerCAmelCase     :				Tuple       =							os.path.join(
                       __UpperCAmelCase   ,						(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file'] )
            else:
                   _lowerCAmelCase     :				Union[str, Any]       =							(filename_prefix + "-" if filename_prefix else "") + save_directory
            _lowerCAmelCase     :				Union[str, Any]       =							0
            if " " in self.encoder:
                   _lowerCAmelCase     :				str       =							self.encoder[" "]
                   del self.encoder[" "]
            if "\n" in self.encoder:
                   _lowerCAmelCase     :				Optional[Any]       =							self.encoder["\n"]
                   del self.encoder["\n"]
            _lowerCAmelCase     :				List[Any]       =							collections.OrderedDict(sorted(self.encoder.items()   ,						key=lambda snake_case__ : x[1] ) )
            with open(__UpperCAmelCase   ,						'w'   ,						encoding='utf-8' ) as writer:
                   for token, token_index in self.encoder.items():
                          if index != token_index:
                                 logger.warning(
                                     F'Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive.'
                                     ' Please check that the vocabulary is not corrupted!' )
                                 _lowerCAmelCase     :				str       =							token_index
                          writer.write(token + '\n' )
                          index += 1
            return (vocab_file,)







     def a						(			self   ,						snake_case__   ,						snake_case__ = None ):

            '''simple docstring'''

            if token_ids_a is None:
                   return [self.bos_token_id] + token_ids_a
            return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a







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

            '''simple docstring'''

            if already_has_special_tokens:
                   return super().get_special_tokens_mask(
                       token_ids_a=__UpperCAmelCase   ,						token_ids_a=__UpperCAmelCase   ,						already_has_special_tokens=__UpperCAmelCase )

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






import argparse
import os

import transformers

from .convert_slow_tokenizer import SLOW_TO_FAST_CONVERTERS
from .utils import logging


logging.set_verbosity_info()

lowerCamelCase		       =      logging.get_logger(__name__)


lowerCamelCase		       =      {name: getattr(transformers, name + """Fast""") for name in SLOW_TO_FAST_CONVERTERS}






def     SCREAMING_SNAKE_CASE(      __UpperCamelCase	,				__UpperCamelCase	,				__UpperCamelCase	,				__UpperCamelCase )	->						Dict:
				if tokenizer_name is not None and tokenizer_name not in TOKENIZER_CLASSES:
								raise ValueError(F'Unrecognized tokenizer name, should be one of {list(TOKENIZER_CLASSES.keys() )}.' )

				if tokenizer_name is None:
								a__	:							str      						= TOKENIZER_CLASSES
				else:
								a__	:							int      						= {tokenizer_name: getattr(__UpperCamelCase	,				tokenizer_name + "Fast" )}

				logger.info(F'Loading tokenizer classes: {tokenizer_names}' )

				for tokenizer_name in tokenizer_names:
								a__	:							Any      						= TOKENIZER_CLASSES[tokenizer_name]

								a__	:							Dict      						= True
								if checkpoint_name is None:
												a__	:							Tuple      						= list(tokenizer_class.max_model_input_sizes.keys() )
								else:
												a__	:							List[str]      						= [checkpoint_name]

								logger.info(F'For tokenizer {tokenizer_class.__class__.__name__} loading checkpoints: {checkpoint_names}' )

								for checkpoint in checkpoint_names:
												logger.info(F'Loading {tokenizer_class.__class__.__name__} {checkpoint}' )

												# Load tokenizer
												a__	:							List[str]      						= tokenizer_class.from_pretrained(__UpperCamelCase	,				force_download=__UpperCamelCase )

												# Save fast tokenizer
												logger.info(F'Save fast tokenizer to {dump_path} with prefix {checkpoint} add_prefix {add_prefix}' )

												# For organization names we create sub-directories
												if "/" in checkpoint:
																a__						,   a__	:							Dict      						= checkpoint.split("/" )
																a__	:							Any      						= os.path.join(__UpperCamelCase	,				__UpperCamelCase )
												elif add_prefix:
																a__	:							int      						= checkpoint
																a__	:							Optional[Any]      						= dump_path
												else:
																a__	:							Dict      						= None
																a__	:							Optional[Any]      						= dump_path

												logger.info(F'=> {dump_path_full} with prefix {checkpoint_prefix_name}, add_prefix {add_prefix}' )

												if checkpoint in list(tokenizer.pretrained_vocab_files_map.values() )[0]:
																a__	:							Dict      						= list(tokenizer.pretrained_vocab_files_map.values() )[0][checkpoint]
																a__	:							Optional[Any]      						= file_path.split(__UpperCamelCase )[-1][0]
																if next_char == "/":
																				a__	:							Union[str, Any]      						= os.path.join(__UpperCamelCase	,				__UpperCamelCase )
																				a__	:							Any      						= None

																logger.info(F'=> {dump_path_full} with prefix {checkpoint_prefix_name}, add_prefix {add_prefix}' )

												a__	:							Any      						= tokenizer.save_pretrained(
												    __UpperCamelCase	,				legacy_format=__UpperCamelCase	,				filename_prefix=__UpperCamelCase )
												logger.info(F'=> File names {file_names}' )

												for file_name in file_names:
																if not file_name.endswith("tokenizer.json" ):
																				os.remove(__UpperCamelCase )
																				logger.info(F'=> removing {file_name}' )


if __name__ == "__main__":
						lowerCamelCase		       =      argparse.ArgumentParser()
						# Required parameters
						parser.add_argument(
						    """--dump_path""", default=None, type=str, required=True, help="""Path to output generated fast tokenizer files."""
						)
						parser.add_argument(
						    """--tokenizer_name""",
						    default=None,
						    type=str,
						    help=(
						        F'Optional tokenizer type selected in the list of {list(TOKENIZER_CLASSES.keys())}. If not given, will '
						        """download and convert all the checkpoints from AWS."""
						    ),
						)
						parser.add_argument(
						    """--checkpoint_name""",
						    default=None,
						    type=str,
						    help="""Optional checkpoint name. If not given, will download and convert the canonical checkpoints from AWS.""",
						)
						parser.add_argument(
						    """--force_download""",
						    action="""store_true""",
						    help="""Re-download checkpoints.""",
						)
						lowerCamelCase		       =      parser.parse_args()

						convert_slow_checkpoint_to_fast(args.tokenizer_name, args.checkpoint_name, args.dump_path, args.force_download)


