import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
      import jieba

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


A   :     Tuple       =				logging.get_logger(__name__)

A   :     Optional[Any]       =				{"vocab_file": "vocab.txt"}

A   :     Tuple       =				{
    "vocab_file": {
        "openbmb/cpm-ant-10b": "https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt",
    },
}

A   :     int       =				{
    "openbmb/cpm-ant-10b": 10_24,
}


def    a__       (						__UpperCamelCase ):
      SCREAMING_SNAKE_CASE_      =     collections.OrderedDict()
      with open(__UpperCamelCase   ,					"r"   ,					encoding="utf-8" ) as reader:
            SCREAMING_SNAKE_CASE_      =     reader.readlines()
      for index, token in enumerate(__UpperCamelCase ):
            SCREAMING_SNAKE_CASE_      =     token.rstrip("\n" )
            SCREAMING_SNAKE_CASE_      =     index
      return vocab


class      lowerCamelCase    (SCREAMING_SNAKE_CASE__    ):
  """simple docstring"""







  def __init__(							self       :				str     ,   __magic_name__       :				Optional[int]     ,   __magic_name__       :				int="<unk>"     ,   __magic_name__       :				List[str]=200      )     ->		int:
        SCREAMING_SNAKE_CASE_      =     vocab
        SCREAMING_SNAKE_CASE_      =     unk_token
        SCREAMING_SNAKE_CASE_      =     max_input_chars_per_word







  def       __A   (							self       :				str     ,   __magic_name__       :				Any      )     ->		List[Any]:
        SCREAMING_SNAKE_CASE_      =     list(__magic_name__      )
        if len(__magic_name__      ) > self.max_input_chars_per_word:
              return [self.unk_token]

        SCREAMING_SNAKE_CASE_      =     0
        SCREAMING_SNAKE_CASE_      =     []
        while start < len(__magic_name__      ):
              SCREAMING_SNAKE_CASE_      =     len(__magic_name__      )
              SCREAMING_SNAKE_CASE_      =     None
              while start < end:
                    SCREAMING_SNAKE_CASE_      =     "".join(chars[start:end]      )
                    if substr in self.vocab:
                          SCREAMING_SNAKE_CASE_      =     substr
                          break
                    end -= 1
              if cur_substr is None:
                    sub_tokens.append(self.unk_token      )
                    start += 1
              else:
                    sub_tokens.append(__magic_name__      )
                    SCREAMING_SNAKE_CASE_      =     end

        return sub_tokens



class      lowerCamelCase    (SCREAMING_SNAKE_CASE__    ):
  """simple docstring"""







  lowerCamelCase__   =  VOCAB_FILES_NAMES
  lowerCamelCase__   =  PRETRAINED_VOCAB_FILES_MAP
  lowerCamelCase__   =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  lowerCamelCase__   =  ['''input_ids''', '''attention_mask''']
  lowerCamelCase__   =  False







  def __init__(							self       :				str     ,   __magic_name__       :				Dict     ,   __magic_name__       :				Optional[int]="<d>"     ,   __magic_name__       :				Optional[Any]="</d>"     ,   __magic_name__       :				str="<s>"     ,   __magic_name__       :				List[str]="</s>"     ,   __magic_name__       :				str="<pad>"     ,   __magic_name__       :				List[str]="<unk>"     ,   __magic_name__       :				List[str]="</n>"     ,   __magic_name__       :				Optional[int]="</_>"     ,   __magic_name__       :				str="left"     ,   **__magic_name__       :				Any     ,   )     ->		Any:
        requires_backends(self     ,   ["jieba"]      )
        super().__init__(
            bod_token=__magic_name__     ,   eod_token=__magic_name__     ,   bos_token=__magic_name__     ,   eos_token=__magic_name__     ,   pad_token=__magic_name__     ,   unk_token=__magic_name__     ,   line_token=__magic_name__     ,   space_token=__magic_name__     ,   padding_side=__magic_name__     ,   **__magic_name__     ,   )
        SCREAMING_SNAKE_CASE_      =     bod_token
        SCREAMING_SNAKE_CASE_      =     eod_token
        SCREAMING_SNAKE_CASE_      =     load_vocab(__magic_name__      )
        SCREAMING_SNAKE_CASE_      =     self.encoder[space_token]
        SCREAMING_SNAKE_CASE_      =     self.encoder[line_token]

        del self.encoder[space_token]
        del self.encoder[line_token]

        SCREAMING_SNAKE_CASE_      =     collections.OrderedDict(sorted(self.encoder.items()     ,   key=lambda __magic_name__      : x[1]      )      )
        SCREAMING_SNAKE_CASE_      =     {v: k for k, v in self.encoder.items()}

        SCREAMING_SNAKE_CASE_      =     WordpieceTokenizer(vocab=self.encoder     ,   unk_token=self.unk_token      )







  @property
  def       __A   (							self       :				Tuple      )     ->		int:
        return self.encoder[self.bod_token]







  @property
  def       __A   (							self       :				Dict      )     ->		List[str]:
        return self.encoder[self.eod_token]







  @property
  def       __A   (							self       :				List[Any]      )     ->		Dict:
        return self.encoder["\n"]







  @property
  def       __A   (							self       :				Optional[int]      )     ->		int:
        return len(self.encoder      )







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







  def       __A   (							self       :				str     ,   __magic_name__       :				Optional[int]      )     ->		Dict:
        SCREAMING_SNAKE_CASE_      =     []
        for x in jieba.cut(__magic_name__     ,   cut_all=__magic_name__      ):
              output_tokens.extend(self.wordpiece_tokenizer.tokenize(__magic_name__      )      )
        return output_tokens







  def       __A   (							self       :				List[str]     ,   __magic_name__       :				Dict     ,   **__magic_name__       :				Dict      )     ->		Any:
        SCREAMING_SNAKE_CASE_      =     [i for i in token_ids if i >= 0]
        SCREAMING_SNAKE_CASE_      =     [
            x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
        ]
        return super()._decode(__magic_name__     ,   **__magic_name__      )







  def       __A   (							self       :				Dict     ,   __magic_name__       :				List[str]      )     ->		str:
        return token in self.encoder







  def       __A   (							self       :				Union[str, Any]     ,   __magic_name__       :				List[str]      )     ->		str:
        return "".join(__magic_name__      )







  def       __A   (							self       :				Optional[Any]     ,   __magic_name__       :				List[str]      )     ->		Optional[int]:
        return self.encoder.get(__magic_name__     ,   self.encoder.get(self.unk_token      )      )







  def       __A   (							self       :				List[Any]     ,   __magic_name__       :				str      )     ->		int:
        return self.decoder.get(__magic_name__     ,   self.unk_token      )







  def       __A   (							self       :				List[Any]     ,   __magic_name__       :				str     ,   __magic_name__       :				Optional[str] = None      )     ->		Tuple[str]:
        if os.path.isdir(__magic_name__      ):
              SCREAMING_SNAKE_CASE_      =     os.path.join(
                  __magic_name__     ,   (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]      )
        else:
              SCREAMING_SNAKE_CASE_      =     (filename_prefix + "-" if filename_prefix else "") + save_directory
        SCREAMING_SNAKE_CASE_      =     0
        if " " in self.encoder:
              SCREAMING_SNAKE_CASE_      =     self.encoder[" "]
              del self.encoder[" "]
        if "\n" in self.encoder:
              SCREAMING_SNAKE_CASE_      =     self.encoder["\n"]
              del self.encoder["\n"]
        SCREAMING_SNAKE_CASE_      =     collections.OrderedDict(sorted(self.encoder.items()     ,   key=lambda __magic_name__      : x[1]      )      )
        with open(__magic_name__     ,   "w"     ,   encoding="utf-8"      ) as writer:
              for token, token_index in self.encoder.items():
                    if index != token_index:
                          logger.warning(
                              F'''Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive.'''
                              " Please check that the vocabulary is not corrupted!"      )
                          SCREAMING_SNAKE_CASE_      =     token_index
                    writer.write(token + "\n"      )
                    index += 1
        return (vocab_file,)







  def       __A   (							self       :				List[str]     ,   __magic_name__       :				List[int]     ,   __magic_name__       :				List[int] = None      )     ->		List[int]:
        if token_ids_a is None:
              return [self.bos_token_id] + token_ids_a
        return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a







  def       __A   (							self       :				str     ,   __magic_name__       :				List[int]     ,   __magic_name__       :				Optional[List[int]] = None     ,   __magic_name__       :				bool = False      )     ->		List[int]:

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

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



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

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

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


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

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


A   :     Tuple       =				get_tests_dir("fixtures")


class      lowerCamelCase    (unittest.TestCase    ):
  """simple docstring"""







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

        # Download this model to make sure it's in the cache.
        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"      )
        # Under the mock environment we get a 500 error when trying to reach the model.
        with mock.patch("requests.Session.request"     ,   return_value=__magic_name__      ) as mock_head:
              SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"      )
              # This check we did call the fake head request
              mock_head.assert_called()







  def       __A   (							self       :				Optional[int]      )     ->		Tuple:
        # This test is for deprecated behavior and can be removed in v5
        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained(
            "https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"      )



@is_staging_test
class      lowerCamelCase    (unittest.TestCase    ):
  """simple docstring"""







  @classmethod
  def       __A   (							cls       :				Any      )     ->		Union[str, Any]:
        SCREAMING_SNAKE_CASE_      =     TOKEN
        HfFolder.save_token(__magic_name__      )







  @classmethod
  def       __A   (							cls       :				int      )     ->		Optional[Any]:
        try:
              delete_repo(token=cls._token     ,   repo_id="test-feature-extractor"      )
        except HTTPError:
              pass

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

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







  def       __A   (							self       :				str      )     ->		List[Any]:
        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained(__magic_name__      )
        feature_extractor.push_to_hub("test-feature-extractor"     ,   use_auth_token=self._token      )

        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''      )
        for k, v in feature_extractor.__dict__.items():
              self.assertEqual(__magic_name__     ,   getattr(__magic_name__     ,   __magic_name__      )      )

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

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

        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''      )
        for k, v in feature_extractor.__dict__.items():
              self.assertEqual(__magic_name__     ,   getattr(__magic_name__     ,   __magic_name__      )      )







  def       __A   (							self       :				Dict      )     ->		str:
        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained(__magic_name__      )
        feature_extractor.push_to_hub("valid_org/test-feature-extractor"     ,   use_auth_token=self._token      )

        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor"      )
        for k, v in feature_extractor.__dict__.items():
              self.assertEqual(__magic_name__     ,   getattr(__magic_name__     ,   __magic_name__      )      )

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

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

        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor-org"      )
        for k, v in feature_extractor.__dict__.items():
              self.assertEqual(__magic_name__     ,   getattr(__magic_name__     ,   __magic_name__      )      )







  def       __A   (							self       :				List[Any]      )     ->		Dict:
        CustomFeatureExtractor.register_for_auto_class()
        SCREAMING_SNAKE_CASE_      =     CustomFeatureExtractor.from_pretrained(__magic_name__      )

        feature_extractor.push_to_hub("test-dynamic-feature-extractor"     ,   use_auth_token=self._token      )

        # This has added the proper auto_map field to the config
        self.assertDictEqual(
            feature_extractor.auto_map     ,   {"AutoFeatureExtractor": "custom_feature_extraction.CustomFeatureExtractor"}     ,   )

        SCREAMING_SNAKE_CASE_      =     AutoFeatureExtractor.from_pretrained(
            F'''{USER}/test-dynamic-feature-extractor'''     ,   trust_remote_code=__magic_name__      )
        # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
        self.assertEqual(new_feature_extractor.__class__.__name__     ,   "CustomFeatureExtractor"      )


