如何在Python中检查文件复制完成？

Question

我希望找到一种方法，可以通知Python脚本文件复制完成。以下是场景：

1. 脚本通过不断使用os.listdir()进行轮询来监视文件夹to_print。 2. 每次os.listdir()返回一个文件列表，其中存在一个之前没有看到的文件，脚本就会对该文件执行一些操作，包括打开并操作其内容。

当文件很小时，这样做没问题，并且从原始源将文件复制到被监视的目录所花费的时间小于下一次os.listdir()轮询的剩余时间。但是，如果轮询到了一个正在复制的文件，那么当脚本尝试对其进行操作时，文件内容将会损坏。

相反，我希望能够（使用os.stat或其他方式）知道文件当前正在复制，并在复制完成后再对其进行操作（如果需要）。 我当前的想法是每次发现新文件时使用os.stat()，然后等待下一次轮询并比较上次轮询以来修改/创建时间，如果它们仍然相同，则该文件是“稳定的”，否则继续轮询，直到它稳定下来。 但是，我不确定这是否有效，因为我对Linux / Unix如何更新这些值并不太熟悉。

- emish

Answer 1

尝试使用 inotify.

这是Linux的一个标准，用于监视文件。对于你的用例，IN_CLOSE_WRITE 事件似乎很有希望。有一个Python inotify库可供使用。以下是一个简单的示例（摘自这里）。你需要修改它以仅捕获IN_CLOSE_WRITE事件。

# Example: loops monitoring events forever.
#
import pyinotify

# Instanciate a new WatchManager (will be used to store watches).

wm = pyinotify.WatchManager()
# Associate this WatchManager with a Notifier (will be used to report and
# process events).

notifier = pyinotify.Notifier(wm)
# Add a new watch on /tmp for ALL_EVENTS.
wm.add_watch('/tmp', pyinotify.ALL_EVENTS) # <-- replace by IN_CLOSE_WRITE

# Loop forever and handle events.
notifier.loop()

这里有一份详尽的API文档：http://seb-m.github.com/pyinotify/

- nalply

Answer 2

由于文件可以在轮询间隔内复制，因此请在检查新文件之前处理上一次轮询找到的新文件。换句话说，不要这样做：

while True:
    newfiles = check_for_new_files()
    process(newfiles)
    time.sleep(pollinterval)

做这个：

newfiles = []

while True:
    process(newfiles)
    newfiles = check_for_new_files()
    time.sleep(pollinterval)

或者将等待放在循环的中间（效果相同）：

while True:
    newfiles = check_for_new_files()
    time.sleep(pollinterval)
    process(newfiles)

- kindall

---

如果没有要处理的文件并且目录为空，则此代码将无法工作。 - emish

@emish，为什么不呢？newfiles难道不只是一个空列表吗？而且process肯定可以合理地处理一个空列表。（如果不能，那么应该进行调整以使其能够处理。） - huon

@kindall 对不起，我之前没有意识到这个区别，直到我尝试了一下。谢谢，这正是我需要的简短技巧！ - emish