


'''simple docstring'''


import html

from ...feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from ...utils import is_bsa_available, logging, requires_backends


if is_bsa_available():
				import bsa
				from bsa import BeautifulSoup


UpperCAmelCase_						:  int				  =						logging.get_logger(__name__)





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							def __init__(       self       ,   **__snake_case    ):
											requires_backends(self       ,   ["""bs4"""]    )
											super().__init__(**__snake_case    )






							def UpperCAmelCase_      (       self       ,   __snake_case    ):
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= []
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= []
											_SCREAMING_SNAKE_CASE					:			str							= element if element.name else element.parent
											for parent in child.parents:  # type: bs4.element.Tag
															_SCREAMING_SNAKE_CASE					:			Any							= parent.find_all(child.name       ,   recursive=__snake_case    )
															xpath_tags.append(child.name    )
															xpath_subscripts.append(
															    0 if 1 == len(__snake_case    ) else next(i for i, s in enumerate(__snake_case       ,   1    ) if s is child    )    )
															_SCREAMING_SNAKE_CASE					:			List[Any]							= parent
											xpath_tags.reverse()
											xpath_subscripts.reverse()
											return xpath_tags, xpath_subscripts






							def UpperCAmelCase_      (       self       ,   __snake_case    ):
											_SCREAMING_SNAKE_CASE					:			str							= BeautifulSoup(__snake_case       ,   """html.parser"""    )

											_SCREAMING_SNAKE_CASE					:			str							= []
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= []
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= []

											for element in html_code.descendants:
															if type(__snake_case    ) == bsa.element.NavigableString:
																			if type(element.parent    ) != bsa.element.Tag:
																							continue

																			_SCREAMING_SNAKE_CASE					:			List[str]							= html.unescape(__snake_case    ).strip()
																			if not text_in_this_tag:
																							continue

																			all_doc_strings.append(__snake_case    )

																			_SCREAMING_SNAKE_CASE						,			_SCREAMING_SNAKE_CASE					:			Dict							= self.xpath_soup(__snake_case    )
																			stringaxtag_seq.append(__snake_case    )
																			stringaxsubs_seq.append(__snake_case    )

											if len(__snake_case    ) != len(__snake_case    ):
															raise ValueError("""Number of doc strings and xtags does not correspond"""    )
											if len(__snake_case    ) != len(__snake_case    ):
															raise ValueError("""Number of doc strings and xsubs does not correspond"""    )

											return all_doc_strings, stringaxtag_seq, stringaxsubs_seq






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case    ):
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= """"""
											for tagname, subs in zip(__snake_case       ,   __snake_case    ):
															xpath += f"""/{tagname}"""
															if subs != 0:
																			xpath += f"""[{subs}]"""
											return xpath






							def __call__(       self       ,   __snake_case    ):
											_SCREAMING_SNAKE_CASE					:			List[Any]							= False

											# Check that strings has a valid type
											if isinstance(__snake_case       ,   __snake_case    ):
															_SCREAMING_SNAKE_CASE					:			Dict							= True
											elif isinstance(__snake_case       ,   (list, tuple)    ):
															if len(__snake_case    ) == 0 or isinstance(html_strings[0]       ,   __snake_case    ):
																			_SCREAMING_SNAKE_CASE					:			Optional[Any]							= True

											if not valid_strings:
															raise ValueError(
															    """HTML strings must of type `str`, `List[str]` (batch of examples), """
															    f"""but is of type {type(__snake_case    )}."""    )

											_SCREAMING_SNAKE_CASE					:			Optional[int]							= bool(isinstance(__snake_case       ,   (list, tuple)    ) and (isinstance(html_strings[0]       ,   __snake_case    ))    )

											if not is_batched:
															_SCREAMING_SNAKE_CASE					:			str							= [html_strings]

											# Get nodes + xpaths
											_SCREAMING_SNAKE_CASE					:			Tuple							= []
											_SCREAMING_SNAKE_CASE					:			Dict							= []
											for html_string in html_strings:
															_SCREAMING_SNAKE_CASE						,			_SCREAMING_SNAKE_CASE						,			_SCREAMING_SNAKE_CASE					:			int							= self.get_three_from_single(__snake_case    )
															nodes.append(__snake_case    )
															_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= []
															for node, tag_list, sub_list in zip(__snake_case       ,   __snake_case       ,   __snake_case    ):
																			_SCREAMING_SNAKE_CASE					:			List[Any]							= self.construct_xpath(__snake_case       ,   __snake_case    )
																			xpath_strings.append(__snake_case    )
															xpaths.append(__snake_case    )

											# return as Dict
											_SCREAMING_SNAKE_CASE					:			Tuple							= {"""nodes""": nodes, """xpaths""": xpaths}
											_SCREAMING_SNAKE_CASE					:			Tuple							= BatchFeature(data=__snake_case       ,   tensor_type=__snake_case    )

											return encoded_inputs








'''simple docstring'''


import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							@require_torch
							def UpperCAmelCase_      (       self    ):
											# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
											# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
											# while running an external program

											# python one-liner segments

											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

											_SCREAMING_SNAKE_CASE					:			Dict							= """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise RuntimeError(\"Offline mode is enabled, we shouldn't access internet\")
socket.socket = offline_socket
        """

											# Force fetching the files so that we can use the cache
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """hf-internal-testing/tiny-random-bert"""
											BertConfig.from_pretrained(__snake_case    )
											BertModel.from_pretrained(__snake_case    )
											BertTokenizer.from_pretrained(__snake_case    )
											pipeline(task="""fill-mask"""       ,   model=__snake_case    )

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= [sys.executable, """-c""", """\n""".join([load, run, mock]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_env()
											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			int							= """1"""
											_SCREAMING_SNAKE_CASE					:			str							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											# python one-liner segments
											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

											_SCREAMING_SNAKE_CASE					:			Any							= """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			List[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Faking flaky internet\")
socket.socket = offline_socket
        """

											# Force fetching the files so that we can use the cache
											_SCREAMING_SNAKE_CASE					:			Tuple							= """hf-internal-testing/tiny-random-bert"""
											BertConfig.from_pretrained(__snake_case    )
											BertModel.from_pretrained(__snake_case    )
											BertTokenizer.from_pretrained(__snake_case    )
											pipeline(task="""fill-mask"""       ,   model=__snake_case    )

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			Tuple							= [sys.executable, """-c""", """\n""".join([load, run, mock]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			str							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Tuple							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
											# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
											# while running an external program

											# python one-liner segments

											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Tuple							= """
from transformers import BertConfig, BertModel, BertTokenizer
        """

											_SCREAMING_SNAKE_CASE					:			str							= """
mname = \"hf-internal-testing/tiny-random-bert-sharded\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			List[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise ValueError(\"Offline mode is enabled\")
socket.socket = offline_socket
        """

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [sys.executable, """-c""", """\n""".join([load, run]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			List[str]							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )

											# next emulate no network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [sys.executable, """-c""", """\n""".join([load, mock, run]    )]

											# Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
											# env["TRANSFORMERS_OFFLINE"] = "0"
											# result = subprocess.run(cmd, env=env, check=False, capture_output=True)
											# self.assertEqual(result.returncode, 1, result.stderr)

											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			Dict							= """1"""
											_SCREAMING_SNAKE_CASE					:			int							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import pipeline
        """
											_SCREAMING_SNAKE_CASE					:			int							= """
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

											_SCREAMING_SNAKE_CASE					:			Any							= """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
											_SCREAMING_SNAKE_CASE					:			Dict							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Any							= """1"""
											_SCREAMING_SNAKE_CASE					:			str							= [sys.executable, """-c""", """\n""".join([load, mock, run]    )]
											_SCREAMING_SNAKE_CASE					:			List[Any]							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   1       ,   result.stderr    )
											self.assertIn(
											    """You cannot infer task automatically within `pipeline` when using offline mode"""       ,   result.stderr.decode().replace("""\n"""       ,   """"""    )       ,   )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import AutoModel
        """
											_SCREAMING_SNAKE_CASE					:			Tuple							= """
mname = \"hf-internal-testing/test_dynamic_model\"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print(\"success\")
        """

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [sys.executable, """-c""", """\n""".join([load, run]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			List[str]							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )

											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """1"""
											_SCREAMING_SNAKE_CASE					:			Any							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )




