



"""simple docstring"""


import logging
import os
import threading
import time


try:
    import warnings
except ImportError:
    snake_case_    					=	None

try:
    import msvcrt
except ImportError:
    snake_case_    					=	None

try:
    import fcntl
except ImportError:
    snake_case_    					=	None


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


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

snake_case_    					=	"""3.0.12"""


snake_case_    					=	None

def 			_lowerCAmelCase				(       ):
      global _logger
      UpperCAmelCase					=			_logger or logging.getLogger(__name__						)
      return _logger







class 				A_      (					SCREAMING_SNAKE_CASE_    ):



      """simple docstring"""






      def __init__( self			:int       ,  lowercase_			:List[Any] )    ->      Optional[int]:
            UpperCAmelCase					=			lock_file
            return None






      def __str__( self			:Dict )    ->      Optional[int]:
            UpperCAmelCase					=			f"""The file lock '{self.lock_file}' could not be acquired."""
            return temp







class 				A_      :



      """simple docstring"""






      def __init__( self			:List[Any]       ,  lowercase_			:Dict )    ->      Dict:
            UpperCAmelCase					=			lock
            return None






      def __enter__( self			:Optional[int] )    ->      Optional[int]:
            return self.lock






      def __exit__( self			:Optional[int]       ,  lowercase_			:Tuple       ,  lowercase_			:Any       ,  lowercase_			:Union[str, Any] )    ->      List[Any]:
            self.lock.release()
            return None







class 				A_      :



      """simple docstring"""






      def __init__( self			:List[str]       ,  lowercase_			:str       ,  lowercase_			:str=-1       ,  lowercase_			:List[Any]=None )    ->      Dict:
            UpperCAmelCase					=			max_filename_length if max_filename_length is not None else 2_55
            # Hash the filename if it's too long
            UpperCAmelCase					=			self.hash_filename_if_too_long(lowercase_       ,  lowercase_ )
            # 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			:Optional[Any] )    ->      int:
            return self._lock_file






      @property
      def      UpperCAmelCase__					( self			:Union[str, Any] )    ->      Any:
            return self._timeout






      @timeout.setter
      def      UpperCAmelCase__					( self			:Union[str, Any]       ,  lowercase_			:Any )    ->      Tuple:
            UpperCAmelCase					=			float(lowercase_ )
            return None






      def      UpperCAmelCase__					( self			:Any )    ->      List[str]:
            raise NotImplementedError()






      def      UpperCAmelCase__					( self			:Union[str, Any] )    ->      List[str]:
            raise NotImplementedError()






      @property
      def      UpperCAmelCase__					( self			:Optional[Any] )    ->      Union[str, Any]:
            return self._lock_file_fd is not None






      def      UpperCAmelCase__					( self			:Union[str, Any]       ,  lowercase_			:Dict=None       ,  lowercase_			:int=0.05 )    ->      Union[str, Any]:
            # 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(lowercase_ )
            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			:Optional[int]       ,  lowercase_			:List[str]=False )    ->      int:
            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			:Any )    ->      Tuple:
            self.acquire()
            return self






      def __exit__( self			:Dict       ,  lowercase_			:Optional[int]       ,  lowercase_			:Union[str, Any]       ,  lowercase_			:Union[str, Any] )    ->      Any:
            self.release()
            return None






      def __del__( self			:Any )    ->      int:
            self.release(force=lowercase_ )
            return None






      def      UpperCAmelCase__					( self			:List[str]       ,  lowercase_			:str       ,  lowercase_			:int )    ->      str:
            UpperCAmelCase					=			os.path.basename(lowercase_ )
            if len(lowercase_ ) > max_length and max_length > 0:
                  UpperCAmelCase					=			os.path.dirname(lowercase_ )
                  UpperCAmelCase					=			str(hash(lowercase_ ) )
                  UpperCAmelCase					=			filename[: max_length - len(lowercase_ ) - 8] + '...' + hashed_filename + '.lock'
                  return os.path.join(lowercase_       ,  lowercase_ )
            else:
                  return path







class 				A_      (					SCREAMING_SNAKE_CASE_    ):



      """simple docstring"""






      def __init__( self			:Tuple       ,  lowercase_			:Tuple       ,  lowercase_			:List[Any]=-1       ,  lowercase_			:Union[str, Any]=None )    ->      Optional[Any]:
            from .file_utils import relative_to_absolute_path

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






      def      UpperCAmelCase__					( self			:List[str] )    ->      str:
            UpperCAmelCase					=			os.O_RDWR | os.O_CREAT | os.O_TRUNC

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






      def      UpperCAmelCase__					( self			:int )    ->      Union[str, Any]:
            UpperCAmelCase					=			self._lock_file_fd
            UpperCAmelCase					=			None
            msvcrt.locking(lowercase_       ,  msvcrt.LK_UNLCK       ,  1 )
            os.close(lowercase_ )

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







class 				A_      (					SCREAMING_SNAKE_CASE_    ):



      """simple docstring"""






      def __init__( self			:Optional[Any]       ,  lowercase_			:str       ,  lowercase_			:List[str]=-1       ,  lowercase_			:str=None )    ->      str:
            UpperCAmelCase					=			os.statvfs(os.path.dirname(lowercase_ ) ).f_namemax
            super().__init__(lowercase_       ,  timeout=lowercase_       ,  max_filename_length=lowercase_ )






      def      UpperCAmelCase__					( self			:str )    ->      Dict:
            UpperCAmelCase					=			os.O_RDWR | os.O_CREAT | os.O_TRUNC
            UpperCAmelCase					=			os.open(self._lock_file       ,  lowercase_ )

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






      def      UpperCAmelCase__					( self			:List[Any] )    ->      List[Any]:
            # 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(lowercase_       ,  fcntl.LOCK_UN )
            os.close(lowercase_ )
            return None







class 				A_      (					SCREAMING_SNAKE_CASE_    ):



      """simple docstring"""






      def      UpperCAmelCase__					( self			:List[Any] )    ->      List[str]:
            UpperCAmelCase					=			os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
            try:
                  UpperCAmelCase					=			os.open(self._lock_file       ,  lowercase_ )
            except OSError:
                  pass
            else:
                  UpperCAmelCase					=			fd
            return None






      def      UpperCAmelCase__					( self			:Optional[int] )    ->      List[Any]:
            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


snake_case_    					=	None

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

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










"""simple docstring"""


import math

def 			_lowerCAmelCase				(       lowercase_						):
						assert isinstance(lowercase_		,     lowercase_						) and (
						    number >= 0
						), "'number' must been an int and positive"

						if 1 < number < 4:
												# 2 and 3 are primes
												return True
						elif number < 2 or not number % 2:
												# Negatives, 0, 1 and all even numbers are not primes
												return False

						UpperCAmelCase					=			range(3		,     int(math.sqrt(lowercase_						) + 1						)		,     2						)
						return not any(not number % i for i in odd_numbers						)






def 			_lowerCAmelCase				(       lowercase_		,     lowercase_=1		,     **lowercase_						):
						UpperCAmelCase					=			factor * value
						UpperCAmelCase					=			value

						while not is_prime(lowercase_						):
												value += 1 if not ("desc" in kwargs and kwargs["desc"] is True) else -1

						if value == first_value_val:
												return next_prime(value + 1		,     **lowercase_						)
						return value





