





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

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
      from .tokenization_pegasus import PegasusTokenizer
else:
      snake_case								=							None


snake_case								=							logging.get_logger(__name__)


snake_case								=							"""▁"""

snake_case								=							{"""vocab_file""": """spiece.model""", """tokenizer_file""": """tokenizer.json"""}

snake_case								=							{
    """vocab_file""": {"""google/pegasus-xsum""": """https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model"""},
    """tokenizer_file""": {
        """google/pegasus-xsum""": """https://huggingface.co/google/pegasus-xsum/resolve/main/tokenizer.json"""
    },
}

snake_case								=							{
    """google/pegasus-xsum""": 5_1_2,
}





class 					A_					(   _a						):







       """simple docstring"""





       SCREAMING_SNAKE_CASE_   :			List[str]  			=	VOCAB_FILES_NAMES
       SCREAMING_SNAKE_CASE_   :			Optional[int]  			=	PRETRAINED_VOCAB_FILES_MAP
       SCREAMING_SNAKE_CASE_   :			Optional[int]  			=	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       SCREAMING_SNAKE_CASE_   :			Dict  			=	PegasusTokenizer
       SCREAMING_SNAKE_CASE_   :			str  			=	['''input_ids''', '''attention_mask''']







       def __init__(     self						:						Tuple    ,__A						:						Optional[Any]=None    ,__A						:						List[str]=None    ,__A						:						Dict="<pad>"    ,__A						:						int="</s>"    ,__A						:						List[str]="<unk>"    ,__A						:						Tuple="<mask_2>"    ,__A						:						Any="<mask_1>"    ,__A						:						Union[str, Any]=None    ,__A						:						Optional[int]=103    ,**__A						:						List[str]    ,)    ->       int:
         _lowercase						    =			offset

         if additional_special_tokens is not None:
           if not isinstance(A_    ,A_	):
             raise TypeError(
                 F"""additional_special_tokens should be of type {type(A_	)}, but is"""
                 F""" {type(A_	)}"""	)

           _lowercase						    =			(
               ([mask_token_sent] + additional_special_tokens)
               if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
               else additional_special_tokens
           )
           # fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
           additional_special_tokens_extended += [
               F"""<unk_{i}>""" for i in range(len(A_	)    ,self.offset - 1	)
           ]

           if len(set(A_	)	) != len(A_	):
             raise ValueError(
                 'Please make sure that the provided additional_special_tokens do not contain an incorrectly'
                 F""" shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}."""	)
           _lowercase						    =			additional_special_tokens_extended
         else:
           _lowercase						    =			[mask_token_sent] if mask_token_sent is not None else []
           additional_special_tokens += [F"""<unk_{i}>""" for i in range(2    ,self.offset	)]

         super().__init__(
             A_    ,tokenizer_file=A_    ,pad_token=A_    ,eos_token=A_    ,unk_token=A_    ,mask_token=A_    ,mask_token_sent=A_    ,offset=A_    ,additional_special_tokens=A_    ,**A_    ,)
         _lowercase						    =			vocab_file
         _lowercase						    =			False if not self.vocab_file else True







       def  __UpperCAmelCase			(     self						:						Optional[Any]    ,__A						:						List[str]	)    ->       Any:
         _lowercase						    =			set(self.all_special_ids	)  # call it once instead of inside list comp
         all_special_ids.remove(self.unk_token_id	)  # <unk> is only sometimes special

         if all_special_ids != set(range(len(self.additional_special_tokens	) + 3	)	):
           raise ValueError(
               'There should be 3 special tokens: mask_token, pad_token, and eos_token +'
               F""" {len(self.additional_special_tokens	)} additional_special_tokens, but got {all_special_ids}"""	)

         return [1 if x in all_special_ids else 0 for x in seq]







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




         if already_has_special_tokens:
           return self._special_token_mask(A_	)
         elif token_ids_a is None:
           return self._special_token_mask(A_	) + [1]
         else:
           return self._special_token_mask(token_ids_a + token_ids_a	) + [1]







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




         if token_ids_a is None:
           return token_ids_a + [self.eos_token_id]
         # We don't expect to process pairs, but leave the pair logic for API consistency
         return token_ids_a + token_ids_a + [self.eos_token_id]







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




         if not self.can_save_slow_tokenizer:
           raise ValueError(
               'Your fast tokenizer does not have the necessary information to save the vocabulary for a slow '
               'tokenizer.'	)

         if not os.path.isdir(A_	):
           logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""	)
           return
         _lowercase						    =			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_	):
           copyfile(self.vocab_file    ,A_	)

         return (out_vocab_file,)



import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / '''utils'''))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


__snake_case   							=						get_tests_dir('''fixtures''')



class 	__lowerCamelCase			(unittest.TestCase						):







       def 		snake_case_	(			self: int					):






            '''simple docstring'''




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

            # Download this model to make sure it's in the cache.
            __UpperCamelCase =			WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'					)
            # Under the mock environment we get a 500 error when trying to reach the model.
            with mock.patch('requests.Session.request',return_value=A_					) as mock_head:
                 __UpperCamelCase =			WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'					)
                 # This check we did call the fake head request
                 mock_head.assert_called()







       def 		snake_case_	(			self: Dict					):






            '''simple docstring'''




            __UpperCamelCase =			WavaVecaFeatureExtractor.from_pretrained(
                'https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'					)



@is_staging_test
class 	__lowerCamelCase			(unittest.TestCase						):







       @classmethod
       def 		snake_case_	(			cls: Tuple					):






            '''simple docstring'''




            __UpperCamelCase =			TOKEN
            HfFolder.save_token(A_					)







       @classmethod
       def 		snake_case_	(			cls: Tuple					):






            '''simple docstring'''




            try:
                 delete_repo(token=cls._token,repo_id='test-feature-extractor'					)
            except HTTPError:
                 pass

            try:
                 delete_repo(token=cls._token,repo_id='valid_org/test-feature-extractor-org'					)
            except HTTPError:
                 pass

            try:
                 delete_repo(token=cls._token,repo_id='test-dynamic-feature-extractor'					)
            except HTTPError:
                 pass







       def 		snake_case_	(			self: Tuple					):






            '''simple docstring'''




            __UpperCamelCase =			WavaVecaFeatureExtractor.from_pretrained(A_					)
            feature_extractor.push_to_hub('test-feature-extractor',use_auth_token=self._token					)

            __UpperCamelCase =			WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''					)
            for k, v in feature_extractor.__dict__.items():
                 self.assertEqual(A_,getattr(A_,A_					)					)

            # Reset repo
            delete_repo(token=self._token,repo_id='test-feature-extractor'					)

            # Push to hub via save_pretrained
            with tempfile.TemporaryDirectory() as tmp_dir:
                 feature_extractor.save_pretrained(
                     A_,repo_id='test-feature-extractor',push_to_hub=A_,use_auth_token=self._token					)

            __UpperCamelCase =			WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''					)
            for k, v in feature_extractor.__dict__.items():
                 self.assertEqual(A_,getattr(A_,A_					)					)







       def 		snake_case_	(			self: List[str]					):






            '''simple docstring'''




            __UpperCamelCase =			WavaVecaFeatureExtractor.from_pretrained(A_					)
            feature_extractor.push_to_hub('valid_org/test-feature-extractor',use_auth_token=self._token					)

            __UpperCamelCase =			WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor'					)
            for k, v in feature_extractor.__dict__.items():
                 self.assertEqual(A_,getattr(A_,A_					)					)

            # Reset repo
            delete_repo(token=self._token,repo_id='valid_org/test-feature-extractor'					)

            # Push to hub via save_pretrained
            with tempfile.TemporaryDirectory() as tmp_dir:
                 feature_extractor.save_pretrained(
                     A_,repo_id='valid_org/test-feature-extractor-org',push_to_hub=A_,use_auth_token=self._token					)

            __UpperCamelCase =			WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor-org'					)
            for k, v in feature_extractor.__dict__.items():
                 self.assertEqual(A_,getattr(A_,A_					)					)







       def 		snake_case_	(			self: int					):






            '''simple docstring'''




            CustomFeatureExtractor.register_for_auto_class()
            __UpperCamelCase =			CustomFeatureExtractor.from_pretrained(A_					)

            feature_extractor.push_to_hub('test-dynamic-feature-extractor',use_auth_token=self._token					)

            # This has added the proper auto_map field to the config
            self.assertDictEqual(
                feature_extractor.auto_map,{'AutoFeatureExtractor': 'custom_feature_extraction.CustomFeatureExtractor'},)

            __UpperCamelCase =			AutoFeatureExtractor.from_pretrained(
                F'''{USER}/test-dynamic-feature-extractor''',trust_remote_code=A_					)
            # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
            self.assertEqual(new_feature_extractor.__class__.__name__,'CustomFeatureExtractor'					)


