在Python中监听文件

Question

我有一个Python脚本，它会对我的数据库进行一些更新。

这个脚本需要的文件是由其他程序在大约凌晨3点保存到一个文件夹里的。

所以我打算安排一个定时任务，让它每天凌晨3点运行；但我想处理一下如果文件在3点时还没有准备好的情况，因为可能会有一些延迟。

所以我基本上需要从凌晨3点开始，每隔5分钟检查一次某个特定名称的文件是否存在。我会尝试大约1个小时，如果还是找不到就放弃。

我该如何在Python中实现这个功能呢？

Answer 1

你可以使用Twisted，它的反应器比无限循环要好得多！另外，你可以用reactor.callLater(myTime, myFunction)这个方法，当myFunction被调用时，你可以调整myTime，并用相同的API callLater()再添加一个回调。

Answer 2

在Python中，你可以检查文件是否存在。

import os.path
os.path.isfile(filename)

然后你可以设置定时任务，让它每5分钟运行一次，从凌晨3点开始：

*/5 3 * * * /path-to-your/script.py

你可以写一个简单的文件来控制你是否已经读取了文件中的数据（或者如果你已经在使用数据库的话，也可以用数据库）。

Answer 3

这就是我首先想到的，挺简单明了的：

from time import sleep
counter = 0
working = True
while counter < 11 and working:
    try:
        # Open file and do whatever you need
        working = False
    except IOError:
        counter +=1
        sleep(5*60)

更好的解决方案

from time import sleep
counter = 0
working = True
while counter < 11 and working:
    if os.path.isfile('path/to/your/file')
        # Open file and do whatever you need
        working = False
    else:
        counter +=1
        sleep(5*60)

Answer 4

对于这种任务，你需要使用watchdog这个库，它可以用来监听和监控系统事件。

它可以监控的一种事件是文件系统事件，这通过FileSystemEventHandler这个类来实现，这个类里面有一个叫on_created()的方法。

你最终会写一个“包装”脚本，这个脚本可以持续运行。这个脚本会使用watchdog来监听特定的文件夹。一旦有文件被创建，这个脚本就会收到通知——然后你需要检查一下这个创建的文件是否符合你想要的文件格式，然后执行你自定义的代码。

幸运的是，因为这是一个常见的任务——已经有一个叫PatternMatchingEventHandler的类可以直接使用，它是从FileSystemEventHandler继承而来的，但它专门用来监控符合特定格式的文件。

所以你的包装脚本就变成了：

from watchdog.observers import Observer  
from watchdog.events import PatternMatchingEventHandler

class FileWatcher(PatternMatchingEventHandler):
    patterns = ["*.dat"] # adjust as required

    def process(self, event):
       # your actual code goes here

       # event.src_path will be the full file path
       # event.event_type will be 'created', 'moved', etc.
       print('{} observed on {}'.format(event.event_type, event.src_path))

    def on_created(self, event):
       self.process(event)

if __name__ == '__main__':
    obs = Observer() # This is what manages running of your code
    obs.schedule(FileWatcher(), path='/the/target/dir')
    obs.start() # Start watching

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        ob.stop()

    obs.join()

Answer 5

试试这样做（如果你使用的是Python 3，你需要把打印语句改成函数调用）。

#!/usr/bin/env python

import os
import time

def watch_file( filename, time_limit=3600, check_interval=60 ):
    '''Return true if filename exists, if not keep checking once every check_interval seconds for time_limit seconds.
    time_limit defaults to 1 hour
    check_interval defaults to 1 minute
    '''

    now = time.time()
    last_time = now + time_limit

    while time.time() <= last_time:
        if os.path.exists( filename ):
             return True
        else:
            # Wait for check interval seconds, then check again.
            time.sleep( check_interval )

    return False

if __name__ == '__main__':
    filename = '/the/file/Im/waiting/for.txt'
    time_limit = 3600 # one hour from now.
    check_interval = 60 # seconds between checking for the file.

    if watch_file( filename, time_limit, check_interval ):
        print "File present!"
    else:
        print "File not found after waiting:", time_limit, " seconds!"