
"""simple docstring"""




import os
import re
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


snake_case			       =		logging.get_logger(__name__)
snake_case			       =		{'vocab_file': 'spiece.model'}

snake_case			       =		{
    'vocab_file': {
        'google/fnet-base': 'https://huggingface.co/google/fnet-base/resolve/main/spiece.model',
        'google/fnet-large': 'https://huggingface.co/google/fnet-large/resolve/main/spiece.model',
    },
}

snake_case			       =		{
    'google/fnet-base': 5_1_2,
    'google/fnet-large': 5_1_2,
}

snake_case			       =		'▁'




class 					UpperCamelCase     (	__magic_name__				):

				"""simple docstring"""



				UpperCAmelCase_    :				List[Any]						  =    VOCAB_FILES_NAMES
				UpperCAmelCase_    :				Optional[Any]						  =    PRETRAINED_VOCAB_FILES_MAP
				UpperCAmelCase_    :				Optional[int]						  =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				UpperCAmelCase_    :				Union[str, Any]						  =    ["input_ids", "token_type_ids"]


				def __init__(    self		,     lowercase__		,     lowercase__=False		,     lowercase__=True		,     lowercase__=True		,     lowercase__="<unk>"		,     lowercase__="[SEP]"		,     lowercase__="<pad>"		,     lowercase__="[CLS]"		,     lowercase__="[MASK]"		,     lowercase__ = None		,     **lowercase__		,     )   ->   None:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     (
									    AddedToken(lowercase__		,     lstrip=lowercase__		,     rstrip=lowercase__		,     normalized=lowercase__      )
									    if isinstance(lowercase__		,     lowercase__      )
									    else mask_token
									)

									SCREAMING_SNAKE_CASE           =     {} if sp_model_kwargs is None else sp_model_kwargs

									super().__init__(
									    do_lower_case=lowercase__		,     remove_space=lowercase__		,     keep_accents=lowercase__		,     unk_token=lowercase__		,     sep_token=lowercase__		,     pad_token=lowercase__		,     cls_token=lowercase__		,     mask_token=lowercase__		,     sp_model_kwargs=self.sp_model_kwargs		,     **lowercase__		,     )

									SCREAMING_SNAKE_CASE           =     do_lower_case
									SCREAMING_SNAKE_CASE           =     remove_space
									SCREAMING_SNAKE_CASE           =     keep_accents
									SCREAMING_SNAKE_CASE           =     vocab_file

									SCREAMING_SNAKE_CASE           =     spm.SentencePieceProcessor(**self.sp_model_kwargs      )
									self.sp_model.Load(lowercase__      )


				@property
				def 							A (    self      )   ->   List[str]:
									"""simple docstring"""







									return len(self.sp_model      )


				def 							A (    self      )   ->   int:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     {self.convert_ids_to_tokens(lowercase__      ): i for i in range(self.vocab_size      )}
									vocab.update(self.added_tokens_encoder      )
									return vocab


				def __getstate__(    self      )   ->   Optional[int]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self.__dict__.copy()
									SCREAMING_SNAKE_CASE           =     None
									return state


				def __setstate__(    self		,     lowercase__      )   ->   Dict:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     d

									# for backward compatibility
									if not hasattr(self		,     'sp_model_kwargs'      ):
														SCREAMING_SNAKE_CASE           =     {}

									SCREAMING_SNAKE_CASE           =     spm.SentencePieceProcessor(**self.sp_model_kwargs      )
									self.sp_model.Load(self.vocab_file      )


				def 							A (    self		,     lowercase__      )   ->   Tuple:
									"""simple docstring"""







									if self.remove_space:
														SCREAMING_SNAKE_CASE           =     ' '.join(inputs.strip().split()      )
									else:
														SCREAMING_SNAKE_CASE           =     inputs
									SCREAMING_SNAKE_CASE           =     outputs.replace('``'		,     '"'      ).replace('\'\''		,     '"'      )

									if not self.keep_accents:
														SCREAMING_SNAKE_CASE           =     unicodedata.normalize('NFKD'		,     lowercase__      )
														SCREAMING_SNAKE_CASE           =     ''.join([c for c in outputs if not unicodedata.combining(lowercase__      )]      )
									if self.do_lower_case:
														SCREAMING_SNAKE_CASE           =     outputs.lower()

									return outputs


				def 							A (    self		,     lowercase__      )   ->   List[str]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self.preprocess_text(lowercase__      )
									SCREAMING_SNAKE_CASE           =     self.sp_model.encode(lowercase__		,     out_type=lowercase__      )
									SCREAMING_SNAKE_CASE           =     []
									for piece in pieces:
														if len(lowercase__      ) > 1 and piece[-1] == str(','      ) and piece[-2].isdigit():
																			SCREAMING_SNAKE_CASE           =     self.sp_model.EncodeAsPieces(piece[:-1].replace(lowercase__		,     ''      )      )
																			if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
																								if len(cur_pieces[0]      ) == 1:
																													SCREAMING_SNAKE_CASE           =     cur_pieces[1:]
																								else:
																													SCREAMING_SNAKE_CASE           =     cur_pieces[0][1:]
																			cur_pieces.append(piece[-1]      )
																			new_pieces.extend(lowercase__      )
														else:
																			new_pieces.append(lowercase__      )

									return new_pieces


				def 							A (    self		,     lowercase__      )   ->   Optional[Any]:
									"""simple docstring"""







									return self.sp_model.PieceToId(lowercase__      )


				def 							A (    self		,     lowercase__      )   ->   Optional[Any]:
									"""simple docstring"""







									return self.sp_model.IdToPiece(lowercase__      )


				def 							A (    self		,     lowercase__      )   ->   List[str]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     []
									SCREAMING_SNAKE_CASE           =     ''
									SCREAMING_SNAKE_CASE           =     False
									for token in tokens:
														# make sure that special tokens are not decoded using sentencepiece model
														if token in self.all_special_tokens:
																			if not prev_is_special:
																								out_string += " "
																			out_string += self.sp_model.decode(lowercase__      ) + token
																			SCREAMING_SNAKE_CASE           =     True
																			SCREAMING_SNAKE_CASE           =     []
														else:
																			current_sub_tokens.append(lowercase__      )
																			SCREAMING_SNAKE_CASE           =     False
									out_string += self.sp_model.decode(lowercase__      )
									return out_string.strip()


				def 							A (    self		,     lowercase__		,     lowercase__ = False		,     lowercase__ = None		,     lowercase__ = True		,     **lowercase__		,     )   ->   str:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     kwargs.pop('use_source_tokenizer'		,     lowercase__      )

									SCREAMING_SNAKE_CASE           =     self.convert_ids_to_tokens(lowercase__		,     skip_special_tokens=lowercase__      )

									# To avoid mixing byte-level and unicode for byte-level BPT
									# we need to build string separately for added tokens and byte-level tokens
									# cf. https://github.com/huggingface/transformers/issues/1133
									SCREAMING_SNAKE_CASE           =     []
									SCREAMING_SNAKE_CASE           =     []
									for token in filtered_tokens:
														if skip_special_tokens and token in self.all_special_ids:
																			continue
														if token in self.added_tokens_encoder:
																			if current_sub_text:
																								sub_texts.append(self.convert_tokens_to_string(lowercase__      )      )
																								SCREAMING_SNAKE_CASE           =     []
																			sub_texts.append(lowercase__      )
														else:
																			current_sub_text.append(lowercase__      )
									if current_sub_text:
														sub_texts.append(self.convert_tokens_to_string(lowercase__      )      )

									# Mimic the behavior of the Rust tokenizer:
									# No space after <unk>
									if spaces_between_special_tokens:
														SCREAMING_SNAKE_CASE           =     re.sub(r'(<unk>) '		,     r'\1'		,     ' '.join(lowercase__      )      )
									else:
														SCREAMING_SNAKE_CASE           =     ''.join(lowercase__      )

									SCREAMING_SNAKE_CASE           =     (
									    clean_up_tokenization_spaces
									    if clean_up_tokenization_spaces is not None
									    else self.clean_up_tokenization_spaces
									)
									if clean_up_tokenization_spaces:
														SCREAMING_SNAKE_CASE           =     self.clean_up_tokenization(lowercase__      )
														return clean_text
									else:
														return text


				def 							A (    self		,     lowercase__		,     lowercase__ = None      )   ->   List[int]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     [self.sep_token_id]
									SCREAMING_SNAKE_CASE           =     [self.cls_token_id]
									if token_ids_a is None:
														return cls + token_ids_a + sep
									return cls + token_ids_a + sep + token_ids_a + sep


				def 							A (    self		,     lowercase__		,     lowercase__ = None		,     lowercase__ = False      )   ->   List[int]:
									"""simple docstring"""







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

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


				def 							A (    self		,     lowercase__		,     lowercase__ = None      )   ->   List[int]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     [self.sep_token_id]
									SCREAMING_SNAKE_CASE           =     [self.cls_token_id]

									if token_ids_a is None:
														return len(cls + token_ids_a + sep      ) * [0]
									return len(cls + token_ids_a + sep      ) * [0] + len(token_ids_a + sep      ) * [1]




				def 							A (    self		,     lowercase__		,     lowercase__ = None      )   ->   Tuple[str]:
									"""simple docstring"""







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

									if os.path.abspath(self.vocab_file      ) != os.path.abspath(lowercase__      ) and os.path.isfile(self.vocab_file      ):
														copyfile(self.vocab_file		,     lowercase__      )
									elif not os.path.isfile(self.vocab_file      ):
														with open(lowercase__		,     'wb'      ) as fi:
																			SCREAMING_SNAKE_CASE           =     self.sp_model.serialized_model_proto()
																			fi.write(lowercase__      )

									return (out_vocab_file,)


"""simple docstring"""




import argparse
import logging

import datasets
from tokenizers import Tokenizer, decoders, normalizers, pre_tokenizers, processors
from tokenizers.models import Unigram
from tokenizers.trainers import UnigramTrainer

from transformers import AlbertTokenizerFast


snake_case			       =		logging.getLogger(__name__)

def   UpperCamelCase_  (		):
					SCREAMING_SNAKE_CASE           =     argparse.ArgumentParser(description='Train a unigram tokenizer on the wikitext dataset.'      )
					parser.add_argument(
					    '--dataset_name',					type=SCREAMING_SNAKE_CASE_,					default='wikitext',					help='Name of the training. Explore datasets at: hf.co/datasets.',					)
					parser.add_argument(
					    '--dataset_config',					type=SCREAMING_SNAKE_CASE_,					default='wikitext-103-raw-v1',					help='Configuration name of the dataset.'      )
					parser.add_argument(
					    '--batch_size',					type=SCREAMING_SNAKE_CASE_,					default=1_0_0_0,					help='Batch size during training.',					)
					parser.add_argument(
					    '--vocab_size',					type=SCREAMING_SNAKE_CASE_,					default=1_0_0_4_8,					help='Size of the desired vocabulary.',					)
					parser.add_argument(
					    '--limit',					default=SCREAMING_SNAKE_CASE_,					type=SCREAMING_SNAKE_CASE_,					help='Limit the number of shards (used for debugging).',					)
					parser.add_argument(
					    '--export_to_hub',					action='store_true',					)

					SCREAMING_SNAKE_CASE           =     parser.parse_args()
					return args

def   UpperCamelCase_  (		SCREAMING_SNAKE_CASE_      ):
					SCREAMING_SNAKE_CASE           =     datasets.load_dataset(args.dataset_name,					args.dataset_config,					split='train'      )

					if args.limit is not None:
										SCREAMING_SNAKE_CASE           =     min(len(SCREAMING_SNAKE_CASE_      ),					args.limit      )
										SCREAMING_SNAKE_CASE           =     dataset.select(range(SCREAMING_SNAKE_CASE_      )      )
										logger.info(f'''Limiting the dataset to {args.limit} entries.'''      )

					def batch_iterator():
										for i in range(0,					len(SCREAMING_SNAKE_CASE_      ),					args.batch_size      ):
															yield dataset[i : i + args.batch_size]["text"]

    # Prepare the tokenizer.
					SCREAMING_SNAKE_CASE           =     Tokenizer(Unigram()      )
					SCREAMING_SNAKE_CASE           =     normalizers.Sequence([normalizers.Replace('``',					'"'      ), normalizers.Replace('\'\'',					'"'      )]      )
					SCREAMING_SNAKE_CASE           =     pre_tokenizers.Metaspace()

					# Prepare the trainer.
					SCREAMING_SNAKE_CASE           =     UnigramTrainer(
					    unk_token='<unk>',					special_tokens=['[CLS]', '[SEP]', '<unk>', '<pad>', '[MASK]'],					vocab_size=args.vocab_size,					)

					logger.info('Training the tokenizer.'      )
					tokenizer.train_from_iterator(batch_iterator(),					trainer=SCREAMING_SNAKE_CASE_      )
					logger.info('Tokenizer training complete!'      )

					SCREAMING_SNAKE_CASE           =     tokenizer.token_to_id('[CLS]'      )
					SCREAMING_SNAKE_CASE           =     tokenizer.token_to_id('[SEP]'      )
					SCREAMING_SNAKE_CASE           =     processors.TemplateProcessing(
					    single='[CLS]:0 $A:0 [SEP]:0',					pair='[CLS]:0 $A:0 [SEP]:0 $B:1 [SEP]:1',					special_tokens=[
					        ('[CLS]', cls_token_id),
					        ('[SEP]', sep_token_id),
					    ],					)
					SCREAMING_SNAKE_CASE           =     decoders.Metaspace()

					if args.export_to_hub:
										logger.info('Exporting the trained tokenzier to Hub.'      )
										SCREAMING_SNAKE_CASE           =     AlbertTokenizerFast(tokenizer_object=SCREAMING_SNAKE_CASE_      )
										new_tokenizer.push_to_hub('unigram-tokenizer-dataset'      )


if __name__ == "__main__":
	snake_case			       =		parse_args()
	main(args)
