在哪里修改这段Python代码以将临时文件保存到tmp文件夹?

0 投票
2 回答
772 浏览
提问于 2025-04-16 15:48

这是一个代码示例,用来实现文件锁定,这样应用程序就只能打开一个实例。现在这个功能是可以正常工作的,但它把锁定文件保存在了主文件夹(Ubuntu系统)。如果应用程序崩溃了,锁定文件就不会被删除,这样就不好了……

我不太清楚该在哪里修改代码,以便把锁定文件保存到c:/tmp文件夹里?

#!/usr/bin/python
# -*- coding: utf-8 -*-


#implements a lockfile if program already is open
import os
import socket
from fcntl import flock

class flock(object):
    '''Class to handle creating and removing (pid) lockfiles'''

    # custom exceptions
    class FileLockAcquisitionError(Exception): pass
    class FileLockReleaseError(Exception): pass

    # convenience callables for formatting
    addr = lambda self: '%d@%s' % (self.pid, self.host)
    fddr = lambda self: '<%s %s>' % (self.path, self.addr())
    pddr = lambda self, lock: '<%s %s@%s>' %\
                              (self.path, lock['pid'], lock['host'])

    def __init__(self, path, debug=None):
        self.pid   = os.getpid()
        self.host  = socket.gethostname()
        self.path  = path
        self.debug = debug # set this to get status messages

    def acquire(self):
        '''Acquire a lock, returning self if successful, False otherwise'''
        if self.islocked():
            if self.debug:
                lock = self._readlock()
                print 'Previous lock detected: %s' % self.pddr(lock)
            return False
        try:
            fh = open(self.path, 'w')
            fh.write(self.addr())
            fh.close()
            if self.debug:
                print 'Acquired lock: %s' % self.fddr()
        except:
            if os.path.isfile(self.path):
                try:
                    os.unlink(self.path)
                except:
                    pass
            raise (self.FileLockAcquisitionError,
                   'Error acquiring lock: %s' % self.fddr())
        return self

    def release(self):
        '''Release lock, returning self'''
        if self.ownlock():
            try:
                os.unlink(self.path)
                if self.debug:
                    print 'Released lock: %s' % self.fddr()
            except:
                raise (self.FileLockReleaseError,
                       'Error releasing lock: %s' % self.fddr())
        return self

    def _readlock(self):
        '''Internal method to read lock info'''
        try:
            lock = {}
            fh   = open(self.path)
            data = fh.read().rstrip().split('@')
            fh.close()
            lock['pid'], lock['host'] = data
            return lock
        except:
            return {'pid': 8**10, 'host': ''}

    def islocked(self):
        '''Check if we already have a lock'''
        try:
            lock = self._readlock()
            os.kill(int(lock['pid']), 0)
            return (lock['host'] == self.host)
        except:
            return False

    def ownlock(self):
        '''Check if we own the lock'''
        lock = self._readlock()
        return (self.fddr() == self.pddr(lock))

    def __del__(self):
        '''Magic method to clean up lock when program exits'''
        self.release()

#now testing to see if file is locked = other instance of this program is running already

lock = flock('tmp.lock', True).acquire()
if lock:
    print 'doing stuff'
else:
    print 'locked!'
    exit()


#end of lockfile

2 个回答

1

使用 tempfile 这个库

0

在脚本的最后部分:

lock = flock('tmp.lock', True).acquire()

'tmp.lock' 是当前目录下一个文件的路径。你可以把它改成你需要的路径,比如 'c:/tmp-folder/tmp.lock'

lock = flock('c:/tmp-folder/tmp.lock', True).acquire()

不过,正如 @g19fanatic 提到的,你是在 Windows 系统('c:/...')还是 Linux(Ubuntu)系统上呢?

撰写回答