
"""simple docstring"""






import time
from dataclasses import dataclass
from multiprocessing import Pool
from unittest import TestCase
from unittest.mock import patch

import multiprocess
import numpy as np
import pytest

from datasets.utils.py_utils import (
    NestedDataStructure,
    asdict,
    iflatmap_unordered,
    map_nested,
    temp_seed,
    temporary_assignment,
    zip_dict,
)

from .utils import require_tf, require_torch



def      lowerCAmelCase__			(		__magic_name__      )						->Tuple:  # picklable for multiprocessing
			return x.sum()



def      lowerCAmelCase__			(		__magic_name__      )						->int:  # picklable for multiprocessing
			return i + 1

@dataclass
class 					__a     :



							'''simple docstring'''







							_lowerCamelCase					:			int
							_lowerCamelCase					:			str





class 					__a     ( __a       ):



							'''simple docstring'''







							def   SCREAMING_SNAKE_CASE						(     self       )			->							Any:




										'''simple docstring'''
										__lowercase       =       {}
										__lowercase       =       []
										__lowercase       =       1
										__lowercase       =       [1, 2]
										__lowercase       =       {"a": 1, "b": 2}
										__lowercase       =       {"a": [1, 2], "b": [3, 4]}
										__lowercase       =       {"a": {"1": 1}, "b": 2}
										__lowercase       =       {"a": 1, "b": 2, "c": 3, "d": 4}
										__lowercase       =       {}
										__lowercase       =       []
										__lowercase       =       2
										__lowercase       =       [2, 3]
										__lowercase       =       {"a": 2, "b": 3}
										__lowercase       =       {"a": [2, 3], "b": [4, 5]}
										__lowercase       =       {"a": {"1": 2}, "b": 3}
										__lowercase       =       {"a": 2, "b": 3, "c": 4, "d": 5}
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase       )  ,    _lowerCamelCase       )

										__lowercase       =       2
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    num_proc=_lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    num_proc=_lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    num_proc=_lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    num_proc=_lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    num_proc=_lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    num_proc=_lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    num_proc=_lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    num_proc=_lowerCamelCase       )  ,    _lowerCamelCase       )

										__lowercase       =       {"a": np.eye(2       ), "b": np.zeros(3       ), "c": np.ones(2       )}
										__lowercase       =       {"a": 2, "b": 0, "c": 2}
										__lowercase       =       {
										    "a": np.eye(2       ).astype(_lowerCamelCase       ),
										    "b": np.zeros(3       ).astype(_lowerCamelCase       ),
										    "c": np.ones(2       ).astype(_lowerCamelCase       ),
										}
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    map_numpy=_lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(
										    {k: v.tolist() for k, v in map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    map_numpy=_lowerCamelCase       ).items()}  ,    {k: v.tolist() for k, v in expected_map_nested_sna_int.items()}  ,    )
										self.assertEqual(map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    map_numpy=_lowerCamelCase  ,    num_proc=_lowerCamelCase       )  ,    _lowerCamelCase       )
										self.assertEqual(
										    {k: v.tolist() for k, v in map_nested(_lowerCamelCase  ,    _lowerCamelCase  ,    map_numpy=_lowerCamelCase  ,    num_proc=_lowerCamelCase       ).items()}  ,    {k: v.tolist() for k, v in expected_map_nested_sna_int.items()}  ,    )
										with self.assertRaises(_lowerCamelCase       ):  # can't pickle a local lambda
													map_nested(lambda _lowerCamelCase       : x + 1  ,    _lowerCamelCase  ,    num_proc=_lowerCamelCase       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''
										__lowercase       =       {"a": 1, "b": 2}
										__lowercase       =       {"a": 3, "b": 4}
										__lowercase       =       {"a": 5, "b": 6}
										__lowercase       =       sorted([("a", (1, 3, 5)), ("b", (2, 4, 6))]       )
										self.assertEqual(sorted(zip_dict(_lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase       )       )  ,    _lowerCamelCase       )





							def   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''

										class 					__a     :



																	'''simple docstring'''







																	_lowerCamelCase					:			Dict          =   """bar"""





										__lowercase       =       Foo()
										self.assertEqual(foo.my_attr  ,    "bar"       )
										with temporary_assignment(_lowerCamelCase  ,    "my_attr"  ,    "BAR"       ):
													self.assertEqual(foo.my_attr  ,    "BAR"       )
										self.assertEqual(foo.my_attr  ,    "bar"       )





@pytest.mark.parametrize(
    "iterable_length, num_proc, expected_num_proc"			,   [
        (1, None, 1),
        (1, 1, 1),
        (2, None, 1),
        (2, 1, 1),
        (2, 2, 1),
        (2, 3, 1),
        (3, 2, 1),
        (1_6, 1_6, 1_6),
        (1_6, 1_7, 1_6),
        (1_7, 1_6, 1_6),
    ]			,   )
def      lowerCAmelCase__			(		__magic_name__			,   __magic_name__			,   __magic_name__      )						->Union[str, Any]:
			with patch("datasets.utils.py_utils._single_map_nested"      ) as mock_single_map_nested, patch(
			    "datasets.parallel.parallel.Pool"      ) as mock_multiprocessing_pool:
						__lowercase       =       {F'''{i}''': i for i in range(__magic_name__      )}
						__lowercase       =       map_nested(lambda __magic_name__      : x + 1_0			,   __magic_name__			,   num_proc=__magic_name__			,   parallel_min_length=1_6      )
						if expected_num_proc == 1:
									assert mock_single_map_nested.called
									assert not mock_multiprocessing_pool.called
						else:
									assert not mock_single_map_nested.called
									assert mock_multiprocessing_pool.called
									assert mock_multiprocessing_pool.call_args[0][0] == expected_num_proc

class 					__a     ( __a       ):



							'''simple docstring'''







							@require_tf
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Optional[int]:




										'''simple docstring'''
										import tensorflow as tf
										from tensorflow.keras import layers

										__lowercase       =       layers.Dense(2       )

										def gen_random_output():
													__lowercase       =       tf.random.uniform((1, 3)       )
													return model(_lowerCamelCase       ).numpy()

										with temp_seed(42  ,    set_tensorflow=_lowerCamelCase       ):
													__lowercase       =       gen_random_output()
										with temp_seed(42  ,    set_tensorflow=_lowerCamelCase       ):
													__lowercase       =       gen_random_output()
										__lowercase       =       gen_random_output()

										np.testing.assert_equal(_lowerCamelCase  ,    _lowerCamelCase       )
										self.assertGreater(np.abs(outa - outa       ).sum()  ,    0       )
							@require_torch
							def   SCREAMING_SNAKE_CASE						(     self       )			->							List[Any]:




										'''simple docstring'''
										import torch

										def gen_random_output():
													__lowercase       =       torch.nn.Linear(3  ,    2       )
													__lowercase       =       torch.rand(1  ,    3       )
													return model(_lowerCamelCase       ).detach().numpy()

										with temp_seed(42  ,    set_pytorch=_lowerCamelCase       ):
													__lowercase       =       gen_random_output()
										with temp_seed(42  ,    set_pytorch=_lowerCamelCase       ):
													__lowercase       =       gen_random_output()
										__lowercase       =       gen_random_output()

										np.testing.assert_equal(_lowerCamelCase  ,    _lowerCamelCase       )
										self.assertGreater(np.abs(outa - outa       ).sum()  ,    0       )





							def   SCREAMING_SNAKE_CASE						(     self       )			->							List[Any]:




										'''simple docstring'''
										def gen_random_output():
													return np.random.rand(1  ,    3       )

										with temp_seed(42       ):
													__lowercase       =       gen_random_output()
										with temp_seed(42       ):
													__lowercase       =       gen_random_output()
										__lowercase       =       gen_random_output()

										np.testing.assert_equal(_lowerCamelCase  ,    _lowerCamelCase       )
										self.assertGreater(np.abs(outa - outa       ).sum()  ,    0       )





@pytest.mark.parametrize("input_data"			,   [{}]      )
def      lowerCAmelCase__			(		__magic_name__      )						->Union[str, Any]:
			__lowercase       =       NestedDataStructure(__magic_name__      ).data
			assert output_data == input_data



@pytest.mark.parametrize(
    "data, expected_output"			,   [
        ({}, []),
        ([], []),
        ("foo", ["foo"]),
        (["foo", "bar"], ["foo", "bar"]),
        ([["foo", "bar"]], ["foo", "bar"]),
        ([[["foo"], ["bar"]]], ["foo", "bar"]),
        ([[["foo"], "bar"]], ["foo", "bar"]),
        ({"a": 1, "b": 2}, [1, 2]),
        ({"a": [1, 2], "b": [3, 4]}, [1, 2, 3, 4]),
        ({"a": [[1, 2]], "b": [[3, 4]]}, [1, 2, 3, 4]),
        ({"a": [[1, 2]], "b": [3, 4]}, [1, 2, 3, 4]),
        ({"a": [[[1], [2]]], "b": [[[3], [4]]]}, [1, 2, 3, 4]),
        ({"a": [[[1], [2]]], "b": [[3, 4]]}, [1, 2, 3, 4]),
        ({"a": [[[1], [2]]], "b": [3, 4]}, [1, 2, 3, 4]),
        ({"a": [[[1], [2]]], "b": [3, [4]]}, [1, 2, 3, 4]),
        ({"a": {"1": 1}, "b": 2}, [1, 2]),
        ({"a": {"1": [1]}, "b": 2}, [1, 2]),
        ({"a": {"1": [1]}, "b": [2]}, [1, 2]),
    ]			,   )
def      lowerCAmelCase__			(		__magic_name__			,   __magic_name__      )						->Tuple:
			__lowercase       =       NestedDataStructure(__magic_name__      ).flatten()
			assert output == expected_output



def      lowerCAmelCase__			(		)						->int:
			__lowercase       =       A(x=1			,   y="foobar"      )
			__lowercase       =       {"x": 1, "y": "foobar"}
			assert asdict(__magic_name__      ) == expected_output

			__lowercase       =       {"a": {"b": A(x=1_0			,   y="foo"      )}, "c": [A(x=2_0			,   y="bar"      )]}
			__lowercase       =       {"a": {"b": {"x": 1_0, "y": "foo"}}, "c": [{"x": 2_0, "y": "bar"}]}
			assert asdict(__magic_name__      ) == expected_output

			with pytest.raises(__magic_name__      ):
						asdict([1, A(x=1_0			,   y="foo"      )]      )



def      lowerCAmelCase__			(		__magic_name__      )						->List[Any]:
			return text.split()



def      lowerCAmelCase__			(		__magic_name__      )						->List[str]:
			yield (time.time(), content)
			time.sleep(2      )
			yield (time.time(), content)







def      lowerCAmelCase__			(		)						->Optional[int]:
			with Pool(2      ) as pool:
						__lowercase       =       list(iflatmap_unordered(__magic_name__			,   _split_text			,   kwargs_iterable=[{"text": "hello there"}] * 1_0      )      )
						assert out.count("hello"      ) == 1_0
						assert out.count("there"      ) == 1_0
						assert len(__magic_name__      ) == 2_0

			# check multiprocess from pathos (uses dill for pickling)
			with multiprocess.Pool(2      ) as pool:
						__lowercase       =       list(iflatmap_unordered(__magic_name__			,   _split_text			,   kwargs_iterable=[{"text": "hello there"}] * 1_0      )      )
						assert out.count("hello"      ) == 1_0
						assert out.count("there"      ) == 1_0
						assert len(__magic_name__      ) == 2_0

			# check that we get items as fast as possible
			with Pool(2      ) as pool:
						__lowercase       =       []
						for yield_time, content in iflatmap_unordered(
						    __magic_name__			,   _aseconds_generator_of_aitems_with_timing			,   kwargs_iterable=[{"content": "a"}, {"content": "b"}]      ):
									assert yield_time < time.time() + 0.1, "we should each item directly after it was yielded"
									out.append(__magic_name__      )
						assert out.count("a"      ) == 2
						assert out.count("b"      ) == 2
						assert len(__magic_name__      ) == 4







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


_lowercase							=				logging.get_logger(__name__)

class 					__a     ( __a       ):



							'''simple docstring'''







							def __init__(     self  ,    **_lowerCamelCase       )			->							List[Any]:




										'''simple docstring'''
										requires_backends(self  ,    ["bs4"]       )
										super().__init__(**_lowerCamelCase       )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							Tuple:




										'''simple docstring'''
										__lowercase       =       []
										__lowercase       =       []
										__lowercase       =       element if element.name else element.parent
										for parent in child.parents:  # type: bs4.element.Tag
													__lowercase       =       parent.find_all(child.name  ,    recursive=_lowerCamelCase       )
													xpath_tags.append(child.name       )
													xpath_subscripts.append(
													    0 if 1 == len(_lowerCamelCase       ) else next(i for i, s in enumerate(_lowerCamelCase  ,    1       ) if s is child       )       )
													__lowercase       =       parent
										xpath_tags.reverse()
										xpath_subscripts.reverse()
										return xpath_tags, xpath_subscripts
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase       )			->							Union[str, Any]:




										'''simple docstring'''
										__lowercase       =       BeautifulSoup(_lowerCamelCase  ,    "html.parser"       )

										__lowercase       =       []
										__lowercase       =       []
										__lowercase       =       []

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

																__lowercase       =       html.unescape(_lowerCamelCase       ).strip()
																if not text_in_this_tag:
																			continue

																all_doc_strings.append(_lowerCamelCase       )

																__lowercase   ,					__lowercase       =       self.xpath_soup(_lowerCamelCase       )
																stringaxtag_seq.append(_lowerCamelCase       )
																stringaxsubs_seq.append(_lowerCamelCase       )

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

										return all_doc_strings, stringaxtag_seq, stringaxsubs_seq
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase       )			->							Tuple:




										'''simple docstring'''
										__lowercase       =       ""
										for tagname, subs in zip(_lowerCamelCase  ,    _lowerCamelCase       ):
													xpath += f'''/{tagname}'''
													if subs != 0:
																xpath += f'''[{subs}]'''
										return xpath





							def __call__(     self  ,    _lowerCamelCase       )			->							BatchFeature:




										'''simple docstring'''
										__lowercase       =       False

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

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

										__lowercase       =       bool(isinstance(_lowerCamelCase  ,    (list, tuple)       ) and (isinstance(html_strings[0]  ,    _lowerCamelCase       ))       )

										if not is_batched:
													__lowercase       =       [html_strings]

										# Get nodes + xpaths
										__lowercase       =       []
										__lowercase       =       []
										for html_string in html_strings:
													__lowercase   ,					__lowercase   ,					__lowercase       =       self.get_three_from_single(_lowerCamelCase       )
													nodes.append(_lowerCamelCase       )
													__lowercase       =       []
													for node, tag_list, sub_list in zip(_lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase       ):
																__lowercase       =       self.construct_xpath(_lowerCamelCase  ,    _lowerCamelCase       )
																xpath_strings.append(_lowerCamelCase       )
													xpaths.append(_lowerCamelCase       )

										# return as Dict
										__lowercase       =       {"nodes": nodes, "xpaths": xpaths}
										__lowercase       =       BatchFeature(data=_lowerCamelCase  ,    tensor_type=_lowerCamelCase       )

										return encoded_inputs





