



'''simple docstring'''


import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

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

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoProcessor,
    AutoTokenizer,
    BertTokenizer,
    ProcessorMixin,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
)
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test
from transformers.tokenization_utils import TOKENIZER_CONFIG_FILE
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_tokenizers_available


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

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402
from test_module.custom_processing import CustomProcessor  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


lowercase_      :				Dict   					=    get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
lowercase_      :				Tuple   					=    get_tests_dir('''fixtures/vocab.json''')
lowercase_      :				Optional[int]   					=    get_tests_dir('''fixtures''')




class      __UpperCamelCase       (unittest.TestCase			):
	__A			  =					['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']







	def   _a   (    self						)    ->    Optional[int]:
							'''simple docstring'''






							lowercase        =							0







	def   _a   (    self						)    ->    Union[str, Any]:
							'''simple docstring'''






							lowercase        =							AutoProcessor.from_pretrained("""facebook/wav2vec2-base-960h"""						)
							self.assertIsInstance(_lowerCAmelCase ,				_lowerCAmelCase						)







	def   _a   (    self						)    ->    int:
							'''simple docstring'''






							with tempfile.TemporaryDirectory() as tmpdirname:
													lowercase        =							WavaVecaConfig()
													lowercase        =							AutoProcessor.from_pretrained("""facebook/wav2vec2-base-960h"""						)

													# save in new folder
													model_config.save_pretrained(_lowerCAmelCase						)
													processor.save_pretrained(_lowerCAmelCase						)

													lowercase        =							AutoProcessor.from_pretrained(_lowerCAmelCase						)

							self.assertIsInstance(_lowerCAmelCase ,				_lowerCAmelCase						)







	def   _a   (    self						)    ->    str:
							'''simple docstring'''






							with tempfile.TemporaryDirectory() as tmpdirname:
													# copy relevant files
													copyfile(_lowerCAmelCase ,				os.path.join(_lowerCAmelCase ,				_lowerCAmelCase						)						)
													copyfile(_lowerCAmelCase ,				os.path.join(_lowerCAmelCase ,				"""vocab.json"""						)						)

													lowercase        =							AutoProcessor.from_pretrained(_lowerCAmelCase						)

							self.assertIsInstance(_lowerCAmelCase ,				_lowerCAmelCase						)







	def   _a   (    self						)    ->    List[str]:
							'''simple docstring'''






							with tempfile.TemporaryDirectory() as tmpdirname:
													lowercase        =							WavaVecaFeatureExtractor()
													lowercase        =							AutoTokenizer.from_pretrained("""facebook/wav2vec2-base-960h"""						)

													lowercase        =							WavaVecaProcessor(_lowerCAmelCase ,				_lowerCAmelCase						)

													# save in new folder
													processor.save_pretrained(_lowerCAmelCase						)

													# drop `processor_class` in tokenizer
													with open(os.path.join(_lowerCAmelCase ,				_lowerCAmelCase						) ,				"""r"""						) as f:
																			lowercase        =							json.load(_lowerCAmelCase						)
																			config_dict.pop("""processor_class"""						)

													with open(os.path.join(_lowerCAmelCase ,				_lowerCAmelCase						) ,				"""w"""						) as f:
																			f.write(json.dumps(_lowerCAmelCase						)						)

													lowercase        =							AutoProcessor.from_pretrained(_lowerCAmelCase						)

							self.assertIsInstance(_lowerCAmelCase ,				_lowerCAmelCase						)







	def   _a   (    self						)    ->    Tuple:
							'''simple docstring'''






							with tempfile.TemporaryDirectory() as tmpdirname:
													lowercase        =							WavaVecaFeatureExtractor()
													lowercase        =							AutoTokenizer.from_pretrained("""facebook/wav2vec2-base-960h"""						)

													lowercase        =							WavaVecaProcessor(_lowerCAmelCase ,				_lowerCAmelCase						)

													# save in new folder
													processor.save_pretrained(_lowerCAmelCase						)

													# drop `processor_class` in feature extractor
													with open(os.path.join(_lowerCAmelCase ,				_lowerCAmelCase						) ,				"""r"""						) as f:
																			lowercase        =							json.load(_lowerCAmelCase						)
																			config_dict.pop("""processor_class"""						)

													with open(os.path.join(_lowerCAmelCase ,				_lowerCAmelCase						) ,				"""w"""						) as f:
																			f.write(json.dumps(_lowerCAmelCase						)						)

													lowercase        =							AutoProcessor.from_pretrained(_lowerCAmelCase						)

							self.assertIsInstance(_lowerCAmelCase ,				_lowerCAmelCase						)







	def   _a   (    self						)    ->    List[str]:
							'''simple docstring'''






							with tempfile.TemporaryDirectory() as tmpdirname:
													lowercase        =							WavaVecaConfig(processor_class="""Wav2Vec2Processor"""						)
													model_config.save_pretrained(_lowerCAmelCase						)
													# copy relevant files
													copyfile(_lowerCAmelCase ,				os.path.join(_lowerCAmelCase ,				"""vocab.json"""						)						)
													# create emtpy sample processor
													with open(os.path.join(_lowerCAmelCase ,				_lowerCAmelCase						) ,				"""w"""						) as f:
																			f.write("""{}"""						)

													lowercase        =							AutoProcessor.from_pretrained(_lowerCAmelCase						)

							self.assertIsInstance(_lowerCAmelCase ,				_lowerCAmelCase						)







	def   _a   (    self						)    ->    int:
							'''simple docstring'''






							with self.assertRaises(_lowerCAmelCase						):
													lowercase        =							AutoProcessor.from_pretrained("""hf-internal-testing/test_dynamic_processor"""						)
							# If remote code is disabled, we can't load this config.
							with self.assertRaises(_lowerCAmelCase						):
													lowercase        =							AutoProcessor.from_pretrained(
													    """hf-internal-testing/test_dynamic_processor""" ,				trust_remote_code=_lowerCAmelCase						)

							lowercase        =							AutoProcessor.from_pretrained("""hf-internal-testing/test_dynamic_processor""" ,				trust_remote_code=_lowerCAmelCase						)
							self.assertTrue(processor.special_attribute_present						)
							self.assertEqual(processor.__class__.__name__ ,				"""NewProcessor"""						)

							lowercase        =							processor.feature_extractor
							self.assertTrue(feature_extractor.special_attribute_present						)
							self.assertEqual(feature_extractor.__class__.__name__ ,				"""NewFeatureExtractor"""						)

							lowercase        =							processor.tokenizer
							self.assertTrue(tokenizer.special_attribute_present						)
							if is_tokenizers_available():
													self.assertEqual(tokenizer.__class__.__name__ ,				"""NewTokenizerFast"""						)

													# Test we can also load the slow version
													lowercase        =							AutoProcessor.from_pretrained(
													    """hf-internal-testing/test_dynamic_processor""" ,				trust_remote_code=_lowerCAmelCase ,				use_fast=_lowerCAmelCase						)
													lowercase        =							new_processor.tokenizer
													self.assertTrue(new_tokenizer.special_attribute_present						)
													self.assertEqual(new_tokenizer.__class__.__name__ ,				"""NewTokenizer"""						)
							else:
													self.assertEqual(tokenizer.__class__.__name__ ,				"""NewTokenizer"""						)







	def   _a   (    self						)    ->    Any:
							'''simple docstring'''






							try:
													AutoConfig.register("""custom""" ,				_lowerCAmelCase						)
													AutoFeatureExtractor.register(_lowerCAmelCase ,				_lowerCAmelCase						)
													AutoTokenizer.register(_lowerCAmelCase ,				slow_tokenizer_class=_lowerCAmelCase						)
													AutoProcessor.register(_lowerCAmelCase ,				_lowerCAmelCase						)
													# Trying to register something existing in the Transformers library will raise an error
													with self.assertRaises(_lowerCAmelCase						):
																			AutoProcessor.register(_lowerCAmelCase ,				_lowerCAmelCase						)

													# Now that the config is registered, it can be used as any other config with the auto-API
													lowercase        =							CustomFeatureExtractor.from_pretrained(_lowerCAmelCase						)

													with tempfile.TemporaryDirectory() as tmp_dir:
																			lowercase        =							os.path.join(_lowerCAmelCase ,				"""vocab.txt"""						)
																			with open(_lowerCAmelCase ,				"""w""" ,				encoding="""utf-8"""						) as vocab_writer:
																									vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)
																			lowercase        =							CustomTokenizer(_lowerCAmelCase						)

													lowercase        =							CustomProcessor(_lowerCAmelCase ,				_lowerCAmelCase						)

													with tempfile.TemporaryDirectory() as tmp_dir:
																			processor.save_pretrained(_lowerCAmelCase						)
																			lowercase        =							AutoProcessor.from_pretrained(_lowerCAmelCase						)
																			self.assertIsInstance(_lowerCAmelCase ,				_lowerCAmelCase						)

							finally:
													if "custom" in CONFIG_MAPPING._extra_content:
																			del CONFIG_MAPPING._extra_content["custom"]
													if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
																			del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]
													if CustomConfig in TOKENIZER_MAPPING._extra_content:
																			del TOKENIZER_MAPPING._extra_content[CustomConfig]
													if CustomConfig in PROCESSOR_MAPPING._extra_content:
																			del PROCESSOR_MAPPING._extra_content[CustomConfig]







	def   _a   (    self						)    ->    int:
							'''simple docstring'''






							class      __UpperCamelCase       (_UpperCAmelCase			):
								__A			  =					False




							class      __UpperCamelCase       (_UpperCAmelCase			):
								__A			  =					False




							class      __UpperCamelCase       (_UpperCAmelCase			):
								__A			  =					'''AutoFeatureExtractor'''
								__A			  =					'''AutoTokenizer'''
								__A			  =					False

							try:
													AutoConfig.register("""custom""" ,				_lowerCAmelCase						)
													AutoFeatureExtractor.register(_lowerCAmelCase ,				_lowerCAmelCase						)
													AutoTokenizer.register(_lowerCAmelCase ,				slow_tokenizer_class=_lowerCAmelCase						)
													AutoProcessor.register(_lowerCAmelCase ,				_lowerCAmelCase						)
													# If remote code is not set, the default is to use local classes.
													lowercase        =							AutoProcessor.from_pretrained("""hf-internal-testing/test_dynamic_processor"""						)
													self.assertEqual(processor.__class__.__name__ ,				"""NewProcessor"""						)
													self.assertFalse(processor.special_attribute_present						)
													self.assertFalse(processor.feature_extractor.special_attribute_present						)
													self.assertFalse(processor.tokenizer.special_attribute_present						)

													# If remote code is disabled, we load the local ones.
													lowercase        =							AutoProcessor.from_pretrained(
													    """hf-internal-testing/test_dynamic_processor""" ,				trust_remote_code=_lowerCAmelCase						)
													self.assertEqual(processor.__class__.__name__ ,				"""NewProcessor"""						)
													self.assertFalse(processor.special_attribute_present						)
													self.assertFalse(processor.feature_extractor.special_attribute_present						)
													self.assertFalse(processor.tokenizer.special_attribute_present						)

													# If remote is enabled, we load from the Hub.
													lowercase        =							AutoProcessor.from_pretrained(
													    """hf-internal-testing/test_dynamic_processor""" ,				trust_remote_code=_lowerCAmelCase						)
													self.assertEqual(processor.__class__.__name__ ,				"""NewProcessor"""						)
													self.assertTrue(processor.special_attribute_present						)
													self.assertTrue(processor.feature_extractor.special_attribute_present						)
													self.assertTrue(processor.tokenizer.special_attribute_present						)

							finally:
													if "custom" in CONFIG_MAPPING._extra_content:
																			del CONFIG_MAPPING._extra_content["custom"]
													if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
																			del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]
													if CustomConfig in TOKENIZER_MAPPING._extra_content:
																			del TOKENIZER_MAPPING._extra_content[CustomConfig]
													if CustomConfig in PROCESSOR_MAPPING._extra_content:
																			del PROCESSOR_MAPPING._extra_content[CustomConfig]







	def   _a   (    self						)    ->    str:
							'''simple docstring'''






							lowercase        =							AutoProcessor.from_pretrained("""hf-internal-testing/tiny-random-bert"""						)
							self.assertEqual(processor.__class__.__name__ ,				"""BertTokenizerFast"""						)







	def   _a   (    self						)    ->    List[str]:
							'''simple docstring'''






							lowercase        =							AutoProcessor.from_pretrained("""hf-internal-testing/tiny-random-convnext"""						)
							self.assertEqual(processor.__class__.__name__ ,				"""ConvNextImageProcessor"""						)




@is_staging_test
class      __UpperCamelCase       (unittest.TestCase			):
	__A			  =					['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']







	@classmethod
	def   _a   (    cls						)    ->    Union[str, Any]:
							'''simple docstring'''






							lowercase        =							TOKEN
							HfFolder.save_token(_lowerCAmelCase						)







	@classmethod
	def   _a   (    cls						)    ->    Optional[Any]:
							'''simple docstring'''






							try:
													delete_repo(token=cls._token ,				repo_id="""test-processor"""						)
							except HTTPError:
													pass

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

							try:
													delete_repo(token=cls._token ,				repo_id="""test-dynamic-processor"""						)
							except HTTPError:
													pass







	def   _a   (    self						)    ->    str:
							'''simple docstring'''






							lowercase        =							WavaVecaProcessor.from_pretrained(_lowerCAmelCase						)
							with tempfile.TemporaryDirectory() as tmp_dir:
													processor.save_pretrained(
													    os.path.join(_lowerCAmelCase ,				"""test-processor"""						) ,				push_to_hub=_lowerCAmelCase ,				use_auth_token=self._token						)

													lowercase        =							WavaVecaProcessor.from_pretrained(F"""{USER}/test-processor"""						)
													for k, v in processor.feature_extractor.__dict__.items():
																			self.assertEqual(_lowerCAmelCase ,				getattr(new_processor.feature_extractor ,				_lowerCAmelCase						)						)
													self.assertDictEqual(new_processor.tokenizer.get_vocab() ,				processor.tokenizer.get_vocab()						)







	def   _a   (    self						)    ->    List[str]:
							'''simple docstring'''






							lowercase        =							WavaVecaProcessor.from_pretrained(_lowerCAmelCase						)

							with tempfile.TemporaryDirectory() as tmp_dir:
													processor.save_pretrained(
													    os.path.join(_lowerCAmelCase ,				"""test-processor-org"""						) ,				push_to_hub=_lowerCAmelCase ,				use_auth_token=self._token ,				organization="""valid_org""" ,				)

													lowercase        =							WavaVecaProcessor.from_pretrained("""valid_org/test-processor-org"""						)
													for k, v in processor.feature_extractor.__dict__.items():
																			self.assertEqual(_lowerCAmelCase ,				getattr(new_processor.feature_extractor ,				_lowerCAmelCase						)						)
													self.assertDictEqual(new_processor.tokenizer.get_vocab() ,				processor.tokenizer.get_vocab()						)







	def   _a   (    self						)    ->    Tuple:
							'''simple docstring'''






							CustomFeatureExtractor.register_for_auto_class()
							CustomTokenizer.register_for_auto_class()
							CustomProcessor.register_for_auto_class()

							lowercase        =							CustomFeatureExtractor.from_pretrained(_lowerCAmelCase						)

							with tempfile.TemporaryDirectory() as tmp_dir:
													lowercase        =							os.path.join(_lowerCAmelCase ,				"""vocab.txt"""						)
													with open(_lowerCAmelCase ,				"""w""" ,				encoding="""utf-8"""						) as vocab_writer:
																			vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)
													lowercase        =							CustomTokenizer(_lowerCAmelCase						)

							lowercase        =							CustomProcessor(_lowerCAmelCase ,				_lowerCAmelCase						)

							with tempfile.TemporaryDirectory() as tmp_dir:
													create_repo(F"""{USER}/test-dynamic-processor""" ,				token=self._token						)
													lowercase        =							Repository(_lowerCAmelCase ,				clone_from=F"""{USER}/test-dynamic-processor""" ,				token=self._token						)
													processor.save_pretrained(_lowerCAmelCase						)

													# This has added the proper auto_map field to the feature extractor config
													self.assertDictEqual(
													    processor.feature_extractor.auto_map ,				{
													        """AutoFeatureExtractor""": """custom_feature_extraction.CustomFeatureExtractor""",
													        """AutoProcessor""": """custom_processing.CustomProcessor""",
													    } ,				)

													# This has added the proper auto_map field to the tokenizer config
													with open(os.path.join(_lowerCAmelCase ,				"""tokenizer_config.json"""						)						) as f:
																			lowercase        =							json.load(_lowerCAmelCase						)
													self.assertDictEqual(
													    tokenizer_config["""auto_map"""] ,				{
													        """AutoTokenizer""": ["""custom_tokenization.CustomTokenizer""", None],
													        """AutoProcessor""": """custom_processing.CustomProcessor""",
													    } ,				)

													# The code has been copied from fixtures
													self.assertTrue(os.path.isfile(os.path.join(_lowerCAmelCase ,				"""custom_feature_extraction.py"""						)						)						)
													self.assertTrue(os.path.isfile(os.path.join(_lowerCAmelCase ,				"""custom_tokenization.py"""						)						)						)
													self.assertTrue(os.path.isfile(os.path.join(_lowerCAmelCase ,				"""custom_processing.py"""						)						)						)

													repo.push_to_hub()

							lowercase        =							AutoProcessor.from_pretrained(F"""{USER}/test-dynamic-processor""" ,				trust_remote_code=_lowerCAmelCase						)
							# Can't make an isinstance check because the new_processor is from the CustomProcessor class of a dynamic module
							self.assertEqual(new_processor.__class__.__name__ ,				"""CustomProcessor"""						)







'''simple docstring'''


from __future__ import annotations

import os
from typing import Any

import requests

lowercase_      :				List[str]   					=    '''https://api.github.com'''

# https://docs.github.com/en/free-pro-team@latest/rest/reference/users#get-the-authenticated-user
lowercase_      :				Any   					=    BASE_URL + '''/user'''

# https://github.com/settings/tokens
lowercase_      :				Union[str, Any]   					=    os.environ.get('''USER_TOKEN''', '''''')







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      str						):
						lowercase        =							{
						    """Authorization""": F"""token {auth_token}""",
						    """Accept""": """application/vnd.github.v3+json""",
						}
						return requests.get(lowercase_ ,  headers=lowercase_						).json()


if __name__ == "__main__":  # pragma: no cover
						if USER_TOKEN:
												for key, value in fetch_github_info(USER_TOKEN).items():
																		print(f'''{key}: {value}''')
						else:
												raise ValueError('''\'USER_TOKEN\' field cannot be empty.''')


