


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			:					Tuple         =							logging.get_logger(__name__)

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

UpperCamelCase			:					int         =							{
    '''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			:					Tuple         =							{
    '''openai-gpt''': 512,
}
def 				UpperCamelCase_	(					__snake_case      :    str  )->   Dict:




						'''simple docstring'''
						__magic_name__					=      text.strip()
						if not text:
												return []
						__magic_name__					=      text.split()
						return tokens

class      __snake_case				(   _UpperCAmelCase):
					def __init__(						self       :						Any	,      _lowerCAmelCase       :						List[Any]=True	,      _lowerCAmelCase       :						Optional[int]=None	,      _lowerCAmelCase       :						int=True	,      _lowerCAmelCase       :						List[Any]=None	,      _lowerCAmelCase       :						Optional[int]=True	,      )							->					List[Any]:



											'''simple docstring'''





											if never_split is None:
																	__magic_name__					=      []
											__magic_name__					=      do_lower_case
											__magic_name__					=      set(_lowerCAmelCase  )
											__magic_name__					=      tokenize_chinese_chars
											__magic_name__					=      strip_accents
											__magic_name__					=      do_split_on_punc
					def 	a			(						self       :						Optional[Any]	,      _lowerCAmelCase       :						Optional[int]	,      _lowerCAmelCase       :						List[str]=None  )							->					int:



											'''simple docstring'''





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

											__magic_name__					=      whitespace_tokenize(' '.join(_lowerCAmelCase  )  )
											return output_tokens
					def 	a			(						self       :						Optional[int]	,      _lowerCAmelCase       :						Optional[int]  )							->					Tuple:



											'''simple docstring'''





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



											'''simple docstring'''





											if not self.do_split_on_punc or (never_split is not None and text in never_split):
																	return [text]
											__magic_name__					=      list(_lowerCAmelCase  )
											__magic_name__					=      0
											__magic_name__					=      True
											__magic_name__					=      []
											while i < len(_lowerCAmelCase  ):
																	__magic_name__					=      chars[i]
																	if _is_punctuation(_lowerCAmelCase  ):
																							output.append([char]  )
																							__magic_name__					=      True
																	else:
																							if start_new_word:
																													output.append([]  )
																							__magic_name__					=      False
																							output[-1].append(_lowerCAmelCase  )
																	i += 1

											return ["".join(_lowerCAmelCase  ) for x in output]
					def 	a			(						self       :						str	,      _lowerCAmelCase       :						Dict  )							->					Dict:



											'''simple docstring'''





											__magic_name__					=      []
											for char in text:
																	__magic_name__					=      ord(_lowerCAmelCase  )
																	if self._is_chinese_char(_lowerCAmelCase  ):
																							output.append(' '  )
																							output.append(_lowerCAmelCase  )
																							output.append(' '  )
																	else:
																							output.append(_lowerCAmelCase  )
											return "".join(_lowerCAmelCase  )
					def 	a			(						self       :						Any	,      _lowerCAmelCase       :						List[str]  )							->					str:



											'''simple docstring'''





											if (
											    (cp >= 0x4E00 and cp <= 0x9FFF)
											    or (cp >= 0x3400 and cp <= 0x4DBF)  #
											    or (cp >= 0x2_0000 and cp <= 0x2_A6DF)  #
											    or (cp >= 0x2_A700 and cp <= 0x2_B73F)  #
											    or (cp >= 0x2_B740 and cp <= 0x2_B81F)  #
											    or (cp >= 0x2_B820 and cp <= 0x2_CEAF)  #
											    or (cp >= 0xF900 and cp <= 0xFAFF)
											    or (cp >= 0x2_F800 and cp <= 0x2_FA1F)  #
											):  #
																	return True

											return False







					def 	a			(						self       :						Union[str, Any]	,      _lowerCAmelCase       :						Optional[int]  )							->					Dict:



											'''simple docstring'''





											__magic_name__					=      []
											for char in text:
																	__magic_name__					=      ord(_lowerCAmelCase  )
																	if cp == 0 or cp == 0xFFFD or _is_control(_lowerCAmelCase  ):
																							continue
																	if _is_whitespace(_lowerCAmelCase  ):
																							output.append(' '  )
																	else:
																							output.append(_lowerCAmelCase  )
											return "".join(_lowerCAmelCase  )







def 				UpperCamelCase_	(					__snake_case      :    Dict  )->   Dict:




						'''simple docstring'''
						__magic_name__					=      set()
						__magic_name__					=      word[0]
						for char in word[1:]:
												pairs.add((prev_char, char)  )
												__magic_name__					=      char
						return pairs
def 				UpperCamelCase_	(					__snake_case      :    List[Any]  )->   str:




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




class      __snake_case				(   _UpperCAmelCase):
					SCREAMING_SNAKE_CASE_		:       Any								=			VOCAB_FILES_NAMES
					SCREAMING_SNAKE_CASE_		:       Optional[Any]								=			PRETRAINED_VOCAB_FILES_MAP
					SCREAMING_SNAKE_CASE_		:       Optional[int]								=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					SCREAMING_SNAKE_CASE_		:       str								=			["input_ids", "attention_mask"]
					def __init__(						self       :						Tuple	,      _lowerCAmelCase       :						str	,      _lowerCAmelCase       :						List[str]	,      _lowerCAmelCase       :						List[Any]="<unk>"	,      **_lowerCAmelCase       :						Optional[Any]  )							->					str:



											'''simple docstring'''





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

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

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

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



											'''simple docstring'''





											return True
					@property
					def 	a			(						self       :						int  )							->					Optional[int]:



											'''simple docstring'''





											return len(self.encoder  )
					def 	a			(						self       :						List[str]  )							->					str:



											'''simple docstring'''





											return dict(self.encoder	,      **self.added_tokens_encoder  )
					def 	a			(						self       :						int	,      _lowerCAmelCase       :						Tuple  )							->					str:



											'''simple docstring'''





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

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

											while True:
																	__magic_name__					=      min(_lowerCAmelCase	,      key=lambda _lowerCAmelCase  : self.bpe_ranks.get(_lowerCAmelCase	,      float('inf'  )  )  )
																	if bigram not in self.bpe_ranks:
																							break
																	__magic_name__		,      __magic_name__					=      bigram
																	__magic_name__					=      []
																	__magic_name__					=      0
																	while i < len(_lowerCAmelCase  ):
																							try:
																													__magic_name__					=      word.index(_lowerCAmelCase	,      _lowerCAmelCase  )
																							except ValueError:
																													new_word.extend(word[i:]  )
																													break
																							else:
																													new_word.extend(word[i:j]  )
																													__magic_name__					=      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
																	__magic_name__					=      tuple(_lowerCAmelCase  )
																	__magic_name__					=      new_word
																	if len(_lowerCAmelCase  ) == 1:
																							break
																	else:
																							__magic_name__					=      get_pairs(_lowerCAmelCase  )
											__magic_name__					=      ' '.join(_lowerCAmelCase  )
											if word == "\n  </w>":
																	__magic_name__					=      '\n</w>'
											__magic_name__					=      word
											return word
					def 	a			(						self       :						Optional[Any]	,      _lowerCAmelCase       :						List[Any]  )							->					Any:



											'''simple docstring'''





											__magic_name__					=      []
											if self.fix_text is None:
																	# Using BERT's BasicTokenizer
																	__magic_name__					=      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)
																	__magic_name__					=      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 	a			(						self       :						int	,      _lowerCAmelCase       :						Dict  )							->					Optional[int]:



											'''simple docstring'''





											return self.encoder.get(_lowerCAmelCase	,      self.encoder.get(self.unk_token  )  )
					def 	a			(						self       :						Optional[int]	,      _lowerCAmelCase       :						List[str]  )							->					Optional[int]:



											'''simple docstring'''





											return self.decoder.get(_lowerCAmelCase	,      self.unk_token  )
					def 	a			(						self       :						Optional[Any]	,      _lowerCAmelCase       :						Any  )							->					Dict:



											'''simple docstring'''





											__magic_name__					=      ''.join(_lowerCAmelCase  ).replace('</w>'	,      ' '  ).strip()
											return out_string







					def 	a			(						self       :						List[Any]	,      _lowerCAmelCase       :						str	,      _lowerCAmelCase       :						Optional[str] = None  )							->					Tuple[str]:



											'''simple docstring'''





											if not os.path.isdir(_lowerCAmelCase  ):
																	logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''  )
																	return
											__magic_name__					=      os.path.join(
											    _lowerCAmelCase	,      (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']  )
											__magic_name__					=      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'  )

											__magic_name__					=      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!'  )
																													__magic_name__					=      token_index
																							writer.write(' '.join(_lowerCAmelCase  ) + '\n'  )
																							index += 1

											return vocab_file, merge_file





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


UpperCamelCase			:					List[str]         =							logging.getLogger(__name__)
def 				UpperCamelCase_	(					)->   Dict:




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

						__magic_name__					=      parser.parse_args()
						return args
def 				UpperCamelCase_	(					__snake_case      :    Any  )->   Optional[int]:




						'''simple docstring'''
						__magic_name__					=      datasets.load_dataset(args.dataset_name						,      args.dataset_config						,      split='train'  )

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

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

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

						# Prepare the trainer.
						__magic_name__					=      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=__snake_case  )
						logger.info('Tokenizer training complete!'  )

						__magic_name__					=      tokenizer.token_to_id('[CLS]'  )
						__magic_name__					=      tokenizer.token_to_id('[SEP]'  )
						__magic_name__					=      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),
						    ]						,      )
						__magic_name__					=      decoders.Metaspace()

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


if __name__ == "__main__":
					UpperCamelCase			:					Dict         =							parse_args()
					main(args)

