
import logging
import os
import threading
import time


try:
					import warnings
except ImportError:
					_UpperCAmelCase :			str              =	None

try:
					import msvcrt
except ImportError:
					_UpperCAmelCase :			List[Any]              =	None

try:
					import fcntl
except ImportError:
					_UpperCAmelCase :			int              =	None


# Backward compatibility
# ------------------------------------------------
try:
					TimeoutError
except NameError:
					_UpperCAmelCase :			Dict              =	OSError


# Data
# ------------------------------------------------
_UpperCAmelCase :			Optional[Any]              =	[
    "Timeout",
    "BaseFileLock",
    "WindowsFileLock",
    "UnixFileLock",
    "SoftFileLock",
    "FileLock",
]

_UpperCAmelCase :			Optional[int]              =	"3.0.12"


_UpperCAmelCase :			int              =	None


def 				A	( )       ->  Tuple:
		'''simple docstring'''

		global _logger
		UpperCamelCase          =					_logger or logging.getLogger(__name__			)
		return _logger


class     lowercase (		_SCREAMING_SNAKE_CASE       ):



		def __init__(			self    ,				A_	)  -> Tuple:
				"""simple docstring"""
				UpperCamelCase          =					lock_file
				return None



		def __str__(			self	)  -> List[str]:
				"""simple docstring"""
				UpperCamelCase          =					F'''The file lock \'{self.lock_file}\' could not be acquired.'''
				return temp






class     lowercase :



		def __init__(			self    ,				A_	)  -> Optional[int]:
				"""simple docstring"""
				UpperCamelCase          =					lock
				return None



		def __enter__(			self	)  -> int:
				"""simple docstring"""
				return self.lock



		def __exit__(			self    ,				A_    ,				A_    ,				A_	)  -> Tuple:
				"""simple docstring"""
				self.lock.release()
				return None






class     lowercase :



		def __init__(			self    ,				A_    ,				A_=-1    ,				A_=None	)  -> Union[str, Any]:
				"""simple docstring"""
				UpperCamelCase          =					max_filename_length if max_filename_length is not None else 255
				# Hash the filename if it's too long
				UpperCamelCase          =					self.hash_filename_if_too_long(A_    ,				A_	)
				# The path to the lock file.
				UpperCamelCase          =					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.
				UpperCamelCase          =					None

				# The default timeout value.
				UpperCamelCase          =					timeout

				# We use this lock primarily for the lock counter.
				UpperCamelCase          =					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.
				UpperCamelCase          =					0
				return None



		@property
		def __UpperCamelCase			(			self	)  -> int:
				"""simple docstring"""
				return self._lock_file



		@property
		def __UpperCamelCase			(			self	)  -> Dict:
				"""simple docstring"""
				return self._timeout



		@timeout.setter
		def __UpperCamelCase			(			self    ,				A_	)  -> Optional[int]:
				"""simple docstring"""
				UpperCamelCase          =					float(A_	)
				return None



		def __UpperCamelCase			(			self	)  -> int:
				"""simple docstring"""
				raise NotImplementedError()



		def __UpperCamelCase			(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				raise NotImplementedError()



		@property
		def __UpperCamelCase			(			self	)  -> Any:
				"""simple docstring"""
				return self._lock_file_fd is not None



		def __UpperCamelCase			(			self    ,				A_=None    ,				A_=0.05	)  -> Union[str, Any]:
				"""simple docstring"""
				# Use the default timeout, if no timeout is provided.
				if timeout is None:
						UpperCamelCase          =					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

				UpperCamelCase          =					id(self	)
				UpperCamelCase          =					self._lock_file
				UpperCamelCase          =					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(A_	)
				except:  # noqa
						# Something did go wrong, so decrement the counter.
						with self._thread_lock:
								UpperCamelCase          =					max(0    ,				self._lock_counter - 1	)

						raise
				return _Acquire_ReturnProxy(lock=self	)



		def __UpperCamelCase			(			self    ,				A_=False	)  -> Optional[Any]:
				"""simple docstring"""
				with self._thread_lock:
						if self.is_locked:
								self._lock_counter -= 1

								if self._lock_counter == 0 or force:
										UpperCamelCase          =					id(self	)
										UpperCamelCase          =					self._lock_file

										logger().debug(F'''Attempting to release lock {lock_id} on {lock_filename}'''	)
										self._release()
										UpperCamelCase          =					0
										logger().debug(F'''Lock {lock_id} released on {lock_filename}'''	)

				return None



		def __enter__(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				self.acquire()
				return self



		def __exit__(			self    ,				A_    ,				A_    ,				A_	)  -> Optional[Any]:
				"""simple docstring"""
				self.release()
				return None



		def __del__(			self	)  -> int:
				"""simple docstring"""
				self.release(force=A_	)
				return None



		def __UpperCamelCase			(			self    ,				A_    ,				A_	)  -> str:
				"""simple docstring"""
				UpperCamelCase          =					os.path.basename(A_	)
				if len(A_	) > max_length and max_length > 0:
						UpperCamelCase          =					os.path.dirname(A_	)
						UpperCamelCase          =					str(hash(A_	)	)
						UpperCamelCase          =					filename[: max_length - len(A_	) - 8] + '...' + hashed_filename + '.lock'
						return os.path.join(A_    ,				A_	)
				else:
						return path






class     lowercase (		_SCREAMING_SNAKE_CASE       ):



		def __init__(			self    ,				A_    ,				A_=-1    ,				A_=None	)  -> List[Any]:
				"""simple docstring"""
				from .file_utils import relative_to_absolute_path

				super().__init__(A_    ,				timeout=A_    ,				max_filename_length=A_	)
				UpperCamelCase          =					'\\\\?\\' + relative_to_absolute_path(self.lock_file	)



		def __UpperCamelCase			(			self	)  -> Tuple:
				"""simple docstring"""
				UpperCamelCase          =					os.O_RDWR | os.O_CREAT | os.O_TRUNC

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



		def __UpperCamelCase			(			self	)  -> int:
				"""simple docstring"""
				UpperCamelCase          =					self._lock_file_fd
				UpperCamelCase          =					None
				msvcrt.locking(A_    ,				msvcrt.LK_UNLCK    ,				1	)
				os.close(A_	)

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






class     lowercase (		_SCREAMING_SNAKE_CASE       ):



		def __init__(			self    ,				A_    ,				A_=-1    ,				A_=None	)  -> Any:
				"""simple docstring"""
				UpperCamelCase          =					os.statvfs(os.path.dirname(A_	)	).f_namemax
				super().__init__(A_    ,				timeout=A_    ,				max_filename_length=A_	)



		def __UpperCamelCase			(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				UpperCamelCase          =					os.O_RDWR | os.O_CREAT | os.O_TRUNC
				UpperCamelCase          =					os.open(self._lock_file    ,				A_	)

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



		def __UpperCamelCase			(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				# 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
				UpperCamelCase          =					self._lock_file_fd
				UpperCamelCase          =					None
				fcntl.flock(A_    ,				fcntl.LOCK_UN	)
				os.close(A_	)
				return None






class     lowercase (		_SCREAMING_SNAKE_CASE       ):



		def __UpperCamelCase			(			self	)  -> List[Any]:
				"""simple docstring"""
				UpperCamelCase          =					os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
				try:
						UpperCamelCase          =					os.open(self._lock_file    ,				A_	)
				except OSError:
						pass
				else:
						UpperCamelCase          =					fd
				return None



		def __UpperCamelCase			(			self	)  -> Union[str, Any]:
				"""simple docstring"""
				os.close(self._lock_file_fd	)
				UpperCamelCase          =					None

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






_UpperCAmelCase :			str              =	None

if msvcrt:
					_UpperCAmelCase :			Optional[int]              =	WindowsFileLock
elif fcntl:
					_UpperCAmelCase :			List[str]              =	UnixFileLock
else:
					_UpperCAmelCase :			Tuple              =	SoftFileLock

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


from __future__ import annotations


def 				A	( lowercase   ,	lowercase			)       ->  tuple[int, int]:
		'''simple docstring'''

		if b == 0:
				return (1, 0)
		((UpperCamelCase)     ,  (UpperCamelCase))          =					extended_euclid(lowercase   ,	a % b			)
		UpperCamelCase          =					a // b
		return (y, x - k * y)


def 				A	( lowercase   ,	lowercase   ,	lowercase   ,	lowercase			)       ->  int:
		'''simple docstring'''

		((UpperCamelCase)     ,  (UpperCamelCase))          =					extended_euclid(lowercase   ,	lowercase			)
		UpperCamelCase          =					na * na
		UpperCamelCase          =					ra * x * na + ra * y * na
		return (n % m + m) % m


def 				A	( lowercase   ,	lowercase			)       ->  int:
		'''simple docstring'''

		((UpperCamelCase)     ,  (UpperCamelCase))          =					extended_euclid(lowercase   ,	lowercase			)
		if b < 0:
				UpperCamelCase          =					(b % n + n) % n
		return b


def 				A	( lowercase   ,	lowercase   ,	lowercase   ,	lowercase			)       ->  int:
		'''simple docstring'''

		UpperCamelCase     ,  UpperCamelCase          =					invert_modulo(lowercase   ,	lowercase			), invert_modulo(lowercase   ,	lowercase			)
		UpperCamelCase          =					na * na
		UpperCamelCase          =					ra * x * na + ra * y * na
		return (n % m + m) % m


if __name__ == "__main__":
					from doctest import testmod

					testmod(name="chinese_remainder_theorem", verbose=True)
					testmod(name="chinese_remainder_theorem2", verbose=True)
					testmod(name="invert_modulo", verbose=True)
					testmod(name="extended_euclid", verbose=True)
