





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

__lowercase				    =      logging.get_logger(__name__)

__lowercase				    =      {
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
}

__lowercase				    =      {
    """vocab_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/vocab.json""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/vocab.json""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/vocab.json""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/vocab.json""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/merges.txt""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/merges.txt""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/merges.txt""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/merges.txt""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/merges.txt""",
    },
}

__lowercase				    =      {
    """gpt2""": 1_024,
    """gpt2-medium""": 1_024,
    """gpt2-large""": 1_024,
    """gpt2-xl""": 1_024,
    """distilgpt2""": 1_024,
}


@lru_cache()
def        __SCREAMING_SNAKE_CASE	(				):



  '''simple docstring'''






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


def        __SCREAMING_SNAKE_CASE	(				__lowercase:		Optional[Any]):



  '''simple docstring'''






  __lowerCAmelCase								=					set()
  __lowerCAmelCase								=					word[0]
  for char in word[1:]:
    pairs.add((prev_char, char))
    __lowerCAmelCase								=					char
  return pairs




class 	__snake_case					(      _a ):






    '''simple docstring'''
    _lowerCAmelCase  =						VOCAB_FILES_NAMES
    _lowerCAmelCase  =						PRETRAINED_VOCAB_FILES_MAP
    _lowerCAmelCase  =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    _lowerCAmelCase  =						["""input_ids""", """attention_mask"""]





    def __init__( self    :		Optional[Any]      ,    snake_case    :		int      ,    snake_case    :		int      ,    snake_case    :		Tuple="replace"      ,    snake_case    :		Union[str, Any]="<|endoftext|>"      ,    snake_case    :		int="<|endoftext|>"      ,    snake_case    :		str="<|endoftext|>"      ,    snake_case    :		Optional[Any]=None      ,    snake_case    :		int=False      ,    snake_case    :		Any=False      ,    **snake_case    :		Optional[int]      ,    ):

      """simple docstring"""





      __lowerCAmelCase								=					AddedToken(snake_case      ,    lstrip=snake_case      ,    rstrip=snake_case				) if isinstance(snake_case      ,    snake_case				) else bos_token
      __lowerCAmelCase								=					AddedToken(snake_case      ,    lstrip=snake_case      ,    rstrip=snake_case				) if isinstance(snake_case      ,    snake_case				) else eos_token
      __lowerCAmelCase								=					AddedToken(snake_case      ,    lstrip=snake_case      ,    rstrip=snake_case				) if isinstance(snake_case      ,    snake_case				) else unk_token
      __lowerCAmelCase								=					AddedToken(snake_case      ,    lstrip=snake_case      ,    rstrip=snake_case				) if isinstance(snake_case      ,    snake_case				) else pad_token
      super().__init__(
          errors=snake_case      ,    unk_token=snake_case      ,    bos_token=snake_case      ,    eos_token=snake_case      ,    pad_token=snake_case      ,    add_prefix_space=snake_case      ,    add_bos_token=snake_case      ,    **snake_case      ,    )
      __lowerCAmelCase								=					add_bos_token

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

      # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
      __lowerCAmelCase								=					re.compile(r'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''				)





    @property
    def 						A		( self    :		int				):

      """simple docstring"""





      return len(self.encoder				)





    def 						A		( self    :		Dict				):

      """simple docstring"""





      return dict(self.encoder      ,    **self.added_tokens_encoder				)





    def 						A		( self    :		int      ,    snake_case    :		str				):

      """simple docstring"""





      if token in self.cache:
        return self.cache[token]
      __lowerCAmelCase								=					tuple(snake_case				)
      __lowerCAmelCase								=					get_pairs(snake_case				)

      if not pairs:
        return token

      while True:
        __lowerCAmelCase								=					min(snake_case      ,    key=lambda snake_case				: self.bpe_ranks.get(snake_case      ,    float('''inf'''				)				)				)
        if bigram not in self.bpe_ranks:
          break
        __lowerCAmelCase						,    __lowerCAmelCase								=					bigram
        __lowerCAmelCase								=					[]
        __lowerCAmelCase								=					0
        while i < len(snake_case				):
          try:
            __lowerCAmelCase								=					word.index(snake_case      ,    snake_case				)
          except ValueError:
            new_word.extend(word[i:]				)
            break
          else:
            new_word.extend(word[i:j]				)
            __lowerCAmelCase								=					j

          if word[i] == first and i < len(snake_case				) - 1 and word[i + 1] == second:
            new_word.append(first + second				)
            i += 2
          else:
            new_word.append(word[i]				)
            i += 1
        __lowerCAmelCase								=					tuple(snake_case				)
        __lowerCAmelCase								=					new_word
        if len(snake_case				) == 1:
          break
        else:
          __lowerCAmelCase								=					get_pairs(snake_case				)
      __lowerCAmelCase								=					''' '''.join(snake_case				)
      __lowerCAmelCase								=					word
      return word





    def 						A		( self    :		Optional[int]      ,    snake_case    :		Dict      ,    snake_case    :		Optional[Any]=None				):

      """simple docstring"""





      if self.add_bos_token:
        __lowerCAmelCase								=					[self.bos_token_id]
      else:
        __lowerCAmelCase								=					[]

      __lowerCAmelCase								=					bos_token_ids + token_ids_a

      if token_ids_a is None:
        return output

      return output + bos_token_ids + token_ids_a





    def 						A		( self    :		Optional[int]      ,    snake_case    :		List[int]      ,    snake_case    :		Optional[List[int]] = None      ,    snake_case    :		bool = False				):

      """simple docstring"""





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

      if not self.add_bos_token:
        return super().get_special_tokens_mask(
            token_ids_a=snake_case      ,    token_ids_a=snake_case      ,    already_has_special_tokens=snake_case				)

      if token_ids_a is None:
        return [1] + ([0] * len(snake_case				))
      return [1] + ([0] * len(snake_case				)) + [1] + ([0] * len(snake_case				))





    def 						A		( self    :		int      ,    snake_case    :		Tuple				):

      """simple docstring"""





      __lowerCAmelCase								=					[]
      for token in re.findall(self.pat      ,    snake_case				):
        __lowerCAmelCase								=					''''''.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(snake_case				).split(''' '''				)				)
      return bpe_tokens





    def 						A		( self    :		Any      ,    snake_case    :		Any				):

      """simple docstring"""





      return self.encoder.get(snake_case      ,    self.encoder.get(self.unk_token				)				)





    def 						A		( self    :		List[str]      ,    snake_case    :		Optional[Any]				):

      """simple docstring"""





      return self.decoder.get(snake_case				)





    def 						A		( self    :		Union[str, Any]      ,    snake_case    :		List[str]				):

      """simple docstring"""





      __lowerCAmelCase								=					''''''.join(snake_case				)
      __lowerCAmelCase								=					bytearray([self.byte_decoder[c] for c in text]				).decode('''utf-8'''      ,    errors=self.errors				)
      return text





    def 						A		( self    :		Optional[Any]      ,    snake_case    :		str      ,    snake_case    :		Optional[str] = None				):

      """simple docstring"""





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

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

      __lowerCAmelCase								=					0
      with open(snake_case      ,    '''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 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!'''				)
            __lowerCAmelCase								=					token_index
          writer.write(''' '''.join(snake_case				) + '''\n'''				)
          index += 1

      return vocab_file, merge_file





    def 						A		( self    :		Any      ,    snake_case    :		Optional[Any]      ,    snake_case    :		Optional[int]=False      ,    **snake_case    :		Optional[int]				):

      """simple docstring"""





      __lowerCAmelCase								=					kwargs.pop('''add_prefix_space'''      ,    self.add_prefix_space				)
      if is_split_into_words or add_prefix_space:
        __lowerCAmelCase								=					''' ''' + text
      return (text, kwargs)





    def 						A		( self    :		Optional[Any]      ,    snake_case    :		"Conversation"				):

      """simple docstring"""





      __lowerCAmelCase								=					[]
      for is_user, text in conversation.iter_texts():
        input_ids.extend(self.encode(snake_case      ,    add_special_tokens=snake_case				) + [self.eos_token_id]				)
      if len(snake_case				) > self.model_max_length:
        __lowerCAmelCase								=					input_ids[-self.model_max_length :]
      return input_ids







def        __SCREAMING_SNAKE_CASE	(				__lowercase:		int):



  '''simple docstring'''






  if not isinstance(__lowercase  ,				__lowercase):
    __lowerCAmelCase								=					F'''Input value of [number={number}] must be an integer'''
    raise TypeError(__lowercase)
  if number < 0 or number % 2 == 1:
    return False
  __lowerCAmelCase								=					int(number**0.5)
  return number == number_sqrt * (number_sqrt + 1)


if __name__ == "__main__":
      import doctest

      doctest.testmod()
