



'''simple docstring'''

import argparse
import logging
import pickle
import random
import time

import numpy as np

from transformers import BertTokenizer, GPTaTokenizer, RobertaTokenizer


logging.basicConfig(
    format='''%(asctime)s - %(levelname)s - %(name)s - %(message)s''', datefmt='''%m/%d/%Y %H:%M:%S''', level=logging.INFO
)
_A				:    Optional[int]     						=logging.getLogger(__name__)







def 						__UpperCamelCase				(							) ->			Optional[int]:
	_lowercase					:       List[Any]       =		argparse.ArgumentParser(
	    description='Preprocess the data to avoid re-doing it several times by (tokenization + token_to_ids).'					)
	parser.add_argument('--file_path',				type=_lowercase,				default='data/dump.txt',				help='The path to the data.'					)
	parser.add_argument('--tokenizer_type',				type=_lowercase,				default='bert',				choices=['bert', 'roberta', 'gpt2']					)
	parser.add_argument('--tokenizer_name',				type=_lowercase,				default='bert-base-uncased',				help='The tokenizer to use.'					)
	parser.add_argument('--dump_file',				type=_lowercase,				default='data/dump',				help='The dump file prefix.'					)
	_lowercase					:       List[str]       =		parser.parse_args()

	logger.info(f'''Loading Tokenizer ({args.tokenizer_name})'''					)
	if args.tokenizer_type == "bert":
		_lowercase					:       Tuple       =		BertTokenizer.from_pretrained(args.tokenizer_name					)
		_lowercase					:       Tuple       =		tokenizer.special_tokens_map['cls_token']  # `[CLS]`
		_lowercase					:       Any       =		tokenizer.special_tokens_map['sep_token']  # `[SEP]`
	elif args.tokenizer_type == "roberta":
		_lowercase					:       Tuple       =		RobertaTokenizer.from_pretrained(args.tokenizer_name					)
		_lowercase					:       Optional[Any]       =		tokenizer.special_tokens_map['cls_token']  # `<s>`
		_lowercase					:       Optional[int]       =		tokenizer.special_tokens_map['sep_token']  # `</s>`
	elif args.tokenizer_type == "gpt2":
		_lowercase					:       Tuple       =		GPTaTokenizer.from_pretrained(args.tokenizer_name					)
		_lowercase					:       Optional[Any]       =		tokenizer.special_tokens_map['bos_token']  # `<|endoftext|>`
		_lowercase					:       List[str]       =		tokenizer.special_tokens_map['eos_token']  # `<|endoftext|>`

	logger.info(f'''Loading text from {args.file_path}'''					)
	with open(args.file_path,				'r',				encoding='utf8'					) as fp:
		_lowercase					:       Optional[int]       =		fp.readlines()

	logger.info('Start encoding'					)
	logger.info(f'''{len(_lowercase					)} examples to process.'''					)

	_lowercase					:       Optional[int]       =		[]
	_lowercase					:       Tuple       =		0
	_lowercase					:       Union[str, Any]       =		1_0000
	_lowercase					:       Tuple       =		time.time()
	for text in data:
		_lowercase					:       int       =		f'''{bos} {text.strip()} {sep}'''
		_lowercase					:       Union[str, Any]       =		tokenizer.encode(_lowercase,				add_special_tokens=_lowercase					)
		rslt.append(_lowercase					)

		iter += 1
		if iter % interval == 0:
			_lowercase					:       Tuple       =		time.time()
			logger.info(f'''{iter} examples processed. - {(end-start):.2f}s/{interval}expl'''					)
			_lowercase					:       Tuple       =		time.time()
	logger.info('Finished binarization'					)
	logger.info(f'''{len(_lowercase					)} examples processed.'''					)

	_lowercase					:       Dict       =		f'''{args.dump_file}.{args.tokenizer_name}.pickle'''
	_lowercase					:       List[Any]       =		tokenizer.vocab_size
	if vocab_size < (1 << 16):
		_lowercase					:       Optional[int]       =		[np.uintaa(_lowercase					) for d in rslt]
	else:
		_lowercase					:       int       =		[np.intaa(_lowercase					) for d in rslt]
	random.shuffle(rslt_					)
	logger.info(f'''Dump to {dp_file}'''					)
	with open(_lowercase,				'wb'					) as handle:
		pickle.dump(rslt_,				_lowercase,				protocol=pickle.HIGHEST_PROTOCOL					)


if __name__ == "__main__":
		main()





'''simple docstring'''

_A				:    Optional[Any]     						='''ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/'''







def 						__UpperCamelCase				(							_lowercase					) ->			bytes:
	# Make sure the supplied data is a bytes-like object
	if not isinstance(_lowercase,				_lowercase					):
		_lowercase					:       Union[str, Any]       =		f'''a bytes-like object is required, not \'{data.__class__.__name__}\''''
		raise TypeError(_lowercase					)

	_lowercase					:       int       =		''.join(bin(_lowercase					)[2:].zfill(8					) for byte in data					)

	_lowercase					:       Dict       =		len(_lowercase					) % 6 != 0

	if padding_needed:
		# The padding that will be added later
		_lowercase					:       Optional[Any]       =		B'=' * ((6 - len(_lowercase					) % 6) // 2)

		# Append binary_stream with arbitrary binary digits (0's by default) to make its
		# length a multiple of 6.
		binary_stream += "0" * (6 - len(_lowercase					) % 6)
	else:
		_lowercase					:       Optional[int]       =		B''

	# Encode every 6 binary digits to their corresponding Base64 character
	return (
	    "".join(
	        B64_CHARSET[int(binary_stream[index : index + 6],				2					)]
	        for index in range(0,				len(_lowercase					),				6					)					).encode()
	    + padding
	)







def 						__UpperCamelCase				(							_lowercase					) ->			bytes:
	# Make sure encoded_data is either a string or a bytes-like object
	if not isinstance(_lowercase,				_lowercase					) and not isinstance(_lowercase,				_lowercase					):
		_lowercase					:       int       =		(
		    'argument should be a bytes-like object or ASCII string, '
		    f'''not \'{encoded_data.__class__.__name__}\''''
		)
		raise TypeError(_lowercase					)

	# In case encoded_data is a bytes-like object, make sure it contains only
	# ASCII characters so we convert it to a string object
	if isinstance(_lowercase,				_lowercase					):
		try:
			_lowercase					:       Optional[int]       =		encoded_data.decode('utf-8'					)
		except UnicodeDecodeError:
			raise ValueError('base64 encoded data should only contain ASCII characters'					)

	_lowercase					:       Optional[int]       =		encoded_data.count('='					)

	# Check if the encoded string contains non base64 characters
	if padding:
		assert all(
		    char in B64_CHARSET for char in encoded_data[:-padding]					), "Invalid base64 character(s) found."
	else:
		assert all(
		    char in B64_CHARSET for char in encoded_data					), "Invalid base64 character(s) found."

	# Check the padding
	assert len(_lowercase					) % 4 == 0 and padding < 3, "Incorrect padding"

	if padding:
		# Remove padding if there is one
		_lowercase					:       str       =		encoded_data[:-padding]

		_lowercase					:       Tuple       =		''.join(
		    bin(B64_CHARSET.index(_lowercase					)					)[2:].zfill(6					) for char in encoded_data					)[: -padding * 2]
	else:
		_lowercase					:       Union[str, Any]       =		''.join(
		    bin(B64_CHARSET.index(_lowercase					)					)[2:].zfill(6					) for char in encoded_data					)

	_lowercase					:       List[str]       =		[
	    int(binary_stream[index : index + 8],				2					)
	    for index in range(0,				len(_lowercase					),				8					)
	]

	return bytes(_lowercase					)


if __name__ == "__main__":
		import doctest

		doctest.testmod()
