





import io
import os
import unicodedata
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


_lowerCamelCase   				=			logging.get_logger(__name__)

_lowerCamelCase   				=			'▁'

_lowerCamelCase   				=			{'vocab_file': 'vocab.txt', 'sentencepiece_model_ckpt': 'sentencepiece.bpe.model'}

_lowerCamelCase   				=			{
    'sentencepiece_model_file': 'sentencepiece.bpe.model',
    'vocab_file': 'vocab.txt',
}

_lowerCamelCase   				=			{
    'vocab_file': {
        'ernie-m-base': 'https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/vocab.txt',
        'ernie-m-large': 'https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/vocab.txt',
    },
    'sentencepiece_model_file': {
        'ernie-m-base': 'https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/sentencepiece.bpe.model',
        'ernie-m-large': 'https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/sentencepiece.bpe.model',
    },
}

_lowerCamelCase   				=			{
    'ernie-m-base': 514,
    'ernie-m-large': 514,
}

_lowerCamelCase   				=			{
    'ernie-m-base': {'do_lower_case': False},
    'ernie-m-large': {'do_lower_case': False},
}




class   UpperCamelCase_       (     UpperCamelCase__    ):
							lowerCamelCase_           =					["input_ids"]

							lowerCamelCase_           =					VOCAB_FILES_NAMES
							lowerCamelCase_           =					PRETRAINED_INIT_CONFIGURATION
							lowerCamelCase_           =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							lowerCamelCase_           =					PRETRAINED_VOCAB_FILES_MAP
							lowerCamelCase_           =					RESOURCE_FILES_NAMES


							def __init__(							self    :List[str]							,				__A    :Optional[int]							,				__A    :List[Any]=None							,				__A    :Any=False							,				__A    :Tuple="utf8"							,				__A    :Tuple="[UNK]"							,				__A    :Dict="[SEP]"							,				__A    :Optional[int]="[PAD]"							,				__A    :List[str]="[CLS]"							,				__A    :Any="[MASK]"							,				__A    :List[Any] = None							,				**__A    :Union[str, Any]							,				)   ->   Tuple:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      {} if sp_model_kwargs is None else sp_model_kwargs
									super().__init__(
									    do_lower_case=__lowerCAmelCase							,				unk_token=__lowerCAmelCase							,				sep_token=__lowerCAmelCase							,				pad_token=__lowerCAmelCase							,				cls_token=__lowerCAmelCase							,				mask_token=__lowerCAmelCase							,				vocab_file=__lowerCAmelCase							,				encoding=__lowerCAmelCase							,				sp_model_kwargs=self.sp_model_kwargs							,				**__lowerCAmelCase							,				)
									SCREAMING_SNAKE_CASE__			=      do_lower_case
									SCREAMING_SNAKE_CASE__			=      sentencepiece_model_ckpt
									SCREAMING_SNAKE_CASE__			=      spm.SentencePieceProcessor(**self.sp_model_kwargs		)
									self.sp_model.Load(__lowerCAmelCase		)

									# to mimic paddlenlp.transformers.ernie_m.tokenizer.ErnieMTokenizer functioning
									if vocab_file is not None:
											SCREAMING_SNAKE_CASE__			=      self.load_vocab(filepath=__lowerCAmelCase		)
									else:
											SCREAMING_SNAKE_CASE__			=      {self.sp_model.id_to_piece(__lowerCAmelCase		): id for id in range(self.sp_model.get_piece_size()		)}
									SCREAMING_SNAKE_CASE__			=      {v: k for k, v in self.vocab.items()}


							def      _snake_case  (							self    :Optional[Any]							,				__A    :Any		)   ->   Optional[Any]:



									"""simple docstring"""



									if text is None:
											return None

									SCREAMING_SNAKE_CASE__			=      self.tokenize(__lowerCAmelCase		)
									SCREAMING_SNAKE_CASE__	,						SCREAMING_SNAKE_CASE__			=      """""", []

									for i, ch in enumerate(__lowerCAmelCase		):
											if ch in self.SP_CHAR_MAPPING:
													SCREAMING_SNAKE_CASE__			=      self.SP_CHAR_MAPPING.get(__lowerCAmelCase		)
											else:
													SCREAMING_SNAKE_CASE__			=      unicodedata.normalize("""NFKC"""							,				__lowerCAmelCase		)
											if self.is_whitespace(__lowerCAmelCase		):
													continue
											normalized_text += ch
											char_mapping.extend([i] * len(__lowerCAmelCase		)		)

									SCREAMING_SNAKE_CASE__	,						SCREAMING_SNAKE_CASE__	,						SCREAMING_SNAKE_CASE__			=      normalized_text, [], 0

									if self.do_lower_case:
											SCREAMING_SNAKE_CASE__			=      text.lower()

									for token in split_tokens:
											if token[:1] == "▁":
													SCREAMING_SNAKE_CASE__			=      token[1:]
											SCREAMING_SNAKE_CASE__			=      text[offset:].index(__lowerCAmelCase		) + offset
											SCREAMING_SNAKE_CASE__			=      start + len(__lowerCAmelCase		)

											token_mapping.append((char_mapping[start], char_mapping[end - 1] + 1)		)
											SCREAMING_SNAKE_CASE__			=      end
									return token_mapping


							@property
							def      _snake_case  (							self    :int		)   ->   str:



									"""simple docstring"""



									return len(self.vocab		)


							def      _snake_case  (							self    :List[Any]		)   ->   Tuple:



									"""simple docstring"""



									return dict(self.vocab							,				**self.added_tokens_encoder		)


							def __getstate__(							self    :Optional[Any]		)   ->   int:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      self.__dict__.copy()
									SCREAMING_SNAKE_CASE__			=      None
									return state


							def __setstate__(							self    :List[Any]							,				__A    :List[str]		)   ->   Union[str, Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      d

									# for backward compatibility
									if not hasattr(self							,				"""sp_model_kwargs"""		):
											SCREAMING_SNAKE_CASE__			=      {}

									SCREAMING_SNAKE_CASE__			=      spm.SentencePieceProcessor(**self.sp_model_kwargs		)
									self.sp_model.Load(self.sentencepiece_model_ckpt		)


							def      _snake_case  (							self    :Union[str, Any]							,				__A    :Tuple		)   ->   Optional[int]:



									"""simple docstring"""



									return "".join((self.SP_CHAR_MAPPING.get(__lowerCAmelCase							,				__lowerCAmelCase		) for c in text)		)


							def      _snake_case  (							self    :Any							,				__A    :Union[str, Any]							,				__A    :Optional[int]=False							,				__A    :Dict=64							,				__A    :Any=0.1		)   ->   Optional[Any]:



									"""simple docstring"""



									if self.sp_model_kwargs.get("""enable_sampling"""		) is True:
											SCREAMING_SNAKE_CASE__			=      True
									if self.sp_model_kwargs.get("""alpha"""		) is not None:
											SCREAMING_SNAKE_CASE__			=      self.sp_model_kwargs.get("""alpha"""		)
									if self.sp_model_kwargs.get("""nbest_size"""		) is not None:
											SCREAMING_SNAKE_CASE__			=      self.sp_model_kwargs.get("""nbest_size"""		)

									if not enable_sampling:
											SCREAMING_SNAKE_CASE__			=      self.sp_model.EncodeAsPieces(__lowerCAmelCase		)
									else:
											SCREAMING_SNAKE_CASE__			=      self.sp_model.SampleEncodeAsPieces(__lowerCAmelCase							,				__lowerCAmelCase							,				__lowerCAmelCase		)
									SCREAMING_SNAKE_CASE__			=      []
									for pi, piece in enumerate(__lowerCAmelCase		):
											if piece == SPIECE_UNDERLINE:
													if not pieces[pi + 1].startswith(__lowerCAmelCase		) and pi != 0:
															new_pieces.append(__lowerCAmelCase		)
															continue
													else:
															continue
											SCREAMING_SNAKE_CASE__			=      0
											for i, chunk in enumerate(__lowerCAmelCase		):
													if chunk == SPIECE_UNDERLINE:
															continue
													if self.is_ch_char(__lowerCAmelCase		) or self.is_punct(__lowerCAmelCase		):
															if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
																	new_pieces.append(piece[lst_i:i]		)
															new_pieces.append(__lowerCAmelCase		)
															SCREAMING_SNAKE_CASE__			=      i + 1
													elif chunk.isdigit() and i > 0 and not piece[i - 1].isdigit():
															if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
																	new_pieces.append(piece[lst_i:i]		)
															SCREAMING_SNAKE_CASE__			=      i
													elif not chunk.isdigit() and i > 0 and piece[i - 1].isdigit():
															if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
																	new_pieces.append(piece[lst_i:i]		)
															SCREAMING_SNAKE_CASE__			=      i
											if len(__lowerCAmelCase		) > lst_i:
													new_pieces.append(piece[lst_i:]		)
									return new_pieces


							def      _snake_case  (							self    :Any							,				__A    :Dict		)   ->   Union[str, Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      """""".join(__lowerCAmelCase		).replace(__lowerCAmelCase							,				""" """		).strip()
									return out_string


							def      _snake_case  (							self    :Any							,				__A    :Any		)   ->   Any:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      self.convert_ids_to_tokens(__lowerCAmelCase		)
									SCREAMING_SNAKE_CASE__			=      """""".join(__lowerCAmelCase		).replace(__lowerCAmelCase							,				""" """		).strip()
									return out_string


							def      _snake_case  (							self    :int							,				__A    :List[Any]		)   ->   int:



									"""simple docstring"""



									return self.vocab.get(__lowerCAmelCase							,				self.vocab.get(self.unk_token		)		)


							def      _snake_case  (							self    :int							,				__A    :Dict		)   ->   Dict:



									"""simple docstring"""



									return self.reverse_vocab.get(__lowerCAmelCase							,				self.unk_token		)


							def      _snake_case  (							self    :Tuple							,				__A    :List[Any]							,				__A    :Dict=None		)   ->   List[str]:



									"""simple docstring"""



									if token_ids_a is None:
											return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
									SCREAMING_SNAKE_CASE__			=      [self.cls_token_id]
									SCREAMING_SNAKE_CASE__			=      [self.sep_token_id]
									return _cls + token_ids_a + _sep + _sep + token_ids_a + _sep


							def      _snake_case  (							self    :str							,				__A    :List[Any]							,				__A    :List[str]=None		)   ->   int:



									"""simple docstring"""



									if offset_mapping_a is None:
											return [(0, 0)] + offset_mapping_a + [(0, 0)]

									return [(0, 0)] + offset_mapping_a + [(0, 0), (0, 0)] + offset_mapping_a + [(0, 0)]


							def      _snake_case  (							self    :Optional[Any]							,				__A    :Optional[Any]							,				__A    :List[str]=None							,				__A    :List[Any]=False		)   ->   Optional[Any]:



									"""simple docstring"""



									if already_has_special_tokens:
											if token_ids_a is not None:
													raise ValueError(
													    """You should not supply a second sequence if the provided sequence of """
													    """ids is already formatted with special tokens for the model."""		)
											return [1 if x in [self.sep_token_id, self.cls_token_id] else 0 for x in token_ids_a]

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


							def      _snake_case  (							self    :Optional[int]							,				__A    :List[str]							,				__A    :Union[str, Any] = None		)   ->   List[Any]:



									"""simple docstring"""



									if token_ids_a is None:
											# [CLS] X [SEP]
											return (len(__lowerCAmelCase		) + 2) * [0]

									# [CLS] A [SEP] [SEP] B [SEP]
									return [0] * (len(__lowerCAmelCase		) + 1) + [1] * (len(__lowerCAmelCase		) + 3)


							def      _snake_case  (							self    :Any							,				__A    :List[Any]		)   ->   Union[str, Any]:



									"""simple docstring"""



									if "\u4e00" <= char <= "\u9fff":
											return True
									return False


							def      _snake_case  (							self    :Tuple							,				__A    :Any		)   ->   Optional[int]:



									"""simple docstring"""



									if ("a" <= char <= "z") or ("A" <= char <= "Z"):
											return True
									return False


							def      _snake_case  (							self    :List[str]							,				__A    :Dict		)   ->   Tuple:



									"""simple docstring"""



									if char in ",;:.?!~，；：。？！《》【】":
											return True
									return False


							def      _snake_case  (							self    :Optional[Any]							,				__A    :Optional[int]		)   ->   List[Any]:



									"""simple docstring"""



									if char == " " or char == "\t" or char == "\n" or char == "\r":
											return True
									if len(__lowerCAmelCase		) == 1:
											SCREAMING_SNAKE_CASE__			=      unicodedata.category(__lowerCAmelCase		)
											if cat == "Zs":
													return True
									return False


							def      _snake_case  (							self    :Union[str, Any]							,				__A    :Any		)   ->   str:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      {}
									with io.open(__lowerCAmelCase							,				"""r"""							,				encoding="""utf-8"""		) as f:
											for index, line in enumerate(__lowerCAmelCase		):
													SCREAMING_SNAKE_CASE__			=      line.rstrip("""\n"""		)
													SCREAMING_SNAKE_CASE__			=      int(__lowerCAmelCase		)

									return token_to_idx


							def      _snake_case  (							self    :Optional[Any]							,				__A    :Optional[Any]							,				__A    :int = None		)   ->   Optional[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      0
									if os.path.isdir(__lowerCAmelCase		):
											SCREAMING_SNAKE_CASE__			=      os.path.join(
											    __lowerCAmelCase							,				(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]		)
									else:
											SCREAMING_SNAKE_CASE__			=      (filename_prefix + """-""" if filename_prefix else """""") + save_directory
									with open(__lowerCAmelCase							,				"""w"""							,				encoding="""utf-8"""		) as writer:
											for token, token_index in sorted(self.vocab.items()							,				key=lambda __A		: kv[1]		):
													if index != token_index:
															logger.warning(
															    f'''Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive.'''
															    """ Please check that the vocabulary is not corrupted!"""		)
															SCREAMING_SNAKE_CASE__			=      token_index
													writer.write(token + """\n"""		)
													index += 1

									SCREAMING_SNAKE_CASE__			=      os.path.join(__lowerCAmelCase							,				"""sentencepiece.bpe.model"""		)
									with open(__lowerCAmelCase							,				"""wb"""		) as fi:
											SCREAMING_SNAKE_CASE__			=      self.sp_model.serialized_model_proto()
											fi.write(__lowerCAmelCase		)

									return (vocab_file,)






import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


_lowerCamelCase   				=			'\\n    Text data.\n    Second line of data.'

_lowerCamelCase   				=			'file'






@pytest.fixture(scope="""session"""    )
def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							Any    ):
		SCREAMING_SNAKE_CASE__			=      tmp_path_factory.mktemp("""data"""    ) / (FILE_PATH + """.zstd""")
		SCREAMING_SNAKE_CASE__			=      bytes(UpperCamelCase__					,   """utf-8"""    )
		with zstd.open(UpperCamelCase__					,   """wb"""    ) as f:
				f.write(UpperCamelCase__    )
		return path






@pytest.fixture
def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							List[str]    ):
		with open(os.path.join(tmpfs.local_root_dir					,   UpperCamelCase__    )					,   """w"""    ) as f:
				f.write(UpperCamelCase__    )
		return FILE_PATH






@pytest.mark.parametrize("""compression_format"""					,   ["""gzip""", """xz""", """zstd"""]    )
def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							int					,   UpperCamelCase__:							Dict					,   UpperCamelCase__:							int					,   UpperCamelCase__:							str					,   UpperCamelCase__:							Optional[int]					,   UpperCamelCase__:							Optional[Any]    ):
		SCREAMING_SNAKE_CASE__			=      {"""gzip""": gz_file, """xz""": xz_file, """zstd""": zstd_path}
		SCREAMING_SNAKE_CASE__			=      input_paths[compression_format]
		SCREAMING_SNAKE_CASE__			=      tmp_path / """cache"""
		SCREAMING_SNAKE_CASE__			=      DownloadConfig(cache_dir=UpperCamelCase__					,   extract_compressed_file=UpperCamelCase__    )
		SCREAMING_SNAKE_CASE__			=      cached_path(UpperCamelCase__					,   download_config=UpperCamelCase__    )
		with open(UpperCamelCase__    ) as f:
				SCREAMING_SNAKE_CASE__			=      f.read()
		with open(UpperCamelCase__    ) as f:
				SCREAMING_SNAKE_CASE__			=      f.read()
		assert extracted_file_content == expected_file_content






@pytest.mark.parametrize("""default_extracted"""					,   [True, False]    )
@pytest.mark.parametrize("""default_cache_dir"""					,   [True, False]    )
def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							Tuple					,   UpperCamelCase__:							List[str]					,   UpperCamelCase__:							Optional[int]					,   UpperCamelCase__:							Any					,   UpperCamelCase__:							Union[str, Any]    ):
		SCREAMING_SNAKE_CASE__			=      """custom_cache"""
		SCREAMING_SNAKE_CASE__			=      """custom_extracted_dir"""
		SCREAMING_SNAKE_CASE__			=      tmp_path / """custom_extracted_path"""
		if default_extracted:
				SCREAMING_SNAKE_CASE__			=      ("""downloads""" if default_cache_dir else custom_cache_dir, """extracted""")
		else:
				monkeypatch.setattr("""datasets.config.EXTRACTED_DATASETS_DIR"""					,   UpperCamelCase__    )
				monkeypatch.setattr("""datasets.config.EXTRACTED_DATASETS_PATH"""					,   str(UpperCamelCase__    )    )
				SCREAMING_SNAKE_CASE__			=      custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

		SCREAMING_SNAKE_CASE__			=      xz_file
		SCREAMING_SNAKE_CASE__			=      (
		    DownloadConfig(extract_compressed_file=UpperCamelCase__    )
		    if default_cache_dir
		    else DownloadConfig(cache_dir=tmp_path / custom_cache_dir					,   extract_compressed_file=UpperCamelCase__    )
		)
		SCREAMING_SNAKE_CASE__			=      cached_path(UpperCamelCase__					,   download_config=UpperCamelCase__    )
		assert Path(UpperCamelCase__    ).parent.parts[-2:] == expected






def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							Optional[int]    ):
		# absolute path
		SCREAMING_SNAKE_CASE__			=      str(Path(UpperCamelCase__    ).resolve()    )
		assert cached_path(UpperCamelCase__    ) == text_file
		# relative path
		SCREAMING_SNAKE_CASE__			=      str(Path(UpperCamelCase__    ).resolve().relative_to(Path(os.getcwd()    )    )    )
		assert cached_path(UpperCamelCase__    ) == text_file






def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							List[str]    ):
		# absolute path
		SCREAMING_SNAKE_CASE__			=      str(tmp_path.resolve() / """__missing_file__.txt"""    )
		with pytest.raises(UpperCamelCase__    ):
				cached_path(UpperCamelCase__    )
		# relative path
		SCREAMING_SNAKE_CASE__			=      """./__missing_file__.txt"""
		with pytest.raises(UpperCamelCase__    ):
				cached_path(UpperCamelCase__    )






def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							List[str]    ):
		SCREAMING_SNAKE_CASE__			=      get_from_cache(f'''tmp://{tmpfs_file}'''    )
		with open(UpperCamelCase__    ) as f:
				SCREAMING_SNAKE_CASE__			=      f.read()
		assert output_file_content == FILE_CONTENT






@patch("""datasets.config.HF_DATASETS_OFFLINE"""					,   UpperCamelCase__    )
def 				SCREAMING_SNAKE_CASE__							(       ):
		with pytest.raises(UpperCamelCase__    ):
				cached_path("""https://huggingface.co"""    )






@patch("""datasets.config.HF_DATASETS_OFFLINE"""					,   UpperCamelCase__    )
def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							Optional[Any]    ):
		SCREAMING_SNAKE_CASE__			=      tmp_path_factory.mktemp("""data"""    ) / """file.html"""
		with pytest.raises(UpperCamelCase__    ):
				http_get("""https://huggingface.co"""					,   temp_file=UpperCamelCase__    )
		with pytest.raises(UpperCamelCase__    ):
				http_head("""https://huggingface.co"""    )






@patch("""datasets.config.HF_DATASETS_OFFLINE"""					,   UpperCamelCase__    )
def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							List[Any]    ):
		SCREAMING_SNAKE_CASE__			=      tmp_path_factory.mktemp("""data"""    ) / """file.html"""
		with pytest.raises(UpperCamelCase__    ):
				ftp_get("""ftp://huggingface.co"""					,   temp_file=UpperCamelCase__    )
		with pytest.raises(UpperCamelCase__    ):
				ftp_head("""ftp://huggingface.co"""    )






@patch("""datasets.config.HF_DATASETS_OFFLINE"""					,   UpperCamelCase__    )
def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							int    ):
		SCREAMING_SNAKE_CASE__			=      tmp_path_factory.mktemp("""data"""    ) / """file.html"""
		with pytest.raises(UpperCamelCase__    ):
				fsspec_get("""s3://huggingface.co"""					,   temp_file=UpperCamelCase__    )
		with pytest.raises(UpperCamelCase__    ):
				fsspec_head("""s3://huggingface.co"""    )