





import os
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 SPIECE_UNDERLINE, logging


lowerCAmelCase_    =    logging.get_logger(__name__)

lowerCAmelCase_    =    {'''vocab_file''': '''spiece.model'''}

lowerCAmelCase_    =    {
    '''vocab_file''': {
        '''xlnet-base-cased''': '''https://huggingface.co/xlnet-base-cased/resolve/main/spiece.model''',
        '''xlnet-large-cased''': '''https://huggingface.co/xlnet-large-cased/resolve/main/spiece.model''',
    }
}

lowerCAmelCase_    =    {
    '''xlnet-base-cased''': None,
    '''xlnet-large-cased''': None,
}

# Segments (not really needed)
lowerCAmelCase_    =    0
lowerCAmelCase_    =    1
lowerCAmelCase_    =    2
lowerCAmelCase_    =    3
lowerCAmelCase_    =    4



class  __lowerCAmelCase	(      _a					):
   lowerCamelCase_					:					int	      =				VOCAB_FILES_NAMES
   lowerCamelCase_					:					Union[str, Any]	      =				PRETRAINED_VOCAB_FILES_MAP
   lowerCamelCase_					:					Dict	      =				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   lowerCamelCase_					:					Optional[Any]	      =				'''left'''






   def __init__(self ,							__magic_name__ ,							__magic_name__=False ,							__magic_name__=True ,							__magic_name__=False ,							__magic_name__="<s>" ,							__magic_name__="</s>" ,							__magic_name__="<unk>" ,							__magic_name__="<sep>" ,							__magic_name__="<pad>" ,							__magic_name__="<cls>" ,							__magic_name__="<mask>" ,							__magic_name__=["<eop>", "<eod>"] ,							__magic_name__ = None ,							**__magic_name__ ,							)					->       None:


       '''simple docstring'''



       snake_case_						:   Dict           =  AddedToken(__magic_name__ ,							lstrip=__magic_name__ ,							rstrip=__magic_name__   ) if isinstance(__magic_name__ ,							__magic_name__   ) else mask_token

       snake_case_						:   List[Any]           =  {} if sp_model_kwargs is None else sp_model_kwargs

       super().__init__(
           do_lower_case=__magic_name__ ,							remove_space=__magic_name__ ,							keep_accents=__magic_name__ ,							bos_token=__magic_name__ ,							eos_token=__magic_name__ ,							unk_token=__magic_name__ ,							sep_token=__magic_name__ ,							pad_token=__magic_name__ ,							cls_token=__magic_name__ ,							mask_token=__magic_name__ ,							additional_special_tokens=__magic_name__ ,							sp_model_kwargs=self.sp_model_kwargs ,							**__magic_name__ ,							)

       snake_case_						:   Union[str, Any]           =  3

       snake_case_						:   List[str]           =  do_lower_case
       snake_case_						:   Union[str, Any]           =  remove_space
       snake_case_						:   List[Any]           =  keep_accents
       snake_case_						:   Union[str, Any]           =  vocab_file

       snake_case_						:   Optional[Any]           =  spm.SentencePieceProcessor(**self.sp_model_kwargs   )
       self.sp_model.Load(__magic_name__   )






   @property
   def       lowerCamelCase  (self   )					->       Dict:


       '''simple docstring'''



       return len(self.sp_model   )






   def       lowerCamelCase  (self   )					->       Any:


       '''simple docstring'''



       snake_case_						:   List[str]           =  {self.convert_ids_to_tokens(__magic_name__   ): i for i in range(self.vocab_size   )}
       vocab.update(self.added_tokens_encoder   )
       return vocab






   def __getstate__(self   )					->       List[str]:


       '''simple docstring'''



       snake_case_						:   Union[str, Any]           =  self.__dict__.copy()
       snake_case_						:   Optional[Any]           =  None
       return state






   def __setstate__(self ,							__magic_name__   )					->       int:


       '''simple docstring'''



       snake_case_						:   Any           =  d

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

       snake_case_						:   Dict           =  spm.SentencePieceProcessor(**self.sp_model_kwargs   )
       self.sp_model.Load(self.vocab_file   )






   def       lowerCamelCase  (self ,							__magic_name__   )					->       int:


       '''simple docstring'''



       if self.remove_space:
           snake_case_						:   List[str]           =  ''' '''.join(inputs.strip().split()   )
       else:
           snake_case_						:   Tuple           =  inputs
       snake_case_						:   List[str]           =  outputs.replace('''``''' ,							'''"'''   ).replace('''\'\'''' ,							'''"'''   )

       if not self.keep_accents:
           snake_case_						:   List[str]           =  unicodedata.normalize('''NFKD''' ,							__magic_name__   )
           snake_case_						:   Tuple           =  ''''''.join([c for c in outputs if not unicodedata.combining(__magic_name__   )]   )
       if self.do_lower_case:
           snake_case_						:   Any           =  outputs.lower()

       return outputs






   def       lowerCamelCase  (self ,							__magic_name__   )					->       List[str]:


       '''simple docstring'''



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

       return new_pieces






   def       lowerCamelCase  (self ,							__magic_name__   )					->       int:


       '''simple docstring'''



       return self.sp_model.PieceToId(__magic_name__   )






   def       lowerCamelCase  (self ,							__magic_name__   )					->       List[Any]:


       '''simple docstring'''



       return self.sp_model.IdToPiece(__magic_name__   )






   def       lowerCamelCase  (self ,							__magic_name__   )					->       Union[str, Any]:


       '''simple docstring'''



       snake_case_						:   str           =  ''''''.join(__magic_name__   ).replace(__magic_name__ ,							''' '''   ).strip()
       return out_string






   def       lowerCamelCase  (self ,							__magic_name__ ,							__magic_name__ = False ,							__magic_name__ = None ,							__magic_name__ = True ,							**__magic_name__ ,							)					->       str:


       '''simple docstring'''



       snake_case_						:   Any           =  kwargs.pop('''use_source_tokenizer''' ,							__magic_name__   )

       snake_case_						:   Union[str, Any]           =  self.convert_ids_to_tokens(__magic_name__ ,							skip_special_tokens=__magic_name__   )

       # 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
       snake_case_						:   List[Any]           =  []
       snake_case_						:   int           =  []
       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(__magic_name__   )   )
                   snake_case_						:   Union[str, Any]           =  []
               sub_texts.append(__magic_name__   )
           else:
               current_sub_text.append(__magic_name__   )
       if current_sub_text:
           sub_texts.append(self.convert_tokens_to_string(__magic_name__   )   )

       # Mimic the behavior of the Rust tokenizer:
       # By default, there are no spaces between special tokens
       snake_case_						:   Union[str, Any]           =  ''''''.join(__magic_name__   )

       snake_case_						:   Optional[Any]           =  (
           clean_up_tokenization_spaces
           if clean_up_tokenization_spaces is not None
           else self.clean_up_tokenization_spaces
       )
       if clean_up_tokenization_spaces:
           snake_case_						:   Union[str, Any]           =  self.clean_up_tokenization(__magic_name__   )
           return clean_text
       else:
           return text






   def       lowerCamelCase  (self ,							__magic_name__ ,							__magic_name__ = None   )					->       List[int]:


       '''simple docstring'''



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






   def       lowerCamelCase  (self ,							__magic_name__ ,							__magic_name__ = None ,							__magic_name__ = False   )					->       List[int]:


       '''simple docstring'''



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

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






   def       lowerCamelCase  (self ,							__magic_name__ ,							__magic_name__ = None   )					->       List[int]:


       '''simple docstring'''



       snake_case_						:   Union[str, Any]           =  [self.sep_token_id]
       snake_case_						:   Optional[int]           =  [2]

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






   def       lowerCamelCase  (self ,							__magic_name__ ,							__magic_name__ = None   )					->       Tuple[str]:


       '''simple docstring'''



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

       if os.path.abspath(self.vocab_file   ) != os.path.abspath(__magic_name__   ) and os.path.isfile(self.vocab_file   ):
           copyfile(self.vocab_file ,							__magic_name__   )
       elif not os.path.isfile(self.vocab_file   ):
           with open(__magic_name__ ,							'''wb'''   ) as fi:
               snake_case_						:   List[Any]           =  self.sp_model.serialized_model_proto()
               fi.write(__magic_name__   )

       return (out_vocab_file,)







import time
from contextlib import contextmanager
from pathlib import Path

import pytest
import requests
from huggingface_hub.hf_api import HfApi, HfFolder


lowerCAmelCase_    =    '''__DUMMY_TRANSFORMERS_USER__'''
lowerCAmelCase_    =    '''Dummy User'''
lowerCAmelCase_    =    '''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'''

lowerCAmelCase_    =    '''https://hub-ci.huggingface.co'''
lowerCAmelCase_    =    CI_HUB_ENDPOINT + '''/datasets/{repo_id}/resolve/{revision}/{path}'''
lowerCAmelCase_    =    CI_HUB_ENDPOINT + '''/{repo_id}/resolve/{revision}/{filename}'''
lowerCAmelCase_    =    Path('''~/.huggingface/hub_ci_token''').expanduser()






@pytest.fixture
def 					lowerCamelCase_						(    _UpperCamelCase			)		->						Optional[int]:






				"""simple docstring"""


				monkeypatch.setattr(
				    '''huggingface_hub.file_download.HUGGINGFACE_CO_URL_TEMPLATE'''			,    _UpperCamelCase			)






@pytest.fixture
def 					lowerCamelCase_						(    _UpperCamelCase			)		->						Optional[int]:






				"""simple docstring"""


				monkeypatch.setattr('''datasets.config.HF_ENDPOINT'''			,    _UpperCamelCase			)
				monkeypatch.setattr('''datasets.config.HUB_DATASETS_URL'''			,    _UpperCamelCase			)






@pytest.fixture
def 					lowerCamelCase_						(    _UpperCamelCase			)		->						Any:






				"""simple docstring"""


				monkeypatch.setattr('''huggingface_hub.hf_api.HfFolder.path_token'''			,    _UpperCamelCase			)






@pytest.fixture
def 					lowerCamelCase_						(    _UpperCamelCase			,    _UpperCamelCase			)		->						Optional[Any]:






				"""simple docstring"""


				HfFolder.save_token(_UpperCamelCase			)
				yield
				HfFolder.delete_token()






@pytest.fixture(scope='''session'''			)
def 					lowerCamelCase_						(    )		->						List[str]:






				"""simple docstring"""


				return HfApi(endpoint=_UpperCamelCase			)






@pytest.fixture(scope='''session'''			)
def 					lowerCamelCase_						(    _UpperCamelCase			)		->						Dict:






				"""simple docstring"""


				snake_case_						:   Union[str, Any]           =  HfFolder.get_token()
				HfFolder.save_token(_UpperCamelCase			)
				yield CI_HUB_USER_TOKEN
				if previous_token is not None:
								HfFolder.save_token(_UpperCamelCase			)






@pytest.fixture
def 					lowerCamelCase_						(    _UpperCamelCase			)		->						List[str]:






				"""simple docstring"""


				def _cleanup_repo(_UpperCamelCase			):
								hf_api.delete_repo(_UpperCamelCase			,    token=_UpperCamelCase			,    repo_type='''dataset'''			)

				return _cleanup_repo






@pytest.fixture
def 					lowerCamelCase_						(    _UpperCamelCase			)		->						List[str]:






				"""simple docstring"""


				@contextmanager
				def _temporary_repo(_UpperCamelCase			):
								try:
												yield repo_id
								finally:
												cleanup_repo(_UpperCamelCase			)

				return _temporary_repo






@pytest.fixture(scope='''session'''			)
def 					lowerCamelCase_						(    _UpperCamelCase			,    _UpperCamelCase			,    _UpperCamelCase			)		->						Dict:






				"""simple docstring"""


				snake_case_						:   Optional[int]           =  f'''repo_txt_data-{int(time.time() * 10E3			)}'''
				snake_case_						:   Any           =  f'''{CI_HUB_USER}/{repo_name}'''
				hf_api.create_repo(_UpperCamelCase			,    token=_UpperCamelCase			,    repo_type='''dataset'''			,    private=_UpperCamelCase			)
				hf_api.upload_file(
				    token=_UpperCamelCase			,    path_or_fileobj=str(_UpperCamelCase			)			,    path_in_repo='''data/text_data.txt'''			,    repo_id=_UpperCamelCase			,    repo_type='''dataset'''			,    )
				yield repo_id
				try:
								hf_api.delete_repo(_UpperCamelCase			,    token=_UpperCamelCase			,    repo_type='''dataset'''			)
				except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
								pass






@pytest.fixture()
def 					lowerCamelCase_						(    _UpperCamelCase			,    _UpperCamelCase			,    _UpperCamelCase			)		->						Dict:






				"""simple docstring"""


				return hf_private_dataset_repo_txt_data_






@pytest.fixture(scope='''session'''			)
def 					lowerCamelCase_						(    _UpperCamelCase			,    _UpperCamelCase			,    _UpperCamelCase			)		->						Optional[Any]:






				"""simple docstring"""


				snake_case_						:   int           =  f'''repo_zipped_txt_data-{int(time.time() * 10E3			)}'''
				snake_case_						:   Tuple           =  f'''{CI_HUB_USER}/{repo_name}'''
				hf_api.create_repo(_UpperCamelCase			,    token=_UpperCamelCase			,    repo_type='''dataset'''			,    private=_UpperCamelCase			)
				hf_api.upload_file(
				    token=_UpperCamelCase			,    path_or_fileobj=str(_UpperCamelCase			)			,    path_in_repo='''data.zip'''			,    repo_id=_UpperCamelCase			,    repo_type='''dataset'''			,    )
				yield repo_id
				try:
								hf_api.delete_repo(_UpperCamelCase			,    token=_UpperCamelCase			,    repo_type='''dataset'''			)
				except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
								pass






@pytest.fixture()
def 					lowerCamelCase_						(    _UpperCamelCase			,    _UpperCamelCase			,    _UpperCamelCase			)		->						Dict:






				"""simple docstring"""


				return hf_private_dataset_repo_zipped_txt_data_






@pytest.fixture(scope='''session'''			)
def 					lowerCamelCase_						(    _UpperCamelCase			,    _UpperCamelCase			,    _UpperCamelCase			)		->						Optional[Any]:






				"""simple docstring"""


				snake_case_						:   List[str]           =  f'''repo_zipped_img_data-{int(time.time() * 10E3			)}'''
				snake_case_						:   str           =  f'''{CI_HUB_USER}/{repo_name}'''
				hf_api.create_repo(_UpperCamelCase			,    token=_UpperCamelCase			,    repo_type='''dataset'''			,    private=_UpperCamelCase			)
				hf_api.upload_file(
				    token=_UpperCamelCase			,    path_or_fileobj=str(_UpperCamelCase			)			,    path_in_repo='''data.zip'''			,    repo_id=_UpperCamelCase			,    repo_type='''dataset'''			,    )
				yield repo_id
				try:
								hf_api.delete_repo(_UpperCamelCase			,    token=_UpperCamelCase			,    repo_type='''dataset'''			)
				except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
								pass






@pytest.fixture()
def 					lowerCamelCase_						(    _UpperCamelCase			,    _UpperCamelCase			,    _UpperCamelCase			)		->						int:






				"""simple docstring"""


				return hf_private_dataset_repo_zipped_img_data_
