






"""simple docstring"""



import os
import zipfile

import pytest

from datasets.utils.extract import (
    BzipaExtractor,
    Extractor,
    GzipExtractor,
    LzaExtractor,
    SevenZipExtractor,
    TarExtractor,
    XzExtractor,
    ZipExtractor,
    ZstdExtractor,
)

from .utils import require_lza, require_pyazr, require_zstandard

@pytest.mark.parametrize(
    "compression_format, is_archive"    ,						[
        ("7z", True),
        ("bz2", False),
        ("gzip", False),
        ("lz4", False),
        ("tar", True),
        ("xz", False),
        ("zip", True),
        ("zstd", False),
    ]    ,						)
def 			_snake_case    (					lowerCamelCase__     :   Union[str, Any]    ,						lowerCamelCase__     :   Tuple    ,						lowerCamelCase__     :   Optional[int]    ,						lowerCamelCase__     :   List[Any]    ,						lowerCamelCase__     :   Optional[Any]    ,						lowerCamelCase__     :   Tuple    ,						lowerCamelCase__     :   Optional[int]    ,						lowerCamelCase__     :   int    ,						lowerCamelCase__     :   Optional[int]    ,						lowerCamelCase__     :   Union[str, Any]    ,						lowerCamelCase__     :   int    ,						lowerCamelCase__     :   str    ,						)    ->     Optional[int]:
							lowerCamelCase_	:							Dict									={
							    "7z": (seven_zip_file, SevenZipExtractor),
							    "bz2": (bza_file, BzipaExtractor),
							    "gzip": (gz_file, GzipExtractor),
							    "lz4": (lza_file, LzaExtractor),
							    "tar": (tar_file, TarExtractor),
							    "xz": (xz_file, XzExtractor),
							    "zip": (zip_file, ZipExtractor),
							    "zstd": (zstd_file, ZstdExtractor),
							}
							lowerCamelCase_	:							int									=input_paths_and_base_extractors[compression_format]
							if input_path is None:
														lowerCamelCase_	:							Optional[Any]									=F"""for '{compression_format}' compression_format, """
														if compression_format == "7z":
																					reason += require_pyazr.kwargs["reason"]
														elif compression_format == "lz4":
																					reason += require_lza.kwargs["reason"]
														elif compression_format == "zstd":
																					reason += require_zstandard.kwargs["reason"]
														pytest.skip(lowerCamelCase__  )
							assert base_extractor.is_extractable(lowerCamelCase__  )
							lowerCamelCase_	:							Tuple									=tmp_path / ("extracted" if is_archive else "extracted.txt")
							base_extractor.extract(lowerCamelCase__    ,						lowerCamelCase__  )
							if is_archive:
														assert output_path.is_dir()
														for file_path in output_path.iterdir():
																					assert file_path.name == text_file.name
																					lowerCamelCase_	:							str									=file_path.read_text(encoding="utf-8"  )
							else:
														lowerCamelCase_	:							str									=output_path.read_text(encoding="utf-8"  )
							lowerCamelCase_	:							Any									=text_file.read_text(encoding="utf-8"  )
							assert extracted_file_content == expected_file_content

@pytest.mark.parametrize(
    "compression_format, is_archive"    ,						[
        ("7z", True),
        ("bz2", False),
        ("gzip", False),
        ("lz4", False),
        ("tar", True),
        ("xz", False),
        ("zip", True),
        ("zstd", False),
    ]    ,						)
def 			_snake_case    (					lowerCamelCase__     :   Optional[int]    ,						lowerCamelCase__     :   Union[str, Any]    ,						lowerCamelCase__     :   str    ,						lowerCamelCase__     :   Tuple    ,						lowerCamelCase__     :   Optional[Any]    ,						lowerCamelCase__     :   int    ,						lowerCamelCase__     :   Optional[Any]    ,						lowerCamelCase__     :   str    ,						lowerCamelCase__     :   List[Any]    ,						lowerCamelCase__     :   Dict    ,						lowerCamelCase__     :   Optional[Any]    ,						lowerCamelCase__     :   Optional[Any]    ,						)    ->     Tuple:
							lowerCamelCase_	:							List[Any]									={
							    "7z": seven_zip_file,
							    "bz2": bza_file,
							    "gzip": gz_file,
							    "lz4": lza_file,
							    "tar": tar_file,
							    "xz": xz_file,
							    "zip": zip_file,
							    "zstd": zstd_file,
							}
							lowerCamelCase_	:							Optional[Any]									=input_paths[compression_format]
							if input_path is None:
														lowerCamelCase_	:							Union[str, Any]									=F"""for '{compression_format}' compression_format, """
														if compression_format == "7z":
																					reason += require_pyazr.kwargs["reason"]
														elif compression_format == "lz4":
																					reason += require_lza.kwargs["reason"]
														elif compression_format == "zstd":
																					reason += require_zstandard.kwargs["reason"]
														pytest.skip(lowerCamelCase__  )
							lowerCamelCase_	:							Optional[int]									=Extractor.infer_extractor_format(lowerCamelCase__  )
							assert extractor_format is not None
							lowerCamelCase_	:							Optional[int]									=tmp_path / ("extracted" if is_archive else "extracted.txt")
							Extractor.extract(lowerCamelCase__    ,						lowerCamelCase__    ,						lowerCamelCase__  )
							if is_archive:
														assert output_path.is_dir()
														for file_path in output_path.iterdir():
																					assert file_path.name == text_file.name
																					lowerCamelCase_	:							Tuple									=file_path.read_text(encoding="utf-8"  )
							else:
														lowerCamelCase_	:							Any									=output_path.read_text(encoding="utf-8"  )
							lowerCamelCase_	:							str									=text_file.read_text(encoding="utf-8"  )
							assert extracted_file_content == expected_file_content

@pytest.fixture
def 			_snake_case    (					lowerCamelCase__     :   Optional[Any]    ,						lowerCamelCase__     :   Dict  )    ->     Dict:
							import tarfile

							lowerCamelCase_	:							List[str]									=tmp_path / "data_dot_dot"
							directory.mkdir()
							lowerCamelCase_	:							Optional[int]									=directory / "tar_file_with_dot_dot.tar"
							with tarfile.TarFile(lowerCamelCase__    ,						"w"  ) as f:
														f.add(lowerCamelCase__    ,						arcname=os.path.join(".."    ,						text_file.name  )  )
							return path

@pytest.fixture
def 			_snake_case    (					lowerCamelCase__     :   List[Any]  )    ->     Optional[Any]:
							import tarfile

							lowerCamelCase_	:							Union[str, Any]									=tmp_path / "data_sym_link"
							directory.mkdir()
							lowerCamelCase_	:							Optional[Any]									=directory / "tar_file_with_sym_link.tar"
							os.symlink(".."    ,						directory / "subdir"    ,						target_is_directory=lowerCamelCase__  )
							with tarfile.TarFile(lowerCamelCase__    ,						"w"  ) as f:
														f.add(str(directory / "subdir"  )    ,						arcname="subdir"  )  # str required by os.readlink on Windows and Python < 3.8
							return path

@pytest.mark.parametrize(
    "insecure_tar_file, error_log"    ,						[("tar_file_with_dot_dot", "illegal path"), ("tar_file_with_sym_link", "Symlink")]    ,						)
def 			_snake_case    (					lowerCamelCase__     :   Union[str, Any]    ,						lowerCamelCase__     :   Tuple    ,						lowerCamelCase__     :   str    ,						lowerCamelCase__     :   List[str]    ,						lowerCamelCase__     :   Union[str, Any]    ,						lowerCamelCase__     :   str  )    ->     List[Any]:
							lowerCamelCase_	:							str									={
							    "tar_file_with_dot_dot": tar_file_with_dot_dot,
							    "tar_file_with_sym_link": tar_file_with_sym_link,
							}
							lowerCamelCase_	:							List[Any]									=insecure_tar_files[insecure_tar_file]
							lowerCamelCase_	:							List[Any]									=tmp_path / "extracted"
							TarExtractor.extract(lowerCamelCase__    ,						lowerCamelCase__  )
							assert caplog.text
							for record in caplog.records:
														assert record.levelname == "ERROR"
														assert error_log in record.msg

def 			_snake_case    (					lowerCamelCase__     :   List[Any]  )    ->     int:
							# We should have less false positives than zipfile.is_zipfile
							# We do that by checking only the magic number
							lowerCamelCase_	:							int									=tmpdir / "not_a_zip_file"
							# From: https://github.com/python/cpython/pull/5053
							lowerCamelCase_	:							Tuple									=(
							    B"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00"
							    B"\x00\x02\x08\x06\x00\x00\x00\x99\x81\xb6'\x00\x00\x00\x15I"
							    B"DATx\x01\x01\n\x00\xf5\xff\x00PK\x05\x06\x00PK\x06\x06\x07"
							    B"\xac\x01N\xc6|a\r\x00\x00\x00\x00IEND\xaeB`\x82"
							)
							with not_a_zip_file.open("wb"  ) as f:
														f.write(lowerCamelCase__  )
							assert zipfile.is_zipfile(str(lowerCamelCase__  )  )  # is a false positive for `zipfile`
							assert not ZipExtractor.is_extractable(lowerCamelCase__  )  # but we're right











"""simple docstring"""



from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_speech_available,
    is_tf_available,
    is_torch_available,
)


A__	:    Optional[int]			 = {
    'configuration_speech_to_text': ['SPEECH_TO_TEXT_PRETRAINED_CONFIG_ARCHIVE_MAP', 'Speech2TextConfig'],
    'processing_speech_to_text': ['Speech2TextProcessor'],
}

try:
						if not is_sentencepiece_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						A__	:    Dict			 = ['Speech2TextTokenizer']

try:
						if not is_speech_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						A__	:    Optional[Any]			 = ['Speech2TextFeatureExtractor']

try:
						if not is_tf_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						A__	:    Dict			 = [
						    'TF_SPEECH_TO_TEXT_PRETRAINED_MODEL_ARCHIVE_LIST',
						    'TFSpeech2TextForConditionalGeneration',
						    'TFSpeech2TextModel',
						    'TFSpeech2TextPreTrainedModel',
						]

try:
						if not is_torch_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						A__	:    Optional[int]			 = [
						    'SPEECH_TO_TEXT_PRETRAINED_MODEL_ARCHIVE_LIST',
						    'Speech2TextForConditionalGeneration',
						    'Speech2TextModel',
						    'Speech2TextPreTrainedModel',
						]


if TYPE_CHECKING:
						from .configuration_speech_to_text import SPEECH_TO_TEXT_PRETRAINED_CONFIG_ARCHIVE_MAP, SpeechaTextConfig
						from .processing_speech_to_text import SpeechaTextProcessor

						try:
												if not is_sentencepiece_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .tokenization_speech_to_text import SpeechaTextTokenizer

						try:
												if not is_speech_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .feature_extraction_speech_to_text import SpeechaTextFeatureExtractor

						try:
												if not is_tf_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_tf_speech_to_text import (
												    TF_SPEECH_TO_TEXT_PRETRAINED_MODEL_ARCHIVE_LIST,
												    TFSpeechaTextForConditionalGeneration,
												    TFSpeechaTextModel,
												    TFSpeechaTextPreTrainedModel,
												)

						try:
												if not is_torch_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_speech_to_text import (
												    SPEECH_TO_TEXT_PRETRAINED_MODEL_ARCHIVE_LIST,
												    SpeechaTextForConditionalGeneration,
												    SpeechaTextModel,
												    SpeechaTextPreTrainedModel,
												)

else:
						import sys

						A__	:    List[Any]			 = _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)



