




"""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

__A 		=			logging.get_logger(__name__)

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

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

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



@lru_cache()
def   UpperCamelCase__     (		):
		snake_case		:						Dict								=				(
		    list(range(ord("!"   )       ,     ord("~"   ) + 1   )   ) + list(range(ord("¡"   )       ,     ord("¬"   ) + 1   )   ) + list(range(ord("®"   )       ,     ord("ÿ"   ) + 1   )   )
		)
		snake_case		:						Dict								=				bs[:]
		snake_case		:						List[Any]								=				0
		for b in range(2**8   ):
				if b not in bs:
						bs.append(lowercase__   )
						cs.append(2**8 + n   )
						n += 1
		snake_case		:						Any								=				[chr(lowercase__   ) for n in cs]
		return dict(zip(lowercase__       ,     lowercase__   )   )



def   UpperCamelCase__     (		lowercase__  :     str   ):
		snake_case		:						int								=				set()
		snake_case		:						int								=				word[0]
		for char in word[1:]:
				pairs.add((prev_char, char)   )
				snake_case		:						Union[str, Any]								=				char
		return pairs

class 			lowerCamelCase__    (  lowerCAmelCase__			):
			a__ :      Optional[int]								=     VOCAB_FILES_NAMES
			a__ :      int								=     PRETRAINED_VOCAB_FILES_MAP
			a__ :      Optional[int]								=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			a__ :      Any								=     ["""input_ids""", """attention_mask"""]


			def __init__(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE="replace"				, SCREAMING_SNAKE_CASE="<|endoftext|>"				, SCREAMING_SNAKE_CASE="<|endoftext|>"				, SCREAMING_SNAKE_CASE="<|endoftext|>"				, SCREAMING_SNAKE_CASE=None				, SCREAMING_SNAKE_CASE=False				, SCREAMING_SNAKE_CASE=False				, **SCREAMING_SNAKE_CASE				, ):



					"""simple docstring"""







					snake_case		:						Optional[int]								=				AddedToken(lowerCAmelCase_				, lstrip=lowerCAmelCase_				, rstrip=lowerCAmelCase_							) if isinstance(lowerCAmelCase_				, lowerCAmelCase_							) else bos_token
					snake_case		:						List[Any]								=				AddedToken(lowerCAmelCase_				, lstrip=lowerCAmelCase_				, rstrip=lowerCAmelCase_							) if isinstance(lowerCAmelCase_				, lowerCAmelCase_							) else eos_token
					snake_case		:						Tuple								=				AddedToken(lowerCAmelCase_				, lstrip=lowerCAmelCase_				, rstrip=lowerCAmelCase_							) if isinstance(lowerCAmelCase_				, lowerCAmelCase_							) else unk_token
					snake_case		:						Union[str, Any]								=				AddedToken(lowerCAmelCase_				, lstrip=lowerCAmelCase_				, rstrip=lowerCAmelCase_							) if isinstance(lowerCAmelCase_				, lowerCAmelCase_							) else pad_token
					super().__init__(
					    errors=lowerCAmelCase_				, unk_token=lowerCAmelCase_				, bos_token=lowerCAmelCase_				, eos_token=lowerCAmelCase_				, pad_token=lowerCAmelCase_				, add_prefix_space=lowerCAmelCase_				, add_bos_token=lowerCAmelCase_				, **lowerCAmelCase_				, )
					snake_case		:						Dict								=				add_bos_token

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

					# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
					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
			def 							lowerCamelCase_				(     self							):



					"""simple docstring"""







					return len(self.encoder							)


			def 							lowerCamelCase_				(     self							):



					"""simple docstring"""







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


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					if token in self.cache:
							return self.cache[token]
					snake_case		:						Dict								=				tuple(lowerCAmelCase_							)
					snake_case		:						Optional[Any]								=				get_pairs(lowerCAmelCase_							)

					if not pairs:
							return token

					while True:
							snake_case		:						Tuple								=				min(lowerCAmelCase_				, key=lambda SCREAMING_SNAKE_CASE							: self.bpe_ranks.get(lowerCAmelCase_				, float("inf"							)							)							)
							if bigram not in self.bpe_ranks:
									break
							snake_case			,				snake_case		:						List[Any]								=				bigram
							snake_case		:						List[Any]								=				[]
							snake_case		:						Optional[Any]								=				0
							while i < len(lowerCAmelCase_							):
									try:
											snake_case		:						Union[str, Any]								=				word.index(lowerCAmelCase_				, lowerCAmelCase_							)
									except ValueError:
											new_word.extend(word[i:]							)
											break
									else:
											new_word.extend(word[i:j]							)
											snake_case		:						int								=				j

									if word[i] == first and i < len(lowerCAmelCase_							) - 1 and word[i + 1] == second:
											new_word.append(first + second							)
											i += 2
									else:
											new_word.append(word[i]							)
											i += 1
							snake_case		:						Dict								=				tuple(lowerCAmelCase_							)
							snake_case		:						Union[str, Any]								=				new_word
							if len(lowerCAmelCase_							) == 1:
									break
							else:
									snake_case		:						Optional[Any]								=				get_pairs(lowerCAmelCase_							)
					snake_case		:						Dict								=				" ".join(lowerCAmelCase_							)
					snake_case		:						int								=				word
					return word


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE=None							):



					"""simple docstring"""







					if self.add_bos_token:
							snake_case		:						Dict								=				[self.bos_token_id]
					else:
							snake_case		:						List[Any]								=				[]

					snake_case		:						Union[str, Any]								=				bos_token_ids + token_ids_a

					if token_ids_a is None:
							return output

					return output + bos_token_ids + token_ids_a


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = False							):



					"""simple docstring"""







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

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

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


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







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


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







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


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					return self.decoder.get(lowerCAmelCase_							)


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					snake_case		:						int								=				"".join(lowerCAmelCase_							)
					snake_case		:						int								=				bytearray([self.byte_decoder[c] for c in text]							).decode("utf-8"				, errors=self.errors							)
					return text


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE = None							):



					"""simple docstring"""







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

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

					snake_case		:						Tuple								=				0
					with open(lowerCAmelCase_				, "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!"							)
											snake_case		:						Any								=				token_index
									writer.write(" ".join(lowerCAmelCase_							) + "\n"							)
									index += 1

					return vocab_file, merge_file


			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE=False				, **SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







					snake_case		:						Optional[Any]								=				kwargs.pop("add_prefix_space"				, self.add_prefix_space							)
					if is_split_into_words or add_prefix_space:
							snake_case		:						List[Any]								=				" " + text
					return (text, kwargs)



			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE							):



					"""simple docstring"""







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







"""simple docstring"""

import textwrap

import pyarrow as pa
import pytest

from datasets import Features, Image
from datasets.packaged_modules.text.text import Text

from ..utils import require_pil



@pytest.fixture
def   UpperCamelCase__     (		lowercase__  :     str   ):
		snake_case		:						List[str]								=				tmp_path / "text.txt"
		snake_case		:						Tuple								=				textwrap.dedent(
		    "\\n        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.\n        Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.\n        Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.\n        Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.\n\n        Second paragraph:\n        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.\n        Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.\n        Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.\n        Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.\n        "   )
		with open(lowercase__       ,     "w"       ,     encoding="utf-8"   ) as f:
				f.write(lowercase__   )
		return str(lowercase__   )



@pytest.fixture
def   UpperCamelCase__     (		lowercase__  :     str       ,     lowercase__  :     Dict   ):
		snake_case		:						Any								=				tmp_path / "text_with_image.txt"
		with open(lowercase__       ,     "w"       ,     encoding="utf-8"   ) as f:
				f.write(lowercase__   )
		return str(lowercase__   )



@pytest.mark.parametrize("keep_linebreaks"       ,     [True, False]   )
def   UpperCamelCase__     (		lowercase__  :     Union[str, Any]       ,     lowercase__  :     Tuple   ):
		with open(lowercase__       ,     encoding="utf-8"   ) as f:
				snake_case		:						List[Any]								=				f.read().splitlines(keepends=lowercase__   )
		snake_case		:						List[Any]								=				Text(keep_linebreaks=lowercase__       ,     encoding="utf-8"   )
		snake_case		:						Optional[int]								=				text._generate_tables([[text_file]]   )
		snake_case		:						int								=				pa.concat_tables([table for _, table in generator]   ).to_pydict()["text"]
		assert generated_content == expected_content



@require_pil
def   UpperCamelCase__     (		lowercase__  :     Optional[Any]   ):
		with open(lowercase__       ,     encoding="utf-8"   ) as f:
				snake_case		:						Union[str, Any]								=				f.read().splitlines()[0]
		snake_case		:						List[Any]								=				Text(encoding="utf-8"       ,     features=Features({"image": Image()}   )   )
		snake_case		:						Tuple								=				text._generate_tables([[text_file_with_image]]   )
		snake_case		:						List[str]								=				pa.concat_tables([table for _, table in generator]   )
		assert pa_table.schema.field("image"   ).type == Image()()
		snake_case		:						Optional[Any]								=				pa_table.to_pydict()["image"]
		assert generated_content == [{"path": image_file, "bytes": None}]



@pytest.mark.parametrize("sample_by"       ,     ["line", "paragraph", "document"]   )
def   UpperCamelCase__     (		lowercase__  :     Union[str, Any]       ,     lowercase__  :     List[str]   ):
		with open(lowercase__       ,     encoding="utf-8"   ) as f:
				snake_case		:						Tuple								=				f.read()
		if sample_by == "line":
				snake_case		:						Optional[int]								=				expected_content.splitlines()
		elif sample_by == "paragraph":
				snake_case		:						str								=				expected_content.split("\n\n"   )
		elif sample_by == "document":
				snake_case		:						List[Any]								=				[expected_content]
		snake_case		:						str								=				Text(sample_by=lowercase__       ,     encoding="utf-8"       ,     chunksize=100   )
		snake_case		:						Optional[int]								=				text._generate_tables([[text_file]]   )
		snake_case		:						Optional[Any]								=				pa.concat_tables([table for _, table in generator]   ).to_pydict()["text"]
		assert generated_content == expected_content

