





import os
import re
from shutil import copyfile
from typing import List, Optional, Tuple

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


_lowerCamelCase   				=			logging.get_logger(__name__)

_lowerCamelCase   				=			{
    'vocab_file': 'vocab.txt',
    'merges_file': 'bpe.codes',
}

_lowerCamelCase   				=			{
    'vocab_file': {
        'vinai/phobert-base': 'https://huggingface.co/vinai/phobert-base/resolve/main/vocab.txt',
        'vinai/phobert-large': 'https://huggingface.co/vinai/phobert-large/resolve/main/vocab.txt',
    },
    'merges_file': {
        'vinai/phobert-base': 'https://huggingface.co/vinai/phobert-base/resolve/main/bpe.codes',
        'vinai/phobert-large': 'https://huggingface.co/vinai/phobert-large/resolve/main/bpe.codes',
    },
}

_lowerCamelCase   				=			{
    'vinai/phobert-base': 256,
    'vinai/phobert-large': 256,
}






def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							Union[str, Any]    ):
		SCREAMING_SNAKE_CASE__			=      set()
		SCREAMING_SNAKE_CASE__			=      word[0]
		for char in word[1:]:
				pairs.add((prev_char, char)    )
				SCREAMING_SNAKE_CASE__			=      char

		SCREAMING_SNAKE_CASE__			=      set(_snake_case    )
		return pairs




class   UpperCamelCase_       (     UpperCamelCase__    ):
							lowerCamelCase_           =					VOCAB_FILES_NAMES
							lowerCamelCase_           =					PRETRAINED_VOCAB_FILES_MAP
							lowerCamelCase_           =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES


							def __init__(							self    :int							,				__A    :int							,				__A    :Optional[Any]							,				__A    :List[Any]="<s>"							,				__A    :List[str]="</s>"							,				__A    :str="</s>"							,				__A    :List[str]="<s>"							,				__A    :Union[str, Any]="<unk>"							,				__A    :Any="<pad>"							,				__A    :Dict="<mask>"							,				**__A    :Union[str, Any]							,				)   ->   str:



									"""simple docstring"""



									super().__init__(
									    bos_token=__A							,				eos_token=__A							,				unk_token=__A							,				sep_token=__A							,				cls_token=__A							,				pad_token=__A							,				mask_token=__A							,				**__A							,				)

									SCREAMING_SNAKE_CASE__			=      vocab_file
									SCREAMING_SNAKE_CASE__			=      merges_file

									SCREAMING_SNAKE_CASE__			=      {}
									SCREAMING_SNAKE_CASE__			=      0
									SCREAMING_SNAKE_CASE__			=      1
									SCREAMING_SNAKE_CASE__			=      2
									SCREAMING_SNAKE_CASE__			=      3

									self.add_from_file(__A		)

									SCREAMING_SNAKE_CASE__			=      {v: k for k, v in self.encoder.items()}

									with open(__A							,				encoding="""utf-8"""		) as merges_handle:
											SCREAMING_SNAKE_CASE__			=      merges_handle.read().split("""\n"""		)[:-1]
									SCREAMING_SNAKE_CASE__			=      [tuple(merge.split()[:-1]		) for merge in merges]
									SCREAMING_SNAKE_CASE__			=      dict(zip(__A							,				range(len(__A		)		)		)		)
									SCREAMING_SNAKE_CASE__			=      {}


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



									"""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    :List[Any]							,				__A    :Optional[int]							,				__A    :str = None							,				__A    :List[str] = False		)   ->   int:



									"""simple docstring"""



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

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


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



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      [self.sep_token_id]
									SCREAMING_SNAKE_CASE__			=      [self.cls_token_id]

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


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



									"""simple docstring"""



									return len(self.encoder		)


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



									"""simple docstring"""



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


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



									"""simple docstring"""



									if token in self.cache:
											return self.cache[token]
									SCREAMING_SNAKE_CASE__			=      tuple(__A		)
									SCREAMING_SNAKE_CASE__			=      tuple(list(word[:-1]		) + [word[-1] + """</w>"""]		)
									SCREAMING_SNAKE_CASE__			=      get_pairs(__A		)

									if not pairs:
											return token

									while True:
											SCREAMING_SNAKE_CASE__			=      min(__A							,				key=lambda __A		: self.bpe_ranks.get(__A							,				float("""inf"""		)		)		)
											if bigram not in self.bpe_ranks:
													break
											SCREAMING_SNAKE_CASE__	,						SCREAMING_SNAKE_CASE__			=      bigram
											SCREAMING_SNAKE_CASE__			=      []
											SCREAMING_SNAKE_CASE__			=      0
											while i < len(__A		):
													try:
															SCREAMING_SNAKE_CASE__			=      word.index(__A							,				__A		)
													except ValueError:
															new_word.extend(word[i:]		)
															break
													else:
															new_word.extend(word[i:j]		)
															SCREAMING_SNAKE_CASE__			=      j

													if word[i] == first and i < len(__A		) - 1 and word[i + 1] == second:
															new_word.append(first + second		)
															i += 2
													else:
															new_word.append(word[i]		)
															i += 1
											SCREAMING_SNAKE_CASE__			=      tuple(__A		)
											SCREAMING_SNAKE_CASE__			=      new_word
											if len(__A		) == 1:
													break
											else:
													SCREAMING_SNAKE_CASE__			=      get_pairs(__A		)
									SCREAMING_SNAKE_CASE__			=      """@@ """.join(__A		)
									SCREAMING_SNAKE_CASE__			=      word[:-4]
									SCREAMING_SNAKE_CASE__			=      word
									return word


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



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      []

									SCREAMING_SNAKE_CASE__			=      re.findall(r"""\S+\n?"""							,				__A		)

									for token in words:
											split_tokens.extend(list(self.bpe(__A		).split(""" """		)		)		)
									return split_tokens


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



									"""simple docstring"""



									return self.encoder.get(__A							,				self.encoder.get(self.unk_token		)		)


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



									"""simple docstring"""



									return self.decoder.get(__A							,				self.unk_token		)


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



									"""simple docstring"""



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


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



									"""simple docstring"""



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

									if os.path.abspath(self.vocab_file		) != os.path.abspath(__A		):
											copyfile(self.vocab_file							,				__A		)

									if os.path.abspath(self.merges_file		) != os.path.abspath(__A		):
											copyfile(self.merges_file							,				__A		)

									return out_vocab_file, out_merge_file


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



									"""simple docstring"""



									if isinstance(__A							,				__A		):
											try:
													with open(__A							,				"""r"""							,				encoding="""utf-8"""		) as fd:
															self.add_from_file(__A		)
											except FileNotFoundError as fnfe:
													raise fnfe
											except UnicodeError:
													raise Exception(f'''Incorrect encoding detected in {f}, please rebuild the dataset'''		)
											return

									SCREAMING_SNAKE_CASE__			=      f.readlines()
									for lineTmp in lines:
											SCREAMING_SNAKE_CASE__			=      lineTmp.strip()
											SCREAMING_SNAKE_CASE__			=      line.rfind(""" """		)
											if idx == -1:
													raise ValueError("""Incorrect dictionary format, expected '<token> <cnt>'"""		)
											SCREAMING_SNAKE_CASE__			=      line[:idx]
											SCREAMING_SNAKE_CASE__			=      len(self.encoder		)






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"""    )