
'''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


UpperCAmelCase            = logging.getLogger(__name__)

def   A			(							):
						snake_case							:  List[Any]			=     argparse.ArgumentParser(description='''Train a unigram tokenizer on the wikitext dataset.'''							)
						parser.add_argument(
						    '''--dataset_name'''     ,				type=A_     ,				default='''wikitext'''     ,				help='''Name of the training. Explore datasets at: hf.co/datasets.'''     ,				)
						parser.add_argument(
						    '''--dataset_config'''     ,				type=A_     ,				default='''wikitext-103-raw-v1'''     ,				help='''Configuration name of the dataset.'''							)
						parser.add_argument(
						    '''--batch_size'''     ,				type=A_     ,				default=1000     ,				help='''Batch size during training.'''     ,				)
						parser.add_argument(
						    '''--vocab_size'''     ,				type=A_     ,				default=10048     ,				help='''Size of the desired vocabulary.'''     ,				)
						parser.add_argument(
						    '''--limit'''     ,				default=A_     ,				type=A_     ,				help='''Limit the number of shards (used for debugging).'''     ,				)
						parser.add_argument(
						    '''--export_to_hub'''     ,				action='''store_true'''     ,				)

						snake_case							:  List[Any]			=     parser.parse_args()
						return args

def   A			(							A_				:       Dict							):
						snake_case							:  List[Any]			=     datasets.load_dataset(args.dataset_name     ,				args.dataset_config     ,				split='''train'''							)

						if args.limit is not None:
												snake_case							:  int			=     min(len(A_							)     ,				args.limit							)
												snake_case							:  Optional[int]			=     dataset.select(range(A_							)							)
												logger.info(F"""Limiting the dataset to {args.limit} entries."""							)

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

    # Prepare the tokenizer.
						snake_case							:  Union[str, Any]			=     Tokenizer(Unigram()							)
						snake_case							:  List[Any]			=     normalizers.Sequence([normalizers.Replace('''``'''     ,				'''"'''							), normalizers.Replace('''\'\''''     ,				'''"'''							)]							)
						snake_case							:  Any			=     pre_tokenizers.Metaspace()

						# Prepare the trainer.
						snake_case							:  int			=     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=A_							)
						logger.info('''Tokenizer training complete!'''							)

						snake_case							:  Union[str, Any]			=     tokenizer.token_to_id('''[CLS]'''							)
						snake_case							:  Tuple			=     tokenizer.token_to_id('''[SEP]'''							)
						snake_case							:  List[str]			=     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),
						    ]     ,				)
						snake_case							:  str			=     decoders.Metaspace()

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


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








from __future__ import annotations

from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any



class  _a  :






							"""simple docstring"""

							snake_case_							=    ("_front", "_back", "_len")



							@dataclass
							class  _a  :






														"""simple docstring"""

														snake_case_							=    None
														snake_case_							=    None
														snake_case_							=    None







							class  _a  :






														"""simple docstring"""

														snake_case_							=    ("_cur",)

														def __init__(   self					:  Tuple				,						a					:  Deque._Node | None     )						->None:
																				SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       cur

														def __iter__(   self					:  Dict     )						->Deque._Iterator:
																				return self






														def   A_	(   self					:  int     )						->Any:
																				if self._cur is None:
																										# finished iterating
																										raise StopIteration
																				SCREAMING_SNAKE_CASE__  :					Tuple =       self._cur.val
																				SCREAMING_SNAKE_CASE__  :					int =       self._cur.next_node

																				return val







							def __init__(   self					:  str				,						a					:  Iterable[Any] | None = None     )						->None:
													SCREAMING_SNAKE_CASE__  :					Any =       None
													SCREAMING_SNAKE_CASE__  :					Any =       None
													SCREAMING_SNAKE_CASE__  :					int =       0

													if iterable is not None:
																			# append every value to the deque
																			for val in iterable:
																									self.append(a     )

							def   A_	(   self					:  Tuple				,						a					:  Any     )						->None:
													SCREAMING_SNAKE_CASE__  :					Optional[Any] =       self._Node(a				,						a				,						a     )
													if self.is_empty():
																			# front = back
																			SCREAMING_SNAKE_CASE__  :					int =       node
																			SCREAMING_SNAKE_CASE__  :					Dict =       1
													else:
																			# connect nodes
																			SCREAMING_SNAKE_CASE__  :					Optional[int] =       node
																			SCREAMING_SNAKE_CASE__  :					Any =       self._back
																			SCREAMING_SNAKE_CASE__  :					Tuple =       node  # assign new back to the new node

																			self._len += 1

																			# make sure there were no errors
																			assert not self.is_empty(), "Error on appending value."

							def   A_	(   self					:  List[str]				,						a					:  Any     )						->None:
													SCREAMING_SNAKE_CASE__  :					List[Any] =       self._Node(a				,						a				,						a     )
													if self.is_empty():
																			# front = back
																			SCREAMING_SNAKE_CASE__  :					str =       node
																			SCREAMING_SNAKE_CASE__  :					Dict =       1
													else:
																			# connect nodes
																			SCREAMING_SNAKE_CASE__  :					Any =       self._front
																			SCREAMING_SNAKE_CASE__  :					Tuple =       node
																			SCREAMING_SNAKE_CASE__  :					int =       node  # assign new front to the new node

																			self._len += 1

																			# make sure there were no errors
																			assert not self.is_empty(), "Error on appending value."

							def   A_	(   self					:  Dict				,						a					:  Iterable[Any]     )						->None:
													for val in iterable:
																			self.append(a     )

							def   A_	(   self					:  Tuple				,						a					:  Iterable[Any]     )						->None:
													for val in iterable:
																			self.appendleft(a     )

							def   A_	(   self					:  Any     )						->Any:
													assert not self.is_empty(), "Deque is empty."

													SCREAMING_SNAKE_CASE__  :					List[Any] =       self._back
													SCREAMING_SNAKE_CASE__  :					Optional[int] =       self._back.prev_node  # set new back
													# drop the last node - python will deallocate memory automatically
													SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       None

													self._len -= 1

													return topop.val

							def   A_	(   self					:  Any     )						->Any:
													assert not self.is_empty(), "Deque is empty."

													SCREAMING_SNAKE_CASE__  :					Any =       self._front
													SCREAMING_SNAKE_CASE__  :					Dict =       self._front.next_node  # set new front and drop the first node
													SCREAMING_SNAKE_CASE__  :					Dict =       None

													self._len -= 1

													return topop.val

							def   A_	(   self					:  str     )						->bool:
													return self._front is None

							def __len__(   self					:  List[str]     )						->int:
													return self._len

							def __eq__(   self					:  List[Any]				,						a					:  object     )						->bool:

													if not isinstance(a				,						a     ):
																			return NotImplemented

													SCREAMING_SNAKE_CASE__  :					List[Any] =       self._front
													SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       other._front

													# if the length of the dequeues are not the same, they are not equal
													if len(self     ) != len(a     ):
																			return False

													while me is not None and oth is not None:
																			# compare every value
																			if me.val != oth.val:
																									return False
																			SCREAMING_SNAKE_CASE__  :					Any =       me.next_node
																			SCREAMING_SNAKE_CASE__  :					Any =       oth.next_node

													return True

							def __iter__(   self					:  Any     )						->Deque._Iterator:
													return Deque._Iterator(self._front     )






							def __repr__(   self					:  Any     )						->str:
													SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       []
													SCREAMING_SNAKE_CASE__  :					int =       self._front
													while aux is not None:
																			# append the values in a list to display
																			values_list.append(aux.val     )
																			SCREAMING_SNAKE_CASE__  :					Union[str, Any] =       aux.next_node

													return f"""[{', '.join(repr(a     ) for val in values_list     )}]"""


if __name__ == "__main__":
				import doctest

				doctest.testmod()