



"""simple docstring"""




import collections
import json
import os
import re
from typing import TYPE_CHECKING, List, Optional, Tuple

import numpy as np

from ...tokenization_utils_fast import PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
 from transformers.pipelines.conversational import Conversation


_lowerCAmelCase    : Optional[int]							=							logging.get_logger(__name__)

_lowerCAmelCase    : Optional[Any]							=							{"vocab_file": "vocab.txt", "emoji_file": "emoji.json"}

_lowerCAmelCase    : str							=							{
    "vocab_file": {
        "abeja/gpt-neox-japanese-2.7b": "https://huggingface.co/abeja/gpt-neox-japanese-2.7b/resolve/main/vocab.txt",
    },
    "emoji_file": {
        "abeja/gpt-neox-japanese-2.7b": "https://huggingface.co/abeja/gpt-neox-japanese-2.7b/resolve/main/emoji.json",
    },
}

_lowerCAmelCase    : Union[str, Any]							=							{
    "abeja/gpt-neox-japanese-2.7b": 20_48,
}

def 						__snake_case						(				SCREAMING_SNAKE_CASE__     :						str						,  SCREAMING_SNAKE_CASE__     :						int    )  ->       Any:





     '''simple docstring'''

     with open(SCREAMING_SNAKE_CASE__						,  "r"						,  encoding="utf-8"    ) as f:
          _UpperCAmelCase      :     str        =   json.loads(f.read()    )

     _UpperCAmelCase      :     List[Any]        =   collections.OrderedDict()
     _UpperCAmelCase      :     Optional[Any]        =   collections.OrderedDict()
     _UpperCAmelCase      :     int        =   collections.OrderedDict()
     with open(SCREAMING_SNAKE_CASE__						,  "r"						,  encoding="utf-8"    ) as f:
          _UpperCAmelCase      :     List[Any]        =   f.readlines()
     _UpperCAmelCase      :     str        =   [[t.rstrip("\n"    )] if (t == "," or "," not in t) else t.rstrip("\n"    ).split(","    ) for t in token]
     for idx, b in enumerate(SCREAMING_SNAKE_CASE__    ):
          _UpperCAmelCase      :     Dict        =   b
          _UpperCAmelCase      :     Any        =   idx
          for wd in b:
               _UpperCAmelCase      :     List[Any]        =   idx

     return vocab, raw_vocab, ids_to_tokens, emoji






class      UpperCAmelCase_  (    _UpperCamelCase						):
      __SCREAMING_SNAKE_CASE						:       int								=			VOCAB_FILES_NAMES
      __SCREAMING_SNAKE_CASE						:       str								=			PRETRAINED_VOCAB_FILES_MAP
      __SCREAMING_SNAKE_CASE						:       int								=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      __SCREAMING_SNAKE_CASE						:       Tuple								=			['input_ids', 'attention_mask']



      def __init__(					self					:     List[Any]					,     A					:     Union[str, Any]					,     A					:     Any					,     A					:     Dict="<|endoftext|>"					,     A					:     Tuple="<|endoftext|>"					,     A					:     List[Any]="<|startoftext|>"					,     A					:     str="<|endoftext|>"					,     A					:     Optional[int]=False					,     **A					:     str					,     ):
           super().__init__(
               unk_token=A					,     pad_token=A					,     bos_token=A					,     eos_token=A					,     do_clean_text=A					,     **A					,     )
           if not os.path.isfile(A						):
                raise ValueError(
                    f'Can\'t find a vocabulary file at path \'{vocab_file}\'. To load the vocabulary from a Google pretrained'
                    " model use `tokenizer = GPTNeoXJapaneseokenizer.from_pretrained(PRETRAINED_MODEL_NAME)`"						)
           if not os.path.isfile(A						):
                raise ValueError(
                    f'Can\'t find a emoji file at path \'{emoji_file}\'. To load the emoji information from a Google'
                    " pretrained model use `tokenizer = GPTNeoXJapaneseokenizer.from_pretrained(PRETRAINED_MODEL_NAME)`"						)
           _UpperCAmelCase      :     List[str]        =   do_clean_text
           _UpperCAmelCase    ,	_UpperCAmelCase    ,	_UpperCAmelCase    ,	_UpperCAmelCase      :     Optional[Any]        =   load_vocab_and_emoji(A					,     A						)
           _UpperCAmelCase      :     Any        =   SubWordJapaneseTokenizer(
               vocab=self.vocab					,     ids_to_tokens=self.ids_to_tokens					,     emoji=self.emoji						)



      @property
      def 						snake_case_ (					self					:     Any						):
           # self.vocab contains support for character fluctuation unique to Japanese, and has a large number of vocab
           return len(self.raw_vocab						)



      def 						snake_case_ (					self					:     Tuple						):
           return dict(self.raw_vocab					,     **self.added_tokens_encoder						)



      def 						snake_case_ (					self					:     Union[str, Any]					,     A					:     int						):
           return self.subword_tokenizer.tokenize(A					,     clean=self.do_clean_text						)



      def 						snake_case_ (					self					:     Any					,     A					:     int						):
           return self.vocab.get(A					,     self.vocab.get(self.unk_token						)						)



      def 						snake_case_ (					self					:     Optional[Any]					,     A					:     Union[str, Any]						):
           return self.subword_tokenizer.convert_id_to_token(A						)



      def 						snake_case_ (					self					:     Any					,     A					:     List[Any]						):
           _UpperCAmelCase      :     List[Any]        =   "".join(A						).strip()
           return out_string



      def 						snake_case_ (					self					:     Dict					,     A					:     "Conversation"						):
           _UpperCAmelCase      :     Tuple        =   []
           for is_user, text in conversation.iter_texts():
                input_ids.extend(self.encode(A					,     add_special_tokens=A						) + [self.eos_token_id]						)

           if len(A						) > self.model_max_length:
                _UpperCAmelCase      :     Tuple        =   input_ids[-self.model_max_length :]
           return input_ids



      def 						snake_case_ (					self					:     List[Any]					,     A					:     str					,     A					:     Optional[str] = None						):
           _UpperCAmelCase      :     int        =   0
           if os.path.isdir(A						):
                _UpperCAmelCase      :     List[Any]        =   os.path.join(
                    A					,     (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]						)
                _UpperCAmelCase      :     int        =   os.path.join(
                    A					,     (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["emoji_file"]						)
           else:
                _UpperCAmelCase      :     Any        =   (
                    (filename_prefix + "-" if filename_prefix else "") + save_directory + VOCAB_FILES_NAMES["vocab_file"]
                )
                _UpperCAmelCase      :     int        =   (
                    (filename_prefix + "-" if filename_prefix else "") + save_directory + VOCAB_FILES_NAMES["emoji_file"]
                )
           with open(A					,     "w"					,     encoding="utf-8"						) as writer:
                for token_index, token in self.ids_to_tokens.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!"						)
                          _UpperCAmelCase      :     str        =   token_index
                     writer.write(",".join(A						) + "\n"						)
                     index += 1
           with open(A					,     "w"					,     encoding="utf-8"						) as writer:
                json.dump(self.emoji					,     A						)
           return vocab_file, emoji_file







class      UpperCAmelCase_  (    _UpperCamelCase						):



      def __init__(					self					:     Optional[int]					,     A					:     Tuple					,     A					:     Dict					,     A					:     Optional[Any]						):
           _UpperCAmelCase      :     Any        =   vocab  # same as swe
           _UpperCAmelCase      :     Optional[int]        =   ids_to_tokens  # same as bpe
           _UpperCAmelCase      :     Tuple        =   emoji
           _UpperCAmelCase      :     Dict        =   np.max([len(A						) for w in self.vocab.keys()]						)
           _UpperCAmelCase      :     int        =   re.compile(R"(https?|ftp)(:\/\/[-_\.!~*\'()a-zA-Z0-9;\/?:\@&=\+$,%#]+)"						)
           _UpperCAmelCase      :     str        =   re.compile(R"[A-Za-z0-9\._+]*@[\-_0-9A-Za-z]+(\.[A-Za-z]+)*"						)
           _UpperCAmelCase      :     Optional[int]        =   re.compile(R"[\(]{0,1}[0-9]{2,4}[\)\-\(]{0,1}[0-9]{2,4}[\)\-]{0,1}[0-9]{3,4}"						)
           _UpperCAmelCase      :     Optional[int]        =   re.compile(
               R"([12]\d{3}[/\-年])*(0?[1-9]|1[0-2])[/\-月]((0?[1-9]|[12][0-9]|3[01])日?)*(\d{1,2}|:|\d{1,2}時|\d{1,2}分|\(日\)|\(月\)|\(火\)|\(水\)|\(木\)|\(金\)|\(土\)|㈰|㈪|㈫|㈬|㈭|㈮|㈯)*"						)
           _UpperCAmelCase      :     Optional[int]        =   re.compile(
               R"(明治|大正|昭和|平成|令和|㍾|㍽|㍼|㍻|\u32ff)\d{1,2}年(0?[1-9]|1[0-2])月(0?[1-9]|[12][0-9]|3[01])日(\d{1,2}|:|\d{1,2}時|\d{1,2}分|\(日\)|\(月\)|\(火\)|\(水\)|\(木\)|\(金\)|\(土\)|㈰|㈪|㈫|㈬|㈭|㈮|㈯)*"						)
           _UpperCAmelCase      :     Optional[int]        =   re.compile(
               R"((0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*億)*((0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*万)*((0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*千)*(0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*(千円|万円|千万円|円|千ドル|万ドル|千万ドル|ドル|千ユーロ|万ユーロ|千万ユーロ|ユーロ)+(\(税込\)|\(税抜\)|\+tax)*"						)
           _UpperCAmelCase      :     Optional[int]        =   "─━│┃┄┅┆┇┈┉┊┋┌┍┎┏┐┑┒┓└┕┖┗┘┙┚┛├┝┞┟┠┡┢┣┤┥┦┧┨┩┪┫┬┭┮┯┰┱┲┳┴┵┶┷┸┹┺┻┼┽┾┿╀╁╂╃╄╅╆╇╈╉╊╋╌╍╎╏═║╒╓╔╕╖╗╘╙╚╛╜╝╞╟╠╡╢╣╤╥╦╧╨╩╪╫╬╭╮╯╰╱╲╳╴╵╶╷╸╹╺╻╼╽╾╿"
           _UpperCAmelCase      :     Union[str, Any]        =   "▀▁▂▃▄▅▆▇█▉▊▋▌▍▎▏▐░▒▓▔▕▖▗▘▙▚▛▜▝▞▟"
           _UpperCAmelCase      :     Dict        =   str.maketrans({k: "<BLOCK>" for k in keisen + blocks}						)



      def __len__(					self					:     Union[str, Any]						):
           return len(self.ids_to_tokens						)



      def 						snake_case_ (					self					:     Optional[Any]					,     A					:     Optional[Any]						):
           _UpperCAmelCase      :     List[Any]        =   self.content_repattera.sub("<URL>"					,     A						)
           _UpperCAmelCase      :     List[Any]        =   self.content_repattera.sub("<EMAIL>"					,     A						)
           _UpperCAmelCase      :     str        =   self.content_repattera.sub("<TEL>"					,     A						)
           _UpperCAmelCase      :     Optional[Any]        =   self.content_repattera.sub("<DATE>"					,     A						)
           _UpperCAmelCase      :     str        =   self.content_repattera.sub("<DATE>"					,     A						)
           _UpperCAmelCase      :     Optional[Any]        =   self.content_repattera.sub("<PRICE>"					,     A						)
           _UpperCAmelCase      :     int        =   content.translate(self.content_transa						)
           while "<BLOCK><BLOCK>" in content:
                _UpperCAmelCase      :     int        =   content.replace("<BLOCK><BLOCK>"					,     "<BLOCK>"						)
           return content



      def 						snake_case_ (					self					:     Optional[int]					,     A					:     Union[str, Any]					,     A					:     List[str]=False						):
           _UpperCAmelCase      :     Union[str, Any]        =   text.replace(" "					,     "<SP>"						)
           _UpperCAmelCase      :     Optional[Any]        =   text.replace("　"					,     "<SP>"						)
           _UpperCAmelCase      :     List[str]        =   text.replace("\r\n"					,     "<BR>"						)
           _UpperCAmelCase      :     Optional[Any]        =   text.replace("\n"					,     "<BR>"						)
           _UpperCAmelCase      :     int        =   text.replace("\r"					,     "<BR>"						)
           _UpperCAmelCase      :     int        =   text.replace("\t"					,     "<TAB>"						)
           _UpperCAmelCase      :     List[Any]        =   text.replace("—"					,     "ー"						)
           _UpperCAmelCase      :     Union[str, Any]        =   text.replace("−"					,     "ー"						)
           for k, v in self.emoji["emoji"].items():
                if k in text:
                     _UpperCAmelCase      :     Union[str, Any]        =   text.replace(A					,     A						)
           if clean:
                _UpperCAmelCase      :     Any        =   self.clean_text(A						)

           def check_simbol(A					:     Optional[int]						):
                _UpperCAmelCase      :     Optional[Any]        =   x.encode()
                if len(A						) == 1 and len(A						) == 2:
                     _UpperCAmelCase      :     List[Any]        =   (int(e[0]						) << 8) + int(e[1]						)
                     if (
                         (c >= 0XC_2A1 and c <= 0XC_2BF)
                         or (c >= 0XC_780 and c <= 0XC_783)
                         or (c >= 0XC_AB9 and c <= 0XC_BBF)
                         or (c >= 0XC_C80 and c <= 0XC_DA2)
                     ):
                          return True
                return False

           def checkuae(A					:     List[str]						):
                _UpperCAmelCase      :     Union[str, Any]        =   x.encode()
                if len(A						) == 1 and len(A						) == 3:
                     _UpperCAmelCase      :     str        =   (int(e[0]						) << 1_6) + (int(e[1]						) << 8) + int(e[2]						)
                     if c >= 0XE28_080 and c <= 0XE2B_07F:
                          return True
                return False

           _UpperCAmelCase      :     str        =   0
           _UpperCAmelCase      :     Optional[Any]        =   []
           while pos < len(A						):
                _UpperCAmelCase      :     Any        =   min(len(A						)					,     pos + self.maxlen + 1						) if text[pos] == "<" else pos + 3
                _UpperCAmelCase      :     Dict        =   []  # (token_id, token, pos)
                for e in range(A					,     A					,     -1						):
                     _UpperCAmelCase      :     Tuple        =   text[pos:e]
                     if wd in self.vocab:
                          if wd[0] == "<" and len(A						) > 2:
                               _UpperCAmelCase      :     int        =   [(self.vocab[wd], wd, e)]
                               break
                          else:
                               candidates.append((self.vocab[wd], wd, e)						)
                if len(A						) > 0:
                     # the smallest token_id is adopted
                     _UpperCAmelCase    ,	_UpperCAmelCase    ,	_UpperCAmelCase      :     Optional[int]        =   sorted(A					,     key=lambda A						: x[0]						)[0]
                     result.append(A						)
                     _UpperCAmelCase      :     Optional[Any]        =   e
                else:
                     _UpperCAmelCase      :     Union[str, Any]        =   pos + 1
                     _UpperCAmelCase      :     Dict        =   text[pos:end]
                     if check_simbol(A						):
                          result.append("<KIGOU>"						)
                     elif checkuae(A						):
                          result.append("<U2000U2BFF>"						)
                     else:
                          for i in wd.encode("utf-8"						):
                               result.append("<|byte%d|>" % i						)
                     _UpperCAmelCase      :     str        =   end
           return result



      def 						snake_case_ (					self					:     str					,     A					:     Dict					,     A					:     Union[str, Any]="\n"						):
           _UpperCAmelCase      :     int        =   []
           _UpperCAmelCase      :     int        =   []
           _UpperCAmelCase      :     Optional[Any]        =   self.ids_to_tokens[index][0]
           if word[:6] == "<|byte" and word[-2:] == "|>":
                byte_tokens.append(int(word[6:-2]						)						)
           else:
                if len(A						) > 0:
                     words.append(bytearray(A						).decode("utf-8"					,     errors="replace"						)						)
                     _UpperCAmelCase      :     Any        =   []
                if word[:7] == "<|emoji" and word[-2:] == "|>":
                     words.append(self.emoji["emoji_inv"][word]						)
                elif word == "<SP>":
                     words.append(" "						)
                elif word == "<BR>":
                     words.append(A						)
                elif word == "<TAB>":
                     words.append("\t"						)
                elif word == "<BLOCK>":
                     words.append("▀"						)
                elif word == "<KIGOU>":
                     words.append("ǀ"						)
                elif word == "<U2000U2BFF>":
                     words.append("‖"						)
                else:
                     words.append(A						)
           if len(A						) > 0:
                words.append(bytearray(A						).decode("utf-8"					,     errors="replace"						)						)
           _UpperCAmelCase      :     str        =   "".join(A						)
           return text





"""simple docstring"""




# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re

from ..utils import cached_file


# docstyle-ignore
_lowerCAmelCase    : Any							=							"\nHuman: <<task>>\n\nAssistant: "


_lowerCAmelCase    : str							=							"huggingface-tools/default-prompts"
_lowerCAmelCase    : Union[str, Any]							=							{"chat": "chat_prompt_template.txt", "run": "run_prompt_template.txt"}

def 						__snake_case						(				SCREAMING_SNAKE_CASE__     :						Union[str, Any]						,  SCREAMING_SNAKE_CASE__     :						Optional[int]						,  SCREAMING_SNAKE_CASE__     :						int="run"    )  ->       int:





     '''simple docstring'''

     if prompt_or_repo_id is None:
          _UpperCAmelCase      :     Optional[int]        =   DEFAULT_PROMPTS_REPO

     # prompt is considered a repo ID when it does not contain any kind of space
     if re.search("\\s"						,  SCREAMING_SNAKE_CASE__    ) is not None:
          return prompt_or_repo_id

     _UpperCAmelCase      :     Dict        =   cached_file(
         SCREAMING_SNAKE_CASE__						,  PROMPT_FILES[mode]						,  repo_type="dataset"						,  user_agent={"agent": agent_name}    )
     with open(SCREAMING_SNAKE_CASE__						,  "r"						,  encoding="utf-8"    ) as f:
          return f.read()
