





'''simple docstring'''




import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
							from .features import FeatureType


__UpperCAmelCase			    =    False, False, False




@dataclass
class 			UpperCamelCase__ :




					"""simple docstring"""
					SCREAMING_SNAKE_CASE__            =     None
					SCREAMING_SNAKE_CASE__            =     True
					SCREAMING_SNAKE_CASE__            =     True
					SCREAMING_SNAKE_CASE__            =     None
					# Automatically constructed
					SCREAMING_SNAKE_CASE__            =     "dict"
					SCREAMING_SNAKE_CASE__            =     pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}       )
					SCREAMING_SNAKE_CASE__            =     field(default='''Audio'''		, init=_a		, repr=_a       )


					def __call__(		self							:       Optional[int]		):

							'''simple docstring'''



							return self.pa_type


					def lowerCamelCase_       (		self							:       List[Any]	,		lowerCamelCase_							:       Union[str, bytes, dict]		):

							'''simple docstring'''



							try:
									import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
							except ImportError as err:
									raise ImportError("""To support encoding audio data, please install 'soundfile'."""		) from err
							if isinstance(__lowerCamelCase	,		__lowerCamelCase		):
									return {"bytes": None, "path": value}
							elif isinstance(__lowerCamelCase	,		__lowerCamelCase		):
									return {"bytes": value, "path": None}
							elif "array" in value:
									# convert the audio array to wav bytes
									SCREAMING_SNAKE_CASE      :							Dict						=       BytesIO()
									sf.write(__lowerCamelCase	,		value["""array"""]	,		value["""sampling_rate"""]	,		format="""wav"""		)
									return {"bytes": buffer.getvalue(), "path": None}
							elif value.get("""path"""		) is not None and os.path.isfile(value["""path"""]		):
									# we set "bytes": None to not duplicate the data if they're already available locally
									if value["path"].endswith("""pcm"""		):
											# "PCM" only has raw audio bytes
											if value.get("""sampling_rate"""		) is None:
													# At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
													raise KeyError("""To use PCM files, please specify a 'sampling_rate' in Audio object"""		)
											if value.get("""bytes"""		):
													# If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
													SCREAMING_SNAKE_CASE      :							List[str]						=       np.frombuffer(value["""bytes"""]	,		dtype=np.intaa		).astype(np.floataa		) / 3_27_67
											else:
													SCREAMING_SNAKE_CASE      :							Any						=       np.memmap(value["""path"""]	,		dtype="""h"""	,		mode="""r"""		).astype(np.floataa		) / 3_27_67

											SCREAMING_SNAKE_CASE      :							Union[str, Any]						=       BytesIO(bytes()		)
											sf.write(__lowerCamelCase	,		__lowerCamelCase	,		value["""sampling_rate"""]	,		format="""wav"""		)
											return {"bytes": buffer.getvalue(), "path": None}
									else:
											return {"bytes": None, "path": value.get("""path"""		)}
							elif value.get("""bytes"""		) is not None or value.get("""path"""		) is not None:
									# store the audio bytes, and path is used to infer the audio format using the file extension
									return {"bytes": value.get("""bytes"""		), "path": value.get("""path"""		)}
							else:
									raise ValueError(
									    f'''An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.'''		)


					def lowerCamelCase_       (		self							:       List[str]	,		lowerCamelCase_							:       dict	,		lowerCamelCase_							:       Optional[Dict[str, Union[str, bool, None]]] = None		):

							'''simple docstring'''



							if not self.decode:
									raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead."""		)

							SCREAMING_SNAKE_CASE      :							int						=       (value["""path"""], BytesIO(value["""bytes"""]		)) if value["""bytes"""] is not None else (value["""path"""], None)
							if path is None and file is None:
									raise ValueError(f'''An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.'''		)

							try:
									import librosa
									import soundfile as sf
							except ImportError as err:
									raise ImportError("""To support decoding audio files, please install 'librosa' and 'soundfile'."""		) from err

							SCREAMING_SNAKE_CASE      :							int						=       xsplitext(__lowerCamelCase		)[1][1:].lower() if path is not None else None
							if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
									raise RuntimeError(
									    """Decoding 'opus' files requires system library 'libsndfile'>=1.0.31, """
									    """You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. """		)
							elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
									raise RuntimeError(
									    """Decoding 'mp3' files requires system library 'libsndfile'>=1.1.0, """
									    """You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. """		)

							if file is None:
									SCREAMING_SNAKE_CASE      :							int						=       token_per_repo_id or {}
									SCREAMING_SNAKE_CASE      :							Any						=       path.split("""::"""		)[-1]
									try:
											SCREAMING_SNAKE_CASE      :							List[str]						=       string_to_dict(__lowerCamelCase	,		config.HUB_DATASETS_URL		)["""repo_id"""]
											SCREAMING_SNAKE_CASE      :							Union[str, Any]						=       token_per_repo_id[repo_id]
									except (ValueError, KeyError):
											SCREAMING_SNAKE_CASE      :							Tuple						=       None

									with xopen(__lowerCamelCase	,		"""rb"""	,		use_auth_token=__lowerCamelCase		) as f:
											SCREAMING_SNAKE_CASE      :							str						=       sf.read(__lowerCamelCase		)

							else:
									SCREAMING_SNAKE_CASE      :							Dict						=       sf.read(__lowerCamelCase		)

							SCREAMING_SNAKE_CASE      :							str						=       array.T
							if self.mono:
									SCREAMING_SNAKE_CASE      :							Dict						=       librosa.to_mono(__lowerCamelCase		)
							if self.sampling_rate and self.sampling_rate != sampling_rate:
									SCREAMING_SNAKE_CASE      :							Optional[int]						=       librosa.resample(__lowerCamelCase	,		orig_sr=__lowerCamelCase	,		target_sr=self.sampling_rate		)
									SCREAMING_SNAKE_CASE      :							Union[str, Any]						=       self.sampling_rate

							return {"path": path, "array": array, "sampling_rate": sampling_rate}


					def lowerCamelCase_       (		self							:       int		):

							'''simple docstring'''



							from .features import Value

							if self.decode:
									raise ValueError("""Cannot flatten a decoded Audio feature."""		)
							return {
							    "bytes": Value("""binary"""		),
							    "path": Value("""string"""		),
							}


					def lowerCamelCase_       (		self							:       Union[str, Any]	,		lowerCamelCase_							:       Union[pa.StringArray, pa.StructArray]		):

							'''simple docstring'''



							if pa.types.is_string(storage.type		):
									SCREAMING_SNAKE_CASE      :							Optional[Any]						=       pa.array([None] * len(__lowerCamelCase		)	,		type=pa.binary()		)
									SCREAMING_SNAKE_CASE      :							List[str]						=       pa.StructArray.from_arrays([bytes_array, storage]	,		["""bytes""", """path"""]	,		mask=storage.is_null()		)
							elif pa.types.is_binary(storage.type		):
									SCREAMING_SNAKE_CASE      :							List[str]						=       pa.array([None] * len(__lowerCamelCase		)	,		type=pa.string()		)
									SCREAMING_SNAKE_CASE      :							Optional[int]						=       pa.StructArray.from_arrays([storage, path_array]	,		["""bytes""", """path"""]	,		mask=storage.is_null()		)
							elif pa.types.is_struct(storage.type		) and storage.type.get_all_field_indices("""array"""		):
									SCREAMING_SNAKE_CASE      :							Optional[Any]						=       pa.array([Audio().encode_example(__lowerCamelCase		) if x is not None else None for x in storage.to_pylist()]		)
							elif pa.types.is_struct(storage.type		):
									if storage.type.get_field_index("""bytes"""		) >= 0:
											SCREAMING_SNAKE_CASE      :							int						=       storage.field("""bytes"""		)
									else:
											SCREAMING_SNAKE_CASE      :							int						=       pa.array([None] * len(__lowerCamelCase		)	,		type=pa.binary()		)
									if storage.type.get_field_index("""path"""		) >= 0:
											SCREAMING_SNAKE_CASE      :							Tuple						=       storage.field("""path"""		)
									else:
											SCREAMING_SNAKE_CASE      :							List[Any]						=       pa.array([None] * len(__lowerCamelCase		)	,		type=pa.string()		)
									SCREAMING_SNAKE_CASE      :							Optional[Any]						=       pa.StructArray.from_arrays([bytes_array, path_array]	,		["""bytes""", """path"""]	,		mask=storage.is_null()		)
							return array_cast(__lowerCamelCase	,		self.pa_type		)



					def lowerCamelCase_       (		self							:       int	,		lowerCamelCase_							:       pa.StructArray		):

							'''simple docstring'''



							@no_op_if_value_is_null
							def path_to_bytes(lowerCamelCase_							:       str		):
									with xopen(__lowerCamelCase	,		"""rb"""		) as f:
											SCREAMING_SNAKE_CASE      :							List[str]						=       f.read()
									return bytes_

							SCREAMING_SNAKE_CASE      :							Optional[int]						=       pa.array(
							    [
							        (path_to_bytes(x["""path"""]		) if x["""bytes"""] is None else x["""bytes"""]) if x is not None else None
							        for x in storage.to_pylist()
							    ]	,		type=pa.binary()	,		)
							SCREAMING_SNAKE_CASE      :							Optional[Any]						=       pa.array(
							    [os.path.basename(__lowerCamelCase		) if path is not None else None for path in storage.field("""path"""		).to_pylist()]	,		type=pa.string()	,		)
							SCREAMING_SNAKE_CASE      :							Tuple						=       pa.StructArray.from_arrays([bytes_array, path_array]	,		["""bytes""", """path"""]	,		mask=bytes_array.is_null()		)
							return array_cast(__lowerCamelCase	,		self.pa_type		)











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_ :						Any      =      logging.get_logger(__name__)

class     _SCREAMING_SNAKE_CASE       (			_a      ):





	def __init__(		self				:					Optional[int] ,  **__lowerCamelCase				:					Optional[int]	):
		requires_backends(self ,  ["""bs4"""]	)
		super().__init__(**__lowerCamelCase	)





	def 						_A (		self				:					List[str] ,  __lowerCamelCase				:					Any	):
		UpperCamelCase     :Optional[int]          =	[]
		UpperCamelCase     :List[str]          =	[]
		UpperCamelCase     :Union[str, Any]          =	element if element.name else element.parent
		for parent in child.parents:  # type: bs4.element.Tag
			UpperCamelCase     :Optional[Any]          =	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	)	)
			UpperCamelCase     :Any          =	parent
		xpath_tags.reverse()
		xpath_subscripts.reverse()
		return xpath_tags, xpath_subscripts





	def 						_A (		self				:					Any ,  __lowerCamelCase				:					Tuple	):
		UpperCamelCase     :Any          =	BeautifulSoup(__lowerCamelCase ,  """html.parser"""	)

		UpperCamelCase     :Union[str, Any]          =	[]
		UpperCamelCase     :Tuple          =	[]
		UpperCamelCase     :Tuple          =	[]

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

				UpperCamelCase     :Any          =	html.unescape(__lowerCamelCase	).strip()
				if not text_in_this_tag:
					continue

				all_doc_strings.append(__lowerCamelCase	)

				UpperCamelCase     ,      UpperCamelCase     :Optional[Any]          =	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 						_A (		self				:					int ,  __lowerCamelCase				:					List[Any] ,  __lowerCamelCase				:					List[str]	):
		UpperCamelCase     :Tuple          =	""""""
		for tagname, subs in zip(__lowerCamelCase ,  __lowerCamelCase	):
			xpath += F"""/{tagname}"""
			if subs != 0:
				xpath += F"""[{subs}]"""
		return xpath





	def __call__(		self				:					Any ,  __lowerCamelCase				:					Dict	):
		UpperCamelCase     :Any          =	False

		# Check that strings has a valid type
		if isinstance(__lowerCamelCase ,  __lowerCamelCase	):
			UpperCamelCase     :List[Any]          =	True
		elif isinstance(__lowerCamelCase ,  (list, tuple)	):
			if len(__lowerCamelCase	) == 0 or isinstance(html_strings[0] ,  __lowerCamelCase	):
				UpperCamelCase     :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(__lowerCamelCase	)}."""	)

		UpperCamelCase     :str          =	bool(isinstance(__lowerCamelCase ,  (list, tuple)	) and (isinstance(html_strings[0] ,  __lowerCamelCase	))	)

		if not is_batched:
			UpperCamelCase     :Any          =	[html_strings]

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

		# return as Dict
		UpperCamelCase     :Optional[int]          =	{"""nodes""": nodes, """xpaths""": xpaths}
		UpperCamelCase     :Any          =	BatchFeature(data=__lowerCamelCase ,  tensor_type=__lowerCamelCase	)

		return encoded_inputs
