




import json
import os
from typing import Dict, List, Optional, Tuple

import regex as re

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


UpperCAmelCase_						=logging.get_logger(__name__)


UpperCAmelCase_						={
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
    """tokenizer_config_file""": """tokenizer_config.json""",
}

UpperCAmelCase_						={
    """vocab_file""": {
        """facebook/blenderbot_small-90M""": """https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/vocab.json"""
    },
    """merges_file""": {
        """facebook/blenderbot_small-90M""": """https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/merges.txt"""
    },
    """tokenizer_config_file""": {
        """facebook/blenderbot_small-90M""": (
            """https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/tokenizer_config.json"""
        )
    },
}

UpperCAmelCase_						={"""facebook/blenderbot_small-90M""": 512}



def       UpperCAmelCase       (     _snake_case							):
							lowerCAmelCase   =		set()
							lowerCAmelCase   =		word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)							)
														lowerCAmelCase   =		char

							lowerCAmelCase   =		set(_snake_case							)
							return pairs


class 							__UpperCamelCase			( __UpperCAmelCase  ):





				'''simple docstring'''



				__a							: int										=VOCAB_FILES_NAMES
				__a							: Optional[Any]										=PRETRAINED_VOCAB_FILES_MAP
				__a							: Optional[int]										=PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				__a							: Union[str, Any]										=["""input_ids""", """attention_mask"""]



				def __init__( self       , UpperCAmelCase_       , UpperCAmelCase_       , UpperCAmelCase_="__start__"       , UpperCAmelCase_="__end__"       , UpperCAmelCase_="__unk__"       , UpperCAmelCase_="__null__"       , **UpperCAmelCase_       , ):
											super().__init__(unk_token=UpperCAmelCase_       , bos_token=UpperCAmelCase_       , eos_token=UpperCAmelCase_       , pad_token=UpperCAmelCase_       , **UpperCAmelCase_						)

											with open(UpperCAmelCase_       , encoding='''utf-8'''						) as vocab_handle:
																		lowerCAmelCase   =		json.load(UpperCAmelCase_						)
											lowerCAmelCase   =		{v: k for k, v in self.encoder.items()}
											with open(UpperCAmelCase_       , encoding='''utf-8'''						) as merges_handle:
																		lowerCAmelCase   =		merges_handle.read().split('''\n'''						)[1:-1]
											lowerCAmelCase   =		[tuple(merge.split()						) for merge in merges]
											lowerCAmelCase   =		dict(zip(UpperCAmelCase_       , range(len(UpperCAmelCase_						)						)						)						)
											lowerCAmelCase   =		{}



				@property
				def       __snake_case		( self						):
											return len(self.encoder						)



				def       __snake_case		( self						):
											return dict(self.encoder       , **self.added_tokens_encoder						)



				def       __snake_case		( self       , UpperCAmelCase_						):
											if token in self.cache:
																		return self.cache[token]
											lowerCAmelCase   =		re.sub('''([.,!?()])'''       , r''' \1'''       , UpperCAmelCase_						)
											lowerCAmelCase   =		re.sub('''(\')'''       , r''' \1 '''       , UpperCAmelCase_						)
											lowerCAmelCase   =		re.sub(r'''\s{2,}'''       , ''' '''       , UpperCAmelCase_						)
											if "\n" in token:
																		lowerCAmelCase   =		token.replace('''\n'''       , ''' __newln__'''						)

											lowerCAmelCase   =		token.split(''' '''						)
											lowerCAmelCase   =		[]
											for token in tokens:
																		if not len(UpperCAmelCase_						):
																									continue

																		lowerCAmelCase   =		token.lower()
																		lowerCAmelCase   =		tuple(UpperCAmelCase_						)
																		lowerCAmelCase   =		tuple(list(word[:-1]						) + [word[-1] + '''</w>''']						)
																		lowerCAmelCase   =		get_pairs(UpperCAmelCase_						)

																		if not pairs:
																									words.append(UpperCAmelCase_						)
																									continue

																		while True:
																									lowerCAmelCase   =		min(UpperCAmelCase_       , key=lambda UpperCAmelCase_						: self.bpe_ranks.get(UpperCAmelCase_       , float('''inf'''						)						)						)
																									if bigram not in self.bpe_ranks:
																																break
																									lowerCAmelCase   ,     lowerCAmelCase   =		bigram
																									lowerCAmelCase   =		[]
																									lowerCAmelCase   =		0

																									while i < len(UpperCAmelCase_						):
																																try:
																																							lowerCAmelCase   =		word.index(UpperCAmelCase_       , UpperCAmelCase_						)
																																							new_word.extend(word[i:j]						)
																																							lowerCAmelCase   =		j
																																except ValueError:
																																							new_word.extend(word[i:]						)
																																							break

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

																		lowerCAmelCase   =		word
																		words.append(UpperCAmelCase_						)
											return " ".join(UpperCAmelCase_						)



				def       __snake_case		( self       , UpperCAmelCase_						):
											lowerCAmelCase   =		[]

											lowerCAmelCase   =		re.findall(r'''\S+\n?'''       , UpperCAmelCase_						)

											for token in words:
																		split_tokens.extend(list(self.bpe(UpperCAmelCase_						).split(''' '''						)						)						)
											return split_tokens



				def       __snake_case		( self       , UpperCAmelCase_						):
											lowerCAmelCase   =		token.lower()
											return self.encoder.get(UpperCAmelCase_       , self.encoder.get(self.unk_token						)						)



				def       __snake_case		( self       , UpperCAmelCase_						):
											return self.decoder.get(UpperCAmelCase_       , self.unk_token						)



				def       __snake_case		( self       , UpperCAmelCase_						):
											lowerCAmelCase   =		''' '''.join(UpperCAmelCase_						).replace('''@@ '''       , ''''''						).strip()
											return out_string






				def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_ = None						):
											if not os.path.isdir(UpperCAmelCase_						):
																		logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""						)
																		return
											lowerCAmelCase   =		os.path.join(
											    UpperCAmelCase_       , (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']						)
											lowerCAmelCase   =		os.path.join(
											    UpperCAmelCase_       , (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']						)

											with open(UpperCAmelCase_       , '''w'''       , encoding='''utf-8'''						) as f:
																		f.write(json.dumps(self.encoder       , indent=2       , sort_keys=UpperCAmelCase_       , ensure_ascii=UpperCAmelCase_						) + '''\n'''						)

											lowerCAmelCase   =		0
											with open(UpperCAmelCase_       , '''w'''       , encoding='''utf-8'''						) as writer:
																		writer.write('''#version: 0.2\n'''						)
																		for bpe_tokens, token_index in sorted(self.bpe_ranks.items()       , key=lambda UpperCAmelCase_						: kv[1]						):
																									if index != token_index:
																																logger.warning(
																																    F"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
																																    ''' Please check that the tokenizer is not corrupted!'''						)
																																lowerCAmelCase   =		token_index
																									writer.write(''' '''.join(UpperCAmelCase_						) + '''\n'''						)
																									index += 1

											return vocab_file, merge_file









import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


UpperCAmelCase_						="""hf-internal-testing/tiny-random-bert"""
UpperCAmelCase_						=os.path.join(TRANSFORMERS_CACHE, """models--hf-internal-testing--tiny-random-bert""")
UpperCAmelCase_						="""9b8c223d42b2188cb49d29af482996f9d0f3e5a6"""


class 							__UpperCamelCase			( unittest.TestCase  ):





				'''simple docstring'''



				def       __snake_case		( self						):
											lowerCAmelCase   =		cached_file(UpperCAmelCase_       , UpperCAmelCase_						)
											# Should have downloaded the file in here
											self.assertTrue(os.path.isdir(UpperCAmelCase_						)						)
											# Cache should contain at least those three subfolders:
											for subfolder in ["blobs", "refs", "snapshots"]:
																		self.assertTrue(os.path.isdir(os.path.join(UpperCAmelCase_       , UpperCAmelCase_						)						)						)
											with open(os.path.join(UpperCAmelCase_       , '''refs'''       , '''main'''						)						) as f:
																		lowerCAmelCase   =		f.read()
											self.assertEqual(UpperCAmelCase_       , os.path.join(UpperCAmelCase_       , '''snapshots'''       , UpperCAmelCase_       , UpperCAmelCase_						)						)
											self.assertTrue(os.path.isfile(UpperCAmelCase_						)						)

											# File is cached at the same place the second time.
											lowerCAmelCase   =		cached_file(UpperCAmelCase_       , UpperCAmelCase_						)
											self.assertEqual(UpperCAmelCase_       , UpperCAmelCase_						)

											# Using a specific revision to test the full commit hash.
											lowerCAmelCase   =		cached_file(UpperCAmelCase_       , UpperCAmelCase_       , revision='''9b8c223'''						)
											self.assertEqual(UpperCAmelCase_       , os.path.join(UpperCAmelCase_       , '''snapshots'''       , UpperCAmelCase_       , UpperCAmelCase_						)						)



				def       __snake_case		( self						):
											with self.assertRaisesRegex(UpperCAmelCase_       , '''is not a valid model identifier'''						):
																		lowerCAmelCase   =		cached_file('''tiny-random-bert'''       , UpperCAmelCase_						)

											with self.assertRaisesRegex(UpperCAmelCase_       , '''is not a valid git identifier'''						):
																		lowerCAmelCase   =		cached_file(UpperCAmelCase_       , UpperCAmelCase_       , revision='''aaaa'''						)

											with self.assertRaisesRegex(UpperCAmelCase_       , '''does not appear to have a file named'''						):
																		lowerCAmelCase   =		cached_file(UpperCAmelCase_       , '''conf'''						)



				def       __snake_case		( self						):
											with self.assertRaisesRegex(UpperCAmelCase_       , '''does not appear to have a file named'''						):
																		lowerCAmelCase   =		cached_file(UpperCAmelCase_       , '''conf'''						)

											with open(os.path.join(UpperCAmelCase_       , '''refs'''       , '''main'''						)						) as f:
																		lowerCAmelCase   =		f.read()
											self.assertTrue(os.path.isfile(os.path.join(UpperCAmelCase_       , '''.no_exist'''       , UpperCAmelCase_       , '''conf'''						)						)						)

											lowerCAmelCase   =		cached_file(UpperCAmelCase_       , '''conf'''       , _raise_exceptions_for_missing_entries=UpperCAmelCase_						)
											self.assertIsNone(UpperCAmelCase_						)

											lowerCAmelCase   =		cached_file(UpperCAmelCase_       , '''conf'''       , local_files_only=UpperCAmelCase_       , _raise_exceptions_for_missing_entries=UpperCAmelCase_						)
											self.assertIsNone(UpperCAmelCase_						)

											lowerCAmelCase   =		mock.Mock()
											lowerCAmelCase   =		5_00
											lowerCAmelCase   =		{}
											lowerCAmelCase   =		HTTPError
											lowerCAmelCase   =		{}

											# Under the mock environment we get a 500 error when trying to reach the tokenizer.
											with mock.patch('''requests.Session.request'''       , return_value=UpperCAmelCase_						) as mock_head:
																		lowerCAmelCase   =		cached_file(UpperCAmelCase_       , '''conf'''       , _raise_exceptions_for_connection_errors=UpperCAmelCase_						)
																		self.assertIsNone(UpperCAmelCase_						)
																		# This check we did call the fake head request
																		mock_head.assert_called()



				def       __snake_case		( self						):
											self.assertTrue(has_file('''hf-internal-testing/tiny-bert-pt-only'''       , UpperCAmelCase_						)						)
											self.assertFalse(has_file('''hf-internal-testing/tiny-bert-pt-only'''       , UpperCAmelCase_						)						)
											self.assertFalse(has_file('''hf-internal-testing/tiny-bert-pt-only'''       , UpperCAmelCase_						)						)



				def       __snake_case		( self						):
											# `get_file_from_repo` returns None if the file does not exist
											self.assertIsNone(get_file_from_repo('''bert-base-cased'''       , '''ahah.txt'''						)						)

											# The function raises if the repository does not exist.
											with self.assertRaisesRegex(UpperCAmelCase_       , '''is not a valid model identifier'''						):
																		get_file_from_repo('''bert-base-case'''       , UpperCAmelCase_						)

											# The function raises if the revision does not exist.
											with self.assertRaisesRegex(UpperCAmelCase_       , '''is not a valid git identifier'''						):
																		get_file_from_repo('''bert-base-cased'''       , UpperCAmelCase_       , revision='''ahaha'''						)

											lowerCAmelCase   =		get_file_from_repo('''bert-base-cased'''       , UpperCAmelCase_						)
											# The name is the cached name which is not very easy to test, so instead we load the content.
											lowerCAmelCase   =		json.loads(open(UpperCAmelCase_       , '''r'''						).read()						)
											self.assertEqual(config['''hidden_size''']       , 7_68						)






				def       __snake_case		( self						):
											with tempfile.TemporaryDirectory() as tmp_dir:
																		lowerCAmelCase   =		Path(UpperCAmelCase_						) / '''a.txt'''
																		filename.touch()
																		self.assertEqual(get_file_from_repo(UpperCAmelCase_       , '''a.txt'''						)       , str(UpperCAmelCase_						)						)

																		self.assertIsNone(get_file_from_repo(UpperCAmelCase_       , '''b.txt'''						)						)



