在Python中侦听文件

Question

我有一个python脚本，它对我的​​数据库做了一些更新。

此脚本需要的文件在某个其他进程的大约凌晨3点保存在一个目录中。

所以我打算安排一个cron工作，每天在凌晨3点运行;但如果文件在凌晨3点不可用，我想处理这种情况，它可能会延迟一段时间。

因此，我基本上需要继续检查从3AM开始每5分钟是否存在某个特定名称的文件。我会尝试大约1小时，如果不能解决问题，请放弃。

我如何在Python中实现这样的事情？

Answer 1

尝试这样的事情（如果你使用的是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!" 

Answer 2

这就是我想到的第一，相当直截了当：

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 3

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

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

然后你设置你的cron从凌晨3点开始每隔5分钟运行一次： */5 3 * * * /path-to-your/script.py

你可以写一个简单的文件来控制你已经读过数据来自文件或不是（或数据库，如果你已经在使用一个）

Answer 4

你可以使用扭曲，它是反应器它比无限循环好得多！你也可以使用reactor.callLater（myTime，myFunction），当myFunction被调用时，你可以调整myTime并用相同的API callLater（）添加另一个回调。

Answer 5

对于这类任务，您需要使用watchdog库监听和监视系统事件。

它可以监视的事件之一是文件系统事件，通过FileSystemEventHandler类，其中有on_created()方法。

你最终会写一个“包装器”脚本，它可以连续运行。该脚本将使用看门狗在该特定目录上进行侦听。当文件被创建时，这个脚本将被通知 - 你必须检查创建的文件是否与目标文件的模式匹配，然后执行你的自定义代码。

幸运的是，因为这是一项常见任务 - 有一个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()