






"""simple docstring"""






import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
     from transformers.pipelines.conversational import Conversation

UpperCAmelCase_     :				str         =				logging.get_logger(__name__)


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

UpperCAmelCase_     :				Optional[Any]         =				{
    """vocab_file""": {"""facebook/blenderbot-3B""": """https://huggingface.co/facebook/blenderbot-3B/resolve/main/vocab.json"""},
    """merges_file""": {"""facebook/blenderbot-3B""": """https://huggingface.co/facebook/blenderbot-3B/resolve/main/merges.txt"""},
    """tokenizer_config_file""": {
        """facebook/blenderbot-3B""": """https://huggingface.co/facebook/blenderbot-3B/resolve/main/tokenizer_config.json"""
    },
}

UpperCAmelCase_     :				Union[str, Any]         =				{"""facebook/blenderbot-3B""": 128}



@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def       _A    ()				-> Optional[Any]:
    """simple docstring"""






    SCREAMING_SNAKE_CASE_   :						int							=	(
        list(range(ord('''!'''							)					,       ord('''~'''							) + 1							)							) + list(range(ord('''¡'''							)					,       ord('''¬'''							) + 1							)							) + list(range(ord('''®'''							)					,       ord('''ÿ'''							) + 1							)							)
    )
    SCREAMING_SNAKE_CASE_   :						Dict							=	bs[:]
    SCREAMING_SNAKE_CASE_   :						Tuple							=	0
    for b in range(2**8							):
        if b not in bs:
            bs.append(__a							)
            cs.append(2**8 + n							)
            n += 1
    SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	[chr(__a							) for n in cs]
    return dict(zip(__a					,       __a							)							)



def       _A    (__a							)				-> str:
    """simple docstring"""






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





class 							lowerCAmelCase__       (    __SCREAMING_SNAKE_CASE				):






      '''simple docstring'''

      __UpperCamelCase							=						VOCAB_FILES_NAMES
      __UpperCamelCase							=						PRETRAINED_VOCAB_FILES_MAP
      __UpperCamelCase							=						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      __UpperCamelCase							=						["input_ids", "attention_mask"]
      def __init__(		self  :						Union[str, Any]		,			lowercase_  :						Tuple		,			lowercase_  :						Dict		,			lowercase_  :						List[str]="replace"		,			lowercase_  :						List[str]="<s>"		,			lowercase_  :						int="</s>"		,			lowercase_  :						Tuple="</s>"		,			lowercase_  :						List[Any]="<s>"		,			lowercase_  :						Union[str, Any]="<unk>"		,			lowercase_  :						List[str]="<pad>"		,			lowercase_  :						Optional[int]="<mask>"		,			lowercase_  :						Optional[Any]=False		,			**lowercase_  :						List[str]		,			):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Any							=	AddedToken(_a		,			lstrip=_a		,			rstrip=_a) if isinstance(_a		,			_a) else bos_token
          SCREAMING_SNAKE_CASE_   :						Dict							=	AddedToken(_a		,			lstrip=_a		,			rstrip=_a) if isinstance(_a		,			_a) else eos_token
          SCREAMING_SNAKE_CASE_   :						List[Any]							=	AddedToken(_a		,			lstrip=_a		,			rstrip=_a) if isinstance(_a		,			_a) else sep_token
          SCREAMING_SNAKE_CASE_   :						Optional[int]							=	AddedToken(_a		,			lstrip=_a		,			rstrip=_a) if isinstance(_a		,			_a) else cls_token
          SCREAMING_SNAKE_CASE_   :						List[Any]							=	AddedToken(_a		,			lstrip=_a		,			rstrip=_a) if isinstance(_a		,			_a) else unk_token
          SCREAMING_SNAKE_CASE_   :						List[str]							=	AddedToken(_a		,			lstrip=_a		,			rstrip=_a) if isinstance(_a		,			_a) else pad_token

          # Mask token behave like a normal word, i.e. include the space before it
          SCREAMING_SNAKE_CASE_   :						Dict							=	AddedToken(_a		,			lstrip=_a		,			rstrip=_a) if isinstance(_a		,			_a) else mask_token

          super().__init__(
              errors=_a		,			bos_token=_a		,			eos_token=_a		,			unk_token=_a		,			sep_token=_a		,			cls_token=_a		,			pad_token=_a		,			mask_token=_a		,			add_prefix_space=_a		,			**_a		,			)

          with open(_a		,			encoding='''utf-8''') as vocab_handle:
              SCREAMING_SNAKE_CASE_   :						Any							=	json.load(_a)
          SCREAMING_SNAKE_CASE_   :						str							=	{v: k for k, v in self.encoder.items()}
          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	errors  # how to handle errors in decoding
          SCREAMING_SNAKE_CASE_   :						Any							=	bytes_to_unicode()
          SCREAMING_SNAKE_CASE_   :						int							=	{v: k for k, v in self.byte_encoder.items()}
          with open(_a		,			encoding='''utf-8''') as merges_handle:
              SCREAMING_SNAKE_CASE_   :						int							=	merges_handle.read().split('''\n''')[1:-1]
          SCREAMING_SNAKE_CASE_   :						Optional[int]							=	[tuple(merge.split()) for merge in bpe_merges]
          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	dict(zip(_a		,			range(len(_a))))
          SCREAMING_SNAKE_CASE_   :						Any							=	{}
          SCREAMING_SNAKE_CASE_   :						int							=	add_prefix_space

          # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
          SCREAMING_SNAKE_CASE_   :						int							=	re.compile(r'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+''')
      @property
      # Copied from transformers.models.roberta.tokenization_roberta.RobertaTokenizer.vocab_size with Roberta->Blenderbot, RoBERTa->Blenderbot
      def _SCREAMING_SNAKE_CASE				(		self  :						Tuple):




          '''simple docstring'''


          return len(self.encoder)
      def _SCREAMING_SNAKE_CASE				(		self  :						Dict):




          '''simple docstring'''


          return dict(self.encoder		,			**self.added_tokens_encoder)
      def _SCREAMING_SNAKE_CASE				(		self  :						Optional[Any]		,			lowercase_  :						Any):




          '''simple docstring'''


          if token in self.cache:
              return self.cache[token]
          SCREAMING_SNAKE_CASE_   :						Optional[int]							=	tuple(_a)
          SCREAMING_SNAKE_CASE_   :						int							=	get_pairs(_a)

          if not pairs:
              return token

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




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	[]
          for token in re.findall(self.pat		,			_a):
              SCREAMING_SNAKE_CASE_   :						str							=	"""""".join(
                  self.byte_encoder[b] for b in token.encode('''utf-8'''))  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
              bpe_tokens.extend(bpe_token for bpe_token in self.bpe(_a).split(''' '''))
          return bpe_tokens
      def _SCREAMING_SNAKE_CASE				(		self  :						str		,			lowercase_  :						List[str]):




          '''simple docstring'''


          return self.encoder.get(_a		,			self.encoder.get(self.unk_token))
      def _SCREAMING_SNAKE_CASE				(		self  :						Any		,			lowercase_  :						List[Any]):




          '''simple docstring'''


          return self.decoder.get(_a)
      def _SCREAMING_SNAKE_CASE				(		self  :						Union[str, Any]		,			lowercase_  :						Dict):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						int							=	"""""".join(_a)
          SCREAMING_SNAKE_CASE_   :						Tuple							=	bytearray([self.byte_decoder[c] for c in text]).decode('''utf-8'''		,			errors=self.errors)
          return text
      def _SCREAMING_SNAKE_CASE				(		self  :						Tuple		,			lowercase_  :						str		,			lowercase_  :						Optional[str] = None):




          '''simple docstring'''


          if not os.path.isdir(_a):
              logger.error(F'Vocabulary path ({save_directory}) should be a directory')
              return
          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	os.path.join(
              _a		,			(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file'''])
          SCREAMING_SNAKE_CASE_   :						Dict							=	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''')

          SCREAMING_SNAKE_CASE_   :						List[Any]							=	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 lowercase_: 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!''')
                      SCREAMING_SNAKE_CASE_   :						Tuple							=	token_index
                  writer.write(''' '''.join(_a) + '''\n''')
                  index += 1

          return vocab_file, merge_file
      def _SCREAMING_SNAKE_CASE				(		self  :						int		,			lowercase_  :						List[int]		,			lowercase_  :						Optional[List[int]] = None		,			lowercase_  :						bool = False):




          '''simple docstring'''


          if already_has_special_tokens:
              return super().get_special_tokens_mask(
                  token_ids_a=_a		,			token_ids_a=_a		,			already_has_special_tokens=_a)

          if token_ids_a is None:
              return [1] + ([0] * len(_a)) + [1]
          return [1] + ([0] * len(_a)) + [1, 1] + ([0] * len(_a)) + [1]
      def _SCREAMING_SNAKE_CASE				(		self  :						int		,			lowercase_  :						List[int]		,			lowercase_  :						Optional[List[int]] = None):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Optional[int]							=	[self.sep_token_id]
          SCREAMING_SNAKE_CASE_   :						List[Any]							=	[self.cls_token_id]

          if token_ids_a is None:
              return len(cls + token_ids_a + sep) * [0]
          return len(cls + token_ids_a + sep + sep + token_ids_a + sep) * [0]
      def _SCREAMING_SNAKE_CASE				(		self  :						Optional[int]		,			lowercase_  :						Dict		,			lowercase_  :						Optional[Any]=False		,			**lowercase_  :						Any):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	kwargs.pop('''add_prefix_space'''		,			self.add_prefix_space)
          if (is_split_into_words or add_prefix_space) and (len(_a) > 0 and not text[0].isspace()):
              SCREAMING_SNAKE_CASE_   :						List[str]							=	""" """ + text
          return (text, kwargs)
      def _SCREAMING_SNAKE_CASE				(		self  :						Optional[Any]		,			lowercase_  :						List[int]		,			lowercase_  :						Optional[List[int]] = None):




          '''simple docstring'''


          return token_ids_a + [self.eos_token_id]



      def _SCREAMING_SNAKE_CASE				(		self  :						List[str]		,			lowercase_  :						"Conversation"):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Tuple							=	[]
          for is_user, text in conversation.iter_texts():
              if is_user:
                  # We need to space prefix as it's being done within blenderbot
                  inputs.append(''' ''' + text)
              else:
                  # Generated responses should contain them already.
                  inputs.append(_a)

          SCREAMING_SNAKE_CASE_   :						List[Any]							=	"""  """.join(_a)
          SCREAMING_SNAKE_CASE_   :						List[str]							=	self.encode(_a)
          if len(_a) > self.model_max_length:
              SCREAMING_SNAKE_CASE_   :						List[Any]							=	input_ids[-self.model_max_length :]
              logger.warning(F'Trimmed input from conversation as it was longer than {self.model_max_length} tokens.')
          return input_ids










"""simple docstring"""






import argparse
import os

import torch

from transformers import (
    XLNetConfig,
    XLNetForQuestionAnswering,
    XLNetForSequenceClassification,
    XLNetLMHeadModel,
    load_tf_weights_in_xlnet,
)
from transformers.utils import CONFIG_NAME, WEIGHTS_NAME, logging


UpperCAmelCase_     :				int         =				{
    """cola""": 2,
    """mnli""": 3,
    """mrpc""": 2,
    """sst-2""": 2,
    """sts-b""": 1,
    """qqp""": 2,
    """qnli""": 2,
    """rte""": 2,
    """wnli""": 2,
}


logging.set_verbosity_info()



def       _A    (__a					,       __a					,       __a					,       __a=None							)				-> int:
    """simple docstring"""






    SCREAMING_SNAKE_CASE_   :						str							=	XLNetConfig.from_json_file(__a							)

    SCREAMING_SNAKE_CASE_   :						List[Any]							=	finetuning_task.lower() if finetuning_task is not None else ''''''
    if finetuning_task in GLUE_TASKS_NUM_LABELS:
        print(f'Building PyTorch XLNetForSequenceClassification model from configuration: {config}'							)
        SCREAMING_SNAKE_CASE_   :						List[Any]							=	finetuning_task
        SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	GLUE_TASKS_NUM_LABELS[finetuning_task]
        SCREAMING_SNAKE_CASE_   :						Dict							=	XLNetForSequenceClassification(__a							)
    elif "squad" in finetuning_task:
        SCREAMING_SNAKE_CASE_   :						str							=	finetuning_task
        SCREAMING_SNAKE_CASE_   :						Dict							=	XLNetForQuestionAnswering(__a							)
    else:
        SCREAMING_SNAKE_CASE_   :						List[str]							=	XLNetLMHeadModel(__a							)

    # Load weights from tf checkpoint
    load_tf_weights_in_xlnet(__a					,       __a					,       __a							)

    # Save pytorch-model
    SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	os.path.join(__a					,       __a							)
    SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	os.path.join(__a					,       __a							)
    print(f'Save PyTorch model to {os.path.abspath(__a							)}'							)
    torch.save(model.state_dict()					,       __a							)
    print(f'Save configuration file to {os.path.abspath(__a							)}'							)
    with open(__a					,       '''w'''					,       encoding='''utf-8'''							) as f:
        f.write(config.to_json_string()							)


if __name__ == "__main__":
  UpperCAmelCase_     :				str         =				argparse.ArgumentParser()
  # Required parameters
  parser.add_argument(
      """--tf_checkpoint_path""", default=None, type=str, required=True, help="""Path to the TensorFlow checkpoint path."""
  )
  parser.add_argument(
      """--xlnet_config_file""",
      default=None,
      type=str,
      required=True,
      help=(
          """The config json file corresponding to the pre-trained XLNet model. \n"""
          """This specifies the model architecture."""
      ),
  )
  parser.add_argument(
      """--pytorch_dump_folder_path""",
      default=None,
      type=str,
      required=True,
      help="""Path to the folder to store the PyTorch model or dataset/vocab.""",
  )
  parser.add_argument(
      """--finetuning_task""",
      default=None,
      type=str,
      help="""Name of a task on which the XLNet TensorFlow model was fine-tuned""",
  )
  UpperCAmelCase_     :				int         =				parser.parse_args()
  print(args)

  convert_xlnet_checkpoint_to_pytorch(
      args.tf_checkpoint_path, args.xlnet_config_file, args.pytorch_dump_folder_path, args.finetuning_task
  )


