在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()