




import os
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from huggingface_hub.file_download import http_get
from requests.exceptions import HTTPError

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


sys.path.append(str(Path(__file__).parent.parent / "utils"))

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
  from test_module.custom_tokenization_fast import CustomTokenizerFast







class 						lowercase							(   unittest.TestCase		):

     def 					a__	(  self				)						->							Any:
           # A mock response for an HTTP head request to emulate server down
           _A							:     Optional[int]			=						mock.Mock()
           _A							:     Optional[Any]			=						500
           _A							:     Dict			=						{}
           _A							:     Union[str, Any]			=						HTTPError
           _A							:     List[Any]			=						{}

           # Download this model to make sure it's in the cache.
           _A							:     int			=						BertTokenizer.from_pretrained("""hf-internal-testing/tiny-random-bert"""				)

           # Under the mock environment we get a 500 error when trying to reach the tokenizer.
           with mock.patch("""requests.Session.request"""		,     return_value=_a				) as mock_head:
                 _A							:     List[Any]			=						BertTokenizer.from_pretrained("""hf-internal-testing/tiny-random-bert"""				)
                 # This check we did call the fake head request
                 mock_head.assert_called()

     @require_tokenizers
     def 					a__	(  self				)						->							Optional[Any]:
           # A mock response for an HTTP head request to emulate server down
           _A							:     str			=						mock.Mock()
           _A							:     Any			=						500
           _A							:     Optional[int]			=						{}
           _A							:     List[str]			=						HTTPError
           _A							:     int			=						{}

           # Download this model to make sure it's in the cache.
           _A							:     int			=						GPTaTokenizerFast.from_pretrained("""gpt2"""				)

           # Under the mock environment we get a 500 error when trying to reach the tokenizer.
           with mock.patch("""requests.Session.request"""		,     return_value=_a				) as mock_head:
                 _A							:     str			=						GPTaTokenizerFast.from_pretrained("""gpt2"""				)
                 # This check we did call the fake head request
                 mock_head.assert_called()

     def 					a__	(  self				)						->							Optional[int]:
           # This test is for deprecated behavior and can be removed in v5
           try:
                 _A							:     Tuple			=						tempfile.mktemp()
                 with open(_a		,     """wb"""				) as f:
                       http_get("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""		,     _a				)

                 _A							:     str			=						AlbertTokenizer.from_pretrained(_a				)
           finally:
                 os.remove(_a				)

           # Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
           # the current folder and have the right name.
           if os.path.isfile("""tokenizer.json"""				):
                 # We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
                 return
           try:
                 with open("""tokenizer.json"""		,     """wb"""				) as f:
                       http_get("""https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json"""		,     _a				)
                 _A							:     Optional[int]			=						AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""				)
                 # The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
                 self.assertEqual(tokenizer.vocab_size		,     1000				)
                 # Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

           finally:
                 os.remove("""tokenizer.json"""				)



     def 					a__	(  self				)						->							Union[str, Any]:
           # This test is for deprecated behavior and can be removed in v5
           _A							:     Dict			=						AlbertTokenizer.from_pretrained("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""				)







@is_staging_test
class 						lowercase							(   unittest.TestCase		):
     _a							 =						["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]

     @classmethod
     def 					a__	(  cls				)						->							Tuple:
           _A							:     int			=						TOKEN
           HfFolder.save_token(_a				)

     @classmethod
     def 					a__	(  cls				)						->							List[Any]:
           try:
                 delete_repo(token=cls._token		,     repo_id="""test-tokenizer"""				)
           except HTTPError:
                 pass

           try:
                 delete_repo(token=cls._token		,     repo_id="""valid_org/test-tokenizer-org"""				)
           except HTTPError:
                 pass

           try:
                 delete_repo(token=cls._token		,     repo_id="""test-dynamic-tokenizer"""				)
           except HTTPError:
                 pass

     def 					a__	(  self				)						->							Dict:
           with tempfile.TemporaryDirectory() as tmp_dir:
                 _A							:     Optional[Any]			=						os.path.join(_a		,     """vocab.txt"""				)
                 with open(_a		,     """w"""		,     encoding="""utf-8"""				) as vocab_writer:
                       vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]				)				)
                 _A							:     Optional[Any]			=						BertTokenizer(_a				)

           tokenizer.push_to_hub("""test-tokenizer"""		,     use_auth_token=self._token				)
           _A							:     List[Any]			=						BertTokenizer.from_pretrained(F'''{USER}/test-tokenizer'''				)
           self.assertDictEqual(new_tokenizer.vocab		,     tokenizer.vocab				)

           # Reset repo
           delete_repo(token=self._token		,     repo_id="""test-tokenizer"""				)

           # Push to hub via save_pretrained
           with tempfile.TemporaryDirectory() as tmp_dir:
                 tokenizer.save_pretrained(_a		,     repo_id="""test-tokenizer"""		,     push_to_hub=_a		,     use_auth_token=self._token				)

           _A							:     List[str]			=						BertTokenizer.from_pretrained(F'''{USER}/test-tokenizer'''				)
           self.assertDictEqual(new_tokenizer.vocab		,     tokenizer.vocab				)

     def 					a__	(  self				)						->							Optional[int]:
           with tempfile.TemporaryDirectory() as tmp_dir:
                 _A							:     Optional[int]			=						os.path.join(_a		,     """vocab.txt"""				)
                 with open(_a		,     """w"""		,     encoding="""utf-8"""				) as vocab_writer:
                       vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]				)				)
                 _A							:     Tuple			=						BertTokenizer(_a				)

           tokenizer.push_to_hub("""valid_org/test-tokenizer-org"""		,     use_auth_token=self._token				)
           _A							:     List[Any]			=						BertTokenizer.from_pretrained("""valid_org/test-tokenizer-org"""				)
           self.assertDictEqual(new_tokenizer.vocab		,     tokenizer.vocab				)

           # Reset repo
           delete_repo(token=self._token		,     repo_id="""valid_org/test-tokenizer-org"""				)

           # Push to hub via save_pretrained
           with tempfile.TemporaryDirectory() as tmp_dir:
                 tokenizer.save_pretrained(
                     _a		,     repo_id="""valid_org/test-tokenizer-org"""		,     push_to_hub=_a		,     use_auth_token=self._token				)

           _A							:     Dict			=						BertTokenizer.from_pretrained("""valid_org/test-tokenizer-org"""				)
           self.assertDictEqual(new_tokenizer.vocab		,     tokenizer.vocab				)



     @require_tokenizers
     def 					a__	(  self				)						->							str:
           CustomTokenizer.register_for_auto_class()
           with tempfile.TemporaryDirectory() as tmp_dir:
                 _A							:     Union[str, Any]			=						os.path.join(_a		,     """vocab.txt"""				)
                 with open(_a		,     """w"""		,     encoding="""utf-8"""				) as vocab_writer:
                       vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]				)				)
                 _A							:     str			=						CustomTokenizer(_a				)

           # No fast custom tokenizer
           tokenizer.push_to_hub("""test-dynamic-tokenizer"""		,     use_auth_token=self._token				)

           _A							:     Optional[int]			=						AutoTokenizer.from_pretrained(F'''{USER}/test-dynamic-tokenizer'''		,     trust_remote_code=_a				)
           # Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
           self.assertEqual(tokenizer.__class__.__name__		,     """CustomTokenizer"""				)

           # Fast and slow custom tokenizer
           CustomTokenizerFast.register_for_auto_class()
           with tempfile.TemporaryDirectory() as tmp_dir:
                 _A							:     Any			=						os.path.join(_a		,     """vocab.txt"""				)
                 with open(_a		,     """w"""		,     encoding="""utf-8"""				) as vocab_writer:
                       vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]				)				)

                 _A							:     Dict			=						BertTokenizerFast.from_pretrained(_a				)
                 bert_tokenizer.save_pretrained(_a				)
                 _A							:     Dict			=						CustomTokenizerFast.from_pretrained(_a				)

           tokenizer.push_to_hub("""test-dynamic-tokenizer"""		,     use_auth_token=self._token				)

           _A							:     Tuple			=						AutoTokenizer.from_pretrained(F'''{USER}/test-dynamic-tokenizer'''		,     trust_remote_code=_a				)
           # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
           self.assertEqual(tokenizer.__class__.__name__		,     """CustomTokenizerFast"""				)
           _A							:     Tuple			=						AutoTokenizer.from_pretrained(
               F'''{USER}/test-dynamic-tokenizer'''		,     use_fast=_a		,     trust_remote_code=_a				)
           # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
           self.assertEqual(tokenizer.__class__.__name__		,     """CustomTokenizer"""				)







class 						lowercase							(   unittest.TestCase		):

     def 					a__	(  self				)						->							List[Any]:
           _A							:     Optional[Any]			=						Trie()
           trie.add("""Hello 友達"""				)
           self.assertEqual(trie.data		,     {"""H""": {"""e""": {"""l""": {"""l""": {"""o""": {""" """: {"""友""": {"""達""": {"""""": 1}}}}}}}}}				)
           trie.add("""Hello"""				)
           trie.data
           self.assertEqual(trie.data		,     {"""H""": {"""e""": {"""l""": {"""l""": {"""o""": {"""""": 1, """ """: {"""友""": {"""達""": {"""""": 1}}}}}}}}}				)

     def 					a__	(  self				)						->							Union[str, Any]:
           _A							:     List[Any]			=						Trie()
           self.assertEqual(trie.split("""[CLS] This is a extra_id_100"""				)		,     ["""[CLS] This is a extra_id_100"""]				)
           trie.add("""[CLS]"""				)
           trie.add("""extra_id_1"""				)
           trie.add("""extra_id_100"""				)
           self.assertEqual(trie.split("""[CLS] This is a extra_id_100"""				)		,     ["""[CLS]""", """ This is a """, """extra_id_100"""]				)

     def 					a__	(  self				)						->							Dict:
           _A							:     List[Any]			=						Trie()
           trie.add("""A"""				)
           self.assertEqual(trie.split("""ABC"""				)		,     ["""A""", """BC"""]				)
           self.assertEqual(trie.split("""BCA"""				)		,     ["""BC""", """A"""]				)

     def 					a__	(  self				)						->							Dict:
           _A							:     Dict			=						Trie()
           trie.add("""TOKEN]"""				)
           trie.add("""[SPECIAL_TOKEN]"""				)
           self.assertEqual(trie.split("""This is something [SPECIAL_TOKEN]"""				)		,     ["""This is something """, """[SPECIAL_TOKEN]"""]				)

     def 					a__	(  self				)						->							List[Any]:
           _A							:     Optional[Any]			=						Trie()
           trie.add("""A"""				)
           trie.add("""P"""				)
           trie.add("""[SPECIAL_TOKEN]"""				)
           self.assertEqual(trie.split("""This is something [SPECIAL_TOKEN]"""				)		,     ["""This is something """, """[SPECIAL_TOKEN]"""]				)

     def 					a__	(  self				)						->							List[Any]:
           _A							:     Dict			=						Trie()
           trie.add("""AB"""				)
           trie.add("""B"""				)
           trie.add("""C"""				)
           self.assertEqual(trie.split("""ABC"""				)		,     ["""AB""", """C"""]				)

     def 					a__	(  self				)						->							int:
           _A							:     Any			=						Trie()
           trie.add("""ABC"""				)
           trie.add("""B"""				)
           trie.add("""CD"""				)
           self.assertEqual(trie.split("""ABCD"""				)		,     ["""ABC""", """D"""]				)



     def 					a__	(  self				)						->							List[Any]:
           # Even if the offsets are wrong, we necessarily output correct string
           # parts.
           _A							:     Tuple			=						Trie()
           _A							:     int			=						trie.cut_text("""ABC"""		,     [0, 0, 2, 1, 2, 3]				)
           self.assertEqual(_a		,     ["""AB""", """C"""]				)







import multiprocessing
from typing import TYPE_CHECKING, Optional, Union

from .. import Dataset, Features, config
from ..formatting import query_table
from ..packaged_modules.sql.sql import Sql
from ..utils import logging
from .abc import AbstractDatasetInputStream


if TYPE_CHECKING:
  import sqlitea

  import sqlalchemy







class 						lowercase							(   UpperCamelCase__		):

     def __init__(  self		,     _a		,     _a		,     _a = None		,     _a = None		,     _a = False		,     **_a		,     )						->							int:
           super().__init__(features=_a		,     cache_dir=_a		,     keep_in_memory=_a		,     **_a				)
           _A							:     Optional[int]			=						Sql(
               cache_dir=_a		,     features=_a		,     sql=_a		,     con=_a		,     **_a		,     )



     def 					a__	(  self				)						->							Optional[Any]:
           _A							:     Tuple			=						None
           _A							:     int			=						None
           _A							:     Tuple			=						None
           _A							:     Union[str, Any]			=						None

           self.builder.download_and_prepare(
               download_config=_a		,     download_mode=_a		,     verification_mode=_a		,     base_path=_a		,     )

           # Build dataset for splits
           _A							:     int			=						self.builder.as_dataset(
               split="""train"""		,     verification_mode=_a		,     in_memory=self.keep_in_memory				)
           return dataset







class 						lowercase							:

     def __init__(  self		,     _a		,     _a		,     _a		,     _a = None		,     _a = None		,     **_a		,     )						->							Union[str, Any]:
           if num_proc is not None and num_proc <= 0:
                 raise ValueError(F'''num_proc {num_proc} must be an integer > 0.'''				)

           _A							:     Dict			=						dataset
           _A							:     int			=						name
           _A							:     Union[str, Any]			=						con
           _A							:     str			=						batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
           _A							:     str			=						num_proc
           _A							:     Optional[Any]			=						to_sql_kwargs

     def 					a__	(  self				)						->							int:
           _A							:     Any			=						self.to_sql_kwargs.pop("""sql"""		,     _a				)
           _A							:     List[str]			=						self.to_sql_kwargs.pop("""con"""		,     _a				)
           _A							:     int			=						self.to_sql_kwargs.pop("""index"""		,     _a				)

           _A							:     List[str]			=						self._write(index=_a		,     **self.to_sql_kwargs				)
           return written

     def 					a__	(  self		,     _a				)						->							Optional[int]:
           _A			,   _A			,   _A							:     List[str]			=						args
           _A							:     int			=						{**to_sql_kwargs, """if_exists""": """append"""} if offset > 0 else to_sql_kwargs
           _A							:     str			=						query_table(
               table=self.dataset.data		,     key=slice(_a		,     offset + self.batch_size				)		,     indices=self.dataset._indices		,     )
           _A							:     Tuple			=						batch.to_pandas()
           _A							:     Union[str, Any]			=						df.to_sql(self.name		,     self.con		,     index=_a		,     **_a				)
           return num_rows or len(_a				)



     def 					a__	(  self		,     _a		,     **_a				)						->							int:
           _A							:     Any			=						0

           if self.num_proc is None or self.num_proc == 1:
                 for offset in logging.tqdm(
                     range(0		,     len(self.dataset				)		,     self.batch_size				)		,     unit="""ba"""		,     disable=not logging.is_progress_bar_enabled()		,     desc="""Creating SQL from Arrow format"""		,     ):
                       written += self._batch_sql((offset, index, to_sql_kwargs)				)
           else:
                 _A			,   _A							:     Tuple			=						len(self.dataset				), self.batch_size
                 with multiprocessing.Pool(self.num_proc				) as pool:
                       for num_rows in logging.tqdm(
                           pool.imap(
                               self._batch_sql		,     [(offset, index, to_sql_kwargs) for offset in range(0		,     _a		,     _a				)]		,     )		,     total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size		,     unit="""ba"""		,     disable=not logging.is_progress_bar_enabled()		,     desc="""Creating SQL from Arrow format"""		,     ):
                             written += num_rows

           return written

