




import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


__a							= logging.get_logger(__name__)

__a							= {
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

__a							= {
    'vocab_file': {'openai-gpt': 'https://huggingface.co/openai-gpt/resolve/main/vocab.json'},
    'merges_file': {'openai-gpt': 'https://huggingface.co/openai-gpt/resolve/main/merges.txt'},
}

__a							= {
    'openai-gpt': 512,
}







def lowerCamelCase__   (			_lowercase      ):

     '''simple docstring'''


     UpperCAmelCase_			:							Any   =			text.strip()
     if not text:
          return []
     UpperCAmelCase_			:							str   =			text.split()
     return tokens

class   __a(     _UpperCamelCase   ):






    """simple docstring"""





    def __init__(      self   ,_SCREAMING_SNAKE_CASE=True   ,_SCREAMING_SNAKE_CASE=None   ,_SCREAMING_SNAKE_CASE=True   ,_SCREAMING_SNAKE_CASE=None   ,_SCREAMING_SNAKE_CASE=True   ,) ->		str:







         if never_split is None:
              UpperCAmelCase_			:							Tuple   =			[]
         UpperCAmelCase_			:							Tuple   =			do_lower_case
         UpperCAmelCase_			:							Union[str, Any]   =			set(a__  )
         UpperCAmelCase_			:							Optional[int]   =			tokenize_chinese_chars
         UpperCAmelCase_			:							str   =			strip_accents
         UpperCAmelCase_			:							str   =			do_split_on_punc





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE=None  ) ->		int:
         UpperCAmelCase_			:							int   =			self.never_split.union(set(a__  )  ) if never_split else self.never_split
         UpperCAmelCase_			:							Dict   =			self._clean_text(a__  )

         # This was added on November 1st, 2018 for the multilingual and Chinese
         # models. This is also applied to the English models now, but it doesn't
         # matter since the English models were not trained on any Chinese data
         # and generally don't have any Chinese data in them (there are Chinese
         # characters in the vocabulary because Wikipedia does have some Chinese
         # words in the English Wikipedia.).
         if self.tokenize_chinese_chars:
              UpperCAmelCase_			:							Tuple   =			self._tokenize_chinese_chars(a__  )
         # prevents treating the same character with different unicode codepoints as different characters
         UpperCAmelCase_			:							Dict   =			unicodedata.normalize('''NFC'''   ,a__  )
         UpperCAmelCase_			:							List[str]   =			whitespace_tokenize(a__  )
         UpperCAmelCase_			:							List[str]   =			[]
         for token in orig_tokens:
              if token not in never_split:
                   if self.do_lower_case:
                        UpperCAmelCase_			:							str   =			token.lower()
                        if self.strip_accents is not False:
                             UpperCAmelCase_			:							Tuple   =			self._run_strip_accents(a__  )
                   elif self.strip_accents:
                        UpperCAmelCase_			:							str   =			self._run_strip_accents(a__  )
              split_tokens.extend(self._run_split_on_punc(a__   ,a__  )  )

         UpperCAmelCase_			:							Any   =			whitespace_tokenize(''' '''.join(a__  )  )
         return output_tokens





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE  ) ->		Tuple:
         UpperCAmelCase_			:							Optional[Any]   =			unicodedata.normalize('''NFD'''   ,a__  )
         UpperCAmelCase_			:							Tuple   =			[]
         for char in text:
              UpperCAmelCase_			:							List[str]   =			unicodedata.category(a__  )
              if cat == "Mn":
                   continue
              output.append(a__  )
         return "".join(a__  )





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE=None  ) ->		List[str]:







         if not self.do_split_on_punc or (never_split is not None and text in never_split):
              return [text]
         UpperCAmelCase_			:							int   =			list(a__  )
         UpperCAmelCase_			:							List[str]   =			0
         UpperCAmelCase_			:							Union[str, Any]   =			True
         UpperCAmelCase_			:							int   =			[]
         while i < len(a__  ):
              UpperCAmelCase_			:							List[Any]   =			chars[i]
              if _is_punctuation(a__  ):
                   output.append([char]  )
                   UpperCAmelCase_			:							List[Any]   =			True
              else:
                   if start_new_word:
                        output.append([]  )
                   UpperCAmelCase_			:							Dict   =			False
                   output[-1].append(a__  )
              i += 1

         return ["".join(a__  ) for x in output]





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE  ) ->		List[Any]:
         UpperCAmelCase_			:							List[str]   =			[]
         for char in text:
              UpperCAmelCase_			:							List[Any]   =			ord(a__  )
              if self._is_chinese_char(a__  ):
                   output.append(''' '''  )
                   output.append(a__  )
                   output.append(''' '''  )
              else:
                   output.append(a__  )
         return "".join(a__  )





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE  ) ->		Optional[int]:







         if (
             (cp >= 0X4e_00 and cp <= 0X9f_ff)
             or (cp >= 0X34_00 and cp <= 0X4d_bf)  #
             or (cp >= 0X2_00_00 and cp <= 0X2_a6_df)  #
             or (cp >= 0X2_a7_00 and cp <= 0X2_b7_3f)  #
             or (cp >= 0X2_b7_40 and cp <= 0X2_b8_1f)  #
             or (cp >= 0X2_b8_20 and cp <= 0X2_ce_af)  #
             or (cp >= 0Xf9_00 and cp <= 0Xfa_ff)
             or (cp >= 0X2_f8_00 and cp <= 0X2_fa_1f)  #
         ):  #
              return True

         return False





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE  ) ->		str:
         UpperCAmelCase_			:							str   =			[]
         for char in text:
              UpperCAmelCase_			:							Dict   =			ord(a__  )
              if cp == 0 or cp == 0Xff_fd or _is_control(a__  ):
                   continue
              if _is_whitespace(a__  ):
                   output.append(''' '''  )
              else:
                   output.append(a__  )
         return "".join(a__  )







def lowerCamelCase__   (			_lowercase      ):

     '''simple docstring'''


     UpperCAmelCase_			:							str   =			set()
     UpperCAmelCase_			:							Union[str, Any]   =			word[0]
     for char in word[1:]:
          pairs.add((prev_char, char)      )
          UpperCAmelCase_			:							Dict   =			char
     return pairs







def lowerCamelCase__   (			_lowercase      ):

     '''simple docstring'''


     UpperCAmelCase_			:							str   =			text.replace('''—''' ,						'''-'''      )
     UpperCAmelCase_			:							Optional[int]   =			text.replace('''–''' ,						'''-'''      )
     UpperCAmelCase_			:							int   =			text.replace('''―''' ,						'''-'''      )
     UpperCAmelCase_			:							Any   =			text.replace('''…''' ,						'''...'''      )
     UpperCAmelCase_			:							str   =			text.replace('''´''' ,						'''\''''      )
     UpperCAmelCase_			:							Dict   =			re.sub(r'''(-+|~+|!+|\"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)''' ,						r''' \1 ''' ,						__lowercase      )
     UpperCAmelCase_			:							str   =			re.sub(r'''\s*\n\s*''' ,						''' \n ''' ,						__lowercase      )
     UpperCAmelCase_			:							str   =			re.sub(r'''[^\S\n]+''' ,						''' ''' ,						__lowercase      )
     return text.strip()



class   __a(     _UpperCamelCase   ):






    """simple docstring"""





    lowerCAmelCase  			=      VOCAB_FILES_NAMES
    lowerCAmelCase  			=      PRETRAINED_VOCAB_FILES_MAP
    lowerCAmelCase  			=      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    lowerCAmelCase  			=      ['''input_ids''', '''attention_mask''']





    def __init__(      self   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE="<unk>"   ,**_SCREAMING_SNAKE_CASE  ) ->		Optional[int]:
         super().__init__(unk_token=a__   ,**a__  )

         try:
              import ftfy
              from spacy.lang.en import English

              UpperCAmelCase_			:							Optional[Any]   =			English()
              UpperCAmelCase_			:							Union[str, Any]   =			_nlp.tokenizer
              UpperCAmelCase_			:							Any   =			ftfy.fix_text
         except ImportError:
              logger.warning('''ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.'''  )
              UpperCAmelCase_			:							Tuple   =			BasicTokenizer(do_lower_case=a__  )
              UpperCAmelCase_			:							Any   =			None

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





    @property
    def 							a__      (      self  ) ->		str:
         return True





    @property
    def 							a__      (      self  ) ->		Tuple:
         return len(self.encoder  )





    def 							a__      (      self  ) ->		int:
         return dict(self.encoder   ,**self.added_tokens_encoder  )





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE  ) ->		Optional[int]:
         UpperCAmelCase_			:							Optional[int]   =			tuple(token[:-1]  ) + (token[-1] + '''</w>''',)
         if token in self.cache:
              return self.cache[token]
         UpperCAmelCase_			:							int   =			get_pairs(a__  )

         if not pairs:
              return token + "</w>"

         while True:
              UpperCAmelCase_			:							Optional[int]   =			min(a__   ,key=lambda _SCREAMING_SNAKE_CASE  : self.bpe_ranks.get(a__   ,float('''inf'''  )  )  )
              if bigram not in self.bpe_ranks:
                   break
              UpperCAmelCase_,    UpperCAmelCase_			:							List[str]   =			bigram
              UpperCAmelCase_			:							Optional[Any]   =			[]
              UpperCAmelCase_			:							Optional[int]   =			0
              while i < len(a__  ):
                   try:
                        UpperCAmelCase_			:							Union[str, Any]   =			word.index(a__   ,a__  )
                   except ValueError:
                        new_word.extend(word[i:]  )
                        break
                   else:
                        new_word.extend(word[i:j]  )
                        UpperCAmelCase_			:							List[str]   =			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
              UpperCAmelCase_			:							List[Any]   =			tuple(a__  )
              UpperCAmelCase_			:							List[Any]   =			new_word
              if len(a__  ) == 1:
                   break
              else:
                   UpperCAmelCase_			:							Optional[int]   =			get_pairs(a__  )
         UpperCAmelCase_			:							Optional[int]   =			''' '''.join(a__  )
         if word == "\n  </w>":
              UpperCAmelCase_			:							int   =			'''\n</w>'''
         UpperCAmelCase_			:							List[str]   =			word
         return word





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE  ) ->		int:
         UpperCAmelCase_			:							List[str]   =			[]
         if self.fix_text is None:
              # Using BERT's BasicTokenizer
              UpperCAmelCase_			:							str   =			self.nlp.tokenize(a__  )
              for token in text:
                   split_tokens.extend(list(self.bpe(a__  ).split(''' '''  )  )  )
         else:
              # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
              UpperCAmelCase_			:							int   =			self.nlp(text_standardize(self.fix_text(a__  )  )  )
              for token in text:
                   split_tokens.extend(list(self.bpe(token.text.lower()  ).split(''' '''  )  )  )
         return split_tokens





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE  ) ->		List[str]:
         return self.encoder.get(a__   ,self.encoder.get(self.unk_token  )  )





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE  ) ->		Optional[int]:
         return self.decoder.get(a__   ,self.unk_token  )





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE  ) ->		List[Any]:
         UpperCAmelCase_			:							Tuple   =			''''''.join(a__  ).replace('''</w>'''   ,''' '''  ).strip()
         return out_string





    def 							a__      (      self   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE = None  ) ->		Tuple[str]:







         if not os.path.isdir(a__  ):
              logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''  )
              return
         UpperCAmelCase_			:							Union[str, Any]   =			os.path.join(
             a__   ,(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']  )
         UpperCAmelCase_			:							Tuple   =			os.path.join(
             a__   ,(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']  )

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

         UpperCAmelCase_			:							int   =			0
         with open(a__   ,'''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 _SCREAMING_SNAKE_CASE  : 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!'''  )
                        UpperCAmelCase_			:							Dict   =			token_index
                   writer.write(''' '''.join(a__  ) + '''\n'''  )
                   index += 1

         return vocab_file, merge_file



"""simple docstring"""


import logging
import os
from dataclasses import dataclass, field
from typing import Dict, List, Optional

import numpy as np
import torch
from utils_hans import HansDataset, InputFeatures, hans_processors, hans_tasks_num_labels

import transformers
from transformers import (
    AutoConfig,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    default_data_collator,
    set_seed,
)
from transformers.trainer_utils import is_main_process


a_											=			logging.getLogger(__name__)
@dataclass
class snake_case					:
		__UpperCamelCase							=	field(
		    metadata={'help': 'Path to pretrained model or model identifier from huggingface.co/models'})
		__UpperCamelCase							=	field(
		    default=_UpperCamelCase   , metadata={'help': 'Pretrained config name or path if not the same as model_name'})
		__UpperCamelCase							=	field(
		    default=_UpperCamelCase   , metadata={'help': 'Pretrained tokenizer name or path if not the same as model_name'})
		__UpperCamelCase							=	field(
		    default=_UpperCamelCase   , metadata={'help': 'Where do you want to store the pretrained models downloaded from huggingface.co'}   , )





@dataclass
class snake_case					:
		__UpperCamelCase							=	field(
		    metadata={'help': 'The name of the task to train selected in the list: ' + ', '.join(hans_processors.keys())})
		__UpperCamelCase							=	field(
		    metadata={'help': 'The input data dir. Should contain the .tsv files (or other data files) for the task.'})
		__UpperCamelCase							=	field(
		    default=128   , metadata={
		        'help': (
		            'The maximum total input sequence length after tokenization. Sequences longer '
		            'than this will be truncated, sequences shorter will be padded.'
		        )
		    }   , )
		__UpperCamelCase							=	field(
		    default=_UpperCamelCase   , metadata={'help': 'Overwrite the cached training and evaluation sets'})






def    a__       ( __lowercase      )		->		Dict[str, torch.Tensor]:
				_A         =	default_data_collator(__lowercase      )
				_A         =	batch.pop("pairID"   ,	__lowercase      )
				return batch






def    a__       ( )		->		Optional[Any]:
				# See all possible arguments in src/transformers/training_args.py
				# or by passing the --help flag to this script.
				# We now keep distinct sets of args, for a cleaner separation of concerns.

				_A         =	HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)      )
				_A     ,						_A     ,						_A         =	parser.parse_args_into_dataclasses()

				if (
				    os.path.exists(training_args.output_dir      )
				    and os.listdir(training_args.output_dir      )
				    and training_args.do_train
				    and not training_args.overwrite_output_dir
				):
								raise ValueError(
								    f"""Output directory ({training_args.output_dir}) already exists and is not empty. Use"""
								    " --overwrite_output_dir to overcome."      )

				# Setup logging
				logging.basicConfig(
				    format="%(asctime)s - %(levelname)s - %(name)s - %(message)s"   ,	datefmt="%m/%d/%Y %H:%M:%S"   ,	level=logging.INFO if training_args.local_rank in [-1, 0] else logging.WARN   ,	)
				logger.warning(
				    "Process rank: %s, device: %s, n_gpu: %s, distributed training: %s, 16-bits training: %s"   ,	training_args.local_rank   ,	training_args.device   ,	training_args.n_gpu   ,	bool(training_args.local_rank != -1      )   ,	training_args.fpaa   ,	)
				# Set the verbosity to info of the Transformers logger (on main process only):
				if is_main_process(training_args.local_rank      ):
								transformers.utils.logging.set_verbosity_info()
								transformers.utils.logging.enable_default_handler()
								transformers.utils.logging.enable_explicit_format()
				logger.info("Training/evaluation parameters %s"   ,	__lowercase      )

				# Set seed
				set_seed(training_args.seed      )

				try:
								_A         =	hans_tasks_num_labels[data_args.task_name]
				except KeyError:
								raise ValueError("Task not found: %s" % (data_args.task_name)      )

				# Load pretrained model and tokenizer
				#
				# Distributed training:
				# The .from_pretrained methods guarantee that only one local process can concurrently
				# download model & vocab.

				_A         =	AutoConfig.from_pretrained(
				    model_args.config_name if model_args.config_name else model_args.model_name_or_path   ,	num_labels=__lowercase   ,	finetuning_task=data_args.task_name   ,	cache_dir=model_args.cache_dir   ,	)
				_A         =	AutoTokenizer.from_pretrained(
				    model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path   ,	cache_dir=model_args.cache_dir   ,	)
				_A         =	AutoModelForSequenceClassification.from_pretrained(
				    model_args.model_name_or_path   ,	from_tf=bool(".ckpt" in model_args.model_name_or_path      )   ,	config=__lowercase   ,	cache_dir=model_args.cache_dir   ,	)

				# Get datasets
				_A         =	(
				    HansDataset(
				        data_dir=data_args.data_dir   ,	tokenizer=__lowercase   ,	task=data_args.task_name   ,	max_seq_length=data_args.max_seq_length   ,	overwrite_cache=data_args.overwrite_cache   ,	)
				    if training_args.do_train
				    else None
				)
				_A         =	(
				    HansDataset(
				        data_dir=data_args.data_dir   ,	tokenizer=__lowercase   ,	task=data_args.task_name   ,	max_seq_length=data_args.max_seq_length   ,	overwrite_cache=data_args.overwrite_cache   ,	evaluate=__lowercase   ,	)
				    if training_args.do_eval
				    else None
				)

				# Initialize our Trainer
				_A         =	Trainer(
				    model=__lowercase   ,	args=__lowercase   ,	train_dataset=__lowercase   ,	eval_dataset=__lowercase   ,	data_collator=__lowercase   ,	)

				# Training
				if training_args.do_train:
								trainer.train(
								    model_path=model_args.model_name_or_path if os.path.isdir(model_args.model_name_or_path      ) else None      )
								trainer.save_model()
								# For convenience, we also re-save the tokenizer to the same directory,
								# so that you can share your model easily on huggingface.co/models =)
								if trainer.is_world_master():
												tokenizer.save_pretrained(training_args.output_dir      )

    # Evaluation
				if training_args.do_eval:
								logger.info("*** Evaluate ***"      )

								_A         =	trainer.predict(__lowercase      )
								_A         =	output.predictions
								_A         =	np.argmax(__lowercase   ,	axis=1      )

								_A         =	[ex.pairID for ex in eval_dataset]
								_A         =	os.path.join(training_args.output_dir   ,	"hans_predictions.txt"      )
								_A         =	eval_dataset.get_labels()
								if trainer.is_world_master():
												with open(__lowercase   ,	"w"      ) as writer:
																writer.write("pairID,gold_label\n"      )
																for pid, pred in zip(__lowercase   ,	__lowercase      ):
																				writer.write("ex" + str(__lowercase      ) + "," + label_list[int(__lowercase      )] + "\n"      )

								trainer._log(output.metrics      )






def    a__       ( __lowercase      )		->		Dict:
				# For xla_spawn (TPUs)
				main()


if __name__ == "__main__":
		main()