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


_a					:					str=			logging.get_logger(__name__)

class 		UpperCamelCase      (   lowercase						):



				def __init__(self    :				Optional[Any]     ,  **_A    :				Optional[Any])   ->     int:
					requires_backends(self     ,  ['bs4'])
					super().__init__(**_A)



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



				def  _lowercase					(self    :				Any     ,  _A    :				Optional[Any])   ->     str:
					__snake_case      :							Optional[Any]						=			BeautifulSoup(_A     ,  'html.parser')

					__snake_case      :							Dict						=			[]
					__snake_case      :							List[Any]						=			[]
					__snake_case      :							Optional[int]						=			[]

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

							__snake_case      :							Any						=			html.unescape(_A).strip()
							if not text_in_this_tag:
								continue

							all_doc_strings.append(_A)

							__snake_case		,    __snake_case      :							List[Any]						=			self.xpath_soup(_A)
							stringaxtag_seq.append(_A)
							stringaxsubs_seq.append(_A)

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

					return all_doc_strings, stringaxtag_seq, stringaxsubs_seq



				def  _lowercase					(self    :				Optional[Any]     ,  _A    :				Dict     ,  _A    :				Optional[int])   ->     Any:
					__snake_case      :							Tuple						=			''
					for tagname, subs in zip(_A     ,  _A):
						xpath += f"/{tagname}"
						if subs != 0:
							xpath += f"[{subs}]"
					return xpath






				def __call__(self    :				str     ,  _A    :				Union[str, Any])   ->     BatchFeature:
					__snake_case      :							str						=			False

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

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

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

					if not is_batched:
						__snake_case      :							List[Any]						=			[html_strings]

					# Get nodes + xpaths
					__snake_case      :							int						=			[]
					__snake_case      :							Optional[Any]						=			[]
					for html_string in html_strings:
						__snake_case		,    __snake_case		,    __snake_case      :							Optional[int]						=			self.get_three_from_single(_A)
						nodes.append(_A)
						__snake_case      :							Optional[Any]						=			[]
						for node, tag_list, sub_list in zip(_A     ,  _A     ,  _A):
							__snake_case      :							Optional[Any]						=			self.construct_xpath(_A     ,  _A)
							xpath_strings.append(_A)
						xpaths.append(_A)

					# return as Dict
					__snake_case      :							List[Any]						=			{'nodes': nodes, 'xpaths': xpaths}
					__snake_case      :							Optional[int]						=			BatchFeature(data=_A     ,  tensor_type=_A)

					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 		UpperCamelCase      (   lowercase						):



				@require_torch
				def  _lowercase					(self    :				Union[str, Any])   ->     Optional[Any]:
					# 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
					__snake_case      :							Any						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

					__snake_case      :							Tuple						=			'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

					__snake_case      :							int						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")\nsocket.socket = offline_socket\n        '

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

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

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



				@require_torch
				def  _lowercase					(self    :				Union[str, Any])   ->     Union[str, Any]:
					# python one-liner segments
					# this must be loaded before socket.socket is monkey-patched
					__snake_case      :							Optional[Any]						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

					__snake_case      :							Optional[Any]						=			'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

					__snake_case      :							Union[str, Any]						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")\nsocket.socket = offline_socket\n        '

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

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

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



				@require_torch
				def  _lowercase					(self    :				int)   ->     Any:
					# 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
					__snake_case      :							int						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

					__snake_case      :							int						=			'\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

					__snake_case      :							Optional[int]						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")\nsocket.socket = offline_socket\n        '

					# baseline - just load from_pretrained with normal network
					__snake_case      :							Optional[Any]						=			[sys.executable, '-c', '\n'.join([load, run])]

					# should succeed
					__snake_case      :							Optional[int]						=			self.get_env()
					__snake_case      :							Dict						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())

					# next emulate no network
					__snake_case      :							Optional[Any]						=			[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
					__snake_case      :							Union[str, Any]						=			'1'
					__snake_case      :							str						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())



				@require_torch
				def  _lowercase					(self    :				str)   ->     Dict:
					__snake_case      :							Dict						=			'\nfrom transformers import pipeline\n        '
					__snake_case      :							List[Any]						=			'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

					__snake_case      :							List[str]						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
					__snake_case      :							str						=			self.get_env()
					__snake_case      :							Tuple						=			'1'
					__snake_case      :							Tuple						=			[sys.executable, '-c', '\n'.join([load, mock, run])]
					__snake_case      :							Optional[Any]						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					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  _lowercase					(self    :				int)   ->     Optional[Any]:
					__snake_case      :							int						=			'\nfrom transformers import AutoModel\n        '
					__snake_case      :							str						=			'\nmname = "hf-internal-testing/test_dynamic_model"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint("success")\n        '

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

					# should succeed
					__snake_case      :							str						=			self.get_env()
					__snake_case      :							Dict						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					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
					__snake_case      :							List[str]						=			'1'
					__snake_case      :							Optional[int]						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())





