



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

_lowerCamelCase						    =  logging.get_logger(__name__)


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

_lowerCamelCase						    =  {
    """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"""
    },
}

_lowerCamelCase						    =  {"""facebook/blenderbot-3B""": 128}






@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def 	_lowerCAmelCase						(				):



 """simple docstring"""





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






def 	_lowerCAmelCase						(				__lowerCamelCase   :     Optional[Any]     ):



 """simple docstring"""





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






class       _SCREAMING_SNAKE_CASE	(UpperCAmelCase_ ):
     lowerCAmelCase					=			VOCAB_FILES_NAMES
     lowerCAmelCase					=			PRETRAINED_VOCAB_FILES_MAP
     lowerCAmelCase					=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     lowerCAmelCase					=			["input_ids", "attention_mask"]





     def __init__(       self :							Tuple     ,  UpperCamelCase :							str     ,  UpperCamelCase :							Tuple     ,  UpperCamelCase :							int="replace"     ,  UpperCamelCase :							List[str]="<s>"     ,  UpperCamelCase :							List[Any]="</s>"     ,  UpperCamelCase :							Optional[Any]="</s>"     ,  UpperCamelCase :							List[str]="<s>"     ,  UpperCamelCase :							int="<unk>"     ,  UpperCamelCase :							List[str]="<pad>"     ,  UpperCamelCase :							int="<mask>"     ,  UpperCamelCase :							Optional[int]=False     ,  **UpperCamelCase :							int     ,  )->str:
      __SCREAMING_SNAKE_CASE       :    int       =	AddedToken(_snake_case     ,  lstrip=_snake_case     ,  rstrip=_snake_case				) if isinstance(_snake_case     ,  _snake_case				) else bos_token
      __SCREAMING_SNAKE_CASE       :    Any       =	AddedToken(_snake_case     ,  lstrip=_snake_case     ,  rstrip=_snake_case				) if isinstance(_snake_case     ,  _snake_case				) else eos_token
      __SCREAMING_SNAKE_CASE       :    int       =	AddedToken(_snake_case     ,  lstrip=_snake_case     ,  rstrip=_snake_case				) if isinstance(_snake_case     ,  _snake_case				) else sep_token
      __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	AddedToken(_snake_case     ,  lstrip=_snake_case     ,  rstrip=_snake_case				) if isinstance(_snake_case     ,  _snake_case				) else cls_token
      __SCREAMING_SNAKE_CASE       :    Any       =	AddedToken(_snake_case     ,  lstrip=_snake_case     ,  rstrip=_snake_case				) if isinstance(_snake_case     ,  _snake_case				) else unk_token
      __SCREAMING_SNAKE_CASE       :    Dict       =	AddedToken(_snake_case     ,  lstrip=_snake_case     ,  rstrip=_snake_case				) if isinstance(_snake_case     ,  _snake_case				) else pad_token

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

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

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

      # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
      __SCREAMING_SNAKE_CASE       :    List[Any]       =	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 	__snake_case    (       self :							int				)->str:
      return len(self.encoder				)





     def 	__snake_case    (       self :							List[Any]				)->Union[str, Any]:
      return dict(self.encoder     ,  **self.added_tokens_encoder				)





     def 	__snake_case    (       self :							Optional[Any]     ,  UpperCamelCase :							Any				)->List[Any]:


      if token in self.cache:
       return self.cache[token]
      __SCREAMING_SNAKE_CASE       :    Tuple       =	tuple(_snake_case				)
      __SCREAMING_SNAKE_CASE       :    Any       =	get_pairs(_snake_case				)

      if not pairs:
       return token

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





     def 	__snake_case    (       self :							int     ,  UpperCamelCase :							Dict				)->Dict:
      __SCREAMING_SNAKE_CASE       :    Dict       =	[]
      for token in re.findall(self.pat     ,  _snake_case				):
       __SCREAMING_SNAKE_CASE       :    Dict       =	"".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 	__snake_case    (       self :							Any     ,  UpperCamelCase :							List[str]				)->Tuple:
      return self.encoder.get(_snake_case     ,  self.encoder.get(self.unk_token				)				)





     def 	__snake_case    (       self :							str     ,  UpperCamelCase :							Optional[Any]				)->str:
      return self.decoder.get(_snake_case				)





     def 	__snake_case    (       self :							Optional[int]     ,  UpperCamelCase :							List[str]				)->str:
      __SCREAMING_SNAKE_CASE       :    List[str]       =	"".join(_snake_case				)
      __SCREAMING_SNAKE_CASE       :    List[str]       =	bytearray([self.byte_decoder[c] for c in text]				).decode("utf-8"     ,  errors=self.errors				)
      return text





     def 	__snake_case    (       self :							List[str]     ,  UpperCamelCase :							str     ,  UpperCamelCase :							Optional[str] = None				)->List[str]:


      if not os.path.isdir(_snake_case				):
       logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""				)
       return
      __SCREAMING_SNAKE_CASE       :    Dict       =	os.path.join(
          _snake_case     ,  (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]				)
      __SCREAMING_SNAKE_CASE       :    Optional[int]       =	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"				)

      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	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 UpperCamelCase				: 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       :    int       =	token_index
        writer.write(" ".join(_snake_case				) + "\n"				)
        index += 1

      return vocab_file, merge_file





     def 	__snake_case    (       self :							Union[str, Any]     ,  UpperCamelCase :							List[int]     ,  UpperCamelCase :							Optional[List[int]] = None     ,  UpperCamelCase :							bool = False				)->int:


      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 token_ids_a is None:
       return [1] + ([0] * len(_snake_case				)) + [1]
      return [1] + ([0] * len(_snake_case				)) + [1, 1] + ([0] * len(_snake_case				)) + [1]





     def 	__snake_case    (       self :							str     ,  UpperCamelCase :							List[int]     ,  UpperCamelCase :							Optional[List[int]] = None				)->Optional[Any]:
      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	[self.sep_token_id]
      __SCREAMING_SNAKE_CASE       :    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 	__snake_case    (       self :							Optional[int]     ,  UpperCamelCase :							Optional[Any]     ,  UpperCamelCase :							str=False     ,  **UpperCamelCase :							int				)->Dict:
      __SCREAMING_SNAKE_CASE       :    List[str]       =	kwargs.pop("add_prefix_space"     ,  self.add_prefix_space				)
      if (is_split_into_words or add_prefix_space) and (len(_snake_case				) > 0 and not text[0].isspace()):
       __SCREAMING_SNAKE_CASE       :    Any       =	" " + text
      return (text, kwargs)





     def 	__snake_case    (       self :							str     ,  UpperCamelCase :							List[int]     ,  UpperCamelCase :							Optional[List[int]] = None				)->int:
      return token_ids_a + [self.eos_token_id]






     def 	__snake_case    (       self :							List[Any]     ,  UpperCamelCase :							"Conversation"				)->str:
      __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	[]
      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(_snake_case				)

      __SCREAMING_SNAKE_CASE       :    int       =	"  ".join(_snake_case				)
      __SCREAMING_SNAKE_CASE       :    Tuple       =	self.encode(_snake_case				)
      if len(_snake_case				) > self.model_max_length:
       __SCREAMING_SNAKE_CASE       :    Union[str, 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











from collections.abc import Generator
from math import sin






def 	_lowerCAmelCase						(				__lowerCamelCase   :     bytes     ):



 """simple docstring"""





 if len(__lowerCamelCase     ) != 32:
  raise ValueError("Input must be of length 32"     )

 __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	b""
 for i in [3, 2, 1, 0]:
  little_endian += string_aa[8 * i : 8 * i + 8]
 return little_endian






def 	_lowerCAmelCase						(				__lowerCamelCase   :     int     ):



 """simple docstring"""





 if i < 0:
  raise ValueError("Input must be non-negative"     )

 __SCREAMING_SNAKE_CASE       :    Any       =	format(__lowerCamelCase		,	"08x"     )[-8:]
 __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	b""
 for i in [3, 2, 1, 0]:
  little_endian_hex += hex_rep[2 * i : 2 * i + 2].encode("utf-8"     )
 return little_endian_hex






def 	_lowerCAmelCase						(				__lowerCamelCase   :     bytes     ):



 """simple docstring"""





 __SCREAMING_SNAKE_CASE       :    Any       =	b""
 for char in message:
  bit_string += format(__lowerCamelCase		,	"08b"     ).encode("utf-8"     )
 __SCREAMING_SNAKE_CASE       :    List[str]       =	format(len(__lowerCamelCase     )		,	"064b"     ).encode("utf-8"     )

 # Pad bit_string to a multiple of 512 chars
 bit_string += b"1"
 while len(__lowerCamelCase     ) % 512 != 448:
  bit_string += b"0"
 bit_string += to_little_endian(start_len[32:]     ) + to_little_endian(start_len[:32]     )

 return bit_string






def 	_lowerCAmelCase						(				__lowerCamelCase   :     bytes     ):



 """simple docstring"""





 if len(__lowerCamelCase     ) % 512 != 0:
  raise ValueError("Input must have length that's a multiple of 512"     )

 for pos in range(0		,	len(__lowerCamelCase     )		,	512     ):
  __SCREAMING_SNAKE_CASE       :    int       =	bit_string[pos : pos + 512]
  __SCREAMING_SNAKE_CASE       :    Tuple       =	[]
  for i in range(0		,	512		,	32     ):
   block_words.append(int(to_little_endian(block[i : i + 32]     )		,	2     )     )
  yield block_words






def 	_lowerCAmelCase						(				__lowerCamelCase   :     int     ):



 """simple docstring"""





 if i < 0:
  raise ValueError("Input must be non-negative"     )

 __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	format(__lowerCamelCase		,	"032b"     )
 __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	""
 for c in i_str:
  new_str += "1" if c == "0" else "0"
 return int(__lowerCamelCase		,	2     )






def 	_lowerCAmelCase						(				__lowerCamelCase   :     int		,	__lowerCamelCase   :     int     ):



 """simple docstring"""





 return (a + b) % 2**32






def 	_lowerCAmelCase						(				__lowerCamelCase   :     int		,	__lowerCamelCase   :     int     ):



 """simple docstring"""





 if i < 0:
  raise ValueError("Input must be non-negative"     )
 if shift < 0:
  raise ValueError("Shift must be non-negative"     )
 return ((i << shift) ^ (i >> (32 - shift))) % 2**32






def 	_lowerCAmelCase						(				__lowerCamelCase   :     bytes     ):



 """simple docstring"""





 __SCREAMING_SNAKE_CASE       :    Tuple       =	preprocess(__lowerCamelCase     )

 __SCREAMING_SNAKE_CASE       :    List[Any]       =	[int(2**32 * abs(sin(i + 1     )     )     ) for i in range(64     )]

 # Starting states
 __SCREAMING_SNAKE_CASE       :    Tuple       =	0X67452301
 __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	0Xefcdab89
 __SCREAMING_SNAKE_CASE       :    Optional[int]       =	0X98badcfe
 __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	0X10325476

 __SCREAMING_SNAKE_CASE       :    List[Any]       =	[
     7,
     12,
     17,
     22,
     7,
     12,
     17,
     22,
     7,
     12,
     17,
     22,
     7,
     12,
     17,
     22,
     5,
     9,
     14,
     20,
     5,
     9,
     14,
     20,
     5,
     9,
     14,
     20,
     5,
     9,
     14,
     20,
     4,
     11,
     16,
     23,
     4,
     11,
     16,
     23,
     4,
     11,
     16,
     23,
     4,
     11,
     16,
     23,
     6,
     10,
     15,
     21,
     6,
     10,
     15,
     21,
     6,
     10,
     15,
     21,
     6,
     10,
     15,
     21,
 ]

 # Process bit string in chunks, each with 16 32-char words
 for block_words in get_block_words(__lowerCamelCase     ):
  __SCREAMING_SNAKE_CASE       :    Any       =	aa
  __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	ba
  __SCREAMING_SNAKE_CASE       :    str       =	ca
  __SCREAMING_SNAKE_CASE       :    Dict       =	da

  # Hash current chunk
  for i in range(64     ):
   if i <= 15:
    # f = (b & c) | (not_32(b) & d)     # Alternate definition for f
    __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	d ^ (b & (c ^ d))
    __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	i
   elif i <= 31:
    # f = (d & b) | (not_32(d) & c)     # Alternate definition for f
    __SCREAMING_SNAKE_CASE       :    int       =	c ^ (d & (b ^ c))
    __SCREAMING_SNAKE_CASE       :    int       =	(5 * i + 1) % 16
   elif i <= 47:
    __SCREAMING_SNAKE_CASE       :    List[str]       =	b ^ c ^ d
    __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	(3 * i + 5) % 16
   else:
    __SCREAMING_SNAKE_CASE       :    Any       =	c ^ (b | not_aa(__lowerCamelCase     ))
    __SCREAMING_SNAKE_CASE       :    str       =	(7 * i) % 16
   __SCREAMING_SNAKE_CASE       :    List[str]       =	(f + a + added_consts[i] + block_words[g]) % 2**32
   __SCREAMING_SNAKE_CASE       :    Dict       =	d
   __SCREAMING_SNAKE_CASE       :    str       =	c
   __SCREAMING_SNAKE_CASE       :    Tuple       =	b
   __SCREAMING_SNAKE_CASE       :    Optional[int]       =	sum_aa(__lowerCamelCase		,	left_rotate_aa(__lowerCamelCase		,	shift_amounts[i]     )     )

  # Add hashed chunk to running total
  __SCREAMING_SNAKE_CASE       :    Dict       =	sum_aa(__lowerCamelCase		,	__lowerCamelCase     )
  __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	sum_aa(__lowerCamelCase		,	__lowerCamelCase     )
  __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	sum_aa(__lowerCamelCase		,	__lowerCamelCase     )
  __SCREAMING_SNAKE_CASE       :    str       =	sum_aa(__lowerCamelCase		,	__lowerCamelCase     )

 __SCREAMING_SNAKE_CASE       :    List[str]       =	reformat_hex(__lowerCamelCase     ) + reformat_hex(__lowerCamelCase     ) + reformat_hex(__lowerCamelCase     ) + reformat_hex(__lowerCamelCase     )
 return digest


if __name__ == "__main__":
      import doctest

      doctest.testmod()






