






"""simple docstring"""
import logging
import os
import threading
import time


try:
							import warnings
except ImportError:
							_lowerCAmelCase         =							None

try:
							import msvcrt
except ImportError:
							_lowerCAmelCase         =							None

try:
							import fcntl
except ImportError:
							_lowerCAmelCase         =							None


# Backward compatibility
# ------------------------------------------------
try:
							TimeoutError
except NameError:
							_lowerCAmelCase         =							OSError


# Data
# ------------------------------------------------
_lowerCAmelCase         =							[
    "Timeout",
    "BaseFileLock",
    "WindowsFileLock",
    "UnixFileLock",
    "SoftFileLock",
    "FileLock",
]

_lowerCAmelCase         =							"3.0.12"


_lowerCAmelCase         =							None






def 			__UpperCamelCase	(		):
	global _logger
	A_		:						Dict	    =   _logger or logging.getLogger(__name__					)
	return _logger



class      SCREAMING_SNAKE_CASE  (				_SCREAMING_SNAKE_CASE						):





	"""simple docstring"""




	def __init__(self     ,     lowerCAmelCase_							):
		A_		:						Union[str, Any]	    =   lock_file
		return None





	def __str__(self							):
		A_		:						Optional[Any]	    =   f"""The file lock '{self.lock_file}' could not be acquired."""
		return temp






class      SCREAMING_SNAKE_CASE  :





	"""simple docstring"""




	def __init__(self     ,     lowerCAmelCase_							):
		A_		:						Optional[Any]	    =   lock
		return None





	def __enter__(self							):
		return self.lock





	def __exit__(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_							):
		self.lock.release()
		return None






class      SCREAMING_SNAKE_CASE  :





	"""simple docstring"""




	def __init__(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_=-1     ,     lowerCAmelCase_=None							):
		A_		:						Tuple	    =   max_filename_length if max_filename_length is not None else 255
		# Hash the filename if it's too long
		A_		:						Tuple	    =   self.hash_filename_if_too_long(lowerCAmelCase_     ,     lowerCAmelCase_							)
		# The path to the lock file.
		A_		:						Optional[Any]	    =   lock_file

		# The file descriptor for the *_lock_file* as it is returned by the
		# os.open() function.
		# This file lock is only NOT None, if the object currently holds the
		# lock.
		A_		:						Any	    =   None

		# The default timeout value.
		A_		:						List[str]	    =   timeout

		# We use this lock primarily for the lock counter.
		A_		:						Tuple	    =   threading.Lock()

		# The lock counter is used for implementing the nested locking
		# mechanism. Whenever the lock is acquired, the counter is increased and
		# the lock is only released, when this value is 0 again.
		A_		:						Dict	    =   0
		return None





	@property
	def    lowerCamelCase(self							):
		return self._lock_file





	@property
	def    lowerCamelCase(self							):
		return self._timeout





	@timeout.setter
	def    lowerCamelCase(self     ,     lowerCAmelCase_							):
		A_		:						Tuple	    =   float(lowerCAmelCase_							)
		return None





	def    lowerCamelCase(self							):
		raise NotImplementedError()





	def    lowerCamelCase(self							):
		raise NotImplementedError()





	@property
	def    lowerCamelCase(self							):
		return self._lock_file_fd is not None





	def    lowerCamelCase(self     ,     lowerCAmelCase_=None     ,     lowerCAmelCase_=0.05							):
		# Use the default timeout, if no timeout is provided.
		if timeout is None:
			A_		:						List[str]	    =   self.timeout

		# Increment the number right at the beginning.
		# We can still undo it, if something fails.
		with self._thread_lock:
			self._lock_counter += 1

		A_		:						str	    =   id(self							)
		A_		:						Optional[int]	    =   self._lock_file
		A_		:						Dict	    =   time.time()
		try:
			while True:
				with self._thread_lock:
					if not self.is_locked:
						logger().debug(f"""Attempting to acquire lock {lock_id} on {lock_filename}"""							)
						self._acquire()

				if self.is_locked:
					logger().debug(f"""Lock {lock_id} acquired on {lock_filename}"""							)
					break
				elif timeout >= 0 and time.time() - start_time > timeout:
					logger().debug(f"""Timeout on acquiring lock {lock_id} on {lock_filename}"""							)
					raise Timeout(self._lock_file							)
				else:
					logger().debug(
					    f"""Lock {lock_id} not acquired on {lock_filename}, waiting {poll_intervall} seconds ..."""							)
					time.sleep(lowerCAmelCase_							)
		except:  # noqa
			# Something did go wrong, so decrement the counter.
			with self._thread_lock:
				A_		:						str	    =   max(0     ,     self._lock_counter - 1							)

			raise
		return _Acquire_ReturnProxy(lock=self							)





	def    lowerCamelCase(self     ,     lowerCAmelCase_=False							):
		with self._thread_lock:
			if self.is_locked:
				self._lock_counter -= 1

				if self._lock_counter == 0 or force:
					A_		:						Union[str, Any]	    =   id(self							)
					A_		:						List[Any]	    =   self._lock_file

					logger().debug(f"""Attempting to release lock {lock_id} on {lock_filename}"""							)
					self._release()
					A_		:						Any	    =   0
					logger().debug(f"""Lock {lock_id} released on {lock_filename}"""							)

		return None





	def __enter__(self							):
		self.acquire()
		return self





	def __exit__(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_							):
		self.release()
		return None





	def __del__(self							):
		self.release(force=lowerCAmelCase_							)
		return None





	def    lowerCamelCase(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_							):
		A_		:						Dict	    =   os.path.basename(lowerCAmelCase_							)
		if len(lowerCAmelCase_							) > max_length and max_length > 0:
			A_		:						str	    =   os.path.dirname(lowerCAmelCase_							)
			A_		:						List[str]	    =   str(hash(lowerCAmelCase_							)							)
			A_		:						List[Any]	    =   filename[: max_length - len(lowerCAmelCase_							) - 8] + """...""" + hashed_filename + """.lock"""
			return os.path.join(lowerCAmelCase_     ,     lowerCAmelCase_							)
		else:
			return path






class      SCREAMING_SNAKE_CASE  (				_SCREAMING_SNAKE_CASE						):





	"""simple docstring"""




	def __init__(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_=-1     ,     lowerCAmelCase_=None							):
		from .file_utils import relative_to_absolute_path

		super().__init__(lowerCAmelCase_     ,     timeout=lowerCAmelCase_     ,     max_filename_length=lowerCAmelCase_							)
		A_		:						Optional[Any]	    =   """\\\\?\\""" + relative_to_absolute_path(self.lock_file							)





	def    lowerCamelCase(self							):
		A_		:						Optional[int]	    =   os.O_RDWR | os.O_CREAT | os.O_TRUNC

		try:
			A_		:						Dict	    =   os.open(self._lock_file     ,     lowerCAmelCase_							)
		except OSError:
			pass
		else:
			try:
				msvcrt.locking(lowerCAmelCase_     ,     msvcrt.LK_NBLCK     ,     1							)
			except OSError:
				os.close(lowerCAmelCase_							)
			else:
				A_		:						Tuple	    =   fd
		return None





	def    lowerCamelCase(self							):
		A_		:						int	    =   self._lock_file_fd
		A_		:						Tuple	    =   None
		msvcrt.locking(lowerCAmelCase_     ,     msvcrt.LK_UNLCK     ,     1							)
		os.close(lowerCAmelCase_							)

		try:
			os.remove(self._lock_file							)
		# Probably another instance of the application
		# that acquired the file lock.
		except OSError:
			pass
		return None






class      SCREAMING_SNAKE_CASE  (				_SCREAMING_SNAKE_CASE						):





	"""simple docstring"""




	def __init__(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_=-1     ,     lowerCAmelCase_=None							):
		A_		:						Tuple	    =   os.statvfs(os.path.dirname(lowerCAmelCase_							)							).f_namemax
		super().__init__(lowerCAmelCase_     ,     timeout=lowerCAmelCase_     ,     max_filename_length=lowerCAmelCase_							)





	def    lowerCamelCase(self							):
		A_		:						Optional[Any]	    =   os.O_RDWR | os.O_CREAT | os.O_TRUNC
		A_		:						int	    =   os.open(self._lock_file     ,     lowerCAmelCase_							)

		try:
			fcntl.flock(lowerCAmelCase_     ,     fcntl.LOCK_EX | fcntl.LOCK_NB							)
		except OSError:
			os.close(lowerCAmelCase_							)
		else:
			A_		:						Any	    =   fd
		return None





	def    lowerCamelCase(self							):
		# Do not remove the lockfile:
		#
		#   https://github.com/benediktschmitt/py-filelock/issues/31
		#   https://stackoverflow.com/questions/17708885/flock-removing-locked-file-without-race-condition
		A_		:						List[str]	    =   self._lock_file_fd
		A_		:						int	    =   None
		fcntl.flock(lowerCAmelCase_     ,     fcntl.LOCK_UN							)
		os.close(lowerCAmelCase_							)
		return None






class      SCREAMING_SNAKE_CASE  (				_SCREAMING_SNAKE_CASE						):





	"""simple docstring"""




	def    lowerCamelCase(self							):
		A_		:						Union[str, Any]	    =   os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
		try:
			A_		:						Any	    =   os.open(self._lock_file     ,     lowerCAmelCase_							)
		except OSError:
			pass
		else:
			A_		:						List[str]	    =   fd
		return None





	def    lowerCamelCase(self							):
		os.close(self._lock_file_fd							)
		A_		:						List[str]	    =   None

		try:
			os.remove(self._lock_file							)
		# The file is already deleted and that's what we want.
		except OSError:
			pass
		return None






_lowerCAmelCase         =							None

if msvcrt:
							_lowerCAmelCase         =							WindowsFileLock
elif fcntl:
							_lowerCAmelCase         =							UnixFileLock
else:
							_lowerCAmelCase         =							SoftFileLock

							if warnings is not None:
														warnings.warn("only soft file lock is available")









"""simple docstring"""
import json
import os
from pathlib import Path

import pytest

from datasets.download.download_config import DownloadConfig
from datasets.download.download_manager import DownloadManager
from datasets.utils.file_utils import hash_url_to_filename


_lowerCAmelCase         =							"http://www.mocksite.com/file1.txt"
_lowerCAmelCase         =							"\"text\": [\"foo\", \"foo\"]"
_lowerCAmelCase         =							"6d8ce9aa78a471c7477201efbeabd3bb01ac2e7d100a6dc024ba1608361f90a8"



class      SCREAMING_SNAKE_CASE  :





	"""simple docstring"""




	_A       :				Optional[int]    =				200
	_A       :				Dict    =				{"""Content-Length""": """100"""}
	_A       :				List[Any]    =				{}





	def    lowerCamelCase(self     ,     **lowerCAmelCase_							):
		return [bytes(lowerCAmelCase_     ,     """utf-8"""							)]






def 			__UpperCamelCase	(		*snake_case__   ,   **snake_case__					):
	return MockResponse()






@pytest.mark.parametrize("""urls_type"""   ,   [str, list, dict]					)
def 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__					):
	import requests

	monkeypatch.setattr(snake_case__   ,   """request"""   ,   snake_case__					)

	A_		:						Optional[int]	    =   URL
	if issubclass(snake_case__   ,   snake_case__					):
		A_		:						str	    =   url
	elif issubclass(snake_case__   ,   snake_case__					):
		A_		:						Union[str, Any]	    =   [url]
	elif issubclass(snake_case__   ,   snake_case__					):
		A_		:						List[Any]	    =   {"""train""": url}
	A_		:						Tuple	    =   """dummy"""
	A_		:						List[str]	    =   """downloads"""
	A_		:						List[str]	    =   tmp_path
	A_		:						str	    =   DownloadConfig(
	    cache_dir=os.path.join(snake_case__   ,   snake_case__					)   ,   use_etag=snake_case__   ,   )
	A_		:						Any	    =   DownloadManager(dataset_name=snake_case__   ,   download_config=snake_case__					)
	A_		:						Dict	    =   dl_manager.download(snake_case__					)
	A_		:						List[str]	    =   urls
	for downloaded_paths in [downloaded_paths]:
		if isinstance(snake_case__   ,   snake_case__					):
			A_		:						Optional[Any]	    =   [downloaded_paths]
			A_		:						Tuple	    =   [urls]
		elif isinstance(snake_case__   ,   snake_case__					):
			assert "train" in downloaded_paths.keys()
			A_		:						Union[str, Any]	    =   downloaded_paths.values()
			A_		:						Tuple	    =   urls.values()
		assert downloaded_paths
		for downloaded_path, input_url in zip(snake_case__   ,   snake_case__					):
			assert downloaded_path == dl_manager.downloaded_paths[input_url]
			A_		:						Union[str, Any]	    =   Path(snake_case__					)
			A_		:						str	    =   downloaded_path.parts
			assert parts[-1] == HASH
			assert parts[-2] == cache_subdir
			assert downloaded_path.exists()
			A_		:						List[str]	    =   downloaded_path.read_text()
			assert content == CONTENT
			A_		:						Union[str, Any]	    =   downloaded_path.with_suffix(""".json"""					)
			assert metadata_downloaded_path.exists()
			A_		:						List[str]	    =   json.loads(metadata_downloaded_path.read_text()					)
			assert metadata_content == {"url": URL, "etag": None}






@pytest.mark.parametrize("""paths_type"""   ,   [str, list, dict]					)
def 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__					):
	A_		:						str	    =   str(snake_case__					)
	if issubclass(snake_case__   ,   snake_case__					):
		A_		:						List[str]	    =   filename
	elif issubclass(snake_case__   ,   snake_case__					):
		A_		:						Tuple	    =   [filename]
	elif issubclass(snake_case__   ,   snake_case__					):
		A_		:						Dict	    =   {"""train""": filename}
	A_		:						Union[str, Any]	    =   """dummy"""
	A_		:						Dict	    =   xz_file.parent
	A_		:						List[str]	    =   """extracted"""
	A_		:						int	    =   DownloadConfig(
	    cache_dir=snake_case__   ,   use_etag=snake_case__   ,   )
	A_		:						str	    =   DownloadManager(dataset_name=snake_case__   ,   download_config=snake_case__					)
	A_		:						int	    =   dl_manager.extract(snake_case__					)
	A_		:						int	    =   paths
	for extracted_paths in [extracted_paths]:
		if isinstance(snake_case__   ,   snake_case__					):
			A_		:						Dict	    =   [extracted_paths]
			A_		:						str	    =   [paths]
		elif isinstance(snake_case__   ,   snake_case__					):
			assert "train" in extracted_paths.keys()
			A_		:						Any	    =   extracted_paths.values()
			A_		:						str	    =   paths.values()
		assert extracted_paths
		for extracted_path, input_path in zip(snake_case__   ,   snake_case__					):
			assert extracted_path == dl_manager.extracted_paths[input_path]
			A_		:						str	    =   Path(snake_case__					)
			A_		:						List[str]	    =   extracted_path.parts
			assert parts[-1] == hash_url_to_filename(snake_case__   ,   etag=snake_case__					)
			assert parts[-2] == extracted_subdir
			assert extracted_path.exists()
			A_		:						Optional[Any]	    =   extracted_path.read_text()
			A_		:						Optional[int]	    =   text_file.read_text()
			assert extracted_file_content == expected_file_content






def 			__UpperCamelCase	(		snake_case__   ,   snake_case__					):
	assert path.endswith(""".jsonl"""					)
	for num_items, line in enumerate(snake_case__   ,   start=1					):
		A_		:						List[str]	    =   json.loads(line.decode("""utf-8"""					)					)
		assert item.keys() == {"col_1", "col_2", "col_3"}
	assert num_items == 4






@pytest.mark.parametrize("""archive_jsonl"""   ,   ["""tar_jsonl_path""", """zip_jsonl_path"""]					)
def 			__UpperCamelCase	(		snake_case__   ,   snake_case__					):
	A_		:						Optional[int]	    =   request.getfixturevalue(snake_case__					)
	A_		:						List[str]	    =   DownloadManager()
	for num_jsonl, (path, file) in enumerate(dl_manager.iter_archive(snake_case__					)   ,   start=1					):
		_test_jsonl(snake_case__   ,   snake_case__					)
	assert num_jsonl == 2






@pytest.mark.parametrize("""archive_nested_jsonl"""   ,   ["""tar_nested_jsonl_path""", """zip_nested_jsonl_path"""]					)
def 			__UpperCamelCase	(		snake_case__   ,   snake_case__					):
	A_		:						int	    =   request.getfixturevalue(snake_case__					)
	A_		:						Union[str, Any]	    =   DownloadManager()
	for num_tar, (path, file) in enumerate(dl_manager.iter_archive(snake_case__					)   ,   start=1					):
		for num_jsonl, (subpath, subfile) in enumerate(dl_manager.iter_archive(snake_case__					)   ,   start=1					):
			_test_jsonl(snake_case__   ,   snake_case__					)
	assert num_tar == 1
	assert num_jsonl == 2






def 			__UpperCamelCase	(		snake_case__					):
	A_		:						int	    =   DownloadManager()
	for num_file, file in enumerate(dl_manager.iter_files(snake_case__					)   ,   start=1					):
		assert os.path.basename(snake_case__					) == ("test.txt" if num_file == 1 else "train.txt")
	assert num_file == 2

