




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


lowercase__				:      List[Any]	   = "hf-internal-testing/tiny-random-bert"
lowercase__				:      Any	   = os.path.join(TRANSFORMERS_CACHE, "models--hf-internal-testing--tiny-random-bert")
lowercase__				:      Any	   = "9b8c223d42b2188cb49d29af482996f9d0f3e5a6"




class       a__	( unittest.TestCase				):







     def       lowerCAmelCase_    (     self			)  ->   Any:



          '''simple docstring'''



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

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

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







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



          '''simple docstring'''



          with self.assertRaisesRegex(A  ,		"is not a valid model identifier"			):
               a     =				cached_file("tiny-random-bert"  ,		A			)

          with self.assertRaisesRegex(A  ,		"is not a valid git identifier"			):
               a     =				cached_file(A  ,		A  ,		revision="aaaa"			)

          with self.assertRaisesRegex(A  ,		"does not appear to have a file named"			):
               a     =				cached_file(A  ,		"conf"			)







     def       lowerCAmelCase_    (     self			)  ->   Tuple:



          '''simple docstring'''



          with self.assertRaisesRegex(A  ,		"does not appear to have a file named"			):
               a     =				cached_file(A  ,		"conf"			)

          with open(os.path.join(A  ,		"refs"  ,		"main"			)			) as f:
               a     =				f.read()
          self.assertTrue(os.path.isfile(os.path.join(A  ,		".no_exist"  ,		A  ,		"conf"			)			)			)

          a     =				cached_file(A  ,		"conf"  ,		_raise_exceptions_for_missing_entries=A			)
          self.assertIsNone(A			)

          a     =				cached_file(A  ,		"conf"  ,		local_files_only=A  ,		_raise_exceptions_for_missing_entries=A			)
          self.assertIsNone(A			)

          a     =				mock.Mock()
          a     =				500
          a     =				{}
          a     =				HTTPError
          a     =				{}

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







     def       lowerCAmelCase_    (     self			)  ->   int:



          '''simple docstring'''



          self.assertTrue(has_file("hf-internal-testing/tiny-bert-pt-only"  ,		A			)			)
          self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"  ,		A			)			)
          self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"  ,		A			)			)







     def       lowerCAmelCase_    (     self			)  ->   Dict:



          '''simple docstring'''



          self.assertIsNone(get_file_from_repo("bert-base-cased"  ,		"ahah.txt"			)			)

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

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

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







     def       lowerCAmelCase_    (     self			)  ->   int:



          '''simple docstring'''



          with tempfile.TemporaryDirectory() as tmp_dir:
               a     =				Path(A			) / "a.txt"
               filename.touch()
               self.assertEqual(get_file_from_repo(A  ,		"a.txt"			)  ,		str(A			)			)

               self.assertIsNone(get_file_from_repo(A  ,		"b.txt"			)			)







import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


lowercase__				:      int	   = logging.get_logger(__name__)


lowercase__				:      Dict	   = "▁"

lowercase__				:      Union[str, Any]	   = {"vocab_file": "spiece.model"}

lowercase__				:      Union[str, Any]	   = {
    "vocab_file": {
        "google/reformer-crime-and-punishment": (
            "https://huggingface.co/google/reformer-crime-and-punishment/resolve/main/spiece.model"
        )
    }
}

lowercase__				:      Tuple	   = {
    "google/reformer-crime-and-punishment": 524_288,
}




class       a__	( UpperCamelCase__				):
     a		:      List[Any]		 =     VOCAB_FILES_NAMES
     a		:      List[Any]		 =     PRETRAINED_VOCAB_FILES_MAP
     a		:      Any		 =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     a		:      Dict		 =     ["""input_ids""", """attention_mask"""]







     def __init__(     self  ,		A  ,		A="</s>"  ,		A="<unk>"  ,		A=[]  ,		A = None  ,		**A  ,		)  ->   None:



          '''simple docstring'''



          a     =				{} if sp_model_kwargs is None else sp_model_kwargs

          super().__init__(
              eos_token=A  ,		unk_token=A  ,		additional_special_tokens=A  ,		sp_model_kwargs=self.sp_model_kwargs  ,		**A  ,		)

          a     =				vocab_file
          a     =				spm.SentencePieceProcessor(**self.sp_model_kwargs			)
          self.sp_model.Load(A			)







     @property
     def       lowerCAmelCase_    (     self			)  ->   Optional[Any]:



          '''simple docstring'''



          return self.sp_model.get_piece_size()







     def       lowerCAmelCase_    (     self			)  ->   Dict[str, int]:



          '''simple docstring'''



          a     =				{self.convert_ids_to_tokens(A			): i for i in range(self.vocab_size			)}
          vocab.update(self.added_tokens_encoder			)
          return vocab







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



          '''simple docstring'''



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







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



          '''simple docstring'''



          a     =				d

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

          a     =				spm.SentencePieceProcessor(**self.sp_model_kwargs			)
          self.sp_model.Load(self.vocab_file			)







     def       lowerCAmelCase_    (     self  ,		A			)  ->   List[str]:



          '''simple docstring'''



          return self.sp_model.encode(A  ,		out_type=A			)







     def       lowerCAmelCase_    (     self  ,		A			)  ->   Union[str, Any]:



          '''simple docstring'''



          return self.sp_model.piece_to_id(A			)







     def       lowerCAmelCase_    (     self  ,		A			)  ->   Optional[int]:



          '''simple docstring'''



          if index < self.sp_model.get_piece_size():
               a     =				self.sp_model.IdToPiece(A			)
          return token







     def       lowerCAmelCase_    (     self  ,		A			)  ->   Union[str, Any]:



          '''simple docstring'''



          a     =				[]
          a     =				""
          for token in tokens:
               # make sure that special tokens are not decoded using sentencepiece model
               if token in self.all_special_tokens:
                    out_string += self.sp_model.decode(A			) + token
                    a     =				[]
               else:
                    current_sub_tokens.append(A			)
          out_string += self.sp_model.decode(A			)
          return out_string.strip()







     def       lowerCAmelCase_    (     self  ,		A  ,		A = None			)  ->   Tuple[str]:



          '''simple docstring'''



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

          if os.path.abspath(self.vocab_file			) != os.path.abspath(A			) and os.path.isfile(self.vocab_file			):
               copyfile(self.vocab_file  ,		A			)
          elif not os.path.isfile(self.vocab_file			):
               with open(A  ,		"wb"			) as fi:
                    a     =				self.sp_model.serialized_model_proto()
                    fi.write(A			)

          return (out_vocab_file,)

