
'''simple docstring'''

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


__snake_case:  Tuple								=			logging.get_logger(__name__)

__snake_case:  int								=			{"vocab_file": "spiece.model"}

__snake_case:  Dict								=			{
    "vocab_file": {
        "TsinghuaAI/CPM-Generate": "https://huggingface.co/TsinghuaAI/CPM-Generate/resolve/main/spiece.model",
    }
}


class  _UpperCAmelCase ( lowerCAmelCase__    ):

     """simple docstring"""

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

       '''simple docstring'''

       a_      :	Optional[int]      =							AddedToken(lowerCAmelCase_							,		lstrip=lowerCAmelCase_							,		rstrip=lowerCAmelCase_ ) if isinstance(lowerCAmelCase_							,		lowerCAmelCase_ ) else mask_token

       a_      :	List[str]      =							{} if sp_model_kwargs is None else sp_model_kwargs

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

       a_      :	str      =							3

       a_      :	Tuple      =							do_lower_case
       a_      :	Dict      =							remove_space
       a_      :	int      =							keep_accents
       a_      :	Union[str, Any]      =							vocab_file

       a_      :	Dict      =							spm.SentencePieceProcessor(**self.sp_model_kwargs )
       self.sp_model.Load(lowerCAmelCase_ )

       try:
         import jieba
       except ModuleNotFoundError as error:
         raise error.__class__(
             """You need to install jieba to use CpmTokenizer or CpmTokenizerFast. """
             """See https://pypi.org/project/jieba/ for installation.""" )
       a_      :	Optional[int]      =							jieba
       a_      :	List[str]      =							str.maketrans(""" \n"""							,		"""\u2582\u2583""" )


     @property
     # Copied from transformers.models.xlnet.tokenization_xlnet.XLNetTokenizer.vocab_size
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return len(self.sp_model )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Any      =							{self.convert_ids_to_tokens(lowerCAmelCase_ ): i for i in range(self.vocab_size )}
       vocab.update(self.added_tokens_encoder )
       return vocab


     def __getstate__(      self ):

       '''simple docstring'''

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


     def __setstate__(      self							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	Any      =							d

       # for backward compatibility
       if not hasattr(self							,		"""sp_model_kwargs""" ):
         a_      :	Union[str, Any]      =							{}

       a_      :	List[str]      =							spm.SentencePieceProcessor(**self.sp_model_kwargs )
       self.sp_model.Load(self.vocab_file )


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_ ):

       '''simple docstring'''

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

       if not self.keep_accents:
         a_      :	int      =							unicodedata.normalize("""NFKD"""							,		lowerCAmelCase_ )
         a_      :	Any      =							"""""".join([c for c in outputs if not unicodedata.combining(lowerCAmelCase_ )] )
       if self.do_lower_case:
         a_      :	Any      =							outputs.lower()

       return outputs


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_ ):

       '''simple docstring'''

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

       return new_pieces


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_ ):

       '''simple docstring'''

       return self.sp_model.PieceToId(lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_ ):

       '''simple docstring'''

       return self.sp_model.IdToPiece(lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	Dict      =							"""""".join(lowerCAmelCase_ ).replace(lowerCAmelCase_							,		""" """ ).strip()
       return out_string


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = None ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							[self.sep_token_id]
       a_      :	Dict      =							[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							,		lowerCAmelCase_							,		lowerCAmelCase_ = None							,		lowerCAmelCase_ = False ):

       '''simple docstring'''

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

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


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = None ):

       '''simple docstring'''

       a_      :	Any      =							[self.sep_token_id]
       a_      :	List[str]      =							[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							,		lowerCAmelCase_							,		lowerCAmelCase_ = None ):

       '''simple docstring'''

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

       if os.path.abspath(self.vocab_file ) != os.path.abspath(lowerCAmelCase_ ) and os.path.isfile(self.vocab_file ):
         copyfile(self.vocab_file							,		lowerCAmelCase_ )
       elif not os.path.isfile(self.vocab_file ):
         with open(lowerCAmelCase_							,		"""wb""" ) as fi:
           a_      :	Union[str, Any]      =							self.sp_model.serialized_model_proto()
           fi.write(lowerCAmelCase_ )

       return (out_vocab_file,)






     def 							_lowerCAmelCase  (      self							,		*lowerCAmelCase_							,		**lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	Any      =							super()._decode(*lowerCAmelCase_							,		**lowerCAmelCase_ )
       a_      :	Dict      =							text.replace(""" """							,		"""""" ).replace("""\u2582"""							,		""" """ ).replace("""\u2583"""							,		"""\n""" )
       return text




'''simple docstring'''

from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


__snake_case:  Optional[Any]								=			logging.get_logger(__name__)


__snake_case:  Tuple								=			TypeVar("DatasetType", Dataset, IterableDataset)






def 	_snake_case						(      A_       :    List[DatasetType]     ,		A_       :    Optional[List[float]] = None     ,		A_       :    Optional[int] = None     ,		A_       :    Optional[DatasetInfo] = None     ,		A_       :    Optional[NamedSplit] = None     ,		A_       :    Literal["first_exhausted", "all_exhausted"] = "first_exhausted"     ,		):
  """simple docstring"""







  from .arrow_dataset import Dataset
  from .iterable_dataset import IterableDataset

  if not datasets:
    raise ValueError("""Unable to interleave an empty list of datasets."""   )
  for i, dataset in enumerate(A_   ):
    if not isinstance(A_     ,		(Dataset, IterableDataset)   ):
      if isinstance(A_     ,		(DatasetDict, IterableDatasetDict)   ):
        if not dataset:
          raise ValueError(
              f'''Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} '''
              """is an empty dataset dictionary."""   )
        raise ValueError(
            f'''Dataset at position {i} has at least one split: {list(A_   )}\n'''
            f'''Please pick one to interleave with the other datasets, for example: dataset[\'{next(iter(A_   )   )}\']'''   )
      raise ValueError(
          f'''Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(A_   ).__name__}.'''   )
    if i == 0:
      a_      , a_      :	Union[str, Any]      =							(
          (Dataset, IterableDataset) if isinstance(A_     ,		A_   ) else (IterableDataset, Dataset)
      )
    elif not isinstance(A_     ,		A_   ):
      raise ValueError(
          f'''Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.'''   )
  if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
    raise ValueError(f'''{stopping_strategy} is not supported. Please enter a valid stopping_strategy.'''   )
  if dataset_type is Dataset:
    return _interleave_map_style_datasets(
        A_     ,		A_     ,		A_     ,		info=A_     ,		split=A_     ,		stopping_strategy=A_   )
  else:
    return _interleave_iterable_datasets(
        A_     ,		A_     ,		A_     ,		info=A_     ,		split=A_     ,		stopping_strategy=A_   )






def 	_snake_case						(      A_       :    List[DatasetType]     ,		A_       :    Optional[DatasetInfo] = None     ,		A_       :    Optional[NamedSplit] = None     ,		A_       :    int = 0     ,		):
  """simple docstring"""







  if not dsets:
    raise ValueError("""Unable to concatenate an empty list of datasets."""   )
  for i, dataset in enumerate(A_   ):
    if not isinstance(A_     ,		(Dataset, IterableDataset)   ):
      if isinstance(A_     ,		(DatasetDict, IterableDatasetDict)   ):
        if not dataset:
          raise ValueError(
              f'''Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} '''
              """is an empty dataset dictionary."""   )
        raise ValueError(
            f'''Dataset at position {i} has at least one split: {list(A_   )}\n'''
            f'''Please pick one to interleave with the other datasets, for example: dataset[\'{next(iter(A_   )   )}\']'''   )
      raise ValueError(
          f'''Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(A_   ).__name__}.'''   )
    if i == 0:
      a_      , a_      :	str      =							(
          (Dataset, IterableDataset) if isinstance(A_     ,		A_   ) else (IterableDataset, Dataset)
      )
    elif not isinstance(A_     ,		A_   ):
      raise ValueError(
          f'''Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.'''   )
  if dataset_type is Dataset:
    return _concatenate_map_style_datasets(A_     ,		info=A_     ,		split=A_     ,		axis=A_   )
  else:
    return _concatenate_iterable_datasets(A_     ,		info=A_     ,		split=A_     ,		axis=A_   )


